Package rosmaster :: Module threadpool

Source Code for Module rosmaster.threadpool

  1  # Software License Agreement (BSD License) 
  2  # 
  3  # Copyright (c) 2008, Willow Garage, Inc. 
  4  # All rights reserved. 
  5  # 
  6  # Redistribution and use in source and binary forms, with or without 
  7  # modification, are permitted provided that the following conditions 
  8  # are met: 
  9  # 
 10  #  * Redistributions of source code must retain the above copyright 
 11  #    notice, this list of conditions and the following disclaimer. 
 12  #  * Redistributions in binary form must reproduce the above 
 13  #    copyright notice, this list of conditions and the following 
 14  #    disclaimer in the documentation and/or other materials provided 
 15  #    with the distribution. 
 16  #  * Neither the name of Willow Garage, Inc. nor the names of its 
 17  #    contributors may be used to endorse or promote products derived 
 18  #    from this software without specific prior written permission. 
 19  # 
 20  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
 21  # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
 22  # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS 
 23  # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE 
 24  # COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, 
 25  # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, 
 26  # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; 
 27  # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 
 28  # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 
 29  # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN 
 30  # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 31  # POSSIBILITY OF SUCH DAMAGE. 
 32  # 
 33  # Revision $Id$ 
 34   
 35  """ 
 36  Internal threadpool library for zenmaster. 
 37   
 38  Adapted from U{http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/203871} 
 39   
 40  Added a 'marker' to tasks so that multiple tasks with the same 
 41  marker are not executed. As we are using the thread pool for i/o 
 42  tasks, the marker is set to the i/o name. This prevents a slow i/o 
 43  for gobbling up all of our threads 
 44  """ 
 45   
 46  import threading, logging, traceback 
 47  from time import sleep 
 48   
49 -class MarkedThreadPool(object):
50 51 """Flexible thread pool class. Creates a pool of threads, then 52 accepts tasks that will be dispatched to the next available 53 thread.""" 54
55 - def __init__(self, numThreads):
56 57 """Initialize the thread pool with numThreads workers.""" 58 59 self.__threads = [] 60 self.__resizeLock = threading.Condition(threading.Lock()) 61 self.__taskLock = threading.Condition(threading.Lock()) 62 self.__tasks = [] 63 self.__markers = set() 64 self.__isJoining = False 65 self.set_thread_count(numThreads)
66
67 - def set_thread_count(self, newNumThreads):
68 69 """ External method to set the current pool size. Acquires 70 the resizing lock, then calls the internal version to do real 71 work.""" 72 73 # Can't change the thread count if we're shutting down the pool! 74 if self.__isJoining: 75 return False 76 77 self.__resizeLock.acquire() 78 try: 79 self.__set_thread_count_nolock(newNumThreads) 80 finally: 81 self.__resizeLock.release() 82 return True
83
84 - def __set_thread_count_nolock(self, newNumThreads):
85 86 """Set the current pool size, spawning or terminating threads 87 if necessary. Internal use only; assumes the resizing lock is 88 held.""" 89 90 # If we need to grow the pool, do so 91 while newNumThreads > len(self.__threads): 92 newThread = ThreadPoolThread(self) 93 self.__threads.append(newThread) 94 newThread.start() 95 # If we need to shrink the pool, do so 96 while newNumThreads < len(self.__threads): 97 self.__threads[0].go_away() 98 del self.__threads[0]
99
100 - def get_thread_count(self):
101 """@return: number of threads in the pool.""" 102 self.__resizeLock.acquire() 103 try: 104 return len(self.__threads) 105 finally: 106 self.__resizeLock.release()
107
108 - def queue_task(self, marker, task, args=None, taskCallback=None):
109 110 """Insert a task into the queue. task must be callable; 111 args and taskCallback can be None.""" 112 113 if self.__isJoining == True: 114 return False 115 if not callable(task): 116 return False 117 118 self.__taskLock.acquire() 119 try: 120 self.__tasks.append((marker, task, args, taskCallback)) 121 return True 122 finally: 123 self.__taskLock.release()
124
125 - def remove_marker(self, marker):
126 """Remove the marker from the currently executing tasks. Only one 127 task with the given marker can be executed at a given time""" 128 if marker is None: 129 return 130 self.__taskLock.acquire() 131 try: 132 self.__markers.remove(marker) 133 finally: 134 self.__taskLock.release()
135
136 - def get_next_task(self):
137 138 """ Retrieve the next task from the task queue. For use 139 only by ThreadPoolThread objects contained in the pool.""" 140 141 self.__taskLock.acquire() 142 try: 143 retval = None 144 for marker, task, args, callback in self.__tasks: 145 # unmarked or not currently executing 146 if marker is None or marker not in self.__markers: 147 retval = (marker, task, args, callback) 148 break 149 if retval: 150 # add the marker so we don't do any similar tasks 151 self.__tasks.remove(retval) 152 if marker is not None: 153 self.__markers.add(marker) 154 return retval 155 else: 156 return (None, None, None, None) 157 finally: 158 self.__taskLock.release()
159
160 - def join_all(self, wait_for_tasks = True, wait_for_threads = True):
161 """ Clear the task queue and terminate all pooled threads, 162 optionally allowing the tasks and threads to finish.""" 163 164 # Mark the pool as joining to prevent any more task queueing 165 self.__isJoining = True 166 167 # Wait for tasks to finish 168 if wait_for_tasks: 169 while self.__tasks != []: 170 sleep(.1) 171 172 # Tell all the threads to quit 173 self.__resizeLock.acquire() 174 try: 175 self.__set_thread_count_nolock(0) 176 self.__isJoining = True 177 178 # Wait until all threads have exited 179 if wait_for_threads: 180 for t in self.__threads: 181 t.join() 182 del t 183 184 # Reset the pool for potential reuse 185 self.__isJoining = False 186 finally: 187 self.__resizeLock.release()
188 189 190
191 -class ThreadPoolThread(threading.Thread):
192 """ 193 Pooled thread class. 194 """ 195 196 threadSleepTime = 0.1 197
198 - def __init__(self, pool):
199 """Initialize the thread and remember the pool.""" 200 threading.Thread.__init__(self) 201 self.daemon = True #don't block program exit 202 self.__pool = pool 203 self.__isDying = False
204
205 - def run(self):
206 """ 207 Until told to quit, retrieve the next task and execute 208 it, calling the callback if any. 209 """ 210 while self.__isDying == False: 211 marker, cmd, args, callback = self.__pool.get_next_task() 212 # If there's nothing to do, just sleep a bit 213 if cmd is None: 214 sleep(ThreadPoolThread.threadSleepTime) 215 else: 216 try: 217 try: 218 result = cmd(*args) 219 finally: 220 self.__pool.remove_marker(marker) 221 if callback is not None: 222 callback(result) 223 except Exception as e: 224 logging.getLogger('rosmaster.threadpool').error(traceback.format_exc())
225
226 - def go_away(self):
227 """ Exit the run loop next time through.""" 228 self.__isDying = True
229