Condition Object - Thread Synchronization in Python

In order to synchronize the access to any resources more efficiently, we can associate a condition with tasks, for any thread to wait until a certain condition is met or notify other threads about the condition being fulfilled so that they may unblock themselves.

Let's take a simple example to understand this. In the Producer Consumer problem, if there is one Produces producing some item and one Consumer consuming it, then until the Producer has produced the item the Consumer cannot consume it. Hence the Consumer waits until the Produced produces an item. And it's the duty of the Producer to inform the Consumer that an item is available for consumption once it is successfully produced.

And if there are multiple Consumers consuming the item produced by the Producer then the Producer must inform all the Consumers about the new item produced.

This is a perfect usecase for the condition object in multithreading in python.

Now that we know what the condition object is used for in python multithreading, let's see the syntax for it:

condition = threading.Condition([lock])

The condition object takes in an optional lock object as argument. If we do not provide anything then it creates a default lock.

A condition object has acquire() and release() methods that call the corresponding methods of the associated lock. It also has a wait() method, and notify() and notifyAll() methods. These three must only be called after the calling thread has acquired the lock.

Following are condition class methods:

acquire(*args) method

This method is used to acquire the lock. This method calls the corresponding acquire() method on the underlying lock present in the condition object; The return value is whatever that method returns.

release() method

this method is used to release the lock. This method calls the corresponding release() method on the underlying lock present in the condition object.

wait([timeout]) method

This method is used to block the thread and make it wait until some other thread notifies it by calling the notify() or notifyAll() method on the same condition object or until the timeout occurs.

This must only be called when the calling thread has acquired the lock.

When called, this method releases the lock and then blocks the thread until it is awakened by a notify() or notifyAll() call for the same condition variable from some other thread, or until the timeout occurs.

This method returns True if it is released because of notify() or notifyAll() method else if timeout occurs this method will return False boolean value.

notify() method

It wakes up any thread waiting on the corresponding condition. This must only be called when the calling thread has acquired the lock. Also, calling this method will wake only one waiting thread.

notifyAll() method

It wakes up all the threads waiting on this condition. This method acts like notify() method, but wakes up all the waiting threads instead of one.

In the code example below we have implemented a simple producer-consumer solution, where the producer produces an item and adds it to a list from which the consumer is consuming the items.

import threading
import time
from random import randint

class SomeItem:
  # init method
  def __init__(self):
    # initialize empty list
    self.list = []
  
  # add to list method for producer
  def produce(self, item):
    print("Adding item to list...")
    self.list.append(item)
    
  # remove item from list method for consumer
  def consume(self):
    print("consuming item from list...")
    item = self.list[0]
    print("Item consumed: ", item)
    self.list.remove(item)

def producer(si, cond):
    r = randint(1,5)
    # creating random number of items
    for i in range(1, r):
      print("working on item creation, it will take: " + str(i) + " seconds")
      time.sleep(i)
      print("acquiring lock...")
      cond.acquire()
      try:
        si.produce(i)
        cond.notify()
      finally:
        cond.release()
      
def consumer(si, cond):
    cond.acquire()
    while True:
      try:
        si.consume()
      except:
        print("No item to consume...")
        # wait with a maximum timeout of 10 sec
        val = cond.wait(10)
        if val:
          print("notification received about item production...")
          continue
        else:
          print("waiting timeout...")
          break
        
    cond.release()
    
if __name__=='__main__':
  # condition object
  cond = threading.Condition()
  # SomeItem object
  si = SomeItem()
  # producer thread
  p = threading.Thread(target=producer, args=(si,cond,))
  p.start()
  # consumer thread
  c = threading.Thread(target=consumer, args=(si,cond,))
  c.start()

  #print('Waiting for producer and consumer threads...')
  p.join()
  c.join()
  print("Done")
  

Few important takeaways from the code example above:

1. We have created a class SomeItem which has a list which acts as the shared resource between the producer and consumer thread.
2. The producer thread is randomly generating some list items and adding it to the list.
3. The consumer thread tries to consume the item, if item is not found, it starts to wait. If the producer sends a notification to the consumer about the item creation before its timeout, then the consumer consumes the item else it exits due to timeout.

This is a very simple example covering all the usecases of condition object. Try running the above program with 2 consumer threads and single producer thread.