home *** CD-ROM | disk | FTP | other *** search
/ OS/2 Shareware BBS: 10 Tools / 10-Tools.zip / fnb101.zip / Lib / site-packages / Fnorb / orb / ThreadPoolQueue.py < prev    next >
Text File  |  1999-06-28  |  3KB  |  133 lines
  1. #!/usr/bin/env python
  2. #############################################################################
  3. # Copyright (C) DSTC Pty Ltd (ACN 052 372 577) 1997, 1998, 1999
  4. # All Rights Reserved.
  5. #
  6. # The software contained on this media is the property of the
  7. # DSTC Pty Ltd.  Use of this software is strictly in accordance
  8. # with the license agreement in the accompanying LICENSE.DOC file.
  9. # If your distribution of this software does not contain a
  10. # LICENSE.DOC file then you have no rights to use this software
  11. # in any manner and should contact DSTC at the address below
  12. # to determine an appropriate licensing arrangement.
  13. # 
  14. #      DSTC Pty Ltd
  15. #      Level 7, Gehrmann Labs
  16. #      University of Queensland
  17. #      St Lucia, 4072
  18. #      Australia
  19. #      Tel: +61 7 3365 4310
  20. #      Fax: +61 7 3365 4311
  21. #      Email: enquiries@dstc.edu.au
  22. # 
  23. # This software is being provided "AS IS" without warranty of
  24. # any kind.  In no event shall DSTC Pty Ltd be liable for
  25. # damage of any kind arising out of or in connection with
  26. # the use or performance of this software.
  27. #
  28. # Project:      Distributed Environment
  29. # File:         $Source: /units/arch/src/Fnorb/orb/RCS/ThreadPoolQueue.py,v $
  30. #
  31. #############################################################################
  32. """ A queue serviced by a thread pool. """
  33.  
  34.  
  35. # Standard/built-in modules.
  36. import thread
  37.  
  38. # Fnorb modules.
  39. import condvar
  40.  
  41.  
  42. class ThreadPoolQueue:
  43.     """ A queue serviced by a thread pool. """
  44.  
  45.     def __init__(self, size, function):
  46.     """ Constructor. """
  47.  
  48.     self.__size = size
  49.     self.__function = function
  50.     self.__stopped = 0
  51.     self.__data = []
  52.     self.__cv = condvar.condvar()
  53.  
  54.     return
  55.  
  56.     def start(self):
  57.     """ Start servicing the queue. """
  58.  
  59.     # Start the appropriate number of worker threads.
  60.     for i in range(self.__size):
  61.         thread.start_new_thread(self.worker_thread, (i,))
  62.  
  63.     return
  64.  
  65.     def stop(self):
  66.     """ Stop servicing the queue. """
  67.  
  68.     self.__cv.acquire()
  69.     self.__stopped = 1
  70.     self.__cv.broadcast()
  71.  
  72.     return
  73.  
  74.     def wait(self):
  75.     """ Wait until all of the worker threads have finished. """
  76.  
  77.     self.__cv.acquire()
  78.     while self.__size > 0:
  79.         self.__cv.wait()
  80.     self.__cv.release()
  81.  
  82.     return
  83.     
  84.     def add_item(self, item):
  85.     """ Add a single item to the queue. """
  86.  
  87.     self.__cv.acquire()
  88.     self.__data.append(item)
  89.     self.__cv.signal()
  90.  
  91.     return
  92.  
  93.     def add_items(self, items):
  94.     """ Add a list of items to the queue. """
  95.  
  96.     self.__cv.acquire()
  97.     self.__data[len(self.__data):] = items
  98.     self.__cv.broadcast()
  99.  
  100.     return
  101.         
  102.     def worker_thread(self, i):
  103.     """ The worker!"""
  104.  
  105.     self.__cv.acquire()
  106.     while not self.__stopped:
  107.         # Is there an item on the queue for me to deal with?
  108.         if len(self.__data) > 0:
  109.         item = self.__data[0]
  110.         del self.__data[0]
  111.         self.__cv.release()
  112.  
  113.         # Do the work!
  114.         apply(self.__function, item)
  115.  
  116.         # Acquire the lock so that I can check to see if I am stopped
  117.         # or if there is some more work for me to do.
  118.         self.__cv.acquire()
  119.  
  120.         # Otherwise, we are not stopped, and there is nothing for me to
  121.         # do, so I'll just wait around a while...
  122.         else:
  123.         self.__cv.wait()
  124.  
  125.     # The thread pool has been stopped so let's get outta here.
  126.     self.__size = self.__size - 1
  127.     self.__cv.signal()
  128.  
  129.     # Explicitly exit the thread!
  130.     thread.exit()
  131.  
  132. #############################################################################
  133.