Open Technical Blog: Multiprocessing - Shared Variables, Lock, Queue

The script shows how to execute 4 processes simultaneously in Python by using the multiprocessing module. In this script, shared variables are defined which can be modified by processes, a message queue is also defined which allows communication between processes.

Four processes defined in this script are:

Finding of Fibonacci numbers
Finding prime numbers
Estimation of Pi
Output of results from other processes

The function getchar() is used in the main program, it can wait for key input from the console without requiring the user to press [Enter].

Reference

Python multiprocessing, ctypes

Source Code

Here is the script.

# Title    : Example for multiprocessing

from multiprocessing import Process, Queue, Value, Lock
import ctypes
import tty, termios, sys

def getchar():
    #Returns a single character from standard input
    fd = sys.stdin.fileno()
    old_settings = termios.tcgetattr(fd)
    try:
        tty.setraw(sys.stdin.fileno())
        ch = sys.stdin.read(1)
        if ch==chr(27):
            ch1 = sys.stdin.read(1)
            ch2 = sys.stdin.read(1)
            if ch2 == 'D':
                ch = 'left'
            elif ch2 == 'C':
                ch = 'right'
            elif ch2 == 'A':
                ch = 'up'
            elif ch2 == 'B':
                ch = 'down'
            else:
                print ch, ch1, ch2
        elif ord(ch)==127:
            ch = 'backspace'
    finally:
       termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
    return ch.strip()

# Processes
# ---------
def FibNum(pRun, pOut, pLock, pQueue):
    mN1 = 1
    mN2 = 1
    mIdx = 3
    print 'Process FibNum(P1) is running...\r'
    while pRun.value==True:
        mN = mN1 + mN2
        if mIdx%1000000==0:
            print 'F(%s)\r'%mIdx
        if pOut.value==True:
            #print 'F(%s)\r'%mIdx
            mmsg = 'P1: F(%s)\r'%mIdx
            pQueue.put(mmsg)
            with pLock:
                pOut.value = False
        mN1 = mN2
        mN2 = mN
        mIdx += 1
    print 'Process FibNum is stopped.'

def PrimeNum(pRun, pOut, pLock, pQueue):
    mN = 3
    mPrimeList = [2,3]
    print 'Process PrimeNum(P2) is running ...\r'
    while pRun.value==True:
        mN += 2
        mNr = mN**0.5
        for mi in mPrimeList:
            if (mN%mi==0):
                break
            elif mi>mNr:
                mPrimeList.append(mN)
                break
        if pOut.value==True:
            #print 'Prime Number(%sth) = %s\r'%(len(mPrimeList),mPrimeList[-1])
            mmsg = 'P2: Prime Number(%sth) = %s\r'%(len(mPrimeList),mPrimeList[-1])
            pQueue.put(mmsg)
            with pLock:
                pOut.value = False
    print 'Process PrimeNum is stopped.'

def Pi(pRun, pOut, pLock, pQueue):
    mIdx = 1
    mSum = 0
    print 'Process Pi(P3) is running ...\r'
    while pRun.value==True:
        mSum = mSum + 1.0/(mIdx**2)
        mPi = (6*mSum)**0.5
        if pOut.value==True:
            #print 'Pi(sum of %s terms) ~ %s\r'%(mIdx,mPi)
            mmsg = 'P3: Pi(sum of %s terms) ~ %s\r'%(mIdx,mPi)
            pQueue.put(mmsg)
            with pLock:
                pOut.value = False
        mIdx += 1
    print 'Process Pi is stopped.'

def OutputMsg(pRun, pQueue):
    while pRun.value==True:
        if pQueue.empty():
            continue
        mmsg = pQueue.get()
        print mmsg

# main program
# -------------------------------------------------
# a message queue for communication between
# process
gQueue = Queue()
# the following are shared variables used in the
# main program and the processes.
gRun = Value(ctypes.c_bool,True) # continue to run
gP1Out = Value(ctypes.c_bool,False) # print out in process 1
gP2Out = Value(ctypes.c_bool,False) # print out in process 2
gP3Out = Value(ctypes.c_bool,False) # print out in process 3
# a lock is required whenever a process wants to update
# a shared value
gLock = Lock()

# define processes
p0 = Process(target=OutputMsg,args=(gRun, gQueue))
p1 = Process(target=FibNum,args=(gRun, gP1Out, gLock, gQueue))
p2 = Process(target=PrimeNum,args=(gRun, gP2Out, gLock, gQueue))
p3 = Process(target=Pi,args=(gRun, gP3Out, gLock, gQueue))

# start all processes
p0.start()
p1.start()
p2.start()
p3.start()

print '  Menu'
print '--------------------------------'
print '    q - quit'
print '    f - show FibNum progress'
print '    p - show PrimeNum progress'
print '    i - show Pi progress'
print
while True:
    ch = getchar()
    if ch == 'q':
        with gLock:
            gRun.value = False
        break
    elif ch=='f':
        with gLock:
            gP1Out.value = True
    elif ch=='p':
        with gLock:
            gP2Out.value = True
    elif ch=='i':
        with gLock:
            gP3Out.value = True
    else:
        print 'You press', ch

# wait for all processes to terminate
p0.join()
p1.join()
p2.join()
p3.join()

print 'bye!'

A sample result of running this program is given below

Open Technical Blog

Friday, 12 June 2020

Multiprocessing - Shared Variables, Lock, Queue

Reference

Source Code

No comments:

Post a Comment