"""

Module-level function that lets you call a function in a dedicated

thread and get a C{Deferred} to its result, with no fuss on your

part.

The thread will remain alive and will be used for further calls to

this function and this function only.

Call with I{f}, I{*args}, and I{**kw} as usual.

This is AsynQueue's single-threaded, queued, I{doNext}-able,

L{iteration.Deferator}-able answer to Twisted's C{deferToThread}.

If you expect a deferred iterator as your result (an instance of

L{iteration.Deferator}), supply an C{IConsumer} implementor via

the I{consumer} keyword. Each iteration will be written to it, and

the deferred will fire when the iterations are done. Otherwise,

the deferred will fire with an L{iteration.Deferator}.

If you want to kill the dedicated thread, just call this function

with no arguments, not even a callable object I{f}. A C{Deferred}

will be returned that fires when the thread is gone.

"""

if not fargs:

tl = _DTL[0]

if tl is None:

return defer.succeed(None)

return tl.stop()

if _DTL[0] is None:

tl = _DTL[0] = ThreadLooper()

reactor.addSystemEventTrigger('before', 'shutdown', tl.stop)

"""

I am a L{TaskQueue} for dispatching arbitrary callables to be run

by a single worker thread.

Having one and just one worker thread is surprisingly useful. It

lets you do synchronous processing without blocking Twisted's

event loop, yet assures you that objects processed during one

queued task will not be disturbed until completion of the Deferred

callback chain from that task.

"""

def __init__(self, **kw):

"""C{ThreadQueue}(**kw)"""

raw = kw.pop('raw', False)

TaskQueue.__init__(self, **kw)

self.worker = ThreadWorker(raw=raw)

self.d = self.attachWorker(self.worker)

def deferToThread(self, f, *args, **kw):

"""

Runs the f-args-kw call in my dedicated worker thread, skipping

past the queue. As with a regular L{TaskQueue.call}, returns a

C{Deferred} that fires with the result and deals with

iterators.

"""

return util.callAfterDeferred(

"""

I implement an L{IWorker} that runs tasks in a dedicated worker

thread.

@cvar cQualified: A task series that all instances of me are

qualified to perform.

@ivar iQualified: A task series that this instance of me is

qualified to perform. Usually left blank, unless you want only

some workers doing certain tasks.

@ivar tasks: A list of pending tasks.

@ivar t: An instance of L{ThreadLooper}.

@keyword series: A list of one or more task series that this

particular instance of me is qualified to handle.

@keyword raw: Set C{True} if you want raw iterators to be returned

instead of L{iteration.Deferator} instances. You can override

this in with the same keyword set C{False} in a call.

"""

cQualified = ['thread', 'local']

def __init__(self, series=[], raw=False):

"""C{ThreadWorker}(series=[], raw=False)"""

self.tasks = []

# Is this really necessary?

# -----------------------------------------

self.tasksPendingBeforeShutdown = {}

# -----------------------------------------

self.iQualified = series

self.t = ThreadLooper(raw)

def setResignator(self, callableObject):

self.t.dLock.addStopper(callableObject)

def run(self, task):

"""

Returns a C{Deferred} that fires only after the threaded call is

done for the supplied I{task}.

I do basic FIFO queuing of calls to this method, but priority

queuing is above my paygrade and you'd best honor my deferred

and let someone like L{tasks.TaskHandler} only call this

method when I say I'm ready.

One simple thing I B{will} do is apply the I{doNext} keyword

to any task with the highest priority, -20 or lower (for a

L{base.TaskQueue.call} with its own I{doNext} keyword set). If

you call this method one task at a time like you're supposed

to, even that won't make a difference, except that it will cut

in front of any existing call with I{doNext} set. So use

judiciously.

"""

def done(statusResult):

if task in self.tasks:

self.tasks.remove(task)

if statusResult[0] == b'i':

# What we got is a Deferator, but if a consumer was

# supplied, we need to couple an IterationProducer to

# it and fire the task callback with the deferred from

# running the producer.

if consumer:

dr = statusResult[1]

ip = iteration.IterationProducer(dr, consumer)

statusResult = (b'i', ip)

task.d.callback(statusResult)

if task in self.tasksPendingBeforeShutdown:

self.tasksPendingBeforeShutdown.pop(task).callback(None)

self.tasks.append(task)

f, args, kw = task.callTuple

consumer = kw.pop('consumer', None)

if task.priority <= -20:

kw['doNext'] = True

return self.t.call(f, *args, **kw).addCallback(done)

def stop(self):

"""

Returns a C{Deferred} that fires when all pending tasks have been

run, the task loop has ended and its thread has terminated.

Waits for all pending tasks to finish because they run in the

task loop and can't do so once the task loop has ended. So the

list of outstanding tasks that is the deferred result should

always be empty.

"""

def tasksDone(null):

return self.t.stop()

dList = []

for task in self.tasks:

d = defer.Deferred()

self.tasksPendingBeforeShutdown[task] = d

dList.append(d)

return defer.DeferredList(dList).addCallback(tasksDone)

def crash(self):

"""

Since a thread can only terminate itself, calling this method only

forces firing of the deferred returned from a previous call to

L{stop} and returns the task that hung the thread.

"""

self.t.stop()

"""

I run function calls in a dedicated thread.

Each call returns a C{Deferred} to its eventual result, which is a

2-tuple containing the status of the last call and its result

according to the format of L{util.CallRunner}.

If the result is an iterable other than one of Python's built-in

ones, the C{Deferred} fires with an instance of

L{iteration.Prefetcherator} instead. Couple it to your own

deferator to iterate over the underlying iterable running in my

thread. You can disable this behavior by setting C{raw=True} in

the constructor, or enable/disable it on an individual call by

setting raw=True/False.

@ivar timeout: The wait timeout, which defaults to 60 (one

minute). This is just how long the thread loop waits before

checking for a pending deferred and firing it with a timeout

error. Otherwise, it simply waits another minute, and it can do

that forever with no problem.

@keyword raw: Set C{True} to have calls return (deferred)

iterators rather than instances of L{Prefetcherator}.

"""

timeout = 60

def __init__(self, raw=False):

"""C{ThreadLooper}(raw=False)"""

# Just a simple attribute to indicate if the thread loop is

# running, mostly for unit testing

self.threadRunning = True

# Tools

self.runner = util.CallRunner(raw)

self.dLock = util.DeferredLock()

self.event = threading.Event()

self.thread = threading.Thread(name=repr(self), target=self.loop)

self.thread.start()

# Deferred Tracker to wait for any running Deferators before

# shutdown

self.dt = util.DeferredTracker()

def loop(self):

"""

Runs a loop in a dedicated thread that waits for new tasks. The loop

exits when a C{None} object is supplied as a task.

"""

def callback(status, result):

reactor.callFromThread(self.d.callback, (status, result))

self.threadRunning = True

while True:

# Wait here for my main-thread caller to release me for

# another call

self.event.wait(self.timeout)

# For Python 2.7 and above, we could have just done

# if not self.event.wait(...):

if not self.event.isSet():

# Timed out waiting for the next call. If there indeed

# was one, we need to let the caller know. That

# shouldn't ever happen, though.

if hasattr(self, 'd') and not self.d.called:

callback('e', "Thread timed out waiting for this call!")

continue

if self.callTuple is None:

# Shutdown was requested

break

status, result = self.runner(self.callTuple)

# We are about to call back the shared deferred, so clear

# the event to force me to wait for the next call at the

# top of the loop. The main thread will not set the event

# again until the callback is done, so this is safe.

self.event.clear()

# OK, now call the shared deferred

callback(status, result)

# Broken out of loop, the thread now dies

self.threadRunning = False

@defer.inlineCallbacks

def call(self, f, *args, **kw):

"""

Runs the supplied callable function with any args and keywords in

a dedicated thread, returning a C{Deferred} that fires with a

status/result tuple.

Calls are done in the order received, unless you set

C{doNext=True}.

Set C{raw=True} to have a raw iterator returned instead of a

Deferator, or C{raw=False} to have a L{Deferator} returned

instead of a raw iterator, contrary to the instance-wide

default set with the constructor keyword 'raw'.

"""

yield self.dLock.acquire(kw.pop('doNext', False))

self.callTuple = f, args, kw

self.d = defer.Deferred()

# The callTuple is set for this call along with the deferred

# to be called back with its result, so release the thread to

# work on it, firing this deferred's callback with its result.

self.event.set()

statusResult = yield self.d

# The deferred lock is released after the call is done so

# that another call can proceed. This is NOT the same as

# the event used as a threading lock. It keeps the main

# thread from setting that event before the thread loop is

# ready for that.

self.dLock.release()

status, result = statusResult

if status == b'i':

ID = str(hash(result))

pf = iteration.Prefetcherator(ID)

ok = yield pf.setup(result)

if ok:

# OK, we can iterate this

result = iteration.Deferator(

repr(pf), self.deferToThread, pf.getNext)

# Make sure Deferator is done before shutting down

self.dt.put(result.d)

else:

# An iterator, but not one we could prefetch

# from. Probably empty.

result = []

# Not an iterator, at least not one being specially

# processed; we already have our result

defer.returnValue((status, result))

def dr2ip(self, dr, consumer=None):

"""

Converts a L{Deferator} into an L{iteration.IterationProducer},

with a consumer registered if you supply one.

Then each iteration will be written to your consumer, and the

deferred returned will fire when the iterations are

done. Otherwise, the deferred will fire with an

L{iteration.IterationProducer} and you will have to register

with and run it yourself.

"""

ip = iteration.IterationProducer(dr)

if consumer:

ip.registerConsumer(consumer)

return ip.run()

return ip

def deferToThread(self, f, *args, **kw):

"""

My single-threaded, queued, doNext-able, Deferator-able answer to

Twisted's deferToThread.

If you expect a deferred iterator as your result (an instance

of L{iteration.Deferator}), supply an C{IConsumer} implementor

via the I{consumer} keyword. Each iteration will be written to

it, and the deferred will fire when the iterations are

done. Otherwise, unless the I{raw} option has been set C{True}

in my constructor or as a keyword to this call, the deferred

will fire with an L{iteration.Deferator}.

"""

def done(statusResult):

status, result = statusResult

if status == b'e':

return Failure(errors.ThreadError(result))

elif status == b'i':

if consumer:

ip = iteration.IterationProducer(dr, consumer)

return ip.run()

return result

return result

consumer = kw.pop('consumer', None)

return self.call(f, *args, **kw).addCallback(done)

def stop(self):

"""

Returns a C{Deferred} that fires when all tasks and instances of

L{Deferator} are done, the task loop has ended, and its thread

has terminated.

"""

def deferatorsDone(null):

if self.threadRunning:

# Tell the thread to quit with a null task

self.callTuple = None

self.event.set()

# Now stop the lock

self.dLock.addStopper(self.thread.join)

return self.dLock.stop()

"""

Abstract base class for objects that access a global thread pool

instead of starting their own threads.

"""

minThreads = 2

maxThreads = 10

@classmethod

def setup(cls, maxThreads=None):

"""

Sets up stuff class-wide, with all the potential pitfalls that

entails.

Sets up all present and future instances of L{Consumerator},

L{Filerator}, L{OrderedItemProducer}, and any other subclasses

of me with a thread pool having at least two and no more than

I{maxThreads} threads.

If this method is called with a thread pool already

instantiated, and I{maxThreads} is specified and different

from the current value, it adjusts the maximum thread pool

size for B{all} instances, current and future, absent yet

another call with still different values.

@note: In the several years that have gone by since I wrote

this, I realized that it's almost always a terrible idea

to make class-wide settings of anything. But it's mature

code and hasn't given me any trouble.

@keyword maxThreads: Set to a maximum number of threads to

use, for all present and future instances.

"""

if maxThreads:

if hasattr(cls, '_pool') and maxThreads != cls.maxThreads:

cls._pool.adjustPoolSize(

minThreads=cls.minThreads,

maxThreads=maxThreads)

cls.maxThreads = maxThreads

if not hasattr(cls, '_pool'):

cls._pool = threadpool.ThreadPool(

minthreads=cls.minThreads, maxthreads=cls.maxThreads,

name="AsynQueue.IterationGetter")

reactor.addSystemEventTrigger('before', 'shutdown', cls.shutdown)

if not hasattr(cls, 'dtp'):

cls.dtp = util.DeferredTracker()

@classmethod

def shutdown(cls):

"""

Shuts down all threads, returning a C{Deferred} that fires when

everything's done, class-wide.

"""

def ready(null):

if hasattr(cls, '_pool'):

cls._pool.stop()

del cls._pool

return cls.dtp.deferToAll().addCallback(ready)

@classmethod

def deferToThreadInPool(cls, f, *args, **kw):

"""

Runs the C{f-args-kw} call combo in one of my threads, returning a

C{Deferred} that fires with the eventual result. Can be run

from the class or any instance of me with the exact same result.

"""

def done(success, result):

f = d.callback if success else d.errback

reactor.callFromThread(f, result)

d = defer.Deferred()

cls.dtp.put(d)

if not cls._pool.started:

cls._pool.start()

cls._pool.callInThreadWithCallback(done, f, *args, **kw)

return d

@property

def pool(self):

"""

Returns a reference to the class-wide threadpool, starting it if

this is the first time it's been used.

"""

if not self._pool.started:

self._pool.start()

"""

Abstract base class for objects that munch data on one end and act

like iterators to yield it on the other end.

@see: L{Consumerator} and L{Filerator}.

"""

class IterationStopper:

pass

def __init__(self, maxThreads=None):

"""C{IterationGetter(maxThreads=None)}"""

self.setup(maxThreads)

self.runState = 'init'

self.dList = []

def start(self):

"""

Call this when I should start listening for iterations.

Sets up locks for my iteration-consuming thread (from my

C{ThreadPool}), the blocking-iterator thread, and the

next-iteration event. Lock both of my iteration-processing

loops until an iteration is received, but leaves the

next-iteration lock I{nLock} unlocked. The iteration-consuming

thread will lock it to overwrite the blocking-iterator

thread's value of each iteration.

Calls my subclass's L{loop} method in its own thread. If too

many threads are currently open, queues the loop call until

one finishes up in some other instance of me.

"""

def done(success, result):

f = d.callback if success else d.errback

reactor.callFromThread(f, result)

self.runState = 'started'

d = defer.Deferred()

self.dtp.put(d)

self.dList.append(d)

# Locks for my iteration-consuming thread, the

# blocking-iterator thread, and the next-iteration event

self.cLock = threading.Semaphore()

self.bLock = threading.Lock()

self.nLock = threading.Lock()

# Lock both of my iteration-processing loops until an

# iteration is received

self.cLock.acquire()

self.bLock.acquire()

# Call my subclass's loop function in a pool thread

self.pool.callInThreadWithCallback(done, self.loop)

def loop(self):

"""

Override this method to generate a value for my iterations. The

method must be synchronized with a series of locking

primitives:

1. First, wait to acquire I{cLock}. This will be released

when a value has been given to my subclass instance,

e.g., through its C{write} method. At this point, you

can retrieve the value and let your value provider know

it is free to provide more.

2. Then wait to acquire I{nLock}. This will be released

when the L{next} method has obtained its next iteration

value from I{bIterationValue}. At this point, overwrite

I{bIterationValue} with the new value you've just

obtained, or with an instance of L{IterationStopper} if

iteration is done.

3. Release I{bLock} to let the L{next} loop know it can

process the next iteration value (or iteration stopper)

now in I{bIterationValue}.

@see: L{Consumerator.loop} and L{Filerator.loop}

"""

raise NotImplementedError("You must override this in a subclass")

def deferUntilDone(self):

"""

Returns a C{Deferred} that fires when I am done iterating.

"""

return defer.DeferredList(self.dList)

# Iterator implementation -------------------------------------------------

# Call in its own thread

def __iter__(self):

return self

def __next__(self):

# Wait for the next iteration to be produced

self.bLock.acquire()

# Get a local reference to the iteration value

value = self.bIterationValue

# Now it can be changed, so release my iteration-consuming

# loop to do so

self.nLock.release()

if isinstance(value, self.IterationStopper):

# We are done iterating. The blocking caller will

# immediately exit its loop.

raise StopIteration

# This is a legit iteration value, return it. Since this

# method runs in the blocking-iterator thread, it won't

# get called again until the caller is ready for another

# iteration.

"""

I act like an C{IConsumer} for your Twisted code and an iterator

for your blocking code running via a L{ThreadWorker}.

This is handy when you are using a conventional library that

relies on an iterator as its input::

def render(request):

w = png.Writer()

c = asynqueue.Consumerator()

c.deferUntilDone().addCallback(lambda _: request.finish())

p = self.producePixelRows(c)

w.write(request, c)

return server.NOT_DONE_YET

I work with either an I{IPushProducer} or an I{IPullProducer}. You

can construct me with an instance of the former and I'll get

started right away. Otherwise, call my L{registerProducer} method

with the producer and whether it is streaming (push) or not.

@ivar runState: 'init', 'started', 'running', 'stopping', 'stopped'

@ivar d: A C{Deferred} that fires when iterations are done.

@keyword producer: The producer for my instance to register, if

you want to supply an C{IPushProducer} one on

instantiation. Otherwise, use L{registerProducer}.

"""

def __init__(self, producer=None, maxThreads=None):

"""C{Consumerator(producer=None, maxThreads=None)}"""

super(Consumerator, self).__init__(maxThreads)

self.dLock = util.DeferredLock()

if producer:

self.registerProducer(producer, True)

def loop(self):

"""

Runs a loop in a dedicated thread that waits for new iterations to

be produced.

When I get an instance of L{IterationGetter.IterationStopper},

the loop exits. I then call my "all done" C{Deferred} and

delete my reference to the producer.

"""

self.runState = 'running'

while True:

# Wait for an iteration

self.cLock.acquire()

# Get a copy of the value

value = self.cIterationValue

# Release the consumer interface to write another

# iteration

reactor.callFromThread(self.dLock.release)

# Wait until it's safe to overwrite the blocking-iterator

# loop's copy

self.nLock.acquire()

# Now do so and release it to work on the new copy

self.bIterationValue = value

self.bLock.release()

if isinstance(value, self.IterationStopper):

# This was the post-iteration signal; this loop is now

# done.

break

# Wait until we know the iteration stopper was noticed and the

# blocking iterations stopped.

self.runState = 'stopping'

self.nLock.acquire()

self.runState = 'stopped'

reactor.callFromThread(delattr, self, 'producer')

def stop(self):

"""

Good manners urge you to call this to cleanly break out of a loop

of my iterations so that my producer doesn't keep working for

nothing.

Calling this method at the Twisted main-loop level is also a

fine way to quit producing and iterating when you know you're

done.

Not part of the official iterator implementation, but

useful for a Twisted way of iterating. You need a way of

letting whatever is producing the iterations know that there

won't be any more of them.

"""

if hasattr(self, 'producer'):

self.producer.stopProducing()

return self.unregisterProducer()

# --- IConsumer implementation --------------------------------------------

def write(self, data):

"""

The producer calls this with a chunk of I{data}. It goes through

two stages to emerge from my blocking end as an iteration, via

L{next}.

"""

def handleData(null, x):

self.cIterationValue = x

if self.runState in ('started', 'running'):

# Release my iteration-consuming loop to work on the next

# iteration value

self.cLock.release()

# The producer can and should write another iteration now

self.producer.resumeProducing()

if self.streaming and self.runState in ('started', 'running'):

# The producer is a IPushProducer, so tell it to hold off

# on any more iteration values for the moment while

# everything it's sent (and may yet send) gets processed

self.producer.pauseProducing()

# Handle the data in the order received

return self.dLock.acquire().addCallback(handleData, data)

def registerProducer(self, producer, streaming):

"""

C{IConsumer} implementation

"""

if hasattr(self, 'producer'):

raise RuntimeError()

self.producer = producer

self.streaming = streaming

if not streaming:

producer.resumeProducing()

self.start()

def unregisterProducer(self):

"""

C{IConsumer} implementation

"""

if not hasattr(self, 'producer'):

return defer.succeed(None)

"""

Stream data to me in one end and I will iterate it out the other.

Acts like a file handle for writing in one thread (even the main

one under the Twisted event loop) and an iterator in another

thread. Hook me up to an L{iteration.Deferator} to stream data

over a worker interface.

You must call my L{close} method to stop me from iterating.

"""

def __init__(self, maxThreads=None):

"""C{Filerator}(maxThreads=None)"""

super(Filerator, self).__init__(maxThreads)

self.itemBuffer = []

self.start()

@property

def closed(self):

return self.runState == 'stopped'

def loop(self):

"""

Runs a loop in a dedicated thread that waits for new iterations to

be written. When I get an instance of

L{IterationGetter.IterationStopper}, the loop exits.

"""

self.runState = 'running'

while True:

# Wait for an iteration

self.cLock.acquire()

# Get the oldest value in the FIFO buffer

value = self.itemBuffer.pop(0)

# Wait until it's safe to overwrite the blocking-iterator

# loop's copy

self.nLock.acquire()

# Now do so and release it to work on the new copy

self.bIterationValue = value

self.bLock.release()

if isinstance(value, self.IterationStopper):

# This was the post-iteration signal; this loop is now

# done.

break

# Wait until we know the iteration stopper was noticed and the

# blocking iterations stopped.

self.runState = 'stopping'

self.nLock.acquire()

self.runState = 'stopped'

def write(self, data):

"""

This is called with a chunk of I{data}. It goes through two stages

to emerge from my blocking end as an iteration, via L{next}.

"""

if self.closed:

raise ValueError("Closed, not accepting writes")

self.itemBuffer.append(data)

if self.runState == 'running':

# Release my iteration-consuming loop to work on the next

# iteration value. The cLock object is actually a

# semaphore, so it's OK if this gets called multiple times

# before the other loop can acquire it again.

self.cLock.release()

def writelines(self, lines):

"""

Adds a list full of data chunks to my buffer.

"""

for line in lines:

self.write(line)

def flush(self):

"""

Doesn't do anything, because I am always trying to flush my buffer

by iterating its contents.

"""

def close(self):

"""

Closing me as a "file" tells me that I can stop iterating once the

buffer is flushed.

"""

if not self.closed:

"""

Produces blocking iterations in the order they are requested via

an asynchronous function call.

I am an implementor of Twisted's C{IPushProducer} interface that

produces an iteration to a blocking call I{fb} for every time you

call a non-blocking item-generating function I{fb} via my

L{produceItem} method. Significantly, the items are buffered as

needed so that the iterations appear in the order of the calls to

L{produceItem} that generated them, B{not} necessarily in the

order in which they are actually generated.

Start things off by constructing an instance of me, with an

existing task queue if you have one you want me to use, and then

running L{start} with your blocking f-args-kw combination. Then

call L{produceItem} repeatedly with whatever f-args-kw combination

results (eventually) in new items to iterate. These calls may

return deferred results and should not block.

When you are done having me produce iterations, call

L{stop}. Whatever loop the blocking-iterator call is in will then

terminate and function I{fb} should end.

@ivar i: My L{Consumerator} instance, which acts like an iterator

for whatever function you supply to L{start}.

@ivar q: The L{TaskQueue} instance I use, either supplied by you

during construction or instantiated by me. Either way, you will

have to call L{TaskQueue.shutdown} on this eventually when

you're done with the queue.

"""

def __init__(self, maxThreads=None):

self.setup(maxThreads)

self.itemBuffer = {}

self.k1, self.k2 = 0, 0

self.seriesID = hash(self)

self.i = Consumerator(self)

self.dLock = defer.DeferredLock()

self.dt = util.DeferredTracker()

self.dLock.acquire()

self.produce = True

def start(self, fb, *args, **kw):

"""

Starts the blocking function call C{fb(i, *args, **kw)} that

relies on my L{Consumerator} instance I{i} for iterations, in

traditional blocking fashion. The function must accept C{i} as

its first argument, and can also accept further arguments

C{*args} and keywords C{**kw}, which you can specify in your

call to L{start}.

@return: A C{Deferred} that fires when the blocking call has

started in its own thread. Shouldn't take long at all,

unless the pool is fully occupied and we need to wait for a

thread to get freed up.

"""

def started():

self.dLock.release()

dStarted.callback(None)

def runner():

reactor.callFromThread(started)

# The actual blocking call

return fb(self.i, *args, **kw)

def finished(success, result):

if not self.dFinished.called:

reactor.callFromThread(

self.dFinished.callback, (success, result))

self.stopProducing()

dStarted = defer.Deferred()

self.dFinished = defer.Deferred()

self.dtp.put(self.dFinished)

self.pool.callInThreadWithCallback(finished, runner)

return dStarted

def produceItem(self, fp, *args, **kw):

"""

Runs C{fp(*args, **kw)} to generate an item that I produce as an

iteration to whatever blocking call was (or will be) set

running via L{start}.

While I am running, the returned C{Deferred} fires after the

call to I{fp} with the item value produced by the call to

I{f}. You don't need to do anything with these deferreds if

you don't want to use them for concurrency limiting; they are

accounted for in L{stop}.

If an exception is raised during the call, the I{dFinished}

callback is called with a corresponding C{Failure} object and

iterations will be stopped. This makes more sense than firing

an errback of the C{Deferred} returning from this

C{produceItem} method, because it's the end result of calling

L{stop} to indicate that iterations are done. *That* is when

the user should expect a status of the overall item-producing

operation.

If my L{stopProducing} method has been called, I no longer

produce iterations and calls to this method do not run

I{fp}. The returned C{Deferred} fires immediately with C{None}.

"""

def gotItem(item):

# We have a result, but we need to wait our turn to

# actually produce it

return self.dLock.acquire().addCallback(gotLock, item)

def oops(failureObj):

self.stop(failureObj)

def gotLock(lock, item):

if self.k2 == k1 and self.produce:

self._writeItem(item)

elif self.produce is not None:

self.itemBuffer[k1] = item

self._flushBuffer()

self.dLock.release()

return item

if self.produce is None:

return defer.succeed(None)

k1 = self.k1

self.k1 += 1

d = defer.maybeDeferred(fp, *args, **kw)

d.addCallback(gotItem)

d.addErrback(oops)

self.dt.put(d)

return d

@defer.inlineCallbacks

def stop(self, failureObj=None):

"""

Call this to indicate that iterations are done. After any pending

calls from L{produceItem} are done, my L{Consumerator} will

raise L{StopIteration} for the blocking iteration-caller in

I{fb} and that function should exit. Whatever value it returns

will fire the C{Deferred} that is returned here.

This method's C{Deferred} may have fired already, if I{fb}

exited early for some reason, or with a C{Failure} object that

may have been generated either by the iteration caller or by a

call to the I{fp} function of L{produceItem}.

Repeated calls to this method make no sense and will be

rewarded with deferreds immediately firing with C{None}.

"""

yield self.dt.deferToAll()

yield self.dLock.acquire()

yield self.i.stop()

if hasattr(self, 'dFinished'):

success, result = yield self.dFinished

del self.dFinished

else:

result = None

if failureObj:

result = failureObj

self.dLock.release()

defer.returnValue(result)

def _writeItem(self, item):

self.i.write(item)

self.k2 += 1

self._flushBuffer()

def _flushBuffer(self):

if self.k2 in self.itemBuffer:

item = self.itemBuffer.pop(self.k2)

# This will result in another call to resumeProducing

self._writeItem(item)

def stopProducing(self):

self.produce = None

def pauseProducing(self):

self.produce = False

def resumeProducing(self):