17. Multiprocessing and Multithreading.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Multiprocessing and Multithreading"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sections\n",
    "\n",
    "- [Multithreading](#Multithreading)\n",
    "- [Multiprocessing](#Multiprocessing)\n",
    "- [Pros and Cons](#Pros-and-Cons)\n",
    "- [Multiprocessing Example](#Multiprocessing-Example)\n",
    "- [Threading Example](#Threading-Example)\n",
    "- [Sharing Data with Processes Using Value](#Sharing-Data-with-Processes-Using-Value)\n",
    "- [Sharing Data with Processes Using Array](#Sharing-Data-with-Processes-Using-Array)\n",
    "- [Implementing Multiprocess Locks](#Implementing-Multiprocess-Locks)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Python is slow comparing to other programming languages.\n",
    "\n",
    "In a Python script execution:\n",
    "- When you type python script.py in your shell you instruct your processor to create and to schedule a single process which is the smallest unit of processing\n",
    "- The allocated process will start to execute the script line by line.\n",
    "- Once the script hit the EOF, the process will be terminated and its resources will be returned to the free pool to be used by other processes.\n",
    "\n",
    "We can overcome the Python speed problem using multiprocessing or multithreading.\n",
    "\n",
    "For example, with multithreding, we can have inside a process multiple threads.\n",
    "\n",
    "But the problem with assigning a lot of threads to one process without special handling is what’s called Race Condition.\n",
    "\n",
    "CPython uses GIL(Global Interpreter Lock) to protect memory amont multiple threads."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Multithreading\n",
    "\n",
    "Threading in python is used to run multiple threads (tasks, function calls) at the same time inside the same process.\n",
    "\n",
    "Thereding is not suitable for CPU intensive application\n",
    "\n",
    "Python threads are best used in cases where the execution of a task involves some waiting, for example working with web servers, networking devices, SQL databases.\n",
    "\n",
    "Python uses the threading module to start multiple threads."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Multiprocessing\n",
    "\n",
    "Multiprocessing achieves true parallelism in Python.\n",
    "\n",
    "Low risk of data-corruption when using multiprocessing.\n",
    "\n",
    "Each spawned process will have their own allocated memory.\n",
    "\n",
    "Each process has it’s owned GIL so there’s no resource conflict or race condition here.\n",
    "\n",
    "Python uses the multiprocessing module to achieve parallel programming.\n",
    "\n",
    "Multiptocessing is used in CPU intensive applications."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pros and Cons\n",
    "\n",
    "__Multiprocessing Pros__\n",
    "\n",
    "- Separate memory space.\n",
    "- Takes advantage of multiple CPUs & cores.\n",
    "- Avoids GIL limitations of CPython.\n",
    "- Child processes are interruptible/killable.\n",
    "- A must with CPython for CPU-bound processing.\n",
    "\n",
    "__Multiprocessing Cons__\n",
    "\n",
    "- Inter Process Communication (IPC) a little more complicated with more overhead.\n",
    "- Larger memory consumption.\n",
    "\n",
    "__Multithreading Pros__\n",
    "\n",
    "- Lightweight , low memory consumption.\n",
    "- Shared memory, makes access to state from another context easier.\n",
    "- Allows you to easily make responsive UIs.\n",
    "- Great option for I/O - bound applications.\n",
    "\n",
    "__Multithreading Cons__\n",
    "\n",
    "- CPython, subject to the GIL.\n",
    "- Threads are not interruptible/killable.\n",
    "- If not following a command queue/message model (using the Queue module), then manual use of synchronization becomes a necessity.\n",
    "- Code is usually harder to understand and to get right due to the potential of race conditions increases dramatically."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Multiprocessing Example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Other instructions of the main module...\n",
      "More instructions of the main module...\n",
      "End of Script\n"
     ]
    }
   ],
   "source": [
    "# Importing the module multiptocessing\n",
    "\n",
    "import multiprocessing as mp\n",
    "import time\n",
    " \n",
    "# Creating the target function. Each process will execute this function in parallel\n",
    "def name_and_time(name):\n",
    "    print(f\"Hello {name}, current time is {time.time()}\")\n",
    "    print(\"Sleeping for 2 seconds ...\")\n",
    "    time.sleep(2)\n",
    "    print(\"Walking up ... finishing function.\")\n",
    "    \n",
    "# This will be run only if the script is run directly (not imported as a module in another script)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    process_list = list()     # list that stores the processes\n",
    " \n",
    "    # Creating 5 processes\n",
    "    for i in range(5):\n",
    "        ## Creating each process. 1st argument is the target function\n",
    "        ## 2nd argument is a tuple (target function's arguments)\n",
    "        process = mp.Process(target=name_and_time, args=(\"Popeye\",))\n",
    "        process_list.append(process)    # Appending each process to the list\n",
    " \n",
    "    # Iterating over the list and start each process\n",
    "    for p in process_list:\n",
    "        p.start()\n",
    " \n",
    "    # Join the processes back to the main process OR\n",
    "    # The main process will wait for forked-processes to finish\n",
    "    for p in process_list:\n",
    "        p.join()\n",
    " \n",
    "    print(\"Other instructions of the main module...\")\n",
    "    print(\"More instructions of the main module...\")\n",
    "    print(\"End of Script\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Threading Example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hello Popeye, current time is 1606880190.5148027\n",
      "Sleeping for 5 seconds ...\n",
      "Hello Popeye, current time is 1606880190.517795\n",
      "Sleeping for 5 seconds ...\n",
      "Hello Popeye, current time is 1606880190.5207868\n",
      "Sleeping for 5 seconds ...\n",
      "\n",
      "Walking up ... finishing function.\n",
      "Walking up ... finishing function.\n",
      "\n",
      "Walking up ... finishing function.\n",
      "\n",
      "Other instructions of the main module...\n",
      "More instructions of the main module...\n",
      "End of Script\n"
     ]
    }
   ],
   "source": [
    "# Importing the module\n",
    "import threading\n",
    "import time\n",
    " \n",
    "# Creating the target function. Each thread will execute this function in parallel\n",
    "def name_and_time(name):\n",
    "    print(f\"Hello {name}, current time is {time.time()}\")\n",
    "    print(\"Sleeping for 5 seconds ...\")\n",
    "    time.sleep(5)\n",
    "    print(\"\\nWalking up ... finishing function.\")\n",
    "\n",
    "# This will be run only if the script is run directly \n",
    "if __name__ == '__main__':\n",
    "    thread_list = list()     # list that stores the threads\n",
    " \n",
    "    # Creating 3 threads\n",
    "    for i in range(3):\n",
    "        ## Creating each thread. 1st argument si the target function\n",
    "        ## 2nd argument is a tuple (target function's arguments)\n",
    "        thread = threading.Thread(target=name_and_time, args=(\"Popeye\",))\n",
    "        thread_list.append(thread)  # Appending each thread to the list\n",
    " \n",
    "    ## Iterating over the list and start each thread\n",
    "    for t in thread_list:\n",
    "        t.start()\n",
    " \n",
    " \n",
    "    ## Join the threads back to the main thread OR\n",
    "    ## The main thread will wait for forked-threads to finish\n",
    "    ## This is optional\n",
    "    for t in thread_list:\n",
    "        t.join()\n",
    " \n",
    "    print(\"Other instructions of the main module...\")\n",
    "    print(\"More instructions of the main module...\")\n",
    "    print(\"End of Script\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sharing Data with Processes Using Value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "counter of type multiprocessing.Value is 1\n",
      "my_counter of type integer is 1\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing as mp\n",
    " \n",
    "# Target function that increments a counter (multiprocessing.Value)\n",
    "def increment(counter):\n",
    "    counter.value += 1\n",
    " \n",
    "# Target function that increments a counter (integer)\n",
    "def my_increment(my_counter):\n",
    "    my_counter  += 1\n",
    "  \n",
    "if __name__ == '__main__':\n",
    "    my_counter = 1      # type integer\n",
    "    counter = mp.Value('i', 1)   # type multiprocessing.Value\n",
    " \n",
    "    # Creating, starting and joining 5 processes. They increment the counter. This is of type multiprocessing.Value\n",
    "    for i in range(5):\n",
    "        process = mp.Process(target=increment, args=(counter,))\n",
    "        process.start()\n",
    "        process.join()\n",
    " \n",
    "    print(f'counter of type multiprocessing.Value is {counter.value}')\n",
    " \n",
    " \n",
    "    # Creating, starting and joining 5 processes. They increment my_counter. This is of type integer\n",
    "    for i in range(5):\n",
    "        process = mp.Process(target=my_increment, args=(my_counter,))\n",
    "        process.start()\n",
    "        process.join()\n",
    " \n",
    "    print(f'my_counter of type integer is {my_counter}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sharing Data with Processes Using Array"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "squares_list outsite process []\n",
      "result Array outside process [0, 0, 0]\n"
     ]
    }
   ],
   "source": [
    "## Sharing data between processes using multiprocessing.Array\n",
    " \n",
    "import multiprocessing as mp\n",
    " \n",
    "# Target function1\n",
    "# The 2nd parameter is a list that will be modified inside the forked process\n",
    "def squares(numbers, squares_list):\n",
    "    for n in numbers:\n",
    "        squares_list.append(n**2)\n",
    "    print(f'square_list inside process {squares_list}')\n",
    " \n",
    "# Target function\n",
    "# The 2nd parameter is an Array used for sharing data between processes\n",
    "def cubes(numbers, result):\n",
    "    i = 0\n",
    "    for num in numbers:\n",
    "        result[i] = num ** 3\n",
    "        i += 1\n",
    "    print(f'result Array inside process/function: {result[::]}')\n",
    " \n",
    " \n",
    "# This will be run only if the script is run directly (not imported as a module in another script)\n",
    "# This is necessary!\n",
    "if __name__ == '__main__':\n",
    " \n",
    "    # Calculating the square of these numbers in a new process\n",
    "    # Adding the result (the squares) in a list inside a process\n",
    "    numbers = [1,2,3]\n",
    "    squares_list = list()\n",
    " \n",
    "    # Creating, Starting and Joining the process\n",
    "    p = mp.Process(target=squares, args=(numbers, squares_list))\n",
    "    p.start()\n",
    "    p.join()\n",
    " \n",
    "    # Printing the squares_list\n",
    "    # !! It hasn't been modified inside the process. The main process and the forked process didn't work on the same data\n",
    "    print(f'squares_list outsite process {squares_list}')\n",
    " \n",
    "    # Creating an Array to share data beetween main process and the forked process\n",
    "    result = mp.Array('i', len(numbers))\n",
    " \n",
    "    # Creating, Starting and Joining the process\n",
    "    p1 = mp.Process(target=cubes, args=(numbers, result))\n",
    "    p1.start()\n",
    "    p1.join()\n",
    " \n",
    "    # Printing the Array. It has been modified inside the forked-process. The array was shared between main and forked process\n",
    "    print(f'result Array outside process {result[::]}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Implementing Multiprocess Locks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Balance BEFORE running processes: 500\n",
      "Balance AFTER running processes: 500\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing as mp\n",
    "import time\n",
    " \n",
    "# Target function. It increments a balance by 0.01  100 times\n",
    "def deposit(balance, lock):\n",
    "    for i in range(100):\n",
    "        time.sleep(0.01)\n",
    "        lock.acquire()\n",
    "        balance.value += 1\n",
    "        lock.release()\n",
    " \n",
    "# Target function. It decrements a balance by 0.01  100 times\n",
    "def withdraw(balance, lock):\n",
    "    for i in range(100):\n",
    "        time.sleep(0.01)\n",
    "        lock.acquire()\n",
    "        balance.value -= 1\n",
    "        lock.release()\n",
    " \n",
    "if __name__ == '__main__':\n",
    "    balance = mp.Value('i', 500)    ## starting balance\n",
    "    print(f'Balance BEFORE running processes: {balance.value}')\n",
    " \n",
    "    lock = mp.Lock()    # lock Object\n",
    " \n",
    "    # Creating, starting and joining 2 processes. They increment and decrement the shared value\n",
    "    p1 = mp.Process(target=deposit, args=(balance, lock))\n",
    "    p2 = mp.Process(target=withdraw, args=(balance, lock))\n",
    " \n",
    "    p1.start()\n",
    "    p2.start()\n",
    " \n",
    "    p1.join()\n",
    "    p2.join()\n",
    " \n",
    "    # The final value of balance\n",
    "    print(f'Balance AFTER running processes: {balance.value}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Concurrent Applications using Asyn IO\n",
    "\n",
    "Async IO is a single threaded, single process design that uses cooperative multitasking.\n",
    "\n",
    "Cooperative  is no OS intervention, each task decides when to give up.\n",
    "\n",
    "Async IO is language agnostc, it is implemented in several laguages.\n",
    "\n",
    "Asyncio is the Python Standar Library that provides support for implementing Asinc IO. Avoids the memry issues from the hreading design.\n",
    "\n",
    "Paramiko that is very common for networking engineers, does not use Async IO, uses threading.\n",
    "\n",
    "AsyncSSH is the option for Asyncronous SSH that uses Async IO."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " one  two  three  one  two  three  one  two  three "
     ]
    }
   ],
   "source": [
    "# Sync code\n",
    "def sync_func():\n",
    "    print(\" one \", end=\"\")\n",
    "    time.sleep(1)\n",
    "    print(\" two \", end=\"\")\n",
    "    time.sleep(1)\n",
    "    print(\" three \", end=\"\")\n",
    "\n",
    "for _ in range(3):\n",
    "    sync_func()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Async code\n",
    "async def async_func():\n",
    "    print(\"one\", end=\"\")\n",
    "    await asyncio.sleep(1)\n",
    "    print(\"two\", end=\"\")\n",
    "    await asyncio.sleep(1)\n",
    "    print(\"three\", end=\"\")\n",
    "    \n",
    "async def main():\n",
    "    #  task = [async_f(), async_f(), asyncio] \n",
    "    task = [async_f() for _ in range(3)]\n",
    "\n",
    "    await asyncio.gather(*tasks)\n",
    "\n",
    "# It seems it seams something else to run on Jupyter Notebook\n",
    "# asyncio.run(main())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Example for iterating with the shell of multiple devices\n",
    "import asyncio\n",
    "\n",
    "async def run(cmd):\n",
    "    proc = await asyncio.create_subprocess-shell(cmd, stdout=asyncio.subprocess.PIP, stderr=asyncio.subprocess.PIPE)\n",
    "    stdout, stderr = await proc.communicate()\n",
    "    \n",
    "    print(f\"{cmd} exited with status code. {proc.returncode}\")\n",
    "    \n",
    "    if stdout:\n",
    "        print(f\"STDOUT:\\n{stdout.decode(\"UTF-8\")}\")\n",
    "              \n",
    "    if stderr:\n",
    "        print(f\"STDERR:\\n{stderr.decode(\"UTF-8\")}\")\n",
    "    \n",
    "async def main(commads):\n",
    "    tasks = []\n",
    "    for cmd in commands:\n",
    "        tasks.append(run(cmd))\n",
    "    \n",
    "    await asyncio.gather(*tasks)\n",
    "\n",
    "commands = (\"ifconfig\", \"ls\", \"who\", \"uname -a\")\n",
    "asyncio.run(main(commands))\n",
    "              "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### References for Async IO\n",
    "\n",
    "[Async IO in Python: A Complete Walkthrough (by Real Python)](https://realpython.com/async-io-python/)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### AsyncSSH\n",
    "\n",
    "Requires the instalation of an external library"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "import asyncio\n",
    "# import asyncssh"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# One command, one client\n",
    "\n",
    "async def connect_and_run(host, username, password, command):\n",
    "    async with async.connect(host=host, username0username, password=password, known-hosts=None) as connection:\n",
    "        result = await conection.run(command)\n",
    "        return result\n",
    "\n",
    "command = \"ifconfig\"\n",
    "result = async.run(connect_and_run(\"192.168.0.50\", \"student\", \"student\", command))\n",
    "print(f\"STDOUT:\\n {result.stdout}\")\n",
    "print(f\"STDERR:\\n {result.stderr}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Multiple commands, one client\n",
    "\n",
    "async def connect_and_run(host, username, password, commands):\n",
    "    async with async.connect(host=host, username0username, password=password, known-hosts=None) as connection:\n",
    "    \n",
    "    results = []\n",
    "    \n",
    "    for cmd in commands:\n",
    "        result = await conection.run(comd)\n",
    "        results.append(result)\n",
    "\n",
    "commands = (\"ifconfig\", \"who -a\", \"uname -a\")\n",
    "results = async.run(connect_and_run(\"192.168.0.50\", \"student\", \"student\", commands))\n",
    "for result in results:\n",
    "    print(f\"STDOUT:\\n {result.stdout}\")\n",
    "    print(f\"STDERR:\\n {result.stderr}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Multiple commands, multiple client\n",
    "\n",
    "async def connect_and_run(host, username, password, command):\n",
    "    async with async.connect(host=host, username0username, password=password, known-hosts=None) as connection:\n",
    "        return await connection.run(command)\n",
    "        \n",
    "async def run_multiple_client(hosts):\n",
    "    tasks = []\n",
    "    for host in hosts:\n",
    "        task = run_client(host[\"host\"], host[\"username\"], host[\"password\"], host[\"command\"])\n",
    "        tasks.append(task)\n",
    "    \n",
    "    result = await asyncio.gather(*taks, return_exceptions=True)\n",
    "\n",
    "    i = 0\n",
    "    for result in results:\n",
    "        i += 1\n",
    "        if isinstance(result, Exception):\n",
    "            print(f\"Task {i} failed. {str(result)}\"\")\n",
    "        elif result.exit_status != 0:\n",
    "            print(f\"taks {i} existed with status. {result.exit_status}\")\n",
    "            print(result.stderr, end=\"\")\n",
    "        else:\n",
    "            print(f\"Task {i} succeeded:\")\n",
    "            print(result.stdout, end=\"\")\n",
    "        print(\"#\" * 50)\n",
    "    \n",
    "hosts = [\n",
    "    {\"host\": \"192.168.1.1\", \"username\": \"student\", \"password\": \"student\", \"command\": \"ifconfif\"},\n",
    "    {\"host\": \"192.168.1.2\", \"username\": \"student\", \"password\": \"student\", \"command\": \"who\"},\n",
    "    {\"host\": \"192.168.1.3\", \"username\": \"student\", \"password\": \"student\", \"command\": \"uname -a\"},    \n",
    "]\n",
    "                  \n",
    "asyncio.run(run_multile-clients(hosts))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}