1  
   2  #!/usr/bin/env python
   3  # -*- coding: utf-8 -*-
   4  # vim:ts=4:sw=4:et
   5  # (c) 2007 under terms of LGPL v 2.1
   6  # by Vsevolod S. Balashov <vsevolod@balashov.name> 
   7  
   8  from threading import Thread
   9  
  10  def threaded(func):
  11      def proxy(*args, **kwargs):
  12          thread = Thread(target=func, args=args, kwargs=kwargs)
  13          thread.start()
  14          return thread
  15      return proxy
  16  
  17  from Queue import Queue
  18  
  19  class Pool(Queue):
  20      def __init__(self, maxsize):
  21          assert maxsize > 0, 'maxsize > 0 required for Pool class'
  22          Queue.__init__(self, maxsize)
  23          for i in range(maxsize):
  24              thread = Thread(target = self._worker)
  25              thread.setDaemon(True)
  26              thread.start()
  27  
  28      def _worker(self):
  29          while True:
  30              try:
  31                  func, args, kwargs = self.get()
  32                  func(*args, **kwargs)
  33              except:
  34                  self.task_done()
  35                  self.join()
  36                  raise
  37              self.task_done()
  38  
  39      def addJob(self, func, *args, **kwargs):
  40          self.put((func, args, kwargs))
  41  
  42      def __enter__(self):
  43          pass
  44  
  45      def __exit__(self, exc_type, exc_value, traceback):
  46          self.join()
  47  
  48  def threadpool(pool):
  49      assert pool.__class__ == Pool, 'threadpool decorator require a Pool object'
  50      def decorator(func):
  51          def proxy(*args, **kwargs):
  52              pool.put((func, args, kwargs))
  53              return pool
  54          return proxy
  55      return decorator

example usage

   1  
   2  from threadpool import *
   3  
   4  from time import sleep
   5  from random import random
   6  
   7  pool = Pool(3)
   8      
   9  @threadpool(pool)
  10  def test_threadpool(i):
  11      print 'threadpool %i enter' % i
  12      sleep(random())
  13      print 'threadpool %i exit' % i
  14  
  15  print 'threadpool example'
  16  for i in range(6):
  17      test_threadpool(i)
  18  pool.join()
  19  print 'done'
  20  print ''
  21  
  22  @threaded
  23  def test_threaded(i):
  24      print 'threaded %i enter' % i
  25      sleep(random())
  26      print 'threaded %i exit' % i
  27  
  28  print 'threaded example'
  29  for i in range(5):
  30      test_threaded(i)
  31  print 'done'

result output

threadpool example
threadpool 0 enter
threadpool 1 enter
threadpool 2 enter
threadpool 0 exit
threadpool 3 enter
threadpool 1 exit
threadpool 4 enter
threadpool 2 exit
threadpool 5 enter
threadpool 3 exit
threadpool 5 exit
threadpool 4 exit
done

threaded example
done
threaded 0 enter
threaded 1 enter
threaded 2 enter
threaded 3 enter
threaded 4 enter
threaded 1 exit
threaded 2 exit
threaded 3 exit
threaded 0 exit
threaded 4 exit