import threading class WorkersLounge ( object ): def __init__ ( self , total_workers_number ): """ @param total_workers_number: the maximum number of worker threads """ self . total_workers_number = total_workers_number self . waiting_place = threading . Condition () self . work_done_event = threading . Event () def rest ( self ): """ When a thread calls this method there are two possible options: - either there are other active threads, in which case the current thread waits - all other threads are already waiting, in which case they all exit @return: True if the caller thread should go back to work False if the caller thread should exit """ with self . waiting_place : if ( len ( self . waiting_place . _Condition__waiters ) == self . total_workers_number - 1 ): # This is the last worker, and it has nothing to do self . work_done_event . set () self . waiting_place . notifyAll () # Notify the caller there is no more work to do return False else : # Wait for a signal self . waiting_place . wait () return not ( self . work_done_event . isSet ()) def back_to_work ( self ): """ Wake up all the waiting threads. Should be called whenever a thread puts new input into the common source. """ # Wake up everybody with self . waiting_place : self . waiting_place . notifyAll () if __name__ == "__main__" : # Run test code import Queue import time print_lock = threading . Lock () def sync_print ( text ): with print_lock : print text def _thread_proc ( input_queue , workers_lounge ): thread_name = threading . currentThread () . name while 1 : try : # Attempt to get input from the common queue input = input_queue . get ( timeout = 0.1 ) # Do something with the input, possibly inserting more jobs # back into the queue sync_print ( " %s got input %s " % ( thread_name , input )) time . sleep ( 1 ) if ( input < 5 ): input_queue . put ( input + 1 ) input_queue . put ( input + 1 ) # Wake up any waiting thread workers_lounge . back_to_work () except Queue . Empty : # The 'rest' method returns False if the thread should stop, # and blocks until someone wakes it up sync_print ( " %s is resting" % thread_name ) if ( workers_lounge . rest () == False ): sync_print ( " %s finished working" % thread_name ) break # Create an initial input source input_queue = Queue . Queue () input_queue . put ( 1 ) input_queue . put ( 1 ) # Run worker threads threads_number = 5 workers_lounge = WorkersLounge ( total_workers_number = threads_number ) for _i in range ( threads_number ): threads . append ( threading . Thread ( target = _thread_proc , args = ( input_queue , workers_lounge ))) for thread in threads : thread . start () for thread in threads : thread . join ()