From f88527eef4175b8e8029f3d8cd6b123094044c3b Mon Sep 17 00:00:00 2001 From: muassif <32828975+muassif@users.noreply.github.com> Date: Wed, 31 Mar 2021 09:22:38 +0400 Subject: [PATCH] adding chapter 7 source code --- Chapter7/asyncio/asyncio1.py | 13 ++++ Chapter7/asyncio/asyncio2.py | 17 +++++ Chapter7/asyncio/asyncio3.py | 39 ++++++++++++ Chapter7/asyncio/asyncio_casestudy.py | 62 +++++++++++++++++++ Chapter7/multiprocessing/process1.py | 31 ++++++++++ Chapter7/multiprocessing/process2.py | 15 +++++ Chapter7/multiprocessing/process3.py | 28 +++++++++ Chapter7/multiprocessing/process4.py | 29 +++++++++ Chapter7/multiprocessing/process5.py | 26 ++++++++ Chapter7/multiprocessing/process6.py | 29 +++++++++ Chapter7/multiprocessing/process7.py | 18 ++++++ .../multiprocessing/processes_casestudy.py | 55 ++++++++++++++++ Chapter7/multithreading/thread1-extra.py | 27 ++++++++ Chapter7/multithreading/thread1.py | 30 +++++++++ Chapter7/multithreading/thread2.py | 29 +++++++++ Chapter7/multithreading/thread3a.py | 27 ++++++++ Chapter7/multithreading/thread3b.py | 27 ++++++++ Chapter7/multithreading/thread4.py | 34 ++++++++++ .../multithreading/thread_casestudy-extra.py | 22 +++++++ Chapter7/multithreading/threads_casestudy.py | 60 ++++++++++++++++++ 20 files changed, 618 insertions(+) create mode 100644 Chapter7/asyncio/asyncio1.py create mode 100644 Chapter7/asyncio/asyncio2.py create mode 100644 Chapter7/asyncio/asyncio3.py create mode 100644 Chapter7/asyncio/asyncio_casestudy.py create mode 100644 Chapter7/multiprocessing/process1.py create mode 100644 Chapter7/multiprocessing/process2.py create mode 100644 Chapter7/multiprocessing/process3.py create mode 100644 Chapter7/multiprocessing/process4.py create mode 100644 Chapter7/multiprocessing/process5.py create mode 100644 Chapter7/multiprocessing/process6.py create mode 100644 Chapter7/multiprocessing/process7.py create mode 100644 Chapter7/multiprocessing/processes_casestudy.py create mode 100644 Chapter7/multithreading/thread1-extra.py create mode 100644 Chapter7/multithreading/thread1.py create mode 100644 Chapter7/multithreading/thread2.py create mode 100644 Chapter7/multithreading/thread3a.py create mode 100644 Chapter7/multithreading/thread3b.py create mode 100644 Chapter7/multithreading/thread4.py create mode 100644 Chapter7/multithreading/thread_casestudy-extra.py create mode 100644 Chapter7/multithreading/threads_casestudy.py diff --git a/Chapter7/asyncio/asyncio1.py b/Chapter7/asyncio/asyncio1.py new file mode 100644 index 0000000..7145b11 --- /dev/null +++ b/Chapter7/asyncio/asyncio1.py @@ -0,0 +1,13 @@ +#asyncio1.py to build a basic coroutine +import asyncio +import time + + +async def say(delay, msg): + await asyncio.sleep(delay) + print(msg) + +print("Started at ", time.strftime("%X")) +asyncio.run(say(1,"Good")) +asyncio.run(say(2, "Morning")) +print("Stopped at ", time.strftime("%X")) diff --git a/Chapter7/asyncio/asyncio2.py b/Chapter7/asyncio/asyncio2.py new file mode 100644 index 0000000..deafef7 --- /dev/null +++ b/Chapter7/asyncio/asyncio2.py @@ -0,0 +1,17 @@ +#asyncio2.py to build and run coroutines in parallel +import asyncio +import time + +async def say(delay, msg): + await asyncio.sleep(delay) + print(msg) + +async def main (): + task1 = asyncio.create_task( say(1, 'Good')) + task2 = asyncio.create_task( say(1, 'Morning')) + print("Started at ", time.strftime("%X")) + await task1 + await task2 + print("Stopped at ", time.strftime("%X")) + +asyncio.run(main()) \ No newline at end of file diff --git a/Chapter7/asyncio/asyncio3.py b/Chapter7/asyncio/asyncio3.py new file mode 100644 index 0000000..b92e174 --- /dev/null +++ b/Chapter7/asyncio/asyncio3.py @@ -0,0 +1,39 @@ +#asyncio3.py to distribute work via queue +import asyncio +import random +import time + +async def executer(name, queue): + while True: + exec_time = await queue.get() + await asyncio.sleep(exec_time) + queue.task_done() + print(f'{name} has taken {exec_time:.2f} seconds') + +async def main (): + + myqueue = asyncio.Queue() + calc_exuection_time = 0 + for _ in range(10): + sleep_for = random.uniform(0.4, 0.8) + calc_exuection_time += sleep_for + myqueue.put_nowait(sleep_for) + + tasks = [] + for id in range(3): + task = asyncio.create_task(executer(f'Task-{id+1}', myqueue)) + tasks.append(task) + + start_time = time.monotonic() + await myqueue.join() + total_exec_time = time.monotonic() - start_time + + for task in tasks: + task.cancel() + + await asyncio.gather(*tasks, return_exceptions=True) + + print(f"Calculated execution time {calc_exuection_time:0.2f}") + print(f"Actual execution time {total_exec_time:0.2f}") + +asyncio.run(main()) \ No newline at end of file diff --git a/Chapter7/asyncio/asyncio_casestudy.py b/Chapter7/asyncio/asyncio_casestudy.py new file mode 100644 index 0000000..02ade69 --- /dev/null +++ b/Chapter7/asyncio/asyncio_casestudy.py @@ -0,0 +1,62 @@ +#asyncio_casestudy.py +import asyncio +import time + +import aiofiles, aiohttp +from getfilelistpy import getfilelist + +TASK_POOL_SIZE = 10 + +#update the resource object as per your API key and the gdriver folder id +resource = { + "api_key": "AIzaSyDYKmm85keqnk41DpYa2bxddKrGns4z0", + "id": "0B8TxHW2Ci6dbckVweTRlV3RUU", + "fields": "files(name,id,webContentLink)", +} + +async def mydownloader(name, queue): + while True: + # Get the file id and name from the queue + item = await queue.get() + try: + async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) as session: + async with session.get(item['webContentLink']) as resp: + if resp.status == 200: + f = await aiofiles.open('./files/{}'.format( + item['name']), mode='wb') + await f.write(await resp.read()) + await f.close() + finally: + print(f"{name}: Download completed for ",item['name']) + queue.task_done() + +def get_files(resource): + res = getfilelist.GetFileList(resource) + files_list = res['fileList'][0] + return files_list + +async def main (): + files = get_files(resource) + #add files info into the queue + myqueue = asyncio.Queue() + for item in files['files']: + myqueue.put_nowait(item) + + tasks = [] + for id in range(TASK_POOL_SIZE): + task = asyncio.create_task( + mydownloader(f'Task-{id+1}', myqueue)) + tasks.append(task) + + start_time = time.monotonic() + await myqueue.join() + total_exec_time = time.monotonic() - start_time + + for task in tasks: + task.cancel() + + await asyncio.gather(*tasks, return_exceptions=True) + + print(f'Time taken to download: {total_exec_time:.2f} seconds') + +asyncio.run(main()) diff --git a/Chapter7/multiprocessing/process1.py b/Chapter7/multiprocessing/process1.py new file mode 100644 index 0000000..92528c9 --- /dev/null +++ b/Chapter7/multiprocessing/process1.py @@ -0,0 +1,31 @@ +# process1.py to create simple processes with function +import os +from multiprocessing import Process, current_process as cp +from time import sleep + +def print_hello(): + sleep(2) + print("{}-{}: Hello".format(os.getpid(), cp().name)) + + +def print_message(msg): + sleep(1) + print("{}-{}: {}".format(os.getpid(), cp().name, msg)) + +processes = [] + +# creating process +processes.append(Process(target=print_hello, name="Process 1")) +processes.append(Process(target=print_hello, name="Process 2")) +processes.append(Process(target=print_message, + args=["Good morning"], name="Process 3")) + +# start the process +for p in processes: + p.start() + +# wait till all are done +for p in processes: + p.join() + +print("Exiting the main process") diff --git a/Chapter7/multiprocessing/process2.py b/Chapter7/multiprocessing/process2.py new file mode 100644 index 0000000..6ea5233 --- /dev/null +++ b/Chapter7/multiprocessing/process2.py @@ -0,0 +1,15 @@ +# process2.py to create processes using a pool +import os +from multiprocessing import Process, Pool, current_process as cp +from time import sleep + +def print_message(msg): + sleep(1) + print("{}-{}: {}".format(os.getpid(), cp().name, msg)) + +# creating process from a pool +with Pool(3) as proc: + proc.map(print_message, ["Orange", "Apple", "Banana", + "Grapes","Pears"]) + +print("Exiting the main process") diff --git a/Chapter7/multiprocessing/process3.py b/Chapter7/multiprocessing/process3.py new file mode 100644 index 0000000..23a18a0 --- /dev/null +++ b/Chapter7/multiprocessing/process3.py @@ -0,0 +1,28 @@ +# process3.py to use shared memory ctype objects + +import multiprocessing +from multiprocessing import Process, Pool, current_process as cp + +def inc_sum_list(list, inc_list, sum): + sum.value = 0 + + for index, num in enumerate(list): + inc_list[index] = num + 1 + sum.value = sum.value + inc_list[index] + + +mylist = [2, 5, 7] + +inc_list = multiprocessing.Array('i', 3) + +sum = multiprocessing.Value('i') + +p = Process(target=inc_sum_list, + args=(mylist, inc_list, sum)) + +p.start() +p.join() +print("incremented list: ", list(inc_list)) +print("sum of inc list: ", sum.value) + +print("Exiting the main process") diff --git a/Chapter7/multiprocessing/process4.py b/Chapter7/multiprocessing/process4.py new file mode 100644 index 0000000..58e293f --- /dev/null +++ b/Chapter7/multiprocessing/process4.py @@ -0,0 +1,29 @@ +# process4.py to use shared memory using the server process + +import multiprocessing +from multiprocessing import Process + +def insert_data (dict1, code, subject): + dict1[code] = subject + +def output(dict1): + print("Dictionary data: ", dict1) + +with multiprocessing.Manager() as mgr: + # create a dictionary in the server process + mydict = mgr.dict({100: "Maths", 200: "Science"}) + + p1 = Process(target=insert_data, args=(mydict, 300, "English")) + p2 = Process(target=insert_data, args=(mydict, 400, "French")) + p3 = Process(target=output, args=(mydict,)) + + p1.start() + p2.start() + + p1.join() + p2.join() + + p3.start() + p3.join() + +print("Exiting the main process") diff --git a/Chapter7/multiprocessing/process5.py b/Chapter7/multiprocessing/process5.py new file mode 100644 index 0000000..7fb189a --- /dev/null +++ b/Chapter7/multiprocessing/process5.py @@ -0,0 +1,26 @@ +# process5.py to use queue to exchange data + +from multiprocessing import Process +from multiprocessing import Queue + +def copy_data (list, myqueue): + for num in list: + myqueue.put(num) + +def output(myqueue): + while not myqueue.empty(): + print(myqueue.get()) + +mylist = [2, 5, 7] +myqueue = Queue() + +p1 = Process(target=copy_data, args=(mylist, myqueue)) +p2 = Process(target=output, args=(myqueue,)) + +p1.start() +p1.join() +p2.start() +p2.join() + +print("Queue is empty: ",myqueue.empty()) +print("Exiting the main process") diff --git a/Chapter7/multiprocessing/process6.py b/Chapter7/multiprocessing/process6.py new file mode 100644 index 0000000..d3a44bd --- /dev/null +++ b/Chapter7/multiprocessing/process6.py @@ -0,0 +1,29 @@ +# process6.py to use Pipe to exchange data + +from multiprocessing import Process, Pipe + +def mysender (s_conn): + s_conn.send({100: "Maths"}) + s_conn.send({200: "Science"}) + s_conn.send("BYE") + s_conn.close() + +def myreceiver(r_conn): + while True: + msg = r_conn.recv() + if msg == "BYE": + break + print("Received message : ", msg) + r_conn.close() +sender_conn, receiver_conn= Pipe() + +p1 = Process(target=mysender, args=(sender_conn, )) +p2 = Process(target=myreceiver, args=(receiver_conn,)) + +p1.start() +p2.start() + +p1.join() +p2.join() + +print("Exiting the main process") diff --git a/Chapter7/multiprocessing/process7.py b/Chapter7/multiprocessing/process7.py new file mode 100644 index 0000000..91a635b --- /dev/null +++ b/Chapter7/multiprocessing/process7.py @@ -0,0 +1,18 @@ +# process7.py to show synchronization and locking +from functools import partial +from multiprocessing import Pool, Manager + +def printme (lock, msg): + lock.acquire() + try: + print(msg) + finally: + lock.release() + +with Pool(3) as proc: + lock = Manager().Lock() + func = partial(printme,lock) + proc.map(func, ["Orange", "Apple", "Banana", + "Grapes","Pears"]) + +print("Exiting the main process") diff --git a/Chapter7/multiprocessing/processes_casestudy.py b/Chapter7/multiprocessing/processes_casestudy.py new file mode 100644 index 0000000..247c92d --- /dev/null +++ b/Chapter7/multiprocessing/processes_casestudy.py @@ -0,0 +1,55 @@ +#processes_casestudy.py + +import time +from multiprocessing import Process, JoinableQueue +from getfilelistpy import getfilelist +import gdown + +PROCESSES_POOL_SIZE = 5 + +#update the resource object as per your API key and the gdriver folder id +resource = { + "api_key": "AIzaSyDYKmm85keqnk4bDpYa2bxddKrGns4z0", + "id": "0B8TxHW2Ci6dbckVweTRtV3RUU", + "fields": "files(name,id,webContentLink)", +} + +def mydownloader( queue): + while True: + # Get the file id and name from the queue + item1 = queue.get() + try: + gdown.download(item1['webContentLink'], + './files/{}'.format(item1['name']), + quiet=False) + finally: + queue.task_done() + + +def get_files(resource): + res = getfilelist.GetFileList(resource) + files_list = res['fileList'][0] + return files_list + +def main (): + + files = get_files(resource) + #add files info into the queue + myqueue = JoinableQueue() + for item in files['files']: + myqueue.put(item) + + processes = [] + for id in range(PROCESSES_POOL_SIZE): + p = Process(target=mydownloader, + args=(myqueue,)) + p.daemon = True + p.start() + + start_time = time.monotonic() + myqueue.join() + total_exec_time = time.monotonic() - start_time + + print(f'Time taken to download: {total_exec_time:.2f} seconds') + +main() diff --git a/Chapter7/multithreading/thread1-extra.py b/Chapter7/multithreading/thread1-extra.py new file mode 100644 index 0000000..b1f745b --- /dev/null +++ b/Chapter7/multithreading/thread1-extra.py @@ -0,0 +1,27 @@ +# thread1-extra.py to create simple threads with function (using threads list to simplly the code) + +import threading +from threading import Thread as Thread +from time import sleep + +def print_hello(): + sleep(2) + print("{}: Hello".format(threading.current_thread().name)) + +def print_message(msg): + sleep(1) + print("{}: {}".format(threading.current_thread().name, msg)) + +threads = [] +# creating threads +threads.append(Thread(target=print_hello, name="Th 1")) +threads.append(Thread(target=print_hello, name="Th 2")) +threads.append(Thread(target=print_message, args=["Good morning"], name="Th 3")) + +# start the threads +for th in threads: + th.start() + +# wait till all are done +for th in threads: + th.join() diff --git a/Chapter7/multithreading/thread1.py b/Chapter7/multithreading/thread1.py new file mode 100644 index 0000000..b5b06e0 --- /dev/null +++ b/Chapter7/multithreading/thread1.py @@ -0,0 +1,30 @@ +# thread1.py to create simple threads with function + +from threading import current_thread, Thread as Thread +from time import sleep + + +def print_hello(): + sleep(2) + print("{}: Hello".format(current_thread().name)) + + +def print_message(msg): + sleep(1) + print("{}: {}".format(current_thread().name, msg)) + + +# creating threads +t1 = Thread(target=print_hello, name="Th 1") +t2 = Thread(target=print_hello, name="Th 2") +t3 = Thread(target=print_message, args=["Good morning"], name="Th 3") + +# start the threads +t1.start() +t2.start() +t3.start() + +# wait till all are done +t1.join() +t2.join() +t3.join() diff --git a/Chapter7/multithreading/thread2.py b/Chapter7/multithreading/thread2.py new file mode 100644 index 0000000..ed8db40 --- /dev/null +++ b/Chapter7/multithreading/thread2.py @@ -0,0 +1,29 @@ +# thread2.py to create daemon and non-daemon threads + +from threading import current_thread, Thread as Thread +from time import sleep + +def daeom_func(): + #print(threading.current_thread().isDaemon()) + sleep(3) + print("{}: Hello from daemon".format + (current_thread().name)) + + +def nondaeom_func(): + #print(threading.current_thread().isDaemon()) + sleep(1) + print("{}: Hello from non-daemon".format( + current_thread().name)) + + +# creating threads +t1 = Thread(target=daeom_func, name="Daemon Thread",daemon=True) +t2 = Thread(target=nondaeom_func, name="Non-Daemon Thread") + +# start the threads +t1.start() +t2.start() + + +print("Exiting the main program") \ No newline at end of file diff --git a/Chapter7/multithreading/thread3a.py b/Chapter7/multithreading/thread3a.py new file mode 100644 index 0000000..d446308 --- /dev/null +++ b/Chapter7/multithreading/thread3a.py @@ -0,0 +1,27 @@ +# thread3a.py when no thread synchronization used + +from threading import Thread as Thread + +def inc(): + global x + for _ in range(1000000): + x+=1 + +#global variable +x = 0 + +# creating threads +t1 = Thread(target=inc, name="Th 1") +t2 = Thread(target=inc, name="Th 2") + +# start the threads +t1.start() +t2.start() + +#wait for the threads +t1.join() +t2.join() + +print("final value of x :", x) + + diff --git a/Chapter7/multithreading/thread3b.py b/Chapter7/multithreading/thread3b.py new file mode 100644 index 0000000..393ea78 --- /dev/null +++ b/Chapter7/multithreading/thread3b.py @@ -0,0 +1,27 @@ +# thread3b.py when thread synchronization is used + +from threading import Lock, Thread as Thread + +def inc_with_lock (lock): + global x + for _ in range(1000000): + lock.acquire() + x+=1 + lock.release() + +x = 0 +mylock = Lock() +# creating threads +t1 = Thread(target=inc_with_lock , args=(mylock,), name="Th 1") +t2 = Thread(target=inc_with_lock , args=(mylock,), name="Th 2") + +# start the threads +t1.start() +t2.start() + +#wait for the threads +t1.join() +t2.join() +print("final value of x :", x) + + diff --git a/Chapter7/multithreading/thread4.py b/Chapter7/multithreading/thread4.py new file mode 100644 index 0000000..8f4ddb8 --- /dev/null +++ b/Chapter7/multithreading/thread4.py @@ -0,0 +1,34 @@ +# thread4.py with queue and custom Thread class + +from queue import Queue +from threading import Thread as Thread +from time import sleep + +class MyWorker (Thread): + def __init__(self, name, q): + Thread.__init__(self) + self.name = name + self.queue = q + + def run(self): + while True: + item = self.queue.get() + sleep(1) + try: + print ("{}: {}".format(self.name, item)) + finally: + self.queue.task_done() + +#filling the queue +myqueue = Queue() +for i in range (10): + myqueue.put("Task {}".format(i+1)) + +# creating threads +for i in range (5): + worker = MyWorker("Th {}".format(i+1), myqueue) + worker.daemon = True + worker.start() + +myqueue.join() + diff --git a/Chapter7/multithreading/thread_casestudy-extra.py b/Chapter7/multithreading/thread_casestudy-extra.py new file mode 100644 index 0000000..93c864d --- /dev/null +++ b/Chapter7/multithreading/thread_casestudy-extra.py @@ -0,0 +1,22 @@ +# this is a code without using any threads +from time import time + +from getfilelistpy import getfilelist +import gdown + +resource = { + "api_key": "AIzaSyDYKmm85keqnk4bF1DpYa2bxddKrGns4z0", + "id": "0B8TxHW2Ci6dbckVweTRtTlV3RUU", + "fields": "files(name,id,webContentLink)", +} + +res = getfilelist.GetFileList(resource) +files_list = res['fileList'][0] + +t1 = time() +for item in files_list['files']: + gdown.download( item['webContentLink'], + './files/{}'.format(item['name']), + quiet=False) + +print('Time taken to download: %s seconds', time() - t1) \ No newline at end of file diff --git a/Chapter7/multithreading/threads_casestudy.py b/Chapter7/multithreading/threads_casestudy.py new file mode 100644 index 0000000..18986eb --- /dev/null +++ b/Chapter7/multithreading/threads_casestudy.py @@ -0,0 +1,60 @@ +#threads_casestudy.py +from queue import Queue +from threading import Thread +import time + +from getfilelistpy import getfilelist +import gdown +THREAD_POOL_SIZE = 5 + +#update the resource object as per your API key and the gdriver folder id +resource = { + "api_key": "AIzaSyDYKmm85keq4bF1DpYa2bxddKrGns4z0", + "id": "0B8TxHW2Ci6dbckVwtTlV3RUU", + "fields": "files(name,id,webContentLink)", +} + +class DownlaodWorker(Thread): + + def __init__(self, name, queue): + Thread.__init__(self) + self.name = name + self.queue = queue + + def run(self): + while True: + # Get the file id and name from the queue + item1 = self.queue.get() + try: + gdown.download( item1['webContentLink'], + './files/{}'.format(item1['name']), + quiet=False) + finally: + self.queue.task_done() + +def main(): + def get_files(resource): + #global files_list + res = getfilelist.GetFileList(resource) + files_list = res['fileList'][0] + return files_list + + start_time = time.monotonic() + + files = get_files(resource) + + #add files info into the queue + queue = Queue() + for item in files['files']: + queue.put(item) + + for i in range (THREAD_POOL_SIZE): + worker = DownlaodWorker("Thread {}".format(i+1),queue) + worker.daemon = True + worker.start() + + queue.join() + end_time = time.monotonic() + print('Time taken to download: {} seconds'. + format( end_time - start_time)) +main()