adding chapter 7 source code

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"))

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())

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())

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())

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")

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")

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")

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")

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")

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")

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")

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()

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()

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()

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")

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)
+
+

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)
+
+

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()
+

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)

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()