Python multiprocessing spawn

Fork Vs Spawn
Fork child inherit all resources from the parent process

Fork is the default method for multi-processing
Spawn child doesn't inherit any resources from the parent process other than those required to execute the specified callable target.
Spawning a process is slower (because it re-imports) than forking a process

Spawn
Spawn is essentially a combination of fork followed by an exec system call
When a child process is spawned, anything imported at the module level in the (above )__main__ module of the parent process gets reimported in the child
Anything below __main__ will not be copied

test1.py
#########
Inside test1

test1.py
#########
Inside test2

spawn_test.py
############
from multiprocessing import Process
import test1
import test2

def process_task():
    print("I am inside child process")


if __name__ == '__main__':

    # Change the method to 'spawn' and verify
    # that the modules are reimported in the child process

    ### Spawn
    multiprocessing.set_start_method('spawn') 
    process = Process(target=process_task)
    process.start()
    process.join()
    print("I am inside parent process")


# Child reimports the test1, test2 module again...

Output:
Inside test1
Inside test2
Inside test1
Inside test2
I am child process
I am parent process

Python multiprocessing fork

Fork
# When we fork, the entire Python process is duplicated in memory including the Python interpreter, code, libraries, current stack, etc.
# This creates a new copy of the python interpreter.
# Fork creates two python interpreters each with its own GIL.
# Fork is faster than Spawn (Fork child inherit all resources from the parent process, Spawn re-imports all above main() method)

# Fork is the default method for multi-processing

 
#Disadvantages of Fork
# It won't work on windows
# When child shares parent libraries, values, data-structures, if a lock acquired by parent, child ends up waiting for that lock ever
# Very hard to debug when you import a third-party module/library that uses threads behind the scenes
# Fork and Multi-threading won't go well

from multiprocessing import Process
import multiprocessing
import os

file_desc = None

def process_task1():
    # write to the file in child process
    file_desc.write(f"\nWritten by child process with id {os.getpid()}")
    file_desc.flush()


if __name__ == '__main__':
    # create a file in the parent process
    file_desc = open("sample.txt", "w")
    file_desc.write(f"\nWritten by parent process with id {os.getpid()}")
    file_desc.flush()

    # Fork is default method to create a process
    # multiprocessing.set_start_method('fork')

    p = Process(target=process_task1)
    p.start()
    p.join()
    file_desc.close()

    file_des = open("sample.txt", "r")
    print(file_des.read())

    os.remove("sample.txt")


Output:
Written by parent process with id 288
Written by child process with id 294

Python AsyncIO

Ref: https://realpython.com/async-io-python/

• Threading Vs Multi-Processing
• Threading is better for I/O based tasks
• Multi-Processing is better for CPU based tasks
• What’s important to know about threading is that it’s better for IO-bound tasks.
• Concurrency Vs Parallelism
• Concurrency is when two tasks can start, run, and complete in overlapping time periods. e.g., Threading, AsyncIO
• Parallelism is when tasks literally run at the same time, eg. multi-processing. 
• While a CPU-bound task is characterised by the computer’s cores continually working hard from start to finish, an IO-bound job is dominated by a lot of waiting on input/output to complete.
• Preemptive multitasking Vs Cooperative multitasking
• OS preempts a thread forcing it to give up the use of CPU (E.g., Threading)
• Cooperative multitasking on the other hand, the running process voluntarily gives up the CPU to other processes E.g., (AsyncIO)
• Coroutine Vs Method/Function/Subroutine
• Method or Function returns a value and don't remember the state between invocations
• A coroutine is a special function that can give up control to its caller without losing its state
• Coroutine Vs Generator
• Generator yield back value to invoker
• Coroutine yields control to another coroutine and can resume execution from point it gave the control
• A generator can't accept arguments once it is started where as a coroutine can accept arguments once it started
• AsyncIO is a single-threaded, single-process design: it uses cooperative multitasking
• AsyncIO gives a feeling of concurrency despite using a single thread in a single process
• Coroutines (a central feature of async IO) can be scheduled concurrently, but they are not inherently concurrent.
• Asynchronous routines are able to “pause” while waiting on their ultimate result and let other routines run in the meantime.

import asyncio

import time

async def count_func():

    print("Line One")

    await asyncio.sleep(1) # await non-blocking call

    print("Line Two")

async def main():

    await asyncio.gather(count_func(), count_func(), count_func())

if __name__ == "__main__":

    t1 = time.time()

    asyncio.run(main())

    elapsed = time.time() - t1

    #This is supposed to take more than 3 secs

    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

Output:

#####

Line One

Line One

Line One

Line Two

Line Two

Line Two

main.py executed in 1.10 seconds.

Python concurrent.futures ProcessPoolExecutor

"""
The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. 

ProcessPoolExecutor uses the multiprocessing module
"""

from concurrent.futures import ProcessPoolExecutor
import math
import multiprocessing
import os
import sys
import time

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    109972689928541]

def is_prime(n):
    if n == 2:

        return True

    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    print('No of CPUs/Processors: {}' . format(multiprocessing.cpu_count()))
    a = time.time()
    #default max_workers is number of processors on the machine
    with ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))
    b = time.time()
    print('Time taken: {:.2f} secs'.format(b-a))

if __name__ == '__main__':
    main()


Output:
No of CPUs/Processors: 4
112272535095293 is prime: True
112582705942171 is prime: True
112272535095293 is prime: True
115280095190773 is prime: True
115797848077099 is prime: True
109972689928541 is prime: False
Time taken: 12.67 secs

Python concurrent.futures ThreadPoolExecutor as_completed

import urllib.request 
from concurrent.futures import ThreadPoolExecutor, as_completed

URLS = ['https://www.google.com', 
               'http://www.cnn.com/',
               'http://europe.wsj.com/', 
               'http://www.bbc.co.uk/', 
               'http://abc.abc.com'   #invalid
             ]

def load_url(url, timeout):
  with urllib.request.urlopen(url, timeout=timeout) as conn:
    txt = conn.read()
    return txt

with ThreadPoolExecutor(max_workers = 5) as executor:
  #Forming Key-Value pairs
  future_to_url = {executor.submit(load_url, url, 50): url for url in URLS}
  print(future_to_url)
  print('----')
  for future in as_completed(future_to_url):
    url = future_to_url[future]
    try:
      data = future.result()
      print('%s length is %d' % (url, len(data)))
    except Exception as e:
      print('Error in URL: %s is %s' % (url, e))


Output:
{<Future at 0x7f40d262d2d0 state=running>: 'https://www.google.com', <Future at 0x7f40cadc4ad0 state=running>: 'http://www.cnn.com/', <Future at 0x7f40cadcd710 state=running>: 'http://europe.wsj.com/', <Future at 0x7f40cadcd410 state=running>: 'http://www.bbc.co.uk/', <Future at 0x7f40cade0a10 state=running>: 'http://abc.abc.com'}
----
Error in URL: http://abc.abc.com is <urlopen error [Errno -2] Name or servicenot known>
https://www.google.com length is 12571
http://www.cnn.com/ length is 1134562
http://europe.wsj.com/ length is 1006417
http://www.bbc.co.uk/ length is 311008

Python ThreadPoolExecutor submit

from concurrent.futures import ThreadPoolExecutor import threading def task(n): print("Processing {} - {}".format(n, threading.current_thread())) def main(): print("Starting ThreadPoolExecutor") with ThreadPoolExecutor(max_workers=3) as executor: future = executor.submit(task, (2)) future = executor.submit(task, (3)) future = executor.submit(task, (4)) print("All tasks complete") if __name__ == '__main__': main()

Output:

Starting ThreadPoolExecutor Processing 2 - <Thread(ThreadPoolExecutor-0_0, started daemon 140052642395904)> Processing 3 - <Thread(ThreadPoolExecutor-0_1, started daemon 140052634003200)> Processing 4 - <Thread(ThreadPoolExecutor-0_2, started daemon 140052625610496)> All tasks complete

Python ThreadPoolExecutor, map

import urllib.request 
from concurrent.futures import ThreadPoolExecutor
import threading

urls = [
  'http://www.python.org', 
  'http://www.python.org/about/',
  'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html',
  'http://www.python.org/doc/',
  'http://www.python.org/download/',
  'http://www.python.org/getit/',
  'http://www.python.org/community/',
  'https://wiki.python.org/moin/',
]

def fun(url):
  print(url, threading.current_thread())
  r = urllib.request.urlopen(url)
  return r

# make the Pool of workers
pool = ThreadPoolExecutor(4) 

results = pool.map(fun, urls)
real_results = list(results)
print('----')
print(real_results)


Output:
http://www.python.org <Thread(ThreadPoolExecutor-0_0, started daemon 139989036611328)>
http://www.python.org/about/ <Thread(ThreadPoolExecutor-0_1, started daemon 139988956083968)>
http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html <Thread(ThreadPoolExecutor-0_2, started daemon 139988947691264)>
http://www.python.org/doc/ <Thread(ThreadPoolExecutor-0_3, started daemon 139988939298560)>
http://www.python.org/download/ <Thread(ThreadPoolExecutor-0_1, started daemon 139988956083968)>
http://www.python.org/getit/ <Thread(ThreadPoolExecutor-0_3, started daemon 139988939298560)>
http://www.python.org/community/ <Thread(ThreadPoolExecutor-0_0, started daemon 139989036611328)>
https://wiki.python.org/moin/ <Thread(ThreadPoolExecutor-0_3, started daemon 139988939298560)>
----
[<http.client.HTTPResponse object at 0x7f51be3d7910>, <http.client.HTTPResponse object at 0x7f51be3c5f90>, <http.client.HTTPResponse object at 0x7f51be3d73d0>, <http.client.HTTPResponse object at 0x7f51be3d77d0>, <http.client.HTTPResponse object at 0x7f51be3b3b90>, <http.client.HTTPResponse object at 0x7f51be3c5610>, <http.client.HTTPResponse object at 0x7f51be3e1c90>, <http.client.HTTPResponse object at 0x7f51be3d7250>]

Python threads using Queue

#Ref:

#https://stackoverflow.com/questions/47900922/split-list-into-n-lists-and-assign-each-list-to-a-worker-in-multithreading

#https://pymotw.com/2/Queue/

#The Queue module provides a FIFO implementation suitable for multi-threaded programming. 

#It can be used to pass messages or other data between producer and consumer threads safely. 

#Locking is handled for the caller, so it is simple to have as many threads as you want working with the same Queue instance. 

#A Queue’s size (number of elements) may be restricted to throttle memory usage or processing.

#### 3 types of queues

# Basic FIFO Queue

# LIFO Queue

# Priority Queue

from queue import Queue, LifoQueue

from threading import Thread, current_thread

from time import sleep

first_names = ['Steve','Jane','Sara','Mary','Jack','tara','bobby']

q = Queue() #FIFO

lq = LifoQueue() #LifoQueue

num_threads = 3

def do_stuff(q):

    while True:

        print(q.get(), current_thread())

        sleep(1)

        q.task_done()

if __name__ == '__main__':

    print('------ FIFO - Basic ------')

    for x in first_names:

        q.put(x)

    

    for i in range(num_threads):

        worker = Thread(target=do_stuff, args=(q,))

        worker.start()

    q.join()

    

    print('------ LIFO - reverse order -----')

    for x in first_names:

        lq.put(x)

    

    for i in range(num_threads):

        worker = Thread(target=do_stuff, args=(lq,))

        worker.start()

    lq.join()

Output:

------ FIFO - Basic ------

Steve <Thread(Thread-1, started 140202362271488)>

Jane <Thread(Thread-2, started 140202353878784)>

Sara <Thread(Thread-3, started 140202345486080)>

Mary <Thread(Thread-1, started 140202362271488)>

Jack <Thread(Thread-2, started 140202353878784)>

tara <Thread(Thread-3, started 140202345486080)>

bobby <Thread(Thread-1, started 140202362271488)>

------ LIFO - reverse order -----

bobby <Thread(Thread-4, started 140202337093376)>

tara <Thread(Thread-5, started 140202328700672)>

Jack <Thread(Thread-6, started 140202320307968)>

Mary <Thread(Thread-4, started 140202337093376)>

Sara <Thread(Thread-5, started 140202328700672)>

Jane <Thread(Thread-6, started 140202320307968)>

Steve <Thread(Thread-4, started 140202337093376)>

Python multiple threads - how to join

from threading import Thread, active_count, current_thread
import time

def fun(val):
  for _ in range(5):
    print(val, current_thread())
    time.sleep(3)

threads = []
for i in range(1, 11):
   t = Thread(target=fun, args=(i*i,))
   threads.append(t)
   t.start()
   print("Current Threads count: %i." % active_count())

#Join threads
for t in threads:
    t.join()

print('bye')


Output:
######
1 <Thread(Thread-1, started 140645849839360)>
Current Threads count: 2.
4 <Thread(Thread-2, started 140645841446656)>
Current Threads count: 3.
9 <Thread(Thread-3, started 140645833053952)>
Current Threads count: 4.
16 <Thread(Thread-4, started 140645616318208)>
Current Threads count: 5.
25 <Thread(Thread-5, started 140645607925504)>
Current Threads count: 6.
36 <Thread(Thread-6, started 140645599532800)>
Current Threads count: 7.
49 <Thread(Thread-7, started 140645591140096)>
Current Threads count: 8.
64 <Thread(Thread-8, started 140645582747392)>
Current Threads count: 9.
81 <Thread(Thread-9, started 140645574354688)>
Current Threads count: 10.
100 <Thread(Thread-10, started 140645565961984)>
Current Threads count: 11.
4 <Thread(Thread-2, started 140645841446656)>
9 <Thread(Thread-3, started 140645833053952)>
16 <Thread(Thread-4, started 140645616318208)>
1 <Thread(Thread-1, started 140645849839360)>
100 <Thread(Thread-10, started 140645565961984)>
25 <Thread(Thread-5, started 140645607925504)>
36 <Thread(Thread-6, started 140645599532800)>
64 <Thread(Thread-8, started 140645582747392)>
49 <Thread(Thread-7, started 140645591140096)>
81 <Thread(Thread-9, started 140645574354688)>
4 <Thread(Thread-2, started 140645841446656)>
9 <Thread(Thread-3, started 140645833053952)>
16 <Thread(Thread-4, started 140645616318208)>
1 <Thread(Thread-1, started 140645849839360)>
100 <Thread(Thread-10, started 140645565961984)>
25 <Thread(Thread-5, started 140645607925504)>
36 <Thread(Thread-6, started 140645599532800)>
64 <Thread(Thread-8, started 140645582747392)>
81 <Thread(Thread-9, started 140645574354688)>
49 <Thread(Thread-7, started 140645591140096)>
4 <Thread(Thread-2, started 140645841446656)>
9 <Thread(Thread-3, started 140645833053952)>
16 <Thread(Thread-4, started 140645616318208)>
100 <Thread(Thread-10, started 140645565961984)>
1 <Thread(Thread-1, started 140645849839360)>
25 <Thread(Thread-5, started 140645607925504)>
36 <Thread(Thread-6, started 140645599532800)>
64 <Thread(Thread-8, started 140645582747392)>
81 <Thread(Thread-9, started 140645574354688)>
49 <Thread(Thread-7, started 140645591140096)>
4 <Thread(Thread-2, started 140645841446656)>
16 <Thread(Thread-4, started 140645616318208)>
9 <Thread(Thread-3, started 140645833053952)>
1 <Thread(Thread-1, started 140645849839360)>
100 <Thread(Thread-10, started 140645565961984)>
64 <Thread(Thread-8, started 140645582747392)>
25 <Thread(Thread-5, started 140645607925504)>
36 <Thread(Thread-6, started 140645599532800)>
81 <Thread(Thread-9, started 140645574354688)>
49 <Thread(Thread-7, started 140645591140096)>
bye

Python multi threading Vs multi processing

Python multi threading Vs multi processing

Ref:
https://medium.com/@nbosco/multithreading-vs-multiprocessing-in-python-c7dc88b50b5b

#Threading
#The Python threading module uses threads instead of processes. Threads run in the same unique memory heap.
#Whereas Processes run in separate memory heaps. This, makes sharing information harder with processes and object instances.
#One problem arises because threads use the same memory heap, multiple threads can write to the same location in the memory heap which is why the global interpreter lock(GIL) in CPython was created as a mutex to prevent it from happening.

#Multiprocessing

#The multiprocessing library uses separate memory space, multiple CPU cores, bypasses GIL limitations in CPython, child processes are killable(ex. function calls in program) and is much easier to use.

#Some caveats of the module are a larger memory footprint and IPC’s a little more complicated with more overhead.

import threading

#Threading
#The Python threading module uses threads instead of processes. Threads run in the same unique memory heap.
#Whereas Processes run in separate memory heaps. This, makes sharing information harder with processes and object instances.
#One problem arises because threads use the same memory heap, multiple threads can write to the same location in the memory heap which is why the global interpreter lock(GIL) in CPython was created as a mutex to prevent it from happening.

def calc_square(number):
    print('Square:' , number * number)

def calc_quad(number):
    print('Quad:' , number * number * number * number)

if __name__ == "__main__":
    number = 7
    thread1 = threading.Thread(target=calc_square, args=(number,))
    thread2 = threading.Thread(target=calc_quad, args=(number,))
    # Will execute both in parallel
    thread1.start()
    thread2.start()
    # Joins threads back to the parent process, which is this
    # program
    thread1.join()
    thread2.join()
    # This program reduces the time of execution by running tasks in parallel

import multiprocessing

#The multiprocessing library uses separate memory space, multiple CPU cores, bypasses GIL limitations in CPython, child processes are killable(ex. function calls in program) and is much easier to use.

#Some caveats of the module are a larger memory footprint and IPC’s a little more complicated with more overhead.

def calc_square(number):

    print('Square:' , number * number)

    result = number * number

    print(result)

def calc_quad(number):

    print('Quad:' , number * number * number * number)

if __name__ == "__main__":

    number = 7

    result = None

    p1 = multiprocessing.Process(target=calc_square, args=(number,))

    p2 = multiprocessing.Process(target=calc_quad, args=(number,))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

    # Wont print because processes run using their own memory location

    print(result)

Python Thread Functions

import time
import threading
from threading import Thread

def sleepFunc(i):
    print("Thread %s going to sleep for 5 seconds..." % threading.current_thread())
    time.sleep(5)
    print("Thread %i is awake now..." % i)


for i in range(10):
    th = Thread(target=sleepFunc, args=(i, ))
    th.start()
    print("Current Threads count: %i." % threading.active_count())


for thread in threading.enumerate():
    print("Thread name is %s ..." % thread.getName())

Python Thread concepts

from threading import Thread
import time

class abc(Thread):
def run(self):
for i in range(5):
print 'abc'
time.sleep(1)

class xyz(Thread):
def run(self):
for i in range(5):
print 'xyz'
time.sleep(1)

a = abc()
b = xyz()
a.start()
time.sleep(0.2)
b.start()

a.join()
b.join()

print 'bye'

python serial

The following example illustrates python serial communication / python read serial port 

What you need to have
Connect the device you want to read the SERIAL logs to the system
To know the port is working, cross check by opening the port and check the logs in TeraTerm
If TeraTerm already using the port, you cannot read the logs from SERIAL port, so make sure before executing the script, stop the TeraTerm

What the script will do
We use python module 'serial'
Create a serial object by passing port, baudrate, bytesize, parity etc.,
To continuously read the logs, use Python Thread concept

from serial import *
from threading import Thread

text_received = ''

def receiveSerialDataFromPort(ser):
    global text_received
    read_buffer = ''

    while True:
        read_buffer += ser.read(ser.inWaiting())
        if '\n' in read_buffer:
            text_received, read_buffer = read_buffer.split('\n')[-2:]
            print text_received

if __name__ ==  '__main__':
    ser = Serial(
        port='COM23',
        baudrate=230400,
        bytesize=EIGHTBITS,
        parity=PARITY_NONE,
        stopbits=STOPBITS_ONE,
        interCharTimeout=None
    )
    
    Thread(target=receiveSerialDataFromPort, args=(ser,)).start()