zc.twist 1.0

Twist: Talking to the ZODB in Twisted Reactor Calls

The twist package contains a few functions and classes, but primarily a helper for having a deferred call on a callable persistent object, or on a method on a persistent object. This lets you have a Twisted reactor call or a Twisted deferred callback affect the ZODB. Everything can be done within the main thread, so it can be full-bore Twisted usage, without threads. There are a few important “gotchas”: see the Gotchas section below for details.

The main API is Partial. You can pass it a callable persistent object, a method of a persistent object, or a normal non-persistent callable, and any arguments or keyword arguments of the same sort. DO NOT use non-persistent data structures (such as lists) of persistent objects with a database connection as arguments. This is your responsibility.

If nothing is persistent, the partial will not bother to get a connection, and will behave normally.

>>> from zc.twist import Partial
>>> def demo():
...     return 42
...
>>> Partial(demo)()
42

Now let’s imagine a demo object that is persistent and part of a database connection. It has a count attribute that starts at 0, a __call__ method that increments count by an amount that defaults to 1, and an decrement method that reduces count by an amount that defaults to 1 [1]. Everything returns the current value of count.

>>> demo.count
0
>>> demo()
1
>>> demo(2)
3
>>> demo.decrement()
2
>>> demo.decrement(2)
0
>>> import transaction
>>> transaction.commit()

Now we can make some deferred calls with these examples. We will use transaction.begin() to sync our connection with what happened in the deferred call. Note that we need to have some adapters set up for this to work. The twist module includes implementations of them that we will also assume have been installed [2].

>>> call = Partial(demo)
>>> demo.count # hasn't been called yet
0
>>> deferred = call()
>>> demo.count # we haven't synced yet
0
>>> t = transaction.begin() # sync the connection
>>> demo.count # ah-ha!
1

We can use the deferred returned from the call to do somethin with the return value. In this case, the deferred is already completed, so adding a callback gets instant execution.

>>> def show_value(res):
...     print res
...
>>> ignore = deferred.addCallback(show_value)
1

We can also pass the method.

>>> call = Partial(demo.decrement)
>>> deferred = call()
>>> demo.count
1
>>> t = transaction.begin()
>>> demo.count
0

Arguments are passed through.

>>> call = Partial(demo)
>>> deferred = call(2)
>>> t = transaction.begin()
>>> demo.count
2
>>> call = Partial(demo.decrement)
>>> deferred = call(amount=2)
>>> t = transaction.begin()
>>> demo.count
0

They can also be set during instantiation.

>>> call = Partial(demo, 3)
>>> deferred = call()
>>> t = transaction.begin()
>>> demo.count
3
>>> call = Partial(demo.decrement, amount=3)
>>> deferred = call()
>>> t = transaction.begin()
>>> demo.count
0

Arguments themselves can be persistent objects. Let’s assume a new demo2 object as well.

>>> demo2.count
0
>>> def mass_increment(d1, d2, value=1):
...     d1(value)
...     d2(value)
...
>>> call = Partial(mass_increment, demo, demo2, value=4)
>>> deferred = call()
>>> t = transaction.begin()
>>> demo.count
4
>>> demo2.count
4
>>> demo.count = demo2.count = 0 # cleanup
>>> transaction.commit()

ConflictErrors make it retry.

In order to have a chance to simulate a ConflictError, this time imagine we have a runner that can switch execution from the call to our code using pause, retry and resume (this is just for tests–remember, calls used in non-threaded Twisted should be non-blocking!) [3].

>>> import sys
>>> demo.count
0
>>> call = Partial(demo)
>>> runner = Runner(call) # it starts paused in the middle of an attempt
>>> call.attempt_count
1
>>> demo.count = 5 # now we will make a conflicting transaction...
>>> transaction.commit()
>>> runner.retry()
>>> call.attempt_count # so it has to retry
2
>>> t = transaction.begin()
>>> demo.count # our value hasn't changed...
5
>>> runner.resume() # but now call will be successful on the second attempt
>>> call.attempt_count
2
>>> t = transaction.begin()
>>> demo.count
6

After five retries (currently hard-coded), the retry fails, raising the last ConflictError. This is returned to the deferred. The failure put on the deferred will have a sanitized traceback. Here, imagine we have a deferred (named deferred) created from such a an event [4].

>>> res = None
>>> def get_result(r):
...     global res
...     res = r # we return None to quiet Twisted down on the command line
...
>>> d = deferred.addErrback(get_result)
>>> print res.getTraceback() # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
Traceback (most recent call last):
...
ZODB.POSException.ConflictError: database conflict error...

Other errors are returned to the deferred as well, as sanitized failures [5].

>>> call = Partial(demo)
>>> d = call('I do not add well with integers')
>>> d = d.addErrback(get_result)
>>> print res.getTraceback() # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
Traceback (most recent call last):
...
...TypeError: unsupported operand type(s) for +=: 'int' and 'str'

The call tries to be a good connection citizen, waiting for a connection if the pool is at its maximum size. This code relies on the twisted reactor; we’ll use a time_flies function, which takes seconds to move ahead, to simulate time passing in the reactor [6].

>>> db.setPoolSize(1)
>>> db.getPoolSize()
1
>>> demo.count = 0
>>> transaction.commit()
>>> call = Partial(demo)
>>> res = None
>>> deferred = call()
>>> d = deferred.addCallback(get_result)
>>> call.attempt_count
0
>>> time_flies(.1) >= 1 # returns number of connection attempts
True
>>> call.attempt_count
0
>>> res # None
>>> db.setPoolSize(2)
>>> db.getPoolSize()
2
>>> time_flies(.2) >= 1
True
>>> call.attempt_count > 0
True
>>> res
1
>>> t = transaction.begin()
>>> demo.count
1

If it takes more than a second or two, it will eventually just decide to grab one. This behavior may change.

>>> db.setPoolSize(1)
>>> db.getPoolSize()
1
>>> call = Partial(demo)
>>> res = None
>>> deferred = call()
>>> d = deferred.addCallback(get_result)
>>> call.attempt_count
0
>>> time_flies(.1) >= 1
True
>>> call.attempt_count
0
>>> res # None
>>> time_flies(1.9) >= 2 # for a total of at least 3
True
>>> res
2
>>> t = transaction.begin()
>>> demo.count
2

Without a running reactor, this functionality will not work [7]. Also, it relies on an undocumented, protected attribute on the ZODB.DB, so is fragile across ZODB versions.

You can also specify a reactor for the partial using setReactor, if you don’t want to use the standard one installed by twisted in twisted.internet.reactor. [8]

Footnotes

If it takes more than a second or two, it will eventually just decide to grab one. This behavior may change.

>>> db.setPoolSize(1)
>>> db.getPoolSize()
1
>>> call = Partial(demo).setReactor(faux)
>>> res = None
>>> deferred = call()
>>> d = deferred.addCallback(get_result)
>>> call.attempt_count
0
>>> time_flies(.1) >= 1
True
>>> call.attempt_count
0
>>> res # None
>>> time_flies(1.9) >= 2 # for a total of at least 3
True
>>> res
2
>>> t = transaction.begin()
>>> demo.count
2

Changes