Parallel Computing

Multi Threading in Python

This article will introduce the most basic 5 different multi-threading methods implementation in Python. We will use a very simple example to show how to make use of these 5 different methods to solve a problem.

Barrier

Barrier objects in python are used to wait for a fixed number of thread to complete execution before any particular thread can proceed forward with the execution of the program. Each thread calls wait() function upon reaching the barrier. The barrier is responsible for keeping track of the number of wait() calls. If this number goes beyond the number of threads for which the barrier was initialized with, then the barrier gives a way to the waiting threads to proceed on with the execution. All the threads at this point of execution, are simultaneously released.

There are several main methods available from this object:

parties()

A number of threads required to reach the common barrier point.

n_waiting()

Number of threads waiting in the common barrier point

broken()

A boolean value, True- if the barrier is in the broken state else False.
####wait( timeout = None)
Wait until notified or a timeout occurs. If the calling thread has not acquired the lock when this method is called, a runtime error is raised.
This method releases the underlying lock and then blocks until it is awakened by a notify() or notify_all() method call for the same condition variable in another thread, or until the optional timeout occurs. Once awakened or timed out, it re-acquires the lock and returns.
When the timeout argument is present and not None, it should be a floating point number specifying a timeout for the operation in seconds (or fractions thereof).

reset()

Set or return the barrier to the default state .i.e. empty state. And threads waiting on it will receive the BrokenBarrierError.

bort()

This will put the barrier into a broken state. This causes all the active threads or any future calls to wait() to fail with the BrokenBarrierError.
Here is a program to show how barrier works in python

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import threading 
  
barrier = threading.Barrier(3) 
  
class thread(threading.Thread): 
    def __init__(self, thread_ID): 
        threading.Thread.__init__(self) 
        self.thread_ID = thread_ID 
    def run(self): 
        print(str(self.thread_ID) + "\n") 
        print("Parties = " + str(barrier.parties) + "\n") 
        print("n_waiting = " + str(barrier.n_waiting) + "\n") 
        barrier.wait() 
          
thread1 = thread(100) 
thread2 = thread(101) 
  
thread1.start() 
thread2.start() 
  
barrier.wait() 
  
print(str(barrier.broken) + "\n") 
barrier.reset() 
print("n_waiting after reset = " + str(barrier.n_waiting)) 
barrier.abort() 
print("End")

Lock

The threading module of Python includes locks as a synchronization tool. A lock has two states:
locked and unlocked.
A lock can be locked using the acquire() method. Once a thread has acquired the lock, all subsequent attempts to acquire the lock are blocked until it is released. The lock can be released using the release() method.

Calling the release() method on a lock, in an unlocked state, results in an error.

Let us take an example to see how to make use lock to solve a race condition problem.

Race condition is a significant problem in concurrent programming. The condition occurs when one thread tries to modify a shared resource at the same time that another thread is modifying that resource – t​his leads to garbled output, which is why threads need to be synchronized.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# Importing the threading module
import threading 
# Declraing a lock
lock = threading.Lock()
deposit = 100
# Function to add profit to the deposit
def add_profit(): 
    global deposit
    for i in range(100000):
        lock.acquire()
        deposit = deposit + 10
        lock.release()
# Function to deduct money from the deposit
def pay_bill(): 
    global deposit
    for i in range(100000):
        lock.acquire()
        deposit = deposit - 10
        lock.release()
# Creating threads
thread1 = threading.Thread(target = add_profit, args = ())
thread2 = threading.Thread(target = pay_bill, args = ())
# Starting the threads  
thread1.start() 
thread2.start() 
# Waiting for both the threads to finish executing 
thread1.join()
thread2.join()
# Displaying the final value of the deposit
print(deposit)

Let us use a list variable to record the whole process how deposit changes alone the way.
If we do not have the lock.acquire() and lock.release(), the result may looks like below when there is no lock.acquire() and lock.release() present:

With the usage of lock.acquire() and lock.release(), it will be like:

Semaphore Object

A semaphore manages an internal counter which is decremented by each acquire() call and incremented by each release() call. The counter can never go below zero; when acquire() finds that it is zero, it blocks, waiting until some other thread calls release().

There are several main methods available from this object:

acquire(blocking=True, timeout=None):
release()

Mutex(Lock) VS Semaphore

As we can tell, the methods we have from Lock instance are pretty similar to the methods we have from Semaphore. Then what is the difference between Mutex and Semaphore ?

Mutex is a mutual exclusion object that synchronizes access to a resource. It is created with a unique name at the start of a program. The Mutex is a locking mechanism that makes sure only one thread can acquire the Mutex at a time and enter the critical section. This thread only releases the Mutex when it exits the critical section.

A semaphore is a signalling mechanism and a thread that is waiting on a semaphore can be signaled by another thread. This is different than a mutex as the mutex can be signaled only by the thread that called the wait function.

There are mainly two types of semaphores i.e. counting semaphores and binary semaphores.

Counting Semaphores are integer value semaphores and have an unrestricted value domain. These semaphores are used to coordinate the resource access, where the semaphore count is the number of available resources.

The binary semaphores are like counting semaphores but their value is restricted to 0 and 1. The wait operation only works when the semaphore is 1 and the signal operation succeeds when semaphore is 0.

Event Object

One thread can signal an event and other threads wait for it.
There are several main methods available from this object:

isSet()

Return true if and only if the internal flag is true

set()

Set the internal flag to true. All threads waiting for it to become true are awakened. Threads that call wait() once the flag is true will not block at all.

clear()

Reset the internal flag to false. Subsequently, threads calling wait() will block until set() is called to set the internal flag to true again.

wait([timeout in seconds])

Block until the internal flag is true.

Condition

There are several main methods available from this object:

Condition([lock])

If the lock argument is given and not None, it must be a Lock or RLock object, and it is used as the underlying lock. Otherwise, a new RLock object is created and used as the underlying lock.

acquire(*args)

Acquire the underlying lock. This method calls the corresponding method on the underlying lock; the return value is whatever that method returns.

release ()

Release the underlying lock. This method calls the corresponding method on the underlying lock; there is no return value.

wait([timeout])
notify()

Wake up a thread waiting on this condition, if any. This must only be called when the calling thread has acquired the lock.

notifyAll()

Wake up all threads waiting on this condition.

Example

Ok. At the end, let see a problem which can be solved by making use Brutal force, Barrier, Event, Condition, Lock and Semaphore.

Problem description
Suppose we have a class:

1
2
3
4
5
6
7
class Foo(object):
    def first():
	printFirst()
    def second():
        printSecond()    
    def third():
        printThird()

Now we have an instance of Foo which will be passed to 3 different threads. Thread A will call first(), thread B will call second(), thread C will call third(). Design a mechanism to ensure that the second() is executed after first(), third() is executed after second().

Brutal Force

We definite can make use of a very brutal force way to solve this problem by setting a flag with in the class. Until the flag was trigger, we are not able to move forward. Refer to the code below:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import time
class Foo(object):
    def __init__(self):
        self.status = 0

    def first(self, printFirst):
        # printFirst() outputs "first". Do not change or remove this line.
        while self.status != 0:
            time.sleep(0.001)
        printFirst()
        self.status = 1
        
    def second(self, printSecond):
        # printSecond() outputs "second". Do not change or remove this line.
        while self.status != 1:
            time.sleep(0.001)
        printSecond()
        self.status = 2
            
    def third(self, printThird):
        # printThird() outputs "third". Do not change or remove this line.
        while self.status != 2:
            time.sleep(0.001)
        printThird()

Barrier

Raise two barriers. Both wait for two threads to reach them.

First thread can print before reaching the first barrier. Second thread can print before reaching the second barrier. Third thread can print after the second barrier.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from threading import Barrier
class Foo:
    def __init__(self):
        self.first_barrier = Barrier(2)
        self.second_barrier = Barrier(2)
            
    def first(self, printFirst):
        printFirst()
        self.first_barrier.wait()
        
    def second(self, printSecond):
        self.first_barrier.wait()
        printSecond()
        self.second_barrier.wait()
            
    def third(self, printThird):
        self.second_barrier.wait()
        printThird()

Lock (Mutex)

Start with two locked locks. First thread unlocks the first lock that the second thread is waiting on. Second thread unlocks the second lock that the third thread is waiting on.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from threading import Lock

class Foo:
    def __init__(self):
        self.locks = (Lock(),Lock())
        self.locks[0].acquire()
        self.locks[1].acquire()
        
    def first(self, printFirst):
        printFirst()
        self.locks[0].release()
        
    def second(self, printSecond):
        with self.locks[0]:
            printSecond()
            self.locks[1].release()
            
            
    def third(self, printThird):
        with self.locks[1]:
            printThird()

Semaphore

Start with two closed gates represented by 0-value semaphores. Second and third thread are waiting behind these gates. When the first thread prints, it opens the gate for the second thread. When the second thread prints, it opens the gate for the third thread.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from threading import Semaphore
class Foo:
    def __init__(self):
        self.gates = (Semaphore(0),Semaphore(0))
        
    def first(self, printFirst):
        printFirst()
        self.gates[0].release()
        
    def second(self, printSecond):
        with self.gates[0]:
            printSecond()
            self.gates[1].release()
            
    def third(self, printThird):
        with self.gates[1]:
            printThird()

Event

Set events from first and second threads when they are done. Have the second thread wait for first one to set its event. Have the third thread wait on the second thread to raise its event.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from threading import Event

class Foo:
    def __init__(self):
        self.done = (Event(),Event())
        
    def first(self, printFirst):
        printFirst()
        self.done[0].set()
        
    def second(self, printSecond):
        self.done[0].wait()
        printSecond()
        self.done[1].set()
            
    def third(self, printThird):
        self.done[1].wait()
        printThird()

Condition

Have all three threads attempt to acquire an RLock via Condition. The first thread can always acquire a lock, while the other two have to wait for the order to be set to the right value. First thread sets the order after printing which signals for the second thread to run. Second thread does the same for the third.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from threading import Condition

class Foo:
    def __init__(self):
        self.exec_condition = Condition()
        self.order = 0
        self.first_finish = lambda: self.order == 1
        self.second_finish = lambda: self.order == 2

    def first(self, printFirst):
        with self.exec_condition:
            printFirst()
            self.order = 1
            self.exec_condition.notify(2)

    def second(self, printSecond):
        with self.exec_condition:
            self.exec_condition.wait_for(self.first_finish)
            printSecond()
            self.order = 2
            self.exec_condition.notify()

    def third(self, printThird):
        with self.exec_condition:
            self.exec_condition.wait_for(self.second_finish)
            printThird()

Referrence:
https://leetcode.com/problems/print-in-order/discuss/335939/5-Python-threading-solutions-(Barrier-Lock-Event-Semaphore-Condition)-with-explanation
https://www.geeksforgeeks.org/barrier-objects-python/
https://www.educative.io/edpresso/what-are-locks-in-python