Barriers and Events : Synchronization Primitives

Event and Barriers In Execution

While playing with threads and processes in a paradigm of concurrency and parallel processing its essential to make sure threads or the processes don't end up messing up each other state. Its impossible to predict the runtime of different processes and threads. So, every programming language provides access to lower level primitives to deal with the scenarios where we need certain execution order to be maintained. This article covers two of those primitives:

• Events: Communication primitives used to a notify threads / process of event that occurred in other in other threads / process. Example: Cars (Threads) Moving from a Zebra Crossing on turning Light Green (Event)
• Barriers: Barriers on the other hand allows threads / processes to wait on each other until a common point of time in execution. Example: Threads waiting for each other (Barrier) to complete execution of tasks to execute next task.

Events

Python provides events from multithreading.Eventor multprocessing.Event it uses internal variable to maintain weather event is set or not. Offers following methods:

• set(): Sets an event to true, all threads waiting for an event are waked. Event emitter process / thread uses this method to sets an event.
• wait(timeout=<timeout_in_sec>): Performs blocking call, waits until event occurs. Every thread wishing to get notified for an event uses this method. Note timeout is optional.
• clear(): Clears or sets event to false.

Lets see an example where we have multiple threads as a Car object which is waiting for an event to turn a traffic light green.

Note, that a with every custom thread class implementation, you must define run method for threads to pick up task from.

import threading
import time

class Car(threading.Thread):

    def __init__(self, name, event, *args, **kwargs):
        self.event = event
        threading.Thread.__init__(self, name=name, *args, **kwargs)
        print(f"Car-{self.name} waiting for light to turn blue...")
    
    def is_light_blue(self):
        """.wait is a blocking call"""
        return self.event.wait()
    
    def run(self):
        if self.is_light_blue(): # waits untill event
            print(f"Car-{self.name} moving..")


event = threading.Event()
cars = [ ]

def countdown(seconds):
    """Performs countdown to seconds and sets event"""
    print(f"Waiting {seconds} seconds to turn light blue.")
    for s in range(0, seconds):
        print(f"{seconds-s}...")
        time.sleep(1)
    event.set()

for i in range(0,5):
    car = Car(name=f"{i}", event=event)
    cars.append(car)

for car in cars:
    car.start()

countdown(5)

for car in cars:
    car.join()

"""
Car-0 waiting for light to turn blue...
Car-1 waiting for light to turn blue...
Car-2 waiting for light to turn blue...
Car-3 waiting for light to turn blue...
Car-4 waiting for light to turn blue...
Waiting 5 seconds to turn light blue.
5...
4...
3...
2...
1...
Car-0 moving..
Car-4 moving..
Car-3 moving..
Car-2 moving..
Car-1 moving..
"""

Here, we have a Car class inheriting threading class. It takes a event (object that notifies if traffic light is set to blue). It has internal method is_light_blue which is a blocking call to wait so, is_light_blue from run method keeps waiting for event to be set.

Barriers

Python provides barrier from multithreading.Barrieror multprocessing.Barrier it uses internal variable to maintain weather event is set or not. Note that barriers must be initialized with integer digit representing how many threads / processes will be in a barrier before they proceed.

Barrier accepts following parameters during initialization:

barrier = threading.Barrier(
    parties=num_threads + 1, 
    action=barrer_passed, 
    timeout=5
)

• parties: No of threads / process waiting on barriers.
• action: Callable / function (without params) called when barrier is released
• timeout: Times in seconds in which barrier must be released otherwise threading.BrokenBarrierError is raised.

Offers following methods:

• wait: Wait is basically used by to when a thread or process had to wait. Note that, a process / thread passes a 3 barrier on calling wait for three times.

Now lets see an example of 3 threads waiting on each other for task 1 and task 2 to be completed by all threads and proceeding with task 3.

import threading
import time

class Worker(threading.Thread):

    def __init__(self, name, barrier, *args, **kwargs):
        self.barrier = barrier
        threading.Thread.__init__(self, name=name, *args, **kwargs)

    def task_1(self):
        print(f"Thread {self.name} Executing task1....")
        time.sleep(1)
        print(f"Thread {self.name} Finished executing task1....")

    def task_2(self):
        print(f"Thread {self.name} Executing task2....")
        time.sleep(1)
        print(f"Thread {self.name} Finished executing task2....")

    def task_3(self):
        print(f"Thread {self.name} Executing task3....")
        time.sleep(1)
        print(f"Thread {self.name} Finished executing task3....")

    
    def run(self):
        self.task_1()
        self.task_2()
        
        # wait for task1 and 2 to finish before runing task3
        self.barrier.wait()
        self.task_3()



num_threads = 3

workers = [ ]

def barrer_passed():
    print("Barrier passed")

barrier = threading.Barrier(parties=num_threads, action=barrer_passed, timeout=5)
for i in range(0, 3):
    worker = Worker(name=f"{i}", barrier=barrier)
    workers.append(worker)

for worker in workers:
    worker.start()

for worker in workers:
    worker.join()


# Output
"""
Thread 0 Executing task1....
Thread 1 Executing task1....
Thread 2 Executing task1....
Thread 0 Finished executing task1....
Thread 0 Executing task2....
Thread 2 Finished executing task1....
Thread 2 Executing task2....
Thread 1 Finished executing task1....
Thread 1 Executing task2....
Thread 0 Finished executing task2....
Thread 2 Finished executing task2....
Thread 1 Finished executing task2....
Barrier passed
Thread 1 Executing task3....
Thread 2 Executing task3....
Thread 0 Executing task3....
Thread 1 Finished executing task3....
Thread 0 Finished executing task3....
Thread 2 Finished executing task3....
"""

In the above example, we have a Worker class with three methods to execute three different task i.e. task1, task2, task3 the only constraint is we want task3 to be ran after running task1 and task2.

We are using barriers with 3 parties so that barriers is released when 3 threads reach a waiting.

Conclusion

In the above two examples, we have used barriers and event within a group of threads initiating from a same class. However in practical implementation events and barriers are often set from different threads, using a single event or threads in multiple thread / process classes.