From Earlham Cluster Department
Back to Robotics Main Page
- Computing device - CPU, RAM, persistant store (two ARM CPUs, etc. in the NXT)
- Stored program; Fetch, decode, execute
- Execution context - instructions, data, program counter, stack
- Processes have all 4 (instructions, data, program counter, stack)
- Threads have PC and stack with shared instructions and data
- Workshop analogy
- Concurrency and locking
- Race conditions
- Locks, semaphores, barriers
Basics of threads in Python
- thread - low level thread library
- threading - higher level library built on thread (addressed on this page)
When creating a thread, you will must always define a class function called
run(). This is the code that will be executed by the thread. When the
run() function exits, the thread is no longer alive.
import threading class hello_world_thread( threading.Thread ): def run( self ): print "Hello World!"
Starting the above thread is done by calling the
start() function. This will execute the run() function you've defined.
>>> hw = hello_world_thread() # initialize >>> hw.start() # run Hello World!
If you want to pass arguments to the thread at initialization, define the constructor function:
import threading class hello_world_thread( threading.Thread ): def __init__( self, message = 'Hello World!' ): # default value for message threading.Thread.__init__( self ) # init the thread self.message = message def run( self ): print self.message
>>> hello_world_thread().start() # initialize and run Hello World! >>> hello_world_thread('Hola Mundo!").start() # init and run with argument Hola Mundo!
Lock Objects (Lock and RLock)
Locks should be used when more than one thread can read or modify a resource. They have two states: locked and unlocked; and they can be blocking or non-blocking. When trying to acquire a lock with blocking enabled (the default behavior), a thread will wait until the lock is available.
- acquire() - acquire a lock, block until available
- release() - release a lock, making it available for other threads to acquire
# defined globally hammer = threading.Lock() saw = threading.Lock() # within a worker thread's run() function hammer.acquire() saw.acquire() try: # do some work finally: # release locks, no matter what hammer.release() saw.release()
The try clause is not strictly necessary, but it is a recommended safety feature that prevents a thread that errors out from preventing another thread from acquiring the lock.
There is another type of lock called an RLock (Re-entrant Lock). See the section called Problems with Simple Locking for a good example of why this more complex lock is sometimes desirable.
Semaphore Objects (Semaphore and BoundedSemaphore)
Semaphores are a locking mechanism that can be used to keep a count of resources available based on the number of times it has been acquired and released
- acquire() - decrement internal counter, block until internal counter is > 0
- release() - increment internal counter
Imagine a queue of people standing outside a restaurant. There is a limited number of available tables in the restaurant (kept track of by a semaphore). When a group of diners are seated (table is acquired), the semaphore is decremented and when the diners leave(table released), it is incremented.
When the number of available tables reaches zero, the diners trying to acquire a table are blocked and must wait until a table becomes available.
import threading # defined globally available_table = threading.BoundedSemaphore( 10 ) # 10 tables in restaurant # in diner thread's run function available_table.acquire() try: # eat finally: available_table.release()
Using a bounded semaphore, as in the above example is usually the best practice since it will throw an error when releasing more than its initializing value (sometimes indicating a bug). In the above example, it would mean that an error would be generated if a table was released when none were acquired (i.e. making an 11th table available). The regular semaphore object will continue to increment the count no matter how many times it is called.
Event Objects (Event)
Event objects can be used for thread synchronization. When a thread calls
wait() for an event that is set, it will continue on immediately. When a thread calls
wait() for an event that is not set, it will wait until that event is set. More than one thread can be looking for the same event.
- set() - set flag to true
- clear() - set flag to false
- wait() - block until flag is true
- is_set() - return True if set; otherwise False
import threading # globally defined green_light = threading.Event() # in traffic controller thread's run function if driver_is_late(): green_light.clear() sleep( frustrating_time * 2 ) green_light.set() # in driver thread's run function green_light.wait() # block until green_light is set # drive on
Condition Objects (Condition)
Conditions are more advanced events that can be used to signify a state change. Threads can either wait for a condition to be true or can be notified of a change.
Condition objects are essentially a combination of locks and events. You can create a condition using an existing lock or a re-entrant lock will be created if one is not specified.
- acquire() - acquire lock
- release() - release lock
- wait() - release the underlying lock and wait for a notification
- notify() - wake up a thread waiting for the condition
- notifyAll() - wake up all threads waiting for the condition
More than one thread can wait for a condition to be true (indefinitely or with a timeout). If a queue of objects is being processed by multiple threads, a condition can be used by a consumer to acquire access after an element has been added by a producer.
# globally defined candy_bucket = threading.Lock() candy_available = threading.Conditional( candy_bucket ) # in homeowner's thread candy_bucket.acquire() # put one black or orange peanut butter candy in bucket candy_available.notify() candy_bucket.release() # in trick-or-treater's thread candy_bucket.acquire() candy_available.wait() # get candy candy_bucket.release()
Generalized kill-switch threading
class thread_wait( threading.Thread ): def __init__( self, condition, action ): threading.Thread.__init__( self ) self.condition = condition self.action = action def run( self ): while not self.condition(): sleep(0.1) self.action
kill_switch_thread = thread_wait( get_kill_switch_function, suicide_function ) kill_switch_thread.start()