zc.async 1.4.0

Overview: Usage

Looking at the design from the perspective of regular usage, your code obtains a queue, which is a place to register jobs to be performed asynchronously.

Your application calls put on the queue to register a job. The job must be a pickleable, callable object. A global function, a callable persistent object, a method of a persistent object, or a special zc.async.job.Job object (discussed later) are all examples of suitable objects. The job by default is registered to be performed as soon as possible, but can be registered to be called at a certain time.

The put call will return a zc.async.job.Job object. This object represents both the callable and its deferred result. It has information about the job requested, the current state of the job, and the result of performing the job.

An example spelling for registering a job might be self.pending_result = queue.put(self.performSpider). The returned object can be stored and polled to see when the job is complete; or the job can be configured to do additional work when it completes (such as storing the result in a data structure).

Overview: Mechanism

Multiple processes, typically spread across multiple machines, can connect to the queue and claim and perform work. As with other collections of processes that share pickled objects, these processes generally should share the same software (though some variations on this constraint should be possible).

A process that should claim and perform work, in addition to a database connection and the necessary software, needs a dispatcher with a reactor to provide a heartbeat. The dispatcher will rely on one or more persistent agents in the queue (in the database) to determine which jobs it should perform.

A dispatcher is in charge of dispatching queued work for a given process to worker threads. It works with one or more queues and a single reactor. It has a universally unique identifier (UUID), which is usually an identifier of the application instance in which it is running. The dispatcher starts jobs in dedicated threads.

A reactor is something that can provide an eternal loop, or heartbeat, to power the dispatcher. It can be the main twisted reactor (in the main thread); another instance of a twisted reactor (in a child thread); or any object that implements a small subset of the twisted reactor interface (see discussion in dispatcher.txt, and example testing reactor in testing.py, used below).

An agent is a persistent object in a queue that is associated with a dispatcher and is responsible for picking jobs and keeping track of them. Zero or more agents within a queue can be associated with a dispatcher. Each agent for a given dispatcher in a given queue is identified uniquely with a name [2].

Generally, these work together as follows. The reactor calls the dispatcher. The dispatcher tries to find the mapping of queues in the database root under a key of zc.async (see constant zc.async.interfaces.KEY). If it finds the mapping, it iterates over the queues (the mapping’s values) and asks each queue for the agents associated with the dispatcher’s UUID. The dispatcher then is responsible for seeing what jobs its agents want to do from the queue, and providing threads and connections for the work to be done. The dispatcher then asks the reactor to call itself again in a few seconds.

Obtaining the queue

First, how do we get the queue? Your installation may have some conveniences. For instance, the Zope 3 configuration described below makes it possible to get the primary queue with an adaptation call like zc.async.interfaces.IQueue(a_persistent_object_with_db_connection).

But failing that, queues are always expected to be in a zc.async.queue.Queues mapping found off the ZODB root in a key defined by the constant zc.async.interfaces.KEY.

>>> import zc.async.interfaces
>>> zc.async.interfaces.KEY
'zc.async'
>>> root = conn.root()
>>> queues = root[zc.async.interfaces.KEY]
>>> import zc.async.queue
>>> isinstance(queues, zc.async.queue.Queues)
True

As the name implies, queues is a collection of queues. As discussed later, it’s possible to have multiple queues, as a tool to distribute and control work. We will assume a convention of a queue being available in the ‘’ (empty string).

>>> queues.keys()
['']
>>> queue = queues['']

queue.put

Now we want to actually get some work done. The simplest case is simple to perform: pass a persistable callable to the queue’s put method and commit the transaction.

>>> def send_message():
...     print "imagine this sent a message to another machine"
>>> job = queue.put(send_message)
>>> import transaction
>>> transaction.commit()

Note that this won’t really work in an interactive session: the callable needs to be picklable, as discussed above, so send_message would need to be a module global, for instance.

The put returned a job. Now we need to wait for the job to be performed. We would normally do this by really waiting. For our examples, we will use a helper method on the testing reactor to wait_for the job to be completed.

>>> reactor.wait_for(job)
imagine this sent a message to another machine

We also could have used the method of a persistent object. Here’s another quick example.

First we define a simple persistent.Persistent subclass and put an instance of it in the database [4].

>>> import persistent
>>> class Demo(persistent.Persistent):
...     counter = 0
...     def increase(self, value=1):
...         self.counter += value
...
>>> root['demo'] = Demo()
>>> transaction.commit()

Now we can put the demo.increase method in the queue.

>>> root['demo'].counter
0
>>> job = queue.put(root['demo'].increase)
>>> transaction.commit()

>>> reactor.wait_for(job)
>>> root['demo'].counter
1

The method was called, and the persistent object modified!

To reiterate, only pickleable callables such as global functions and the methods of persistent objects can be used. This rules out, for instance, lambdas and other functions created dynamically. As we’ll see below, the job instance can help us out there somewhat by offering closure-like features.

queue.pull and queue.remove

If you put a job into a queue and it hasn’t been claimed yet and you want to cancel the job, pull or remove it from the queue.

The pull method removes the first job, or takes an integer index.

>>> len(queue)
0
>>> job1 = queue.put(send_message)
>>> job2 = queue.put(send_message)
>>> len(queue)
2
>>> job1 is queue.pull()
True
>>> list(queue) == [job2]
True
>>> job1 is queue.put(job1)
True
>>> list(queue) == [job2, job1]
True
>>> job1 is queue.pull(-1)
True
>>> job2 is queue.pull()
True
>>> len(queue)
0

The remove method removes the specific given job.

>>> job1 = queue.put(send_message)
>>> job2 = queue.put(send_message)
>>> len(queue)
2
>>> queue.remove(job1)
>>> list(queue) == [job2]
True
>>> job1 is queue.put(job1)
True
>>> list(queue) == [job2, job1]
True
>>> queue.remove(job1)
>>> list(queue) == [job2]
True
>>> queue.remove(job2)
>>> len(queue)
0

Scheduled Calls

When using put, you can also pass a datetime.datetime to schedule a call. A datetime without a timezone is considered to be in the UTC timezone.

>>> t = transaction.begin()
>>> import datetime
>>> import pytz
>>> datetime.datetime.now(pytz.UTC)
datetime.datetime(2006, 8, 10, 15, 44, 33, 211, tzinfo=<UTC>)
>>> job = queue.put(
...     send_message, begin_after=datetime.datetime(
...         2006, 8, 10, 15, 56, tzinfo=pytz.UTC))
>>> job.begin_after
datetime.datetime(2006, 8, 10, 15, 56, tzinfo=<UTC>)
>>> transaction.commit()
>>> reactor.wait_for(job, attempts=2) # +5 virtual seconds
TIME OUT
>>> reactor.wait_for(job, attempts=2) # +5 virtual seconds
TIME OUT
>>> datetime.datetime.now(pytz.UTC)
datetime.datetime(2006, 8, 10, 15, 44, 43, 211, tzinfo=<UTC>)

>>> zc.async.testing.set_now(datetime.datetime(
...     2006, 8, 10, 15, 56, tzinfo=pytz.UTC))
>>> reactor.wait_for(job)
imagine this sent a message to another machine
>>> datetime.datetime.now(pytz.UTC) >= job.begin_after
True

If you set a time that has already passed, it will be run as if it had been set to run as soon as possible [5]…unless the job has already timed out, in which case the job fails with an abort [6].

The queue’s put method is the essential API. pull is used rarely. Other methods are used to introspect, but are not needed for basic usage.

But what is that result of the put call in the examples above? A job? What do you do with that?

Overview

The result of a call to put returns an IJob. The job represents the pending result. This object has a lot of functionality that’s explored in other documents in this package, and demonstrated a bit below, but here’s a summary.

• You can introspect, and even modify, the call and its arguments.

• You can specify that the job should be run serially with others of a given identifier.

• You can specify other calls that should be made on the basis of the result of this call.

• You can persist a reference to it, and periodically (after syncing your connection with the database, which happens whenever you begin or commit a transaction) check its status to see if it is equal to zc.async.interfaces.COMPLETED. When it is, the call has run to completion, either to success or an exception.

• You can look at the result of the call (once COMPLETED). It might be the result you expect, or a zc.twist.Failure, a subclass of twisted.python.failure.Failure, which is a way to safely communicate exceptions across connections and machines and processes.

Results

So here’s a simple story. What if you want to get a result back from a call? Look at the job.result after the call is COMPLETED.

>>> def imaginaryNetworkCall():
...     # let's imagine this makes a network call...
...     return "200 OK"
...
>>> job = queue.put(imaginaryNetworkCall)
>>> print job.result
None
>>> job.status == zc.async.interfaces.PENDING
True
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> t = transaction.begin()
>>> job.result
'200 OK'
>>> job.status == zc.async.interfaces.COMPLETED
True

Closures

What’s more, you can pass a Job to the put call. This means that you aren’t constrained to simply having simple non-argument calls performed asynchronously, but you can pass a job with a call, arguments, and keyword arguments–effectively, a kind of closure. Here’s a quick example. We’ll use the demo object, and its increase method, that we introduced above, but this time we’ll include some arguments [7].

With positional arguments:

>>> t = transaction.begin()
>>> job = queue.put(
...     zc.async.job.Job(root['demo'].increase, 5))
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> t = transaction.begin()
>>> root['demo'].counter
6

With keyword arguments (value):

>>> job = queue.put(
...     zc.async.job.Job(root['demo'].increase, value=10))
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> t = transaction.begin()
>>> root['demo'].counter
16

Note that arguments to these jobs can be any persistable object.

Failures

What happens if a call raises an exception? The return value is a Failure.

>>> def I_am_a_bad_bad_function():
...     return foo + bar
...
>>> job = queue.put(I_am_a_bad_bad_function)
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> t = transaction.begin()
>>> job.result
<zc.twist.Failure exceptions.NameError>

Failures can provide useful information such as tracebacks.

>>> print job.result.getTraceback()
... # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
Traceback (most recent call last):
...
exceptions.NameError: global name 'foo' is not defined
<BLANKLINE>

Callbacks

You can register callbacks to handle the result of a job, whether a Failure or another result.

Note that, unlike callbacks on a Twisted deferred, these callbacks do not change the result of the original job. Since callbacks are jobs, you can chain results, but generally callbacks for the same job all get the same result as input.

Also note that, during execution of a callback, there is no guarantee that the callback will be processed on the same machine as the main call. Also, some of the local functions, discussed below, will not work as desired.

Here’s a simple example of reacting to a success.

>>> def I_scribble_on_strings(string):
...     return string + ": SCRIBBLED"
...
>>> job = queue.put(imaginaryNetworkCall)
>>> callback = job.addCallback(I_scribble_on_strings)
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> job.result
'200 OK'
>>> callback.result
'200 OK: SCRIBBLED'

Here’s a more complex example of handling a Failure, and then chaining a subsequent callback.

>>> def I_handle_NameErrors(failure):
...     failure.trap(NameError) # see twisted.python.failure.Failure docs
...     return 'I handled a name error'
...
>>> job = queue.put(I_am_a_bad_bad_function)
>>> callback1 = job.addCallbacks(failure=I_handle_NameErrors)
>>> callback2 = callback1.addCallback(I_scribble_on_strings)
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> job.result
<zc.twist.Failure exceptions.NameError>
>>> callback1.result
'I handled a name error'
>>> callback2.result
'I handled a name error: SCRIBBLED'

zc.async.local

Jobs always run their callables in a thread, within the context of a connection to the ZODB. The callables have access to five special thread-local functions if they need them for special uses. These are available off of zc.async.local.

zc.async.local.getJob()

The getJob function can be used to examine the job, to get a connection off of _p_jar, to get the queue into which the job was put, or other uses.

zc.async.local.getQueue()

The getQueue function can be used to examine the queue, to put another task into the queue, or other uses. It is sugar for zc.async.local.getJob().queue.

zc.async.local.setLiveAnnotation(name, value, job=None)

The setLiveAnnotation tells the agent to set an annotation on a job, by default the current job, in another connection. This makes it possible to send messages about progress or for coordination while in the middle of other work.

As a simple rule, only send immutable objects like strings or numbers as values [8].

zc.async.local.getLiveAnnotation(name, default=None, timeout=0, poll=1, job=None)

The getLiveAnnotation tells the agent to get an annotation for a job, by default the current job, from another connection. This makes it possible to send messages about progress or for coordination while in the middle of other work.

As a simple rule, only ask for annotation values that will be immutable objects like strings or numbers [9].

If the timeout argument is set to a positive float or int, the function will wait at least that number of seconds until an annotation of the given name is available. Otherwise, it will return the default if the name is not present in the annotations. The poll argument specifies approximately how often to poll for the annotation, in seconds (to be more precise, a subsequent poll will be min(poll, remaining seconds until timeout) seconds away).

zc.async.local.getReactor()

The getReactor function returns the job’s dispatcher’s reactor. The getLiveAnnotation and setLiveAnnotation functions use this, along with the zc.twist package, to work their magic; if you are feeling adventurous, you can do the same.

zc.async.local.getDispatcher()

The getDispatcher function returns the job’s dispatcher. This might be used to analyze its non-persistent poll data structure, for instance (described later in configuration discussions).

Let’s give three of those a whirl. We will write a function that examines the job’s state while it is being called, and sets the state in an annotation, then waits for our flag to finish.

>>> def annotateStatus():
...     zc.async.local.setLiveAnnotation(
...         'zc.async.test.status',
...         zc.async.local.getJob().status)
...     zc.async.local.getLiveAnnotation(
...         'zc.async.test.flag', timeout=5)
...     return 42
...
>>> job = queue.put(annotateStatus)
>>> transaction.commit()
>>> import time
>>> def wait_for_annotation(job, key):
...     reactor.time_flies(dispatcher.poll_interval) # starts thread
...     for i in range(10):
...         while reactor.time_passes():
...             pass
...         transaction.begin()
...         if key in job.annotations:
...             break
...         time.sleep(0.1)
...     else:
...         print 'Timed out' + repr(dict(job.annotations))
...
>>> wait_for_annotation(job, 'zc.async.test.status')
>>> job.annotations['zc.async.test.status'] == (
...     zc.async.interfaces.ACTIVE)
True
>>> job.status == zc.async.interfaces.ACTIVE
True

[10]

>>> job.annotations['zc.async.test.flag'] = True
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> job.result
42

[11] getReactor and getDispatcher are for advanced use cases and are not explored further here.

Job Quotas

One class of asynchronous jobs are ideally serialized. For instance, you may want to reduce or eliminate the chance of conflict errors when updating a text index. One way to do this kind of serialization is to use the quota_names attribute of the job.

For example, let’s first show two non-serialized jobs running at the same time, and then two serialized jobs created at the same time. The first part of the example does not use queue_names, to show a contrast.

For our parallel jobs, we’ll do something that would create a deadlock if they were serial. Notice that we are mutating the job arguments after creation to accomplish this, which is supported.

>>> def waitForParallel(other):
...     zc.async.local.setLiveAnnotation(
...         'zc.async.test.flag', True)
...     zc.async.local.getLiveAnnotation(
...         'zc.async.test.flag', job=other, timeout=0.4, poll=0)
...
>>> job1 = queue.put(waitForParallel)
>>> job2 = queue.put(waitForParallel)
>>> job1.args.append(job2)
>>> job2.args.append(job1)
>>> transaction.commit()
>>> reactor.wait_for(job1, job2)
>>> job1.status == zc.async.interfaces.COMPLETED
True
>>> job2.status == zc.async.interfaces.COMPLETED
True
>>> job1.result is job2.result is None
True

On the other hand, for our serial jobs, we’ll do something that would fail if it were parallel. We’ll rely on quota_names.

Quotas verge on configuration, which is not what this section is about, because they must be configured on the queue. However, they also affect usage, so we show them here.

>>> def pause(other):
...     zc.async.local.setLiveAnnotation(
...         'zc.async.test.flag', True)
...     res = zc.async.local.getLiveAnnotation(
...         'zc.async.test.flag', timeout=0.4, poll=0.1, job=other)
...
>>> job1 = queue.put(pause)
>>> job2 = queue.put(imaginaryNetworkCall)

You can’t put a name in quota_names unless the quota has been created in the queue.

>>> job1.quota_names = ('test',)
Traceback (most recent call last):
...
ValueError: ('unknown quota name', 'test')
>>> queue.quotas.create('test')
>>> job1.quota_names = ('test',)
>>> job2.quota_names = ('test',)

Now we can see the two jobs being performed serially.

>>> job1.args.append(job2)
>>> transaction.commit()
>>> reactor.time_flies(dispatcher.poll_interval)
1
>>> for i in range(10):
...     t = transaction.begin()
...     if job1.status == zc.async.interfaces.ACTIVE:
...         break
...     time.sleep(0.1)
... else:
...     print 'TIME OUT'
...
>>> job2.status == zc.async.interfaces.PENDING
True
>>> job2.annotations['zc.async.test.flag'] = False
>>> transaction.commit()
>>> reactor.wait_for(job1)
>>> reactor.wait_for(job2)
>>> print job1.result
None
>>> print job2.result
200 OK

Quotas can be configured for limits greater than one at a time, if desired. This may be valuable when a needed resource is only available in limited numbers at a time.

Note that, while quotas are valuable tools for doing serialized work such as updating a text index, other optimization features sometimes useful for this sort of task, such as collapsing similar jobs, are not provided directly by this package. This functionality could be trivially built on top of zc.async, however [12].

Returning Jobs

Our examples so far have done work directly. What if the job wants to orchestrate other work? One way this can be done is to return another job. The result of the inner job will be the result of the first job once the inner job is finished. This approach can be used to break up the work of long running processes; to be more cooperative to other jobs; and to make parts of a job that can be parallelized available to more workers.

Returning Deferreds

What if you want to do work that doesn’t require a ZODB connection? You can also return a Twisted deferred (twisted.internet.defer.Deferred). When you then callback the deferred with the eventual result, the agent will be responsible for setting that value on the original deferred and calling its callbacks. This can be a useful trick for making network calls using Twisted or zc.ngi, for instance.

>>> def imaginaryNetworkCall2(deferred):
...     # make a network call...
...     deferred.callback('200 OK')
...
>>> import twisted.internet.defer
>>> import threading
>>> def delegator():
...     deferred = twisted.internet.defer.Deferred()
...     t = threading.Thread(
...         target=imaginaryNetworkCall2, args=(deferred,))
...     t.run()
...     return deferred
...
>>> job = queue.put(delegator)
>>> transaction.commit()
>>> reactor.wait_for(job)
>>> job.result
'200 OK'

Required Component Registrations

The required registrations can be installed for you by the zc.async.configure.base function. Most other examples in this package, such as those in the Usage section, use this in their test setup.

Again, for a quick start, you might just want to use the helper zc.async.configure.base function, and move on to the Required ZODB Set Up section below.

Here, though, we will go over each required registration to briefly explain what they are.

You must have three adapter registrations: IConnection to ITransactionManager, IPersistent to IConnection, and IPersistent to ITransactionManager.

The zc.twist package provides all of these adapters. However, zope.app.keyreference also provides a version of the connection adapter that is identical or very similar, and that should work fine if you are already using that package in your application.

>>> import zc.twist
>>> import zope.component
>>> zope.component.provideAdapter(zc.twist.transactionManager)
>>> zope.component.provideAdapter(zc.twist.connection)
>>> import ZODB.interfaces
>>> zope.component.provideAdapter(
...     zc.twist.transactionManager, adapts=(ZODB.interfaces.IConnection,))

We also need to be able to adapt functions and methods to jobs. The zc.async.job.Job class is the expected implementation.

>>> import types
>>> import zc.async.interfaces
>>> import zc.async.job
>>> zope.component.provideAdapter(
...     zc.async.job.Job,
...     adapts=(types.FunctionType,),
...     provides=zc.async.interfaces.IJob)
>>> zope.component.provideAdapter(
...     zc.async.job.Job,
...     adapts=(types.MethodType,),
...     provides=zc.async.interfaces.IJob)
>>> zope.component.provideAdapter( # optional, rarely used
...     zc.async.job.Job,
...     adapts=(zc.twist.METHOD_WRAPPER_TYPE,),
...     provides=zc.async.interfaces.IJob)

The queue looks for the UUID utility to set the assignerUUID job attribute, and may want to use it to optionally filter jobs during claim in the future. Also, the dispatcher will look for a UUID utility if a UUID is not specifically provided to its constructor.

>>> from zc.async.instanceuuid import UUID
>>> zope.component.provideUtility(
...     UUID, zc.async.interfaces.IUUID, '')

The UUID we register here is a UUID of the instance, which is expected to uniquely identify the process when in production. It is stored in the file specified by the ZC_ASYNC_UUID environment variable (or in os.join(os.getcwd(), 'uuid.txt') if this is not specified, for easy initial experimentation with the package).

>>> import uuid
>>> import os
>>> f = open(os.environ["ZC_ASYNC_UUID"])
>>> uuid_hex = f.readline().strip()
>>> f.close()
>>> uuid = uuid.UUID(uuid_hex)
>>> UUID == uuid
True

The uuid.txt file is intended to stay in the instance home as a persistent identifier.

Again, all of the required registrations above can be accomplished quickly with zc.async.configure.base.

Required ZODB Set Up

On a basic level, zc.async needs a setup that supports good conflict resolution. Most or all production ZODB storages now have the necessary APIs to support MVCC.

Of course, if you want to run multiple processes, you need ZEO. You should also then make sure that your ZEO server installation has all the code that includes conflict resolution, such as zc.queue, because, as of this writing, conflict resolution happens in the ZEO server, not in clients.

A more subtle decision is whether to use multiple databases. The zc.async dispatcher can generate a lot of database churn. It may be wise to put the queue in a separate database from your content database(s).

The downsides to this option include the fact that you must be careful to specify to which database objects belong; and that broken cross-database references are not handled gracefully in the ZODB as of this writing.

We will use multiple databases for our example here, because we are trying to demonstrate production-quality examples. We will show this with a pure-Python approach, rather than the ZConfig approach usually used by Zope. If you know ZConfig, that will be a reasonable approach as well; see zope.app.appsetup for how Zope uses ZConfig to set up multidatabases.

In our example, we create two file storages. In production, you might likely use ZEO; hooking ClientStorage up instead of FileStorage should be straight forward.

>>> databases = {}
>>> import ZODB.FileStorage
>>> storage = ZODB.FileStorage.FileStorage(
...     'main.fs', create=True)

>>> async_storage = ZODB.FileStorage.FileStorage(
...     'async.fs', create=True)

>>> from ZODB.DB import DB
>>> databases[''] = db = DB(storage)
>>> databases['async'] = async_db = DB(async_storage)
>>> async_db.databases = db.databases = databases
>>> db.database_name = ''
>>> async_db.database_name = 'async'
>>> conn = db.open()
>>> root = conn.root()

ZODB Configuration

Process Configuration

Optional Component Registrations for a Client Process

The only optional component registration potentially valuable for client instances that only put jobs in the queue is registering an adapter from persistent objects to a queue. The zc.async.queue.getDefaultQueue adapter does this for an adapter to the queue named ‘’ (empty string). Since that’s what we have from the ZODB Configuration above section, we’ll register it. Writing your own adapter is trivial, as you can see if you look at the implementation of this function.

>>> zope.component.provideAdapter(zc.async.queue.getDefaultQueue)
>>> zc.async.interfaces.IQueue(root) is queue
True

Starting the Dispatcher

Now we can start the reactor, and start the dispatcher. In some applications this may be done with an event subscriber to DatabaseOpened, as is done in zc.async.subscribers. Here, we will do it inline.

Any object that conforms to the specification of zc.async.interfaces.IReactor will be usable by the dispatcher. For our example, we will use our own instance of the Twisted select-based reactor running in a separate thread. This is separate from the Twisted reactor installed in twisted.internet.reactor, and so this approach can be used with an application that does not otherwise use Twisted (for instance, a Zope application using the “classic” zope publisher).

The testing module also has a reactor on which the Usage section relies, if you would like to see a minimal contract.

Configuring the basics is fairly simple, as we’ll see in a moment. The trickiest part is to handle signals cleanly. It is also optional! The dispatcher will eventually figure out that there was not a clean shut down before and take care of it. Here, though, essentially as an optimization, we install signal handlers in the main thread using reactor._handleSignals. reactor._handleSignals may work in some real-world applications, but if your application already needs to handle signals you may need a more careful approach. Again, see zc.async.subscribers for some options you can explore.

>>> import twisted.internet.selectreactor
>>> reactor = twisted.internet.selectreactor.SelectReactor()
>>> reactor._handleSignals()

Now we are ready to instantiate our dispatcher.

>>> import zc.async.dispatcher
>>> dispatcher = zc.async.dispatcher.Dispatcher(db, reactor)

Notice it has the uuid defined in instanceuuid.

>>> dispatcher.UUID == UUID
True

Now we can start the reactor and the dispatcher in a thread.

>>> import threading
>>> def start():
...     dispatcher.activate()
...     reactor.run(installSignalHandlers=0)
...
>>> thread = threading.Thread(target=start)
>>> thread.setDaemon(True)

>>> thread.start()

The dispatcher should be starting up now. Let’s wait for it to activate. We’re using a test convenience, get_poll, defined in the testing module.

>>> from zc.async.testing import get_poll
>>> poll = get_poll(dispatcher, 0)

We’re off! The events have been fired for registering and activating the dispatcher. Therefore, our subscriber to add our agent has fired.

We need to begin our transaction to synchronize our view of the database.

>>> t = transaction.begin()

We get the collection of dispatcher agents from the queue, using the UUID.

>>> dispatcher_agents = queue.dispatchers[UUID]

It has one agent–the one placed by our subscriber.

>>> dispatcher_agents.keys()
['main']
>>> agent = dispatcher_agents['main']

Now we have our agent! But…what is it [23]?

Agents

Agents are the way you control what a dispatcher’s worker threads do. They pick the jobs and assign them to their dispatcher when the dispatcher asks.

If a dispatcher does not have any agents in a given queue, it will not perform any tasks for that queue.

We currently have an agent that simply asks for the next available FIFO job. We are using an agent implementation that allows you to specify a callable to choose the job. That callable is now zc.async.agent.chooseFirst.

>>> agent.chooser is zc.async.agent.chooseFirst
True

Here’s the entire implementation of that function:

def chooseFirst(agent):
    return agent.queue.claim()

What would another agent do? Well, it might pass a filter function to claim. This function takes a job and returns a value evaluated as a boolean. For instance, let’s say we always wanted a certain number of threads available for working on a particular call; for the purpose of example, we’ll use operator.mul, though a more real-world example might be a network call or a particular call in your application.

>>> import operator
>>> def chooseMul(agent):
...     return agent.queue.claim(lambda job: job.callable is operator.mul)
...

Another variant would prefer operator.mul, but if one is not in the queue, it will take any.

>>> def preferMul(agent):
...     res = agent.queue.claim(lambda job: job.callable is operator.mul)
...     if res is None:
...         res = agent.queue.claim()
...     return res
...

Other approaches might look at the current jobs in the agent, or the agent’s dispatcher, and decide what jobs to prefer on that basis. The agent should support many ideas.

Let’s set up another agent, in addition to the chooseFirst one, that has the preferMul policy.

>>> agent2 = dispatcher_agents['mul'] = zc.async.agent.Agent(preferMul)

Another characteristic of agents is that they specify how many jobs they should pick at a time. The dispatcher actually adjusts the size of the ZODB connection pool to accommodate its agents’ size. The default is 3.

>>> agent.size
3
>>> agent2.size
3

We can change that at creation or later.

Finally, it’s worth noting that agents contain the jobs that are currently worked on by the dispatcher, on their behalf; and have a completed collection of the more recent completed jobs, beginning with the most recently completed job.

Logging and Monitoring

Logs are sent to the zc.async.events log for big events, like startup and shutdown, and errors. Poll and job logs are sent to zc.async.trace. Configure the standard Python logging module as usual to send these logs where you need. Be sure to auto-rotate the trace logs.

The package supports monitoring using zc.z3monitor, but using this package includes more Zope 3 dependencies, so it is not included here. If you would like to use it, see monitor.txt in the package and our next section: Configuration with Zope 3. Otherwise, if you want to roll your own monitoring, glance at monitor.py–you’ll see that most of the heavy lifting for the monitor support is done in the dispatcher, so it should be pretty easy to hook up the basic data another way.

>>> reactor.stop()

Shared Single Database Set Up

As described above, using a shared single database will probably be the quickest way to get started. Large-scale production usage will probably prefer to use the Two Database Set Up described later.

So, without further ado, here is the text of our zope.conf-alike, and of our site.zcml-alike [25].

>>> zope_conf = """
... site-definition %(site_zcml_file)s
...
... <zodb main>
...   <filestorage>
...     create true
...     path %(main_storage_path)s
...   </filestorage>
... </zodb>
...
... <product-config zc.z3monitor>
...   port %(monitor_port)s
... </product-config>
...
... <logger>
...   level debug
...   name zc.async
...   propagate no
...
...   <logfile>
...     path %(async_event_log)s
...   </logfile>
... </logger>
...
... <logger>
...   level debug
...   name zc.async.trace
...   propagate no
...
...   <logfile>
...     path %(async_trace_log)s
...   </logfile>
... </logger>
...
... <eventlog>
...   <logfile>
...     formatter zope.exceptions.log.Formatter
...     path STDOUT
...   </logfile>
...   <logfile>
...     formatter zope.exceptions.log.Formatter
...     path %(event_log)s
...   </logfile>
... </eventlog>
... """ % {'site_zcml_file': site_zcml_file,
...        'main_storage_path': os.path.join(dir, 'main.fs'),
...        'async_storage_path': os.path.join(dir, 'async.fs'),
...        'monitor_port': monitor_port,
...        'event_log': os.path.join(dir, 'z3.log'),
...        'async_event_log': os.path.join(dir, 'async.log'),
...        'async_trace_log': os.path.join(dir, 'async_trace.log'),}
...

In a non-trivial production system, you will also probably want to replace the file storage with a <zeoclient> stanza.

Also note that an open monitor port should be behind a firewall, of course.

We’ll assume that zdaemon.conf has been set up to put ZC_ASYNC_UUID in the proper place too. It would have looked something like this in the zdaemon.conf:

<environment>
  ZC_ASYNC_UUID /path/to/uuid.txt
</environment>

(Other tools, such as supervisor, also can work, of course; their spellings are different and are “left as an exercise to the reader” at the moment.)

We’ll do that by hand:

>>> os.environ['ZC_ASYNC_UUID'] = os.path.join(dir, 'uuid.txt')

Now let’s define our site-zcml-alike.

>>> site_zcml = """
... <configure xmlns='http://namespaces.zope.org/zope'
...            xmlns:meta="http://namespaces.zope.org/meta"
...            >
... <include package="zope.component" file="meta.zcml" />
... <include package="zope.component" />
... <include package="zc.z3monitor" />
... <include package="zc.async" file="basic_dispatcher_policy.zcml" />
...
... <!-- this is usually handled in Zope applications by the
...      zope.app.keyreference.persistent.connectionOfPersistent adapter -->
... <adapter factory="zc.twist.connection" />
... </configure>
... """

Now we’re done.

If we process these files, and wait for a poll, we’ve got a working set up [26].

>>> import zc.async.dispatcher
>>> dispatcher = zc.async.dispatcher.get()
>>> import pprint
>>> pprint.pprint(get_poll(dispatcher, 0))
{'': {'main': {'active jobs': [],
               'error': None,
               'len': 0,
               'new jobs': [],
               'size': 3}}}
>>> bool(dispatcher.activated)
True

We can ask for a job to be performed, and get the result.

>>> conn = db.open()
>>> root = conn.root()
>>> import zc.async.interfaces
>>> queue = zc.async.interfaces.IQueue(root)
>>> import operator
>>> import zc.async.job
>>> job = queue.put(zc.async.job.Job(operator.mul, 21, 2))
>>> import transaction
>>> transaction.commit()
>>> wait_for_result(job)
42

We can connect to the monitor server with telnet.

>>> import telnetlib
>>> tn = telnetlib.Telnet('127.0.0.1', monitor_port)
>>> tn.write('async status\n') # immediately disconnects
>>> print tn.read_all() # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
{
    "poll interval": {
        "seconds": ...
    },
    "status": "RUNNING",
    "time since last poll": {
        "seconds": ...
    },
    "uptime": {
        "seconds": ...
    },
    "uuid": "..."
}
<BLANKLINE>

Now we’ll “shut down” with a CTRL-C, or SIGINT, and clean up.

>>> import signal
>>> if getattr(os, 'getpid', None) is not None: # UNIXEN, not Windows
...     pid = os.getpid()
...     try:
...         os.kill(pid, signal.SIGINT)
...     except KeyboardInterrupt:
...         if dispatcher.activated:
...             assert False, 'dispatcher did not deactivate'
...     else:
...         print "failed to send SIGINT, or something"
... else:
...     dispatcher.reactor.callFromThread(dispatcher.reactor.stop)
...     for i in range(30):
...         if not dispatcher.activated:
...             break
...         time.sleep(0.1)
...     else:
...         assert False, 'dispatcher did not deactivate'
...
>>> import transaction
>>> t = transaction.begin() # sync
>>> import zope.component
>>> import zc.async.interfaces
>>> uuid = zope.component.getUtility(zc.async.interfaces.IUUID)
>>> da = queue.dispatchers[uuid]
>>> bool(da.activated)
False

>>> db.close()
>>> import shutil
>>> shutil.rmtree(dir)

These instructions are very similar to the Two Database Set Up.

Two Database Set Up

Even though it is a bit more trouble to set up, large-scale production usage will probably prefer to use this approach, over the shared single database described above.

For our zope.conf, we only need one additional stanza to the one seen above:

<zodb async>
  <filestorage>
    create true
    path REPLACE_THIS_WITH_PATH_TO_STORAGE
  </filestorage>
</zodb>

(You would replace “REPLACE_THIS_WITH_PATH_TO_STORAGE” with the path to the storage file.)

As before, you will probably prefer to use ZEO rather than FileStorage in production.

The zdaemon.conf instructions are the same: set the ZC_ASYNC_UUID environment variable properly in the zdaemon.conf file.

For our site.zcml, the only difference is that we use the multidb_dispatcher_policy.zcml file rather than the basic_dispatcher_policy.zcml file.

If you want to change policy, change “multidb_dispatcher_policy.zcml” to “dispatcher.zcml” in the example above and register your replacement bits for the policy in “multidb_dispatcher_policy.zcml”. You’ll see that most of that comes from code in subscribers.py, which can be adjusted easily.

If we process the files described above, and wait for a poll, we’ve got a working set up [27].

>>> import zc.async.dispatcher
>>> dispatcher = zc.async.dispatcher.get()
>>> import pprint
>>> pprint.pprint(get_poll(dispatcher, 0))
{'': {'main': {'active jobs': [],
               'error': None,
               'len': 0,
               'new jobs': [],
               'size': 3}}}
>>> bool(dispatcher.activated)
True

As before, we can ask for a job to be performed, and get the result.

>>> conn = db.open()
>>> root = conn.root()
>>> import zc.async.interfaces
>>> queue = zc.async.interfaces.IQueue(root)
>>> import operator
>>> import zc.async.job
>>> job = queue.put(zc.async.job.Job(operator.mul, 21, 2))
>>> import transaction
>>> transaction.commit()
>>> wait_for_result(job)
42

Hopefully zc.async will be an easy-to-configure, easy-to-use, and useful tool for you! Good luck! [28]

What Might Go Wrong?

Sometimes bad things happen in the course of processing tasks. What might go wrong? How does zc.async handle these errors? What are your responsibilities?

First, what might go wrong?

• zc.async could have a problem while polling for jobs. We’ll call this a “polling exception.”

• zc.async could have a problem while performing a particular job. We’ll call this a “job-related exception.”

For the purpose of this discussion, we will omit the possibility that zc.async has a bug. That is certainly a possibility, but the recovery story is not predictable, and if we knew of a bug, we’d try to fix it, rather than discuss it here!

We’ll discuss both polling exceptions and job related exceptions, then drill down into some specific scenarios. This will illuminate how your code and zc.async’s can work together to handle them.

Scenarios: Polling Errors

Scenarios: Job-Related Errors

Summary

• Make sure you are using zope.app.testing version 3.4.2 or newer, or else ftests may intermittently raise spurious errors having to do with a missing _result attribute on a request’s response.

• The Zope 3 tests use DemoStorage, which does not use MVCC. This can lead to your tests having occasional ConflictErrors that will not occur in production. In common cases for ftests, you won’t notice these because of Zope’s usual retry policy. Unit or integration tests may show these problems.

• Set up the basic configuration in zcml or Python (see examples below), but you need to make sure that ftests do not use dispatchers started in the application. Start up ftest (or integration test) dispatchers separately, using zc.async.ftesting.setUp, and then tear down after the tests are done with zc.async.ftesting.tearDown.

  The ftest dispatcher polls every tenth of a second, so you shouldn’t need to wait long for you job to get started in your tests.

• General zc.async testing tools such as zc.async.dispatcher.get, zc.async.testing.get_poll and zc.async.testing.wait_for_result can still be useful for in-depth zc.async tests.

• If you don’t want to dig into guts in your functional tests to use the tools described in the previous point, consider making a view to check on job status using a data structure like JSON, and looking at that in your tests.

• The setUp code by default sends critical log messages to __stdout__ so it can help diagnose why a callback might never complete.

Discussion

Normally, in a Zope 3 configuration that uses zc.async, you configure it when you start your application. For instance, you might include a zc.async zcml file like basic_dispatcher_policy.zcml that performs the necessary set up.

However, the Zope 3 ftesting layer database dance doesn’t play well with zc.async unless you take a bit of extra care.

This is because zc.async will be started with the ftests’ underlying database, and then the test will be run with a DemoStorage wrapper. The zc.async dispatcher will run, then, but it will never see the changes that you make in the wrapper DemoStorage that your test manipulates. This can be mystifying and frustrating.

Because of this, when you write a Zope 3 app that wants to use both layered ftests and zc.async, you have to set things up in a mildly inconvenient way.

When you start your application normally, use configuration (zcml or grok or whatever) to register subscribers like the ones in subscribers.py: adding queues, starting dispatchers, and adding agents.

But don’t have this configuration registered for your ftests. Instead, bypass that part of your site’s configuration in your ftesting layer, and use the zc.async.ftesting.setUp function to set zc.async up in tests when you need it, in a footnote of your test or in a similar spot.

You’ll still want the basic adapters registered, as found in zc.async’s configure.zcml or configure.py files; and maybe the zc.async.queue.getDefaultQueue adapter too. This can be registered in ftesting.zcml with this snippet:

<include package="zc.async" />
<adapter factory="zc.async.queue.getDefaultQueue" />

Or in Python, you might want to do something like this:

>>> import zc.async.configure
>>> zc.async.configure.base() # or, more likely, ``minimal`` for Zope 3
>>> import zope.component
>>> import zc.async.queue
>>> zope.component.provideAdapter(zc.async.queue.getDefaultQueue)

Don’t forget to call tearDown (see below) at the end of your test!

Here’s a usage example.

As mentioned above, setUp does expect the necessary basic adapters to already be installed.

Zope 3 ftests generally have a getRootObject hanging around to give you the root object in the Zope application (but not in the ZODB). Therefore, this function tries to be helpful, for better and worse, and muck around in the locals to find it. If you want it to leave your locals alone, pass it a database connection.

So, here’s some set up. We create a database and make our stub getRootFolder function in the globals.

>>> import transaction
>>> import BTrees
>>> import ZODB.FileStorage
>>> storage = ZODB.FileStorage.FileStorage(
...     'zc_async.fs', create=True)
>>> from ZODB.DB import DB
>>> db = DB(storage)
>>> conn = db.open()
>>> root = conn.root()
>>> PseudoZopeRoot = root['Application'] = BTrees.family32.OO.BTree()
>>> transaction.commit()
>>> def _getRootObject():
...     return PseudoZopeRoot
...
>>> globals()['getRootFolder'] = _getRootObject

Notice we are using a real FileStorage, and not a DemoStorage, as is usually used in ftests. The fact that DemoStorage does not have MVCC can sometimes lead standard ftests to raise spurious ReadConflictErrors that will not actually occur in production. The ConflictErrors will generally be retried, so your tests should usually pass, even though you might see some “complaints”.

Now we can call setUp as if we were in a functional test.

>>> import zc.async.ftesting
>>> zc.async.ftesting.setUp()

Now the dispatcher is activated and the polls are running. The function sets up a dispatcher that polls much more frequently than usual–every 0.1 seconds rather than every 5, so that tests might run faster–but otherwise uses typical zc.async default values.

It’s worth noting a few tricks that are particularly useful for tests here. We’ll also use a couple of them to verify that setUp did its work.

zc.async.dispatcher.get() returns the currently installed dispatcher. This can let you check if it is activated and polling and use its simple statistical methods, if you want.

>>> import zc.async.dispatcher
>>> dispatcher = zc.async.dispatcher.get()

For now, we’ll just see that the dispatcher is activated.

>>> bool(dispatcher.activated)
True

See the dispatcher.txt for information on information you can get from the dispatcher object.

zc.async.testing has a number of helpful functions for testing. get_poll is the most pertinent here: given a dispatcher, it will give you the next poll. This is a good way to make sure that a job you just put in has had a chance to be claimed by a dispatcher. It’s also a reasonable way to verify that the dispatcher has started. setUp already gets the first two polls, so it’s definitely all started.

>>> import zc.async.testing
>>> import pprint
>>> pprint.pprint(zc.async.testing.get_poll(dispatcher))
{'': {'main': {'active jobs': [],
               'error': None,
               'len': 0,
               'new jobs': [],
               'size': 3}}}

Other useful testing functions are zc.async.testing.wait_for_result, which waits for the result on a give job and returns it; and zc.async.testing.wait_for_annotation, which waits for a given annotation on a given job. These are demonstrated in various doctests in this package, but should also be reasonably simple and self-explanatory.

Callbacks will retry some errors forever, by default. The logic is that callbacks are often the “cleanup” and must be run. This can lead to confusion in debugging tests, though, because the retry warnings are sent to the log, and the log is not usually monitored in functional tests.

setUp tries to help with this by adding logging of CRITICAL log messages in the “zc.async” logger to stdout.

>>> import logging
>>> logging.getLogger('zc.async.event').critical('Foo!')
Foo!
>>> logging.getLogger('zc.async.event').error('Bar!')

Once you have finished your tests, make sure to shut down your dispatcher, or the testing framework will complain about an unstopped daemon thread. zc.async.ftesting.tearDown will do the trick.

>>> zc.async.ftesting.tearDown()
>>> dispatcher.activated
False

You can then start another async-enabled functional test up again later in the same layer, of course.

>>> zc.async.ftesting.setUp()
>>> dispatcher = zc.async.dispatcher.get()
>>> bool(dispatcher.activated)
True

>>> zc.async.ftesting.tearDown()
>>> dispatcher.activated
False

Also worth noting, as mentioned in the summary, is that you should use zope.app.testing version 3.4.2 or higher to avoid getting spurious, intermittent bug reports from ftests that use zc.async.

In your test or your test’s tearDown, if you used a FileStorage, as we did here, you’ll need to clean up as well. We normally do this in our tests’ tearDowns, but we do it here, now, to make the point.

>>> db.close()
>>> storage.close()
>>> storage.cleanup()

>>> del storage # we just do this to not confuse our own tearDown code.
>>> del globals()['getRootFolder'] # clean up globals