Multithreading in Python | Part-1

This article discusses the concept of thread synchronization in case of multithreading in Python programming language.

Synchronization between threads

Thread synchronization is defined as a mechanism which ensures that two or more concurrent threads do not simultaneously execute some particular program segment known as critical section.

Critical section refers to the parts of the program where the shared resource is accessed.

For example, in the diagram below, 3 threads try to access shared resource or critical section at the same time.

Concurrent accesses to shared resource can lead to race condition.

A race condition occurs when two or more threads can access shared data and they try to change it at the same time. As a result, the values of variables may be unpredictable and vary depending on the timings of context switches of the processes.

Consider the program below to understand the concept of race condition:

import threading

# global variable x
x = 0

def increment():
"""
function to increment global variable x
"""
global x
x += 1

def thread_task():
"""
task for thread
calls increment function 100000 times.
"""
for _ in range(100000):
increment()

def main_task():
global x
# setting global variable x as 0
x = 0

# creating threads
t1 = threading.Thread(target=thread_task)
t2 = threading.Thread(target=thread_task)

# start threads
t1.start()
t2.start()

# wait until threads finish their job
t1.join()
t2.join()

if __name__ == "__main__":
for i in range(10):
main_task()
print("Iteration {0}: x = {1}".format(i,x))

Output:

Iteration 0: x = 175005
Iteration 1: x = 200000
Iteration 2: x = 200000
Iteration 3: x = 169432
Iteration 4: x = 153316
Iteration 5: x = 200000
Iteration 6: x = 167322
Iteration 7: x = 200000
Iteration 8: x = 169917
Iteration 9: x = 153589

In above program:

• Two threads t1 and t2 are created in main_task function and global variable x is set to 0.
• Each thread has a target function thread_task in which increment function is called 100000 times.
• increment function will increment the global variable x by 1 in each call.

The expected final value of x is 200000 but what we get in 10 iterations of main_task function is some different values.

This happens due to concurrent access of threads to the shared variable x. This unpredictability in value of x is nothing but race condition.

Given below is a diagram which shows how can race condition occur in above program:

Notice that expected value of x in above diagram is 12 but due to race condition, it turns out to be 11!

Hence, we need a tool for proper synchronization between multiple threads.

Using Locks

threading module provides a Lock class to deal with the race conditions. Lock is implemented using a Semaphore object provided by the Operating System.

A semaphore is a synchronization object that controls access by multiple processes/threads to a common resource in a parallel programming environment. It is simply a value in a designated place in operating system (or kernel) storage that each process/thread can check and then change. Depending on the value that is found, the process/thread can use the resource or will find that it is already in use and must wait for some period before trying again. Semaphores can be binary (0 or 1) or can have additional values. Typically, a process/thread using semaphores checks the value and then, if it using the resource, changes the value to reflect this so that subsequent semaphore users will know to wait.

Lock class provides following methods:

• acquire([blocking]) : To acquire a lock. A lock can be blocking or non-blocking.
  • When invoked with the blocking argument set to True (the default), thread execution is blocked until the lock is unlocked, then lock is set to locked and return True.
  • When invoked with the blocking argument set to False, thread execution is not blocked. If lock is unlocked, then set it to locked and return True else return False immediately.
• release() : To release a lock.
  • When the lock is locked, reset it to unlocked, and return. If any other threads are blocked waiting for the lock to become unlocked, allow exactly one of them to proceed.
  • If lock is already unlocked, a ThreadError is raised.

Consider the example given below:

import threading

# global variable x
x = 0

def increment():
"""
function to increment global variable x
"""
global x
x += 1

def thread_task(lock):
"""
task for thread
calls increment function 100000 times.
"""
for _ in range(100000):
lock.acquire()
increment()
lock.release()

def main_task():
global x
# setting global variable x as 0
x = 0

# creating a lock
lock = threading.Lock()

# creating threads
t1 = threading.Thread(target=thread_task, args=(lock,))
t2 = threading.Thread(target=thread_task, args=(lock,))

# start threads
t1.start()
t2.start()

# wait until threads finish their job
t1.join()
t2.join()

if __name__ == "__main__":
for i in range(10):
main_task()
print("Iteration {0}: x = {1}".format(i,x))

Output:

Iteration 0: x = 200000
Iteration 1: x = 200000
Iteration 2: x = 200000
Iteration 3: x = 200000
Iteration 4: x = 200000
Iteration 5: x = 200000
Iteration 6: x = 200000
Iteration 7: x = 200000
Iteration 8: x = 200000
Iteration 9: x = 200000

Let us try to understand the above code step by step:

• Firstly, a Lock object is created using:

    lock = threading.Lock()
  

• Then, lock is passed as target function argument:

    t1 = threading.Thread(target=thread_task, args=(lock,))
    t2 = threading.Thread(target=thread_task, args=(lock,))
  

• In the critical section of target function, we apply lock using lock.acquire() method. As soon as a lock is acquired, no other thread can access the critical section (here, increment function) until the lock is released using lock.release() method.

    lock.acquire()
    increment()
    lock.release()
  

  As you can see in the results, the final value of x comes out to be 200000 every time (which is the expected final result).

Here is a diagram given below which depicts the implementation of locks in above program:

This brings us to the end of this tutorial series on Multithreading in Python.
Finally, here are are a few advantages and disadvantages of multithreading:

Advantages:

• It doesn’t block the user. This is because threads are independent of each other.
• Better use of system resources is possible since threads execute tasks parallely.
• Enhanced performance on multi-processor machines.
• Multi-threaded servers and interactive GUIs use multithreading exclusively.

Disadvantages:

• As number of threads increase, complexity increases.
• Synchronization of shared resources (objects, data) is necessary.
• It is difficult to debug, result is sometimes unpredictable.
• Potential deadlocks which leads to starvation, i.e. some threads may not be served with a bad design
• Constructing and synchronizing threads is CPU/memory intensive.

Upcoming articles are gonna cover asyncio!

This article covers the basics of multithreading in Python programming language.

Just like multiprocessing, multithreading is a way of achieving multitasking. In multithreading, the concept of threads is used.

Let us first understand the concept of thread in computer architecture.

Thread

In computing, a process is an instance of a computer program that is being executed. Any process has 3 basic components:

• An executable program.
• The associated data needed by the program (variables, work space, buffers, etc.)
• The execution context of the program (State of process)

A thread is an entity within a process that can be scheduled for execution. Also, it is the smallest unit of processing that can be performed in an OS (Operating System).

In simple words, a thread is a sequence of such instructions within a program that can be executed independently of other code. For simplicity, you can assume that a thread is simply a subset of a process!

A thread contains all this information in a Thread Control Block (TCB):

• Thread Identifier: Unique id (TID) is assigned to every new thread
• Stack pointer: Points to thread’s stack in the process. Stack contains the local variables under thread’s scope.
• Program counter: a register which stores the address of the instruction currently being executed by thread.
• Thread state: can be running, ready, waiting, start or done.
• Thread’s register set: registers assigned to thread for computations.
• Parent process Pointer: A pointer to the Process control block (PCB) of the process that the thread lives on.

Consider the diagram below to understand the relation between process and its thread:

Multithreading

Multiple threads can exist within one process where:

• Each thread contains its own register set and local variables (stored in stack).
• All thread of a process share global variables (stored in heap) and the program code.

Consider the diagram below to understand how multiple threads exist in memory:

Multithreading is defined as the ability of a processor to execute multiple threads concurrently.

In a simple, single-core CPU, it is achieved using frequent switching between threads. This is termed as context switching. In context switching, the state of a thread is saved and state of another thread is loaded whenever any interrupt (due to I/O or manually set) takes place. Context switching takes place so frequently that all the threads appear to be running parallely (this is termed as multitasking).

Consider the diagram below in which a process contains two active threads:

Multithreading in Python

In Python, the threading module provides a very simple and intuitive API for spawning multiple threads in a program.

Let us consider a simple example using threading module:

# importing the threading module
import threading

def print_cube(num):
    """
    function to print cube of given num
    """
    print("Cube: {}".format(num * num * num))

def print_square(num):
    """
    function to print square of given num
    """
    print("Square: {}".format(num * num))

if __name__ == "__main__":
    # creating thread
    t1 = threading.Thread(target=print_square, args=(10,))
    t2 = threading.Thread(target=print_cube, args=(10,))

    # starting thread 1
    t1.start()
    # starting thread 2
    t2.start()

    # wait until thread 1 is completely executed
    t1.join()
    # wait until thread 2 is completely executed
    t2.join()

    # both threads completely executed
    print("Done!")

Square: 100
Cube: 1000
Done!

Let us try to understand the above code:

• To import the threading module, we do:

  import threading
  

• To create a new thread, we create an object of Thread class. It takes following arguments:
  • target: the function to be executed by thread
  • args: the arguments to be passed to the target function

  In above example, we created 2 threads with different target functions:

  t1 = threading.Thread(target=print_square, args=(10,))
  t2 = threading.Thread(target=print_cube, args=(10,))
  

• To start a thread, we use start method of Thread class.

  t1.start()
  t2.start()
  

• Once the threads start, the current program (you can think of it like a main thread) also keeps on executing. In order to stop execution of current program until a thread is complete, we use join method.

  t1.join()
  t2.join()
  

  As a result, the current program will first wait for the completion of t1 and then t2. Once, they are finished, the remaining statements of current program are executed.

Consider the diagram below for a better understanding of how above program works:

Now, consider the python program given below in which we print thread name and corresponding process for each task:

import threading
import os

def task1():
    print("Task 1 assigned to thread: {}".format(threading.current_thread().name))
    print("ID of process running task 1: {}".format(os.getpid()))

def task2():
    print("Task 2 assigned to thread: {}".format(threading.current_thread().name))
    print("ID of process running task 2: {}".format(os.getpid()))

if __name__ == "__main__":

    # print ID of current process
    print("ID of process running main program: {}".format(os.getpid()))

    # print name of main thread
    print("Main thread name: {}".format(threading.main_thread().name))

    # creating threads
    t1 = threading.Thread(target=task1, name='t1')
    t2 = threading.Thread(target=task2, name='t2')  

    # starting threads
    t1.start()
    t2.start()

    # wait until all threads finish
    t1.join()
    t2.join()

ID of process running main program: 11758
Main thread name: MainThread
Task 1 assigned to thread: t1
ID of process running task 1: 11758
Task 2 assigned to thread: t2
ID of process running task 2: 11758

Let us try to understand the above code:

• We use os.getpid() function to get ID of current process.

  print("ID of process running main program: {}".format(os.getpid()))
  

  As it is clear from the output, the process ID remains same for all threads.

• We use threading.main_thread() function to get the main thread object. In normal conditions, the main thread is the thread from which the Python interpreter was started. name attribute of thread object is used to get the name of thread.

  print("Main thread name: {}".format(threading.main_thread().name))
  

• We use the threading.current_thread() function to get the current thread object.

  print("Task 1 assigned to thread: {}".format(threading.current_thread().name))
  

The diagram given below clears the above concept:

So, this was a brief introduction to multithreading in Python. The next article in this series covers synchronization between multiple threads.

This article discusses two important concepts related to multiprocessing in Python:

• Synchronization between processes
• Pooling of processes

Synchronization between processes

Process synchronization is defined as a mechanism which ensures that two or more concurrent processes do not simultaneously execute some particular program segment known as critical section.

Critical section refers to the parts of the program where the shared resource is accessed.

For example, in the diagram below, 3 processes try to access shared resource or critical section at the same time.

Concurrent accesses to shared resource can lead to race condition.

A race condition occurs when two or more processes can access shared data and they try to change it at the same time. As a result, the values of variables may be unpredictable and vary depending on the timings of context switches of the processes.

Consider the program below to understand the concept of race condition:

import multiprocessing

def withdraw(balance):
    # function to withdraw from account
    for _ in range(10000):
        balance.value = balance.value - 1

def deposit(balance):
    # function to deposit to account
    for _ in range(10000):
        balance.value = balance.value + 1

def perform_transactions():
    # initial balance (in shared memory)
    balance = multiprocessing.Value('i', 100)

    # creating new processes
    p1 = multiprocessing.Process(target=withdraw, args=(balance,))
    p2 = multiprocessing.Process(target=deposit, args=(balance,))

    # starting processes
    p1.start()
    p2.start()

    # wait until processes are finished
    p1.join()
    p2.join()

    # print final balance
    print("Final balance = {}".format(balance.value))

if __name__ == "__main__":
    for _ in range(10):
        # perform same transaction process 10 times
        perform_transactions()

If you run above program, you will get some unexpected values like this:

Final balance = 1311
Final balance = 199
Final balance = 558
Final balance = -2265
Final balance = 1371
Final balance = 1158
Final balance = -577
Final balance = -1300
Final balance = -341
Final balance = 157

In above program, 10000 withdraw and 10000 deposit transactions are carried out with initial balance as 100. The expected final balance is 100 but what we get in 10 iterations of perform_transactions function is some different values.

This happens due to concurrent access of processes to the shared data balance. This unpredictability in balance value is nothing but race condition.

Let us try to understand it better using the sequence diagrams given below. These are the different sequences which can be produced in above example for a single withdraw and deposit action.

• This is a possible sequence which gives wrong answer as both processes read the same value and write it back accordingly.
  
• These are 2 possible sequences which are desired in above scenario.
  

Using Locks

multiprocessing module provides a Lock class to deal with the race conditions. Lock is implemented using a Semaphore object provided by the Operating System.

A semaphore is a synchronization object that controls access by multiple processes to a common resource in a parallel programming environment. It is simply a value in a designated place in operating system (or kernel) storage that each process can check and then change. Depending on the value that is found, the process can use the resource or will find that it is already in use and must wait for some period before trying again. Semaphores can be binary (0 or 1) or can have additional values. Typically, a process using semaphores checks the value and then, if it using the resource, changes the value to reflect this so that subsequent semaphore users will know to wait.

Consider the example given below:

import multiprocessing

def withdraw(balance, lock):
    # function to withdraw from account
    for _ in range(10000):
        lock.acquire()
        balance.value = balance.value - 1
        lock.release()

def deposit(balance, lock):
    # function to deposit to account
    for _ in range(10000):
        lock.acquire()
        balance.value = balance.value + 1
        lock.release()

def perform_transactions():
    # intitial balance (in shared memory)
    balance = multiprocessing.Value('i', 100)

    # creating a lock object
    lock = multiprocessing.Lock()

    # creating new processes
    p1 = multiprocessing.Process(target=withdraw, args=(balance,lock))
    p2 = multiprocessing.Process(target=deposit, args=(balance,lock))

    # starting processes
    p1.start()
    p2.start()

    # wait until processes are finished
    p1.join()
    p2.join()

    # print final balance
    print("Final balance = {}".format(balance.value))

if __name__ == "__main__":
    for _ in range(10):
        # perform same transaction process 10 times
        perform_transactions()

Output:

Final balance = 100
Final balance = 100
Final balance = 100
Final balance = 100
Final balance = 100
Final balance = 100
Final balance = 100
Final balance = 100
Final balance = 100
Final balance = 100

Let us try to understand the above code step by step:

• Firstly, a Lock object is created using:

  lock = multiprocessing.Lock()
  

• Then, lock is passed as target function argument:

   p1 = multiprocessing.Process(target=withdraw, args=(balance,lock))
   p2 = multiprocessing.Process(target=deposit, args=(balance,lock))
  

• In the critical section of target function, we apply lock using lock.acquire() method. As soon as a lock is acquired, no other process can access its critical section until the lock is released using lock.release() method.

  lock.acquire()
  balance.value = balance.value - 1
  lock.release()
  

  As you can see in the results, the final balance comes out to be 100 every time (which is the expected final result).

Pooling between processes

Let us consider a simple program to find squares of numbers in a given list.

def square(n):
    return (n*n)

if __name__ == "__main__":
    # input list
    mylist = [1,2,3,4,5]
    # empty list to store result
    result = []

    for num in mylist:
        result.append(square(num))

    print(result)

Output:

[1, 4, 9, 16, 25]

It is a simple program to calculate squares of elements of a given list. In a multi-core/multi-processor system, consider the diagram below to understand how above program will work:

Only one of the cores is used for program execution and it’s quite possible that other cores remain idle.

In order to utilize all the cores, multiprocessing module provides a Pool class. The Pool class represents a pool of worker processes. It has methods which allows tasks to be offloaded to the worker processes in a few different ways. Consider the diagram below:

Here, the task is offloaded/distributed among the cores/processes automatically by Pool object. User doesn’t need to worry about creating processes explicitly.

Consider the program given below:

import multiprocessing
import os

def square(n):
    print("Worker process id for {0}: {1}".format(n, os.getpid()))
    return (n*n)

if __name__ == "__main__":
    # input list
    mylist = [1,2,3,4,5]

    # creating a pool object
    p = multiprocessing.Pool()

    # map list to target function
    result = p.map(square, mylist)

    print(result)

Output:

Worker process id for 2: 4152
Worker process id for 1: 4151
Worker process id for 4: 4151
Worker process id for 3: 4153
Worker process id for 5: 4152
[1, 4, 9, 16, 25]

Let us try to understand above code step by step:

• We create a Pool object using:

    p = multiprocessing.Pool()
  

  There are a few arguments for gaining more control over offloading of task. These are:

  • processes: specify the number of worker processes.
  • maxtasksperchild: specify the maximum number of task to be assigned per child.

  All the processes in a pool can be made to perform some initialization using these arguments:

  • initializer: specify an initialization function for worker processes.
  • initargs: arguments to be passed to initializer.
• Now, in order to perform some task, we have to map it to some function. In the example above, we map mylist to square function. As a result, the contents of mylist and definition of square will be distributed among the cores.

    result = p.map(square, mylist)
  

• Once all the worker processes finish their task, a list is returned with the final result.

This brings us to the end of this tutorial series on multiprocessing in Python!
Next few articles will cover multithreading and asyncio libraries in Python.

• run independently
• have their own memory space.

Consider the program below to understand this concept:

import multiprocessing

# empty list with global scope
result = []

def square_list(mylist):
    """
    function to square a given list
    """
    global result
    # append squares of mylist to global list result
    for num in mylist:
        result.append(num * num)
    # print global list result
    print("Result(in process p1): {}".format(result))

if __name__ == "__main__":
    # input list
    mylist = [1,2,3,4]

    # creating new process
    p1 = multiprocessing.Process(target=square_list, args=(mylist,))
    # starting process
    p1.start()
    # wait until process is finished
    p1.join()

    # print global result list
    print("Result(in main program): {}".format(result))

Result(in process p1): [1, 4, 9, 16]
Result(in main program): []

In above example, we try to print contents of global list result at two places:

• In square_list function. Since, this function is called by process p1, result list is changed in memory space of process p1 only.
• After the completion of process p1 in main program. Since main program is run by a different process, its memory space still contains the empty result list.

Diagram shown below clears this concept:

Sharing data between processes

1. Shared memory : multiprocessing module provides Array and Value objects to share data between processes.
   • Array: a ctypes array allocated from shared memory.
   • Value: a ctypes object allocated from shared memory.

   Given below is a simple example showing use of Array and Value for sharing data between processes.

   import multiprocessing
   
   def square_list(mylist, result, square_sum):
       """
       function to square a given list
       """
       # append squares of mylist to result array
       for idx, num in enumerate(mylist):
           result[idx] = num * num
   
       # square_sum value
       square_sum.value = sum(result)
   
       # print result Array
       print("Result(in process p1): {}".format(result[:]))
   
       # print square_sum Value
       print("Sum of squares(in process p1): {}".format(square_sum.value))
   
   if __name__ == "__main__":
       # input list
       mylist = [1,2,3,4]
   
       # creating Array of int data type with space for 4 integers
       result = multiprocessing.Array('i', 4)
   
       # creating Value of int data type
       square_sum = multiprocessing.Value('i')
   
       # creating new process
       p1 = multiprocessing.Process(target=square_list, args=(mylist, result, square_sum))
   
       # starting process
       p1.start()
   
       # wait until process is finished
       p1.join()
   
       # print result array
       print("Result(in main program): {}".format(result[:]))
   
       # print square_sum Value
       print("Sum of squares(in main program): {}".format(square_sum.value))
   

   Result(in process p1): [1, 4, 9, 16]
   Sum of squares(in process p1): 30
   Result(in main program): [1, 4, 9, 16]
   Sum of squares(in main program): 30
   

   Let us try to understand the above code line by line:

   • First of all, we create an Array result like this:

       result = multiprocessing.Array('i', 4)
     

     • First argument is the data type. ‘i’ stands for integer whereas ‘d’ stands for float data type.
     • Second argument is the size of array. Here, we create an array of 4 elements.

     Similarly, we create a Value square_sum like this:

       square_sum = multiprocessing.Value('i')
     

     Here, we only need to specify data type. The value can be given an initial value(say 10) like this:

       square_sum = multiprocessing.Value('i', 10)
     

   • Secondly, we pass result and square_sum as arguments while creating Process object.

       p1 = multiprocessing.Process(target=square_list, args=(mylist, result, square_sum))
     

   • result array elements are given a value by specifying index of array element.

       for idx, num in enumerate(mylist):
           result[idx] = num * num
     

     square_sum is given a value by using its value attribute:

       square_sum.value = sum(result)
     

   • In order to print result array elements, we use result[:] to print complete array.

       print("Result(in process p1): {}".format(result[:]))
     

     Value of square_sum is simply printed as:

       print("Sum of squares(in process p1): {}".format(square_sum.value))
     

   Here is a diagram depicting how processes share Array and Value object:
   

2. Server process : Whenever a python program starts, a server process is also started. From there on, whenever a new process is needed, the parent process connects to the server and requests it to fork a new process.
   A server process can hold Python objects and allows other processes to manipulate them using proxies.
   multiprocessing module provides a Manager class which controls a server process. Hence, managers provide a way to create data which can be shared between different processes.

Server process managers are more flexible than using shared memory objects because they can be made to support arbitrary object types like lists, dictionaries, Queue, Value, Array, etc. Also, a single manager can be shared by processes on different computers over a network. They are, however, slower than using shared memory.

Consider the example given below:

import multiprocessing

def print_records(records):
    """
    function to print record(tuples) in records(list)
    """
    for record in records:
        print("Name: {0}\nScore: {1}\n".format(record[0], record[1]))

def insert_record(record, records):
    """
    function to add a new record to records(list)
    """
    records.append(record)
    print("New record added!\n")

if __name__ == '__main__':
    with multiprocessing.Manager() as manager:
        # creating a list in server process memory
        records = manager.list([('Sam', 10), ('Adam', 9), ('Kevin',9)])
        # new record to be inserted in records
        new_record = ('Jeff', 8)

        # creating new processes
        p1 = multiprocessing.Process(target=insert_record, args=(new_record, records))
        p2 = multiprocessing.Process(target=print_records, args=(records,))

        # running process p1 to insert new record
        p1.start()
        p1.join()

        # running process p2 to print records
        p2.start()
        p2.join()

New record added!

Name: Sam
Score: 10

Name: Adam
Score: 9

Name: Kevin
Score: 9

Name: Jeff
Score: 8

Let us try to understand above piece of code:

• First of all, we create a manager object using:

    with multiprocessing.Manager() as manager:
  

  All the lines under with statement block are under the scope of manager object.

• Then, we create a list records in server process memory using:

    records = manager.list([('Sam', 10), ('Adam', 9), ('Kevin',9)])
  

  Similarly, you can create a dictionary as manager.dict method.

• Finally, we create to processes p1 (to insert a new record in records list) and p2 (to print records) and run them while passing records as one of the arguments.

The concept of server process is depicted in the diagram shown below:

Communication between processes

Effective use of multiple processes usually requires some communication between them, so that work can be divided and results can be aggregated.
multiprocessing supports two types of communication channel between processes:

• Queue
• Pipe

1. Queue : A simple way to communicate between process with multiprocessing is to use a Queue to pass messages back and forth. Any Python object can pass through a Queue.
   Note: The multiprocessing.Queue class is a near clone of queue.Queue.
   Consider the example program given below:

   import multiprocessing
   
   def square_list(mylist, q):
       """
       function to square a given list
       """
       # append squares of mylist to queue
       for num in mylist:
           q.put(num * num)
   
   def print_queue(q):
       """
       function to print queue elements
       """
       print("Queue elements:")
       while not q.empty():
           print(q.get())
       print("Queue is now empty!")
   
   if __name__ == "__main__":
       # input list
       mylist = [1,2,3,4]
   
       # creating multiprocessing Queue
       q = multiprocessing.Queue()
   
       # creating new processes
       p1 = multiprocessing.Process(target=square_list, args=(mylist, q))
       p2 = multiprocessing.Process(target=print_queue, args=(q,))
   
       # running process p1 to square list
       p1.start()
       p1.join()
   
       # running process p2 to get queue elements
       p2.start()
       p2.join()
   

   Queue elements:
   1
   4
   9
   16
   Queue is now empty!
   

   Let us try to understand the above code step by step:

   • Firstly, we create a multiprocessing Queue using:

       q = multiprocessing.Queue()
     

   • Then we pass empty queue q to square_list function through process p1. Elements are inserted to queue using put method.

       q.put(num * num)
     

   • In order to print queue elements, we use get method until queue is not empty.

       while not q.empty():
           print(q.get())
     

   Given below is a simple diagram depicting the operations on queue:
   

2. Pipes : A pipe can have only two endpoints. Hence, it is preferred over queue when only two-way communication is required.multiprocessing module provides Pipe() function which returns a pair of connection objects connected by a pipe. The two connection objects returned by Pipe() represent the two ends of the pipe. Each connection object has send() and recv() methods (among others).
   Consider the program given below:

   import multiprocessing
   
   def sender(conn, msgs):
       """
       function to send messages to other end of pipe
       """
       for msg in msgs:
           conn.send(msg)
           print("Sent the message: {}".format(msg))
       conn.close()
   
   def receiver(conn):
       """
       function to print the messages received from other
       end of pipe
       """
       while 1:
           msg = conn.recv()
           if msg == "END":
               break
           print("Received the message: {}".format(msg))
   
   if __name__ == "__main__":
       # messages to be sent
       msgs = ["hello", "hey", "hru?", "END"]
   
       # creating a pipe
       parent_conn, child_conn = multiprocessing.Pipe()
   
       # creating new processes
       p1 = multiprocessing.Process(target=sender, args=(parent_conn,msgs))
       p2 = multiprocessing.Process(target=receiver, args=(child_conn,))
   
       # running processes
       p1.start()
       p2.start()
   
       # wait until processes finish
       p1.join()
       p2.join()
   

   Sent the message: hello
   Sent the message: hey
   Sent the message: hru?
   Received the message: hello
   Sent the message: END
   Received the message: hey
   Received the message: hru?
   

   Let us try to understand above code:

   • A pipe was created simply using:

       parent_conn, child_conn = multiprocessing.Pipe()
     

     The function returned two connection objects for the two ends of the pipe.

   • Message is sent from one end of pipe to another using send method.

       conn.send(msg)
     

   • To receive any messages at one end of a pipe, we use recv method.

       msg = conn.recv()
     

   • In above program, we send a list of messages from one end to another. At the other end, we read messages until we receive “END” message.

   Consider the diagram given below which shows the relation b/w pipe and processes:
   

Note: Data in a pipe may become corrupted if two processes (or threads) try to read from or write to the same end of the pipe at the same time. Of course there is no risk of corruption from processes using different ends of the pipe at the same time. Also note that, Queues do proper synchronization between processes, at the expense of more complexity. Hence, queues are said to be thread and process safe!

This article is a brief yet concise introduction to multiprocessing in Python programming language.

What is multiprocessing?

Multiprocessing refers to the ability of a system to support more than one processor at the same time. Applications in a multiprocessing system are broken to smaller routines that run independently. The operating system allocates these threads to the processors improving performance of the system.

Why multiprocessing?

Consider a computer system with a single processor. If it is assigned several processes at the same time, it will have to interrupt each task and switch briefly to another, to keep all of the processes going.
This situation is just like a chef working in a kitchen alone. He has to do several tasks like baking, stirring, kneading dough, etc.

So the gist is that: The more tasks you must do at once, the more difficult it gets to keep track of them all, and keeping the timing right becomes more of a challenge.
This is where the concept of multiprocessing arises!

A multiprocessing system can have:

• multiprocessor, i.e. a computer with more than one central processor.
• multi-core processor, i.e. a single computing component with two or more independent actual processing units (called “cores”).

Here, the CPU can easily executes several tasks at once, with each task using its own processor.

It is just like the chef in last situation being assisted by his assistants. Now, they can divide the tasks among themselves and chef doesn’t need to switch between his tasks.

Multiprocessing in Python

In Python, the multiprocessing module includes a very simple and intuitive API for dividing work between multiple processes.
Let us consider a simple example using multiprocessing module:

# importing the multiprocessing module
import multiprocessing

def print_cube(num):
    """
    function to print cube of given num
    """
    print("Cube: {}".format(num * num * num))

def print_square(num):
    """
    function to print square of given num
    """
    print("Square: {}".format(num * num))

if __name__ == "__main__":
    # creating processes
    p1 = multiprocessing.Process(target = print_square, args =(10, ))
    p2 = multiprocessing.Process(target = print_cube, args =(10, ))

    # starting process 1
    p1.start()
    # starting process 2
    p2.start()

    # wait until process 1 is finished
    p1.join()
    # wait until process 2 is finished
    p2.join()

    # both processes finished
    print("Done !")

Square: 100
Cube: 1000
Done!

Let us try to understand the above code:

• To import the multiprocessing module, we do:

  import multiprocessing
  

• To create a process, we create an object of Process class. It takes following arguments:
  • target: the function to be executed by process
  • args: the arguments to be passed to the target function

  Note: Process constructor takes many other arguments also which will be discussed later. In above example, we created 2 processes with different target functions:

  p1 = multiprocessing.Process(target=print_square, args=(10, ))
  p2 = multiprocessing.Process(target=print_cube, args=(10, ))
  

• To start a process, we use start method of Process class.

  p1.start()
  p2.start()
  

• Once the processes start, the current program also keeps on executing. In order to stop execution of current program until a process is complete, we use join method.

  p1.join()
  p2.join()
  

  As a result, the current program will first wait for the completion of p1 and then p2. Once, they are completed, the next statements of current program are executed.

Let us consider another program to understand the concept of different processes running on same python script. In this example below, we print the ID of the processes running the target functions:

# importing the multiprocessing module
import multiprocessing
import os

def worker1():
    # printing process id
    print("ID of process running worker1: {}".format(os.getpid()))

def worker2():
    # printing process id
    print("ID of process running worker2: {}".format(os.getpid()))

if __name__ == "__main__":
    # printing main program process id
    print("ID of main process: {}".format(os.getpid()))

    # creating processes
    p1 = multiprocessing.Process(target = worker1)
    p2 = multiprocessing.Process(target = worker2)

    # starting processes
    p1.start()
    p2.start()

    # process IDs
    print("ID of process p1: {}".format(p1.pid))
    print("ID of process p2: {}".format(p2.pid))

    # wait until processes are finished
    p1.join()
    p2.join()

    # both processes finished
    print("Both processes finished execution !")

    # check if processes are alive
    print("Process p1 is alive: {}".format(p1.is_alive()))
    print("Process p2 is alive: {}".format(p2.is_alive()))

ID of main process: 28628
ID of process running worker1: 29305
ID of process running worker2: 29306
ID of process p1: 29305
ID of process p2: 29306
Both processes finished execution!
Process p1 is alive: False
Process p2 is alive: False

• The main python script has a different process ID and multiprocessing module spawns new processes with different process IDs as we create Process objects p1 and p2. In above program, we use os.getpid() function to get ID of process running the current target function.Notice that it matches with the process IDs of p1 and p2 which we obtain using pid attribute of Process class.
• Each process runs independently and has its own memory space.
• As soon as the execution of target function is finished, the processes get terminated. In above program we used is_alive method of Process class to check if a process is still active or not.

Consider the diagram below to understand how new processes are different from main Python script:

So, this was a brief introduction to multiprocessing in Python.
Find next part here: Multiprocessing in Python | Part-2

References:

• http://learn.parallax.com/tutorials/language/blocklyprop/blocklyprop-functions-and-multicore/bit-about-multicore
• https://docs.python.org/3/library/multiprocessing.html

Wit.ai is a NLP (natural language processing) interface for applications capable of turning sentences into structured data. And most importantly, it is free! So, there are no API call limits!

Wit.ai API provides many kind of NLP services including Speech Recognition.

In this article, I am going to show how to consume the Wit Speech API using Python with minimum dependencies.

Step 1: Create an API Key

In order to use Wit.ai API, you need to create a Wit.ai app. Every Wit.ai app has a server access token which can be used as an API Key.

Follow these steps to create a Wit.ai app and generate API Key:

• Go to Wit.ai home page and sign in with your GitHub or Facebook account.
• Click on ‘+’ sign in menu bar on top and create a new app.
• Now, open the app dashboard and got to Settings of your app.
• In Settings, under API Details, copy the Server Access Token and use it as API key.

Step 2: Python script to record audio

Obviously, we need to pass audio data to the Wit API for speech recognition. For this, we create a Python script to record audio from microphone.

For this purpose, we will be using PyAudio module.

Installations

• Windows: Just install PyAudio module using a simple pip command:

  pip install pyaudio
  

• MAC OS X: Install portaudio library using Homebrew and then install PyAudio module using pip:

  brew install portaudio
  pip install pyaudio
  

• Linux: Install portaudio library development package using this command:

  sudo apt-get install portaudio19-dev
  

  Then, install PyAudio module using pip:

  pip install pyaudio
  

Now, consider the code below to record audio from microphone:

import pyaudio
import wave

def record_audio(RECORD_SECONDS, WAVE_OUTPUT_FILENAME):
    #--------- SETTING PARAMS FOR OUR AUDIO FILE ------------#
    FORMAT = pyaudio.paInt16    # format of wave
    CHANNELS = 2                # no. of audio channels
    RATE = 44100                # frame rate
    CHUNK = 1024                # frames per audio sample
    #--------------------------------------------------------#

    # creating PyAudio object
    audio = pyaudio.PyAudio()

    # open a new stream for microphone
    # It creates a PortAudio Stream Wrapper class object
    stream = audio.open(format=FORMAT,channels=CHANNELS,
                        rate=RATE, input=True,
                        frames_per_buffer=CHUNK)

    #----------------- start of recording -------------------#
    print("Listening...")

    # list to save all audio frames
    frames = []

    for i in range(int(RATE / CHUNK * RECORD_SECONDS)):
        # read audio stream from microphone
        data = stream.read(CHUNK)
        # append audio data to frames list
        frames.append(data)

    #------------------ end of recording --------------------#
    print("Finished recording.")

    stream.stop_stream()    # stop the stream object
    stream.close()          # close the stream object
    audio.terminate()       # terminate PortAudio

    #------------------ saving audio ------------------------#

    # create wave file object
    waveFile = wave.open(WAVE_OUTPUT_FILENAME, 'wb')

    # settings for wave file object
    waveFile.setnchannels(CHANNELS)
    waveFile.setsampwidth(audio.get_sample_size(FORMAT))
    waveFile.setframerate(RATE)
    waveFile.writeframes(b''.join(frames))

    # closing the wave file object
    waveFile.close()

def read_audio(WAVE_FILENAME):
    # function to read audio(wav) file
    with open(WAVE_FILENAME, 'rb') as f:
        audio = f.read()
    return audio

Here, we use PyAudio file to record audio in WAV format.

For writing audio stream to a WaveFile, we use in-built Python library wave. Once audio is recorded using PyAudio, it is saved as a wav file in current directory.

Save this Python script as Recorder.py (as we will import this Python script by this name in main Python script).

Step 3: Python script to interact with Wit Speech API

Now, its time to write Python script for interacting with Wit Speech API.

Consider the code below:

import requests
import json
from Recorder import record_audio, read_audio

# Wit speech API endpoint
API_ENDPOINT = 'https://api.wit.ai/speech'

# Wit.ai api access token
wit_access_token = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXX'

def RecognizeSpeech(AUDIO_FILENAME, num_seconds = 5):

    # record audio of specified length in specified audio file
    record_audio(num_seconds, AUDIO_FILENAME)

    # reading audio
    audio = read_audio(AUDIO_FILENAME)

    # defining headers for HTTP request
    headers = {'authorization': 'Bearer ' + wit_access_token,
               'Content-Type': 'audio/wav'}

    # making an HTTP post request
    resp = requests.post(API_ENDPOINT, headers = headers,
                         data = audio)

    # converting response content to JSON format
    data = json.loads(resp.content)

    # get text from data
    text = data['_text']

    # return the text
    return text

if __name__ == "__main__":
    text =  RecognizeSpeech('myspeech.wav', 4)
    print("\nYou said: {}".format(text))

Wit Speech API accepts HTTP POST request.

The POST request must contain:

• headers

  headers = {'authorization': 'Bearer ' + wit_access_token,
               'Content-Type': 'audio/wav'}
  

  where wit_access_token is the API Key we generated in Step 1.

• data

  data = audio
  

  The data to be passed is the audio stream in wav format. As you can notice, the recorded audio is saved in a file called myspeech.wav. We read audio back again form this file using read_audio method.

And we send this HTTP request to this endpoint: https://api.wit.ai/speech

A sample response of HTTP request looks like this:

{u'_text': u'hey how are you',
 u'entities': {},
 u'msg_id': u'1ca8f790-4e83-443c-915c-914bc1a42100'}

Wit.ai is not just limited to speech recognition.

It also allows you to create a chatbot which can recognize any user-defined entity in the provided text! Since I have not created any entities for my current Wit app, the entities section of HTTP response is empty.

Github repository: Wit Speech API Wrapper

Here is a demo video of how above scripts work:

This article demonstrates how to create a simple Desktop Notifier application using Python.

A desktop notifier is a simple application which produces a notification in form of a pop-up message on desktop.

Notification content

In the example we use in this article, the content that will appear as notification on desktop is the top news headlines of the day.

So, in order to fetch the top headlines, we will be using this Python script to scrape news headlines:

import requests
import xml.etree.ElementTree as ET

# url of news rss feed
RSS_FEED_URL = "http://www.hindustantimes.com/rss/topnews/rssfeed.xml"    

def loadRSS():
    '''
    utility function to load RSS feed
    '''
    # create HTTP request response object
    resp = requests.get(RSS_FEED_URL)
    # return response content
    return resp.content

def parseXML(rss):
    '''
    utility function to parse XML format rss feed
    '''
    # create element tree root object
    root = ET.fromstring(rss)
    # create empty list for news items
    newsitems = []

    # iterate news items
    for item in root.findall('./channel/item'):
        news = {}
        # iterate child elements of item
        for child in item:
            # special checking for namespace object content:media
            if child.tag == '{http://search.yahoo.com/mrss/}content':
                news['media'] = child.attrib['url']
            else:
                news[child.tag] = child.text.encode('utf8')
        newsitems.append(news)

    # return news items list
    return newsitems

def topStories():
    '''
    main function to generate and return news items
    '''
    # load rss feed
    rss = loadRSS()
    # parse XML
    newsitems = parseXML(rss)
    return newsitems

{'description': 'Months after it was first reported, the feud between Dwayne Johnson and 
                 Vin Diesel continues to rage on, with a new report saying that the two are 
                 being kept apart during the promotions of The Fate of the Furious.',
 'link': 'http://www.hindustantimes.com/hollywood/vin-diesel-dwayne-johnson-feud-rages-on-they-re-being-kept-apart-for-fast-8-tour/story-Bwl2Nx8gja9T15aMvcrcvL.html',
 'media': 'http://www.hindustantimes.com/rf/image_size_630x354/HT/p2/2017/04/01/Pictures/_fbcbdc10-1697-11e7-9d7a-cd3db232b835.jpg',
 'pubDate': b'Sat, 01 Apr 2017 05:22:51 GMT ',
 'title': "Vin Diesel, Dwayne Johnson feud rages on; they're being deliberately kept apart"}

pip install notify2

import time
import notify2
from topnews import topStories

# path to notification window icon
ICON_PATH = "put full path to icon image here"

# fetch news items
newsitems = topStories()

# initialise the d-bus connection
notify2.init("News Notifier")
# create Notification object
n = notify2.Notification(None, icon = ICON_PATH)
# set urgency level
n.set_urgency(notify2.URGENCY_NORMAL)
# set timeout for a notification
n.set_timeout(10000)

for newsitem in newsitems:
    # update notification data for Notification object
    n.update(newsitem['title'], newsitem['description'])
    # show notification on screen
    n.show()
    # short delay between notifications
    time.sleep(15)