;ò íN¡Jc@sýdZdZddddddgZdZd Zd Zd Zd Zd kZd k Z d k Z d k Z de fd„ƒYZ de fd„ƒYZd„Zeed„Zde ifd„ƒYZdfd„ƒYZdfd„ƒYZedjod kZd kZd„Zd„Zd„ZgiZedƒD]Zeei ddƒƒq7[Z!eee!eeƒZ"giZedƒD]%Zeei ddƒfhfƒq…[Z!e"i#eee!eeƒƒdGHedƒZ$x'e"D]Z%e$i&e%ƒde%i'GHqçWdZxÅe(o½y…ei)d ƒe$i*ƒd!Gd"e i+ƒdfGHedjod#GHe$i,dƒnedjod$GHe$i-d%ƒned7ZWqe.j o d&GHPqe j o d'GHPqXqWe$i/od(GHe$i0ƒqùnd S()sWEasy to use object-oriented thread pool framework. A thread pool is an object that maintains a pool of worker threads to perform time consuming operations in parallel. It assigns jobs to the threads by putting them in a work request queue, where they are picked up by the next available thread. This then performs the requested operation in the background and puts the results in another queue. The thread pool object can then collect the results from all threads from this queue as soon as they become available or after all threads have finished their work. It's also possible, to define callbacks to handle each result as it comes in. The basic concept and some code was taken from the book "Python in a Nutshell, 2nd edition" by Alex Martelli, O'Reilly 2006, ISBN 0-596-10046-9, from section 14.5 "Threaded Program Architecture". I wrapped the main program logic in the ThreadPool class, added the WorkRequest class and the callback system and tweaked the code here and there. Kudos also to Florent Aide for the exception handling mechanism. Basic usage:: >>> pool = ThreadPool(poolsize) >>> requests = makeRequests(some_callable, list_of_args, callback) >>> [pool.putRequest(req) for req in requests] >>> pool.wait() See the end of the module code for a brief, annotated usage example. Website : http://chrisarndt.de/projects/threadpool/ srestructuredtext ens makeRequestssNoResultsPendingsNoWorkersAvailables ThreadPools WorkRequests WorkerThreadsChristopher Arndts1.2.5s$Revision: 354 $s5$Date: 2008-11-19 18:34:46 +0100 (Wed, 19 Nov 2008) $s MIT licenseNcBstZdZRS(s&All work requests have been processed.(s__name__s __module__s__doc__(((s(/home/summariz/public_html/threadpool.pysNoResultsPending<s cBstZdZRS(s:No worker threads available to process remaining requests.(s__name__s __module__s__doc__(((s(/home/summariz/public_html/threadpool.pysNoWorkersAvailable@s cCsti|ŒdS(s~Default exception handler callback function. This just prints the exception info via ``traceback.print_exception``. N(s tracebacksprint_exceptionsexc_info(srequestsexc_info((s(/home/summariz/public_html/threadpool.pys_handle_thread_exceptionFsc Csˆg}xw|D]o}t|tƒo1|it||d|dd|d|ƒƒq |it||gt d|d|ƒƒq W|SdS(sTCreate several work requests for same callable with different arguments. Convenience function for creating several work requests for the same callable where each invocation of the callable receives different values for its arguments. ``args_list`` contains the parameters for each invocation of callable. Each item in ``args_list`` should be either a 2-item tuple of the list of positional arguments and a dictionary of keyword arguments or a single, non-tuple argument. See docstring for ``WorkRequest`` for info on ``callback`` and ``exc_callback``. iiscallbacks exc_callbackN( srequestss args_listsitems isinstancestuplesappends WorkRequests callable_scallbacks exc_callbacksNone(s callable_s args_listscallbacks exc_callbacksitemsrequests((s(/home/summariz/public_html/threadpool.pys makeRequestsPs cBs,tZdZdd„Zd„Zd„ZRS(sÒBackground thread connected to the requests/results queues. A worker thread sits in the background and picks up work requests from one queue and puts the results in another until it is dismissed. icKsXtii|||idƒ||_||_ ||_ ti ƒ|_ |iƒdS(sîSet up thread in daemonic mode and start it immediatedly. ``requests_queue`` and ``results_queue`` are instances of ``Queue.Queue`` passed by the ``ThreadPool`` class when it creates a new worker thread. iN(s threadingsThreads__init__sselfskwdss setDaemonsrequests_queues_requests_queues results_queues_results_queues poll_timeouts _poll_timeoutsEvents _dismissedsstart(sselfsrequests_queues results_queues poll_timeoutskwds((s(/home/summariz/public_html/threadpool.pys__init__ys    cCsæxßto×|iiƒoPny|iit|iƒ}Wnti j o qqX|iiƒo|ii |ƒPny2|i |i |i Ž}|ii ||fƒWqt|_|ii |tiƒfƒqXqWdS(s4Repeatedly process the job queue until told to exit.N(sTruesselfs _dismissedsisSets_requests_queuesgets _poll_timeoutsrequestsQueuesEmptysputscallablesargsskwdssresults_results_queues exceptionssyssexc_info(sselfsrequestsresult((s(/home/summariz/public_html/threadpool.pysrun‰s$ cCs|iiƒdS(sBSets a flag to tell the thread to exit when done with current job.N(sselfs _dismissedsset(sself((s(/home/summariz/public_html/threadpool.pysdismiss¢s(s__name__s __module__s__doc__s__init__srunsdismiss(((s(/home/summariz/public_html/threadpool.pys WorkerThreadqs   cBs/tZdZeeeeed„Zd„ZRS(sA request to execute a callable for putting in the request queue later. See the module function ``makeRequests`` for the common case where you want to build several ``WorkRequest`` objects for the same callable but with different arguments for each call. cCs|tjot|ƒ|_n6yt|ƒ|_Wntj otdƒ‚nXt|_||_||_ ||_ |pg|_ |ph|_ dS(s#Create a work request for a callable and attach callbacks. A work request consists of the a callable to be executed by a worker thread, a list of positional arguments, a dictionary of keyword arguments. A ``callback`` function can be specified, that is called when the results of the request are picked up from the result queue. It must accept two anonymous arguments, the ``WorkRequest`` object and the results of the callable, in that order. If you want to pass additional information to the callback, just stick it on the request object. You can also give custom callback for when an exception occurs with the ``exc_callback`` keyword parameter. It should also accept two anonymous arguments, the ``WorkRequest`` and a tuple with the exception details as returned by ``sys.exc_info()``. The default implementation of this callback just prints the exception info via ``traceback.print_exception``. If you want no exception handler callback, just pass in ``None``. ``requestID``, if given, must be hashable since it is used by ``ThreadPool`` object to store the results of that work request in a dictionary. It defaults to the return value of ``id(self)``. srequestID must be hashable.N(s requestIDsNonesidsselfshashs TypeErrorsFalses exceptionscallbacks exc_callbacks callable_scallablesargsskwds(sselfs callable_sargsskwdss requestIDscallbacks exc_callback((s(/home/summariz/public_html/threadpool.pys__init__°s     cCs$d|i|i|i|ifSdS(Ns2(sselfs requestIDsargsskwdss exception(sself((s(/home/summariz/public_html/threadpool.pys__str__Ùs(s__name__s __module__s__doc__sNones_handle_thread_exceptions__init__s__str__(((s(/home/summariz/public_html/threadpool.pys WorkRequest§s )cBsetZdZdddd„Zdd„Zed„Zd„Zedd„Z ed„Z d „Z RS( s{A thread pool, distributing work requests and collecting results. See the module docstring for more information. iicCsSti|ƒ|_ti|ƒ|_g|_g|_h|_|i ||ƒdS(sSet up the thread pool and start num_workers worker threads. ``num_workers`` is the number of worker threads to start initially. If ``q_size > 0`` the size of the work *request queue* is limited and the thread pool blocks when the queue is full and it tries to put more work requests in it (see ``putRequest`` method), unless you also use a positive ``timeout`` value for ``putRequest``. If ``resq_size > 0`` the size of the *results queue* is limited and the worker threads will block when the queue is full and they try to put new results in it. .. warning: If you set both ``q_size`` and ``resq_size`` to ``!= 0`` there is the possibilty of a deadlock, when the results queue is not pulled regularly and too many jobs are put in the work requests queue. To prevent this, always set ``timeout > 0`` when calling ``ThreadPool.putRequest()`` and catch ``Queue.Full`` exceptions. N( sQueuesq_sizesselfs_requests_queues resq_sizes_results_queuesworkerssdismissedWorkerss workRequestss createWorkerss num_workerss poll_timeout(sselfs num_workerssq_sizes resq_sizes poll_timeout((s(/home/summariz/public_html/threadpool.pys__init__äs   cCs@x9t|ƒD]+}|iit|i|id|ƒƒq WdS(säAdd num_workers worker threads to the pool. ``poll_timout`` sets the interval in seconds (int or float) for how ofte threads should check whether they are dismissed, while waiting for requests. s poll_timeoutN( sranges num_workerssisselfsworkerssappends WorkerThreads_requests_queues_results_queues poll_timeout(sselfs num_workerss poll_timeoutsi((s(/home/summariz/public_html/threadpool.pys createWorkerss  cCsg}xLtt|t|iƒƒƒD],}|iiƒ}|i ƒ|i |ƒq%W|ox,|D]}|i ƒqcWn|ii|ƒdS(sATell num_workers worker threads to quit after their current task.N(s dismiss_listsrangesmins num_workersslensselfsworkerssispopsworkersdismisssappendsdo_joinsjoinsdismissedWorkerssextend(sselfs num_workerssdo_joins dismiss_listsisworker((s(/home/summariz/public_html/threadpool.pysdismissWorkers s cCs+x|iD]}|iƒq Wg|_dS(sNPerform Thread.join() on all worker threads that have been dismissed. N(sselfsdismissedWorkerssworkersjoin(sselfsworker((s(/home/summariz/public_html/threadpool.pysjoinAllDismissedWorkerss  cCs\t|tƒpt‚t|dtƒ pt‚|ii|||ƒ||i |i