Mutltithreading_Multiprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e4a20b58",
   "metadata": {},
   "source": [
    "# Mutitasking in python"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1bb24e71",
   "metadata": {},
   "source": [
    "## Multithreading"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8fba946d",
   "metadata": {},
   "outputs": [],
   "source": [
    "Example1 - "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c8048c39",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "import threading\n",
    "  \n",
    "def cube(num):\n",
    "\n",
    "    return \"Cube: {}\".format(num * num * num)\n",
    "\n",
    "def square(num):\n",
    "\n",
    "    return \"Square: {}\".format(num * num)\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    # creating thread\n",
    "    t1 = threading.Thread(target=square, args=(10,))\n",
    "    t2 = threading.Thread(target=cube, args=(10,))\n",
    "  \n",
    "    # starting thread 1\n",
    "    t1.start()\n",
    "    # starting thread 2\n",
    "    t2.start()\n",
    "    \n",
    "  \n",
    "    # both threads completely executed\n",
    "    print(\"Done!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c2ae3e7a",
   "metadata": {},
   "outputs": [],
   "source": [
    "https://docs.python.org/3/library/threading.html#module-threading"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8480c86f",
   "metadata": {},
   "outputs": [],
   "source": [
    "Example 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "91f23974",
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "\n",
    "class Addition(threading.Thread):\n",
    "    def __init__(self, a, b, name=\"Addition\"):\n",
    "        super().__init__() \n",
    "        self.name = name\n",
    "        self.a = a\n",
    "        self.b = b\n",
    "\n",
    "    def run(self):\n",
    "        time.sleep(3)\n",
    "        print(\"Addition of {} & {} is {}\".format(self.a, self.b, self.a + self.b))\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "\n",
    "    thread1 = Addition(10,20, name=\"Addition1\")\n",
    "    thread2 = Addition(20,20, name=\"Addition2\")\n",
    "\n",
    "    thread1.start()\n",
    "    thread2.start()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "db9860e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "### Some methods and attributes of thread"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "08593710",
   "metadata": {},
   "source": [
    "List of methods:\n",
    "\n",
    "- threading.current_thread() - This method returns Thread instance which represents current thread which is running the code.\n",
    "- threading.main_thread() - This method returns Thread instance which represents the main thread of the process which spawned other threads.\n",
    "- threading.active_count() - This method returns integer representing number of threads which are currently alive.\n",
    "- threading.get_ident() - This method returns integer which represents ID of the thread in which its called.\n",
    "- threading.get_native_id() - This method returns integer which represents ID of thread assigned by underlying OS."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46b00484",
   "metadata": {},
   "source": [
    "List of new attributes:\n",
    "\n",
    "- Thread.name - It returns the name of the thread.\n",
    "- Thread.daemon - It returns a boolean value representing whether the thread is a daemon or not. We can even set this attribute and the thread will become a daemon thread from the normal thread but it needs to be set before starting the thread (by calling start() method).\n",
    "- Thread.ident - This attribute returns an integer that represents the ID of the thread in which it’s called.\n",
    "- Thread.native_id - This attribute returns an integer that represents the ID of the thread assigned by the underlying OS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a31136e8",
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "\n",
    "def addition(a, b):\n",
    "    curr_thread = threading.current_thread()\n",
    "    print(\"Thread Name : {} (Daemon : {}), Thread Identifier : {}, Native Identifier : {}\".\n",
    "                 format(curr_thread.name, curr_thread.daemon, curr_thread.ident, curr_thread.native_id))\n",
    "    time.sleep(2)\n",
    "\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "    print(\"\\n\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    thread1 = threading.Thread(target=addition, args=(10,20), name=\"Addition1\", )\n",
    "    thread2 = threading.Thread(target=addition, kwargs={\"a\": 20, \"b\": 20}, name=\"Addition2\", )\n",
    "\n",
    "    thread1.start()\n",
    "    thread2.start()\n",
    "\n",
    "    main_thread = threading.main_thread()\n",
    "    print(\"Thread Name : {} (Daemon : {}), Thread Identifier : {}, Native Identified : {}\".\n",
    "                 format(main_thread.name, main_thread.daemon, threading.get_ident(), threading.get_native_id()))\n",
    "\n",
    "    print(\"Currently Active Threads Count (Including Main Thread ): {}\".format(threading.active_count()))\n",
    "    print(\"End   Time : {}\".format(datetime.now()))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "22acd4cc",
   "metadata": {},
   "outputs": [],
   "source": [
    "## Wait for single thread to complete"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "e22e97a0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start Time : 2022-05-04 16:32:49.899477\n",
      "Thread Name : Addition1 (Daemon : False), Thread Identifier : 123145450360832, Native Identifier : 731967\n",
      "Thread Name : Addition2 (Daemon : False), Thread Identifier : 123145483939840, Native Identifier : 731968Thread Name : MainThread (Daemon : False), Thread Identifier : 4541033920, Native Identified : 682843\n",
      "\n",
      "Currently Active Threads Count (Including Main Thread ): 11\n",
      "Addition of 10 & 20 is 30\n",
      "End   Time : 2022-05-04 16:32:51.906958\n",
      "Addition of 20 & 20 is 40\n"
     ]
    }
   ],
   "source": [
    "import threading\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "\n",
    "def addition(a, b):\n",
    "    curr_thread = threading.current_thread()\n",
    "    print(\"Thread Name : {} (Daemon : {}), Thread Identifier : {}, Native Identifier : {}\".\n",
    "          format(curr_thread.name, curr_thread.daemon, curr_thread.ident, curr_thread.native_id))\n",
    "    time.sleep(2)\n",
    "\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    thread1 = threading.Thread(target=addition, args=(10,20), name=\"Addition1\", )\n",
    "    thread2 = threading.Thread(target=addition, args=(20,20), name=\"Addition2\", )\n",
    "\n",
    "    thread1.start()\n",
    "    thread2.start()\n",
    "\n",
    "    main_thread = threading.main_thread()\n",
    "    print(\"Thread Name : {} (Daemon : {}), Thread Identifier : {}, Native Identified : {}\".format(main_thread.name, main_thread.daemon, threading.get_ident(), threading.get_native_id()))\n",
    "\n",
    "    print(\"Currently Active Threads Count (Including Main Thread ): {}\".format(threading.active_count()))\n",
    "\n",
    "    thread1.join()\n",
    "    \n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4023e038",
   "metadata": {},
   "outputs": [],
   "source": [
    "## wait for all threads to complete"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ba3c2d8a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "\n",
    "def addition(a, b):\n",
    "    curr_thread = threading.current_thread()\n",
    "    print(\"Thread Name : {} (Daemon : {}), Thread Identifier : {}, Native Identifier : {}\".\n",
    "          format(curr_thread.name, curr_thread.daemon, curr_thread.ident, curr_thread.native_id))\n",
    "    time.sleep(2)\n",
    "\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    thread1 = threading.Thread(target=addition, args=(10,20), name=\"Addition1\", )\n",
    "    thread2 = threading.Thread(target=addition, args=(20,20), name=\"Addition2\", )\n",
    "\n",
    "    thread1.start()\n",
    "    thread2.start()\n",
    "\n",
    "    main_thread = threading.main_thread()\n",
    "    print(\"Thread Name : {} (Daemon : {}), Thread Identifier : {}, Native Identified : {}\".format(main_thread.name, main_thread.daemon, threading.get_ident(), threading.get_native_id()))\n",
    "\n",
    "    print(\"Currently Active Threads Count (Including Main Thread ): {}\".format(threading.active_count()))\n",
    "\n",
    "    thread1.join()\n",
    "    thread2.join()\n",
    "    \n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f38a38e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "## enumerate() and is_alive()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fbeb275c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "\n",
    "def addition(a, b):\n",
    "    curr_thread = threading.current_thread()\n",
    "    print(\"Thread Name : {} (Daemon : {}), Thread Identifier : {}, Native Identifier : {}\".\n",
    "          format(curr_thread.name, curr_thread.daemon, curr_thread.ident, curr_thread.native_id))\n",
    "    time.sleep(2)\n",
    "\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    thread1 = threading.Thread(target=addition, args=(10,20), name=\"Addition1\", )\n",
    "    thread2 = threading.Thread(target=addition, args=(20,20), name=\"Addition2\", )\n",
    "\n",
    "    thread1.start()\n",
    "    thread2.start()\n",
    "\n",
    "    for thread in threading.enumerate():\n",
    "        print(\"Thread Name : {}, Is Thread Alive? : {}\".format(thread.name, thread.is_alive()))\n",
    "    print()\n",
    "\n",
    "    thread1.join()\n",
    "    thread2.join()\n",
    "\n",
    "    for thread in threading.enumerate():\n",
    "        print(\"Thread Name : {}, Is Thread Alive? : {}\".format(thread.name, thread.is_alive()))\n",
    "    print()\n",
    "    \n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f6763a4c",
   "metadata": {},
   "source": [
    "## Multiprocessing"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7667f3ec",
   "metadata": {},
   "source": [
    "#### New process using process constructor"
   ]
  },
  {
   "cell_type": "raw",
   "id": "05d814a6",
   "metadata": {},
   "source": [
    "Process(target=None,name=None,args=(),kwargs={},daemon=None) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ec58c8b",
   "metadata": {},
   "outputs": [],
   "source": [
    "start demo for multproc1-6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "7a6b83e7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start Time : 2022-05-04 15:09:26.791296\n",
      "End   Time : 2022-05-04 15:09:26.800710\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"<string>\", line 1, in <module>\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 116, in spawn_main\n",
      "    exitcode = _main(fd, parent_sentinel)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 126, in _main\n",
      "    self = reduction.pickle.load(from_parent)\n",
      "AttributeError: Can't get attribute 'addition' on <module '__main__' (built-in)>\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "def addition(a, b):\n",
    "    time.sleep(3)\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    \n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    p = multiprocessing.Process(target=addition, args = (10,20))\n",
    "    \n",
    "    p.start()\n",
    "\n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "9484e53a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start Time : 2022-05-04 16:34:44.938801\n",
      "End   Time : 2022-05-04 16:34:44.963674\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"<string>\", line 1, in <module>\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 116, in spawn_main\n",
      "    exitcode = _main(fd, parent_sentinel)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 126, in _main\n",
      "    self = reduction.pickle.load(from_parent)\n",
      "AttributeError: Can't get attribute 'Addition' on <module '__main__' (built-in)>\n",
      "Traceback (most recent call last):\n",
      "  File \"<string>\", line 1, in <module>\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 116, in spawn_main\n",
      "    exitcode = _main(fd, parent_sentinel)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 126, in _main\n",
      "    self = reduction.pickle.load(from_parent)\n",
      "AttributeError: Can't get attribute 'Addition' on <module '__main__' (built-in)>\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "\n",
    "class Addition(multiprocessing.Process):\n",
    "    def __init__(self, a, b, name=\"Addition\"):\n",
    "        super().__init__() ## Thread.__init__() will do just same.\n",
    "        self.name = name\n",
    "        self.a = a\n",
    "        self.b = b\n",
    "\n",
    "    def run(self):\n",
    "        time.sleep(3)\n",
    "        print(\"Addition of {} & {} is {}\".format(self.a, self.b, self.a + self.b))\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    process1 = Addition(10,20, name=\"Addition1\")\n",
    "    process2 = Addition(20,20, name=\"Addition2\")\n",
    "\n",
    "    process1.start()\n",
    "    process2.start()\n",
    "\n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3b799401",
   "metadata": {},
   "source": [
    "Important Methods of multiprocessing Module\n",
    "- current_process() - It returns a Process object representing the current process in which it was called.\n",
    "- parent_process() - It returns a Process object representing the parent process of the process in which it was called.\n",
    "- get_start_method() - It returns the method which was used to start the process. It can be one of the fork, spawn and forkserver.\n",
    "- cpu_count() - It returns count of computer cores on the system.\n",
    "- get_all_start_methods() - It returns all the start methods supported by the system. On windows only spawn method is available whereas on unix fork and forkserver are available.\n",
    "- active_children() - It returns list of Process instances representing the child processes of the process in which this method was called.\n",
    "\n",
    "Important Methods of Process Instance\n",
    "- is_alive() - It returns a boolean value specifying whether the process is alive or not.\n",
    "\n",
    "Important Attributes of Process Instance\n",
    "- name - It returns the name of the process.\n",
    "- pid - It return process ID."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "988d4b30",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "4f37b4b5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start Time : 2022-05-04 15:07:25.727459\n",
      "CPU Count : 4\n",
      "Available Methods to Create Processes : ['spawn', 'fork', 'forkserver']\n",
      "\n",
      "Process Name : MainProcess (Daemon : False), Process Identifier : 16043, Start Method : spawn\n",
      "\n",
      "Currently Active Children of Main Process Count : [<Process name='Addition1' pid=16051 parent=16043 started>, <Process name='Addition2' pid=16052 parent=16043 started>]\n",
      "Is Process Addition1 alive? : True\n",
      "Is Process Addition2 alive? : True\n",
      "\n",
      "End   Time : 2022-05-04 15:07:25.876094\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"<string>\", line 1, in <module>\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 116, in spawn_main\n",
      "    exitcode = _main(fd, parent_sentinel)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 126, in _main\n",
      "    self = reduction.pickle.load(from_parent)\n",
      "AttributeError: Can't get attribute 'addition' on <module '__main__' (built-in)>\n",
      "Traceback (most recent call last):\n",
      "  File \"<string>\", line 1, in <module>\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 116, in spawn_main\n",
      "    exitcode = _main(fd, parent_sentinel)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 126, in _main\n",
      "    self = reduction.pickle.load(from_parent)\n",
      "AttributeError: Can't get attribute 'addition' on <module '__main__' (built-in)>\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "\n",
    "\n",
    "def addition(a, b):\n",
    "    curr_process = multiprocessing.current_process()\n",
    "    parent_process = multiprocessing.parent_process()\n",
    "    print(\"Process Name : {} (Daemon : {}), Process Identifier : {}, Parent Process : {}, Start Method : {}\\n\".\n",
    "          format(curr_process.name, curr_process.daemon, curr_process.pid, parent_process.name, multiprocessing.get_start_method()))\n",
    "\n",
    "    time.sleep(3)\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    print(\"CPU Count : {}\".format(multiprocessing.cpu_count()))\n",
    "    print(\"Available Methods to Create Processes : {}\\n\".format(multiprocessing.get_all_start_methods()))\n",
    "\n",
    "    process1 = multiprocessing.Process(target=addition, args = (10,20), name=\"Addition1\")\n",
    "    process2 = multiprocessing.Process(target=addition, args = (20,30), name=\"Addition2\")\n",
    "\n",
    "    process1.start()\n",
    "    process2.start()\n",
    "\n",
    "    main_process = multiprocessing.current_process()\n",
    "    print(\"Process Name : {} (Daemon : {}), Process Identifier : {}, Start Method : {}\\n\".\n",
    "          format(main_process.name, main_process.daemon, main_process.pid, multiprocessing.get_start_method()))\n",
    "\n",
    "    children = multiprocessing.active_children()\n",
    "    print(\"Currently Active Children of Main Process Count : {}\".format(children))\n",
    "    for child_process in children:\n",
    "        print(\"Is Process {} alive? : {}\".format(child_process.name, child_process.is_alive()))\n",
    "    print()\n",
    "    \n",
    "    process1.join()\n",
    "    process2.join()\n",
    "\n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "raw",
   "id": "6971dc3d",
   "metadata": {},
   "source": [
    "Contexts and start methods\n",
    "Depending on the platform, multiprocessing supports three ways to start a process. These start methods are\n",
    "\n",
    "spawn\n",
    "The parent process starts a fresh python interpreter process. The child process will only inherit those resources necessary to run the process object’s run() method. In particular, unnecessary file descriptors and handles from the parent process will not be inherited. Starting a process using this method is rather slow compared to using fork or forkserver.\n",
    "\n",
    "Available on Unix and Windows. The default on Windows and macOS.\n",
    "\n",
    "fork\n",
    "The parent process uses os.fork() to fork the Python interpreter. The child process, when it begins, is effectively identical to the parent process. All resources of the parent are inherited by the child process. Note that safely forking a multithreaded process is problematic.\n",
    "\n",
    "Available on Unix only. The default on Unix.\n",
    "\n",
    "forkserver\n",
    "When the program starts and selects the forkserver start method, a server process is started. From then on, whenever a new process is needed, the parent process connects to the server and requests that it fork a new process. The fork server process is single threaded so it is safe for it to use os.fork(). No unnecessary resources are inherited.\n",
    "\n",
    "Available on Unix platforms which support passing file descriptors over Unix pipes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "917f81f3",
   "metadata": {},
   "outputs": [],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "\n",
    "def addition(a, b):\n",
    "    time.sleep(3)\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    process1 = multiprocessing.Process(target=addition, args = (10,20), name=\"Addition1\")\n",
    "    process2 = multiprocessing.Process(target=addition, args = (20,30), name=\"Addition2\")\n",
    "\n",
    "    process1.start()\n",
    "    process2.start()\n",
    "\n",
    "\n",
    "    time.sleep(1)\n",
    "\n",
    "    children = multiprocessing.active_children() #Return list of all live children of the current process.\n",
    "    print(\"Currently Active Children of Main Process Count : {}\".format(children))\n",
    "    for child_process in children:\n",
    "        if child_process.is_alive():\n",
    "            child_process.terminate()\n",
    "            print(\"Terminating Process : {}\".format(child_process.name))\n",
    "    time.sleep(1)\n",
    "    for child_process in children:\n",
    "        print(\"Process : {}, ExitCode : {}\".format(child_process.name, child_process.exitcode))\n",
    "    print()\n",
    "\n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0a8d9ee6",
   "metadata": {},
   "source": [
    "terminate/kill process\n",
    "\n",
    "terminate is a signal that can be intercepted by any BG program and hence be used to saaefly close proceses\n",
    "kill is not a sigterm and hence any unfinished business can corrupt the process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "19c2edfc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start Time : 2022-05-04 15:45:53.953473\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Traceback (most recent call last):\n",
      "  File \"<string>\", line 1, in <module>\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 116, in spawn_main\n",
      "    exitcode = _main(fd, parent_sentinel)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 126, in _main\n",
      "    self = reduction.pickle.load(from_parent)\n",
      "AttributeError: Can't get attribute 'addition' on <module '__main__' (built-in)>\n",
      "Traceback (most recent call last):\n",
      "  File \"<string>\", line 1, in <module>\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 116, in spawn_main\n",
      "    exitcode = _main(fd, parent_sentinel)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/spawn.py\", line 126, in _main\n",
      "    self = reduction.pickle.load(from_parent)\n",
      "AttributeError: Can't get attribute 'addition' on <module '__main__' (built-in)>\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Currently Active Children of Main Process Count : []\n",
      "\n",
      "End   Time : 2022-05-04 15:45:55.986716\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing\n",
    "import time\n",
    "from datetime import datetime\n",
    "\n",
    "def addition(a, b):\n",
    "    time.sleep(3)\n",
    "    print(\"Addition of {} & {} is {}\".format(a,b, a + b))\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print(\"Start Time : {}\".format(datetime.now()))\n",
    "\n",
    "    process1 = multiprocessing.Process(target=addition, args = (10,20), name=\"Addition1\")\n",
    "    process2 = multiprocessing.Process(target=addition, args = (20,30), name=\"Addition2\")\n",
    "\n",
    "    process1.start()\n",
    "    process2.start()\n",
    "\n",
    "    time.sleep(1)\n",
    "\n",
    "    children = multiprocessing.active_children()\n",
    "    print(\"Currently Active Children of Main Process Count : {}\".format(children))\n",
    "    for child_process in children:\n",
    "        if child_process.is_alive():\n",
    "            child_process.kill()\n",
    "            print(\"Killing Process : {}\".format(child_process.name))\n",
    "    time.sleep(1)\n",
    "    for child_process in children:\n",
    "        print(\"Process : {}, ExitCode : {}\".format(child_process.name, child_process.exitcode))\n",
    "    print()\n",
    "\n",
    "    print(\"End   Time : {}\".format(datetime.now()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b302df57",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "b6c7355c",
   "metadata": {},
   "source": [
    "concurrent.futures is a module to execute concurrent threads\n",
    "\n",
    "Two main executors:\n",
    "    - ThreadPoolExecutor\n",
    "    - ProcessPoolExecutor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ef131d1f",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "8b4d6d68",
   "metadata": {},
   "outputs": [],
   "source": [
    "import concurrent.futures\n",
    "from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor\n",
    "import threading\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "ef805f92",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Thread_0', 1), ('Thread_1', 2), ('Thread_2', 3), ('Thread_3', 4), ('Thread_4', 5), ('Thread_0', 6), ('Thread_1', 7), ('Thread_2', 8), ('Thread_3', 9), ('Thread_4', 10)]\n",
      "CPU times: user 5.02 ms, sys: 3.34 ms, total: 8.36 ms\n",
      "Wall time: 15 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "def cube(x):\n",
    "    curr_thread = threading.current_thread()\n",
    "    time.sleep(x)\n",
    "    return curr_thread.name, pow(x,1)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    tpe = ThreadPoolExecutor(max_workers=5, thread_name_prefix = 'Thread')\n",
    "    futures = []\n",
    "    for i in range(1,11):\n",
    "        futures.append(tpe.submit(cube, i))\n",
    "\n",
    "    print([future.result() for future in futures])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "b0707505",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Thread_0', 1), ('Thread_1', 8), ('Thread_2', 27), ('Thread_3', 64), ('Thread_4', 125), ('Thread_0', 216), ('Thread_1', 343), ('Thread_2', 512), ('Thread_3', 729), ('Thread_4', 1000)]\n",
      "CPU times: user 5.23 ms, sys: 3.95 ms, total: 9.18 ms\n",
      "Wall time: 15 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "def cube(x):\n",
    "    curr_thread = threading.current_thread()\n",
    "    time.sleep(x)\n",
    "    return curr_thread.name, pow(x,3)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    tpe = ThreadPoolExecutor(max_workers=5, thread_name_prefix = 'Thread')\n",
    "    result = tpe.map(cube, range(1,11))\n",
    "    print(list(result))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "206a93cf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Thread_0', 1), ('Thread_1', 8), ('Thread_2', 27), ('Thread_3', 64), ('Thread_4', 125), ('Thread_0', 216), ('Thread_1', 343), ('Thread_2', 512), ('Thread_3', 729), ('Thread_4', 1000)]\n",
      "CPU times: user 4.6 ms, sys: 3.54 ms, total: 8.14 ms\n",
      "Wall time: 15 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "def cube(x):\n",
    "    curr_thread = threading.current_thread()\n",
    "    time.sleep(x)\n",
    "    return curr_thread.name, pow(x,3)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    tpe = ThreadPoolExecutor(max_workers=5, thread_name_prefix = 'Thread')\n",
    "    result = tpe.map(cube, range(1,11))\n",
    "    print(list(result))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "33989ef4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Thread_0', 1), ('Thread_1', 8), ('Thread_2', 27), ('Thread_3', 64), ('Thread_4', 125), ('Thread_0', 216), ('Thread_1', 343), ('Thread_2', 512), ('Thread_3', 729), ('Thread_4', 1000)]\n",
      "CPU times: user 5.34 ms, sys: 4.14 ms, total: 9.48 ms\n",
      "Wall time: 15 s\n"
     ]
    }
   ],
   "source": [
    "## using with instead of shutdown()\n",
    "%%time\n",
    "def cube(x):\n",
    "    curr_thread = threading.current_thread()\n",
    "    time.sleep(x)\n",
    "    return curr_thread.name, pow(x,3)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    with ThreadPoolExecutor(max_workers=5, thread_name_prefix = 'Thread') as tpe:\n",
    "        result = tpe.map(cube, range(1,11))\n",
    "        print(list(result))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "a087be94",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Task has not started yet. We can cancel it.\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Process SpawnProcess-32:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 315, in _bootstrap\n",
      "    self.run()\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 108, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/process.py\", line 233, in _process_worker\n",
      "    call_item = call_queue.get(block=True)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/queues.py\", line 116, in get\n",
      "    return _ForkingPickler.loads(res)\n",
      "AttributeError: Can't get attribute 'cube' on <module '__main__' (built-in)>\n",
      "Exception in thread QueueManagerThread:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/threading.py\", line 932, in _bootstrap_inner\n",
      "    self.run()\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/threading.py\", line 870, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/process.py\", line 394, in _queue_management_worker\n",
      "    work_item.future.set_exception(bpe)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/_base.py\", line 539, in set_exception\n",
      "    raise InvalidStateError('{}: {!r}'.format(self._state, self))\n",
      "concurrent.futures._base.InvalidStateError: CANCELLED: <Future at 0x7fc2708500a0 state=cancelled>\n"
     ]
    },
    {
     "ename": "BrokenProcessPool",
     "evalue": "A process in the process pool was terminated abruptly while the future was running or pending.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mBrokenProcessPool\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m<timed exec>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n",
      "\u001b[0;32m<timed exec>\u001b[0m in \u001b[0;36m<listcomp>\u001b[0;34m(.0)\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/concurrent/futures/_base.py\u001b[0m in \u001b[0;36mresult\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    437\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mCancelledError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    438\u001b[0m             \u001b[0;32melif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_state\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0mFINISHED\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 439\u001b[0;31m                 \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__get_result\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    440\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    441\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mTimeoutError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/concurrent/futures/_base.py\u001b[0m in \u001b[0;36m__get_result\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    386\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m__get_result\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    387\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exception\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 388\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exception\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    389\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    390\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_result\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mBrokenProcessPool\u001b[0m: A process in the process pool was terminated abruptly while the future was running or pending."
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Process SpawnProcess-33:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 315, in _bootstrap\n",
      "    self.run()\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 108, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/process.py\", line 233, in _process_worker\n",
      "    call_item = call_queue.get(block=True)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/queues.py\", line 116, in get\n",
      "    return _ForkingPickler.loads(res)\n",
      "AttributeError: Can't get attribute 'cube' on <module '__main__' (built-in)>\n",
      "Process SpawnProcess-34:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 315, in _bootstrap\n",
      "    self.run()\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 108, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/process.py\", line 233, in _process_worker\n",
      "    call_item = call_queue.get(block=True)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/queues.py\", line 116, in get\n",
      "    return _ForkingPickler.loads(res)\n",
      "AttributeError: Can't get attribute 'cube' on <module '__main__' (built-in)>\n",
      "Process SpawnProcess-35:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 315, in _bootstrap\n",
      "    self.run()\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 108, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/process.py\", line 233, in _process_worker\n",
      "    call_item = call_queue.get(block=True)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/queues.py\", line 116, in get\n",
      "    return _ForkingPickler.loads(res)\n",
      "AttributeError: Can't get attribute 'cube' on <module '__main__' (built-in)>\n",
      "Process SpawnProcess-36:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 315, in _bootstrap\n",
      "    self.run()\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 108, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/process.py\", line 233, in _process_worker\n",
      "    call_item = call_queue.get(block=True)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/queues.py\", line 116, in get\n",
      "    return _ForkingPickler.loads(res)\n",
      "AttributeError: Can't get attribute 'cube' on <module '__main__' (built-in)>\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "def cube(x):\n",
    "    time.sleep(x)\n",
    "    return pow(x,3)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    tpe = ProcessPoolExecutor(max_workers=5)\n",
    "    futures = []\n",
    "    for i in range(1,12):\n",
    "        futures.append(tpe.submit(cube, i))\n",
    "\n",
    "    for i, future in enumerate(futures):\n",
    "        if not future.running():\n",
    "            print(\"Task has not started yet. We can cancel it.\")\n",
    "            cancelled = future.cancel()\n",
    "            if cancelled:\n",
    "                print(\"Task cancelled : \"+str(future.cancelled()))\n",
    "\n",
    "    completed = set([future.result() for future in futures if not future.cancelled()])\n",
    "    actual = set([pow(i,3) for i in range(1,12)])\n",
    "    cancelled = actual.difference(completed)\n",
    "    print('Completed successfully : '+str(completed))\n",
    "    print('Cancelled ones : '+str(cancelled))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "cd794ea1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Task has not started yet. We can cancel it.\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n",
      "Task has not started yet. We can cancel it.\n",
      "Task cancelled : True\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Process SpawnProcess-25:\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 315, in _bootstrap\n",
      "    self.run()\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/process.py\", line 108, in run\n",
      "    self._target(*self._args, **self._kwargs)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/concurrent/futures/process.py\", line 233, in _process_worker\n",
      "    call_item = call_queue.get(block=True)\n",
      "  File \"/Users/siddhart/opt/anaconda3/lib/python3.8/multiprocessing/queues.py\", line 116, in get\n",
      "    return _ForkingPickler.loads(res)\n",
      "AttributeError: Can't get attribute 'cube' on <module '__main__' (built-in)>\n"
     ]
    },
    {
     "ename": "BrokenProcessPool",
     "evalue": "A process in the process pool was terminated abruptly while the future was running or pending.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mBrokenProcessPool\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m<timed exec>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n",
      "\u001b[0;32m<timed exec>\u001b[0m in \u001b[0;36m<listcomp>\u001b[0;34m(.0)\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/concurrent/futures/_base.py\u001b[0m in \u001b[0;36mresult\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    437\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mCancelledError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    438\u001b[0m             \u001b[0;32melif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_state\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0mFINISHED\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 439\u001b[0;31m                 \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__get_result\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    440\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    441\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mTimeoutError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/opt/anaconda3/lib/python3.8/concurrent/futures/_base.py\u001b[0m in \u001b[0;36m__get_result\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    386\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m__get_result\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    387\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exception\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 388\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_exception\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    389\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    390\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_result\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mBrokenProcessPool\u001b[0m: A process in the process pool was terminated abruptly while the future was running or pending."
     ]
    }
   ],
   "source": [
    "%%time\n",
    "def cube(x):\n",
    "    time.sleep(x)\n",
    "    return pow(x,3)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    tpe = ProcessPoolExecutor(max_workers=5)\n",
    "    futures = []\n",
    "    for i in range(1,12):\n",
    "        futures.append(tpe.submit(cube, i))\n",
    "\n",
    "    for i, future in enumerate(futures):\n",
    "        if not future.running():\n",
    "            print(\"Task has not started yet. We can cancel it.\")\n",
    "            cancelled = future.cancel()\n",
    "            if cancelled:\n",
    "                print(\"Task cancelled : \"+str(future.cancelled()))\n",
    "\n",
    "    completed = set([future.result() for future in futures if not future.cancelled()])\n",
    "    actual = set([pow(i,3) for i in range(1,12)])\n",
    "    cancelled = actual.difference(completed)\n",
    "    print('Completed successfully : '+str(completed))\n",
    "    print('Cancelled ones : '+str(cancelled))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0ed084f8",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}