twisted/internet/threads.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Extended thread dispatching support.
   6
   7 For basic support see reactor threading API docs.
   8
   9 Maintainer: Itamar Shtull-Trauring
  10 """
  11
  12 import Queue
  13
  14 from twisted.python import failure
  15 from twisted.internet import defer
  16
  17
  18 def deferToThreadPool(reactor, threadpool, f, *args, **kwargs):
  19     """
  20     Call the function C{f} using a thread from the given threadpool and return
  21     the result as a Deferred.
  22
  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
  25     C{deferToThread}.
  26
  27     @param reactor: The reactor in whose main thread the Deferred will be
  28         invoked.
  29
  30     @param threadpool: An object which supports the C{callInThreadWithCallback}
  31         method of C{twisted.python.threadpool.ThreadPool}.
  32
  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.
  36
  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
  39         exception.
  40     """
  41     d = defer.Deferred()
  42
  43     def onResult(success, result):
  44         if success:
  45             reactor.callFromThread(d.callback, result)
  46         else:
  47             reactor.callFromThread(d.errback, result)
  48
  49     threadpool.callInThreadWithCallback(onResult, f, *args, **kwargs)
  50
  51     return d
  52
  53
  54 def deferToThread(f, *args, **kwargs):
  55     """
  56     Run a function in a thread and return the result as a Deferred.
  57
  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.
  61
  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
  64     an exception.
  65     """
  66     from twisted.internet import reactor
  67     return deferToThreadPool(reactor, reactor.getThreadPool(),
  68                              f, *args, **kwargs)
  69
  70
  71 def _runMultiple(tupleList):
  72     """
  73     Run a list of functions.
  74     """
  75     for f, args, kwargs in tupleList:
  76         f(*args, **kwargs)
  77
  78
  79 def callMultipleInThread(tupleList):
  80     """
  81     Run a list of functions in the same thread.
  82
  83     tupleList should be a list of (function, argsList, kwargsDict) tuples.
  84     """
  85     from twisted.internet import reactor
  86     reactor.callInThread(_runMultiple, tupleList)
  87
  88
  89 def blockingCallFromThread(reactor, f, *a, **kw):
  90     """
  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.
  94
  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}.
 101
 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}.
 104
 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}).
 110     """
 111     queue = Queue.Queue()
 112     def _callFromThread():
 113         result = defer.maybeDeferred(f, *a, **kw)
 114         result.addBoth(queue.put)
 115     reactor.callFromThread(_callFromThread)
 116     result = queue.get()
 117     if isinstance(result, failure.Failure):
 118         result.raiseException()
 119     return result
 120
 121
 122 __all__ = ["deferToThread", "deferToThreadPool", "callMultipleInThread",
 123            "blockingCallFromThread"]