Synchronization in Python – Synchronize Threads in Python

Synchronization In Python

Let’s talk about synchronization in Python. Multithreading allows your computer to perform actions in parallel, utilizing multiple cores/ multiple CPUs present on your system. However, when it comes to reading and updating shared variables at the same time, it can lead to erroneous results. We will learn how to synchronize threads to give correct results.

Understanding the Race Condition in Multithreading

When 2 or more threads try to access a shared resource simultaneously and change the data, the final value of such variables is unpredictable. This is because the thread scheduling algorithm can swap between threads at any time and you don’t know which thread will execute first. This scenario is called a race condition.

Let us take an example where we transfer some amount from one bank account to another using threads. We will create 100 threads to transfer 1 unit from account1 to account2.

import threading
import time

class BankAccount():
  def __init__(self, name, balance):
    self.name = name
    self.balance = balance

  def __str__(self):
    return self.name

# These accounts are our shared resources
account1 = BankAccount("account1", 100)
account2 = BankAccount("account2", 0)

class BankTransferThread(threading.Thread):
  def __init__(self, sender, receiver, amount):
    threading.Thread.__init__(self)
    self.sender = sender
    self.receiver = receiver
    self.amount = amount
  
  def run(self):
    sender_initial_balance = self.sender.balance
    sender_initial_balance -= self.amount
    # Inserting delay to allow switch between threads
    time.sleep(0.001)
    self.sender.balance = sender_initial_balance
    
    receiver_initial_balance = self.receiver.balance
    receiver_initial_balance += self.amount
    # Inserting delay to allow switch between threads
    time.sleep(0.001)
    self.receiver.balance = receiver_initial_balance

if __name__ == "__main__":
  
  threads = []

  for i in range(100):
    threads.append(BankTransferThread(account1, account2, 1))

  for thread in threads:
    thread.start()

  for thread in threads:
    thread.join()

  print(account1.balance)
  print(account2.balance)
account1 98
account2 3

Initially, account1 has 100 units and account2 has 0 units.

After 100 transfers of 1 unit, account1 should have 0 units and the account2 should have 100 units. However, we got different results. If we run this multiple times, we will get different results.

Synchronization in Python – Different Methods to Synchronize Threads

Lets see how to synchronize threads to avoid race conditions.

1. Lock Objects

A Lock object is the most basic synchronization primitive which is not owned by a particular thread when locked. A Lock object does not keep information about which thread has a permit of the lock and any thread can release the lock.

The Lock object is in one of the 2 states, “locked” and “unlocked”. When the Lock object is created, it is in the “unlocked” state. There are only 3 methods in the lock object:

  • acquire(): This method changes the Lock object from an “unlocked” state to a “locked” state and allows the calling thread to continue execution. If the Lock object is already in a “locked” state, the calling thread will be blocked until the lock comes in an “unlocked” state.
  • release(): This method changes the Lock object state from “locked” to “unlocked” state. If the Lock object is already in an “unlocked” state, a RuntimeError is raised. The method can be called from any thread, not only the thread which has acquired the lock.
  • locked(): This method returns true if the Lock object is acquired.

Let’s see how to use Lock object to add synchronization in Python to our bank transfer example.

import threading
import time

class BankAccount():
  def __init__(self, name, balance):
    self.name = name
    self.balance = balance

  def __str__(self):
    return self.name

# These accounts are our shared resources
account1 = BankAccount("account1", 100)
account2 = BankAccount("account2", 0)

# Creating lock for threads
lock = threading.Lock()

class BankTransferThread(threading.Thread):
  def __init__(self, sender, receiver, amount):
    threading.Thread.__init__(self)
    self.sender = sender
    self.receiver = receiver
    self.amount = amount
  
  def run(self):
    lock.acquire()
    
    sender_initial_balance = self.sender.balance
    sender_initial_balance -= self.amount
    # Inserting delay to allow switch between threads
    time.sleep(0.001)
    self.sender.balance = sender_initial_balance
    
    receiver_initial_balance = self.receiver.balance
    receiver_initial_balance += self.amount
    # Inserting delay to allow switch between threads
    time.sleep(0.001)
    self.receiver.balance = receiver_initial_balance
    
    lock.release()
    

if __name__ == "__main__":
  
  threads = []

  for i in range(100):
    threads.append(BankTransferThread(account1, account2, 1))

  for thread in threads:
    thread.start()

  for thread in threads:
    thread.join()

  print(account1.name, account1.balance)
  print(account2.name, account2.balance)
account1 0
account2 100

The Lock object does not know which thread calls the acquire() method and any thread can call release() on the lock which can take permit from the thread that calls the acquire().

Also if same thread calls acquire() method again without release(), the thread will be in the deadlock state.

import threading

lock = threading.Lock()

def funcA():
  print("In A, acquiring lock")
  lock.acquire()
  
  print("In A, lock acquired")
  
  print("In A, lock acquiring again and entering into deadlock")
  lock.acquire()
  
  print("In A, releasing lock")
  lock.release()
  
  print("In A, lock released")

def funcB():
  print("In B, releasing lock acquired by A")
  lock.release()
  
  print("In B, lock released")

if __name__ == "__main__":
  thread1 = threading.Thread(target=funcA)
  thread2 = threading.Thread(target=funcB)

  thread1.start()
  thread2.start()

  thread1.join()
  thread2.join()
In A, acquiring lock
In A, lock acquired
In A, lock acquiring again and entering into deadlock
In B, releasing lock acquired by A
In B, lock released
In A, releasing lock
In A, lock released

2. RLock Objects

A reentrant lock (RLock) is another synchronization primitive that may be acquired multiple times by the same thread without entering into a deadlock state. The RLock object knows which thread has the permission of the lock and the same thread can unlock it.

The RLock object is in one of the 2 states, “locked” and “unlocked”. When the RLock object is created, it is in the “unlocked” state. There are only 2 methods in the RLock object:

  • acquire(): This method changes the Lock object from an “unlocked” state to a “locked” state and allows the calling thread to continue execution. If the same thread calls this method again, it increases the recursion level by one. To fully release the lock, the same thread needs to call release() the same number of times. If another thread calls this method in a “locked” state, the thread will be blocked.
  • release(): This method releases the lock and decreases the recursion level by one. If the recursion level becomes 0 after decrement, the lock state is changed to an “unlocked” state. If after the decrement the recursion level is still nonzero, the lock remains “locked” and owned by the calling thread. If the RLock object is already in an “unlocked” state, a RuntimeError is raised.
import threading

lock = threading.RLock()

def funcA():
  print("In A, acquiring lock")
  lock.acquire()
  
  print("In A, lock acquired, recursion level = 1")
  
  print("In A, acquiring lock again")
  lock.acquire()
  
  print("In A, lock acquired again, recursion level = 2")
  
  print("In A, releasing lock")
  lock.release()
  
  print("In A, lock released, recursion level = 1")
  

def funcB():
  print("In B, trying to acquire lock, but A released only once, so entering in deadlock state")
  lock.acquire()
  print("This statement won't be executed")
  
if __name__ == "__main__":
  thread1 = threading.Thread(target=funcA)
  thread2 = threading.Thread(target=funcB)

  thread1.start()
  thread2.start()

  thread1.join()
  thread2.join()
In A, acquiring l
In A, lock acquired, recursion level = 1
In A, acquiring lock again
In A, lock acquired again, recursion level = 2
In A, releasing lock
In A, lock released, recursion level = 1
In B, trying to acquire lock, but A released only once, so entering in deadlock state

3. Semaphores

Semaphore is simply a variable that is non-negative and shared between threads. While the Lock and RLock objects allow only one thread to execute, Semaphore allows more than one thread to execute at a time. Semaphores are used to protect resources that have a limited capacity by specifying the number of threads allowed to execute when creating a Semaphore object. If this initial count is 1, Semaphores can help in the synchronization of threads.

  • Creating a Semaphore: To create a Semaphore object, call Semaphore(count) in threading module, where count is the number of threads allowed to access simultaneously. The default value of the count is 1.
  • acquire(): When a thread calls this method
    • If the count value of Semaphore is 0, the thread is blocked until awoken by a call to release().
    • If the count value of Semaphore is greater than 0, it is decremented by 1 and the thread continues its execution.
  • release(): This method increments the count value by 1. If any thread is blocked on acquire(), it unblocks one of the threads.

Let us take an example, where 10 threads are trying to read a shared resource, but we limit the concurrent reads on shared resource to 3 using Semaphores.

import threading
import time

read_mutex = threading.Semaphore(3)

# Our shared resource
data = "A Data Stream"

class ReaderThread(threading.Thread):
  def __init__(self):
    threading.Thread.__init__(self)

  def run(self):
    
    read_mutex.acquire()
    
    output = self.getName() + " starts reading"
    print(output)
    
    # threads take time to read a data
    time.sleep(0.5)
    some_data = data
    
    output = self.getName() + " ends reading"
    print(output)
    
    read_mutex.release()
  
  
if __name__ == "__main__":
  
  threads = []
  for i in range(10):
    threads.append(ReaderThread())

  for thread in threads:
    thread.start()

  for thread in threads:
    thread.join()
Thread-6 starts reading
Thread-7 starts reading
Thread-8 starts reading
Thread-8 ends readingThread-7 ends readingThread-6 ends reading


Thread-10 starts reading
Thread-11 starts reading
Thread-9 starts reading
Thread-11 ends readingThread-10 ends reading
Thread-12 starts reading

Thread-13 starts reading
Thread-9 ends reading
Thread-14 starts reading
Thread-13 ends readingThread-12 ends reading

Thread-15 starts reading
Thread-14 ends reading
Thread-15 ends reading

Conclusion

In this tutorial, we have learned synchronization in Python to avoid race conditions by using the threading module in Python. We used Lock, RLock, and Semaphores to achieve synchronization in Python. Thanks for reading!!