1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Extended thread dispatching support.
7 For basic support see reactor threading API docs.
9 Maintainer: Itamar Shtull-Trauring
14 from twisted.python import failure
15 from twisted.internet import defer
18 def deferToThreadPool(reactor, threadpool, f, *args, **kwargs):
20 Call the function C{f} using a thread from the given threadpool and return
21 the result as a Deferred.
23 This function is only used by client code which is maintaining its own
24 threadpool. To run a function in the reactor's threadpool, use
27 @param reactor: The reactor in whose main thread the Deferred will be
30 @param threadpool: An object which supports the C{callInThreadWithCallback}
31 method of C{twisted.python.threadpool.ThreadPool}.
33 @param f: The function to call.
34 @param *args: positional arguments to pass to f.
35 @param **kwargs: keyword arguments to pass to f.
37 @return: A Deferred which fires a callback with the result of f, or an
38 errback with a L{twisted.python.failure.Failure} if f throws an
43 def onResult(success, result):
45 reactor.callFromThread(d.callback, result)
47 reactor.callFromThread(d.errback, result)
49 threadpool.callInThreadWithCallback(onResult, f, *args, **kwargs)
54 def deferToThread(f, *args, **kwargs):
56 Run a function in a thread and return the result as a Deferred.
58 @param f: The function to call.
59 @param *args: positional arguments to pass to f.
60 @param **kwargs: keyword arguments to pass to f.
62 @return: A Deferred which fires a callback with the result of f,
63 or an errback with a L{twisted.python.failure.Failure} if f throws
66 from twisted.internet import reactor
67 return deferToThreadPool(reactor, reactor.getThreadPool(),
71 def _runMultiple(tupleList):
73 Run a list of functions.
75 for f, args, kwargs in tupleList:
79 def callMultipleInThread(tupleList):
81 Run a list of functions in the same thread.
83 tupleList should be a list of (function, argsList, kwargsDict) tuples.
85 from twisted.internet import reactor
86 reactor.callInThread(_runMultiple, tupleList)
89 def blockingCallFromThread(reactor, f, *a, **kw):
91 Run a function in the reactor from a thread, and wait for the result
92 synchronously. If the function returns a L{Deferred}, wait for its
93 result and return that.
95 @param reactor: The L{IReactorThreads} provider which will be used to
96 schedule the function call.
97 @param f: the callable to run in the reactor thread
98 @type f: any callable.
99 @param a: the arguments to pass to C{f}.
100 @param kw: the keyword arguments to pass to C{f}.
102 @return: the result of the L{Deferred} returned by C{f}, or the result
103 of C{f} if it returns anything other than a L{Deferred}.
105 @raise: If C{f} raises a synchronous exception,
106 C{blockingCallFromThread} will raise that exception. If C{f}
107 returns a L{Deferred} which fires with a L{Failure},
108 C{blockingCallFromThread} will raise that failure's exception (see
109 L{Failure.raiseException}).
111 queue = Queue.Queue()
112 def _callFromThread():
113 result = defer.maybeDeferred(f, *a, **kw)
114 result.addBoth(queue.put)
115 reactor.callFromThread(_callFromThread)
117 if isinstance(result, failure.Failure):
118 result.raiseException()
122 __all__ = ["deferToThread", "deferToThreadPool", "callMultipleInThread",
123 "blockingCallFromThread"]