advanced-python/17-Multiprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Multiprocessing in Python\n",
    "In this article we talk about how to use the `multiprocessing` module in Python.\n",
    "\n",
    "- How to create and start multiple processes\n",
    "- How to wait for processes to complete\n",
    "- How to share data between processes\n",
    "- How to use `Locks` to prevent race conditions\n",
    "- How to use a `Queue` for process-safe data/task processing.\n",
    "- How to use a `Pool` to manage multiple worker processes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create and run processes\n",
    "You create a process with `multiprocessing.Process()`. It takes two important arguments:\n",
    "    \n",
    "- `target`: a callable object (function) for this process to be invoked when the process starts\n",
    "- `args`: the (function) arguments for the target function. This must be a tuple\n",
    "\n",
    "Start a process with `process.start()`\n",
    "\n",
    "Call `process.join()` to tell the program that it should wait for this process to complete before it continues with the rest of the code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from multiprocessing import Process\n",
    "import os\n",
    "\n",
    "def square_numbers():\n",
    "    for i in range(1000):\n",
    "        result = i * i\n",
    "\n",
    "        \n",
    "if __name__ == \"__main__\":        \n",
    "    processes = []\n",
    "    num_processes = os.cpu_count()\n",
    "    # number of CPUs on the machine. Usually a good choise for the number of processes\n",
    "\n",
    "    # create processes and asign a function for each process\n",
    "    for i in range(num_processes):\n",
    "        process = Process(target=square_numbers)\n",
    "        processes.append(process)\n",
    "\n",
    "    # start all processes\n",
    "    for process in processes:\n",
    "        process.start()\n",
    "\n",
    "    # wait for all processes to finish\n",
    "    # block the main programm until these processes are finished\n",
    "    for process in processes:\n",
    "        process.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Share data between processes\n",
    "Since processes don't live in the same memory space, they do not have access to the same (public) data. Thus, they need special shared memory objects to share data.\n",
    "\n",
    "Data can be stored in a shared memory variable using `Value` or `Array`.\n",
    "\n",
    "- `Value(type, value)`: Create a `ctypes` object of type `type`. Access the value with `.target`.\n",
    "- `Array(type, value)`: Create a `ctypes` array with elements of type `type`. Access the values with `[]`.\n",
    "\n",
    "Task: Create two processes, each process should have access to a shared variable and modify it (in this case only increase it repeatedly by 1 for 100 times). Create another two processes that share an array and modify (increase) all the elements in the array."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Value at beginning: 0\n",
      "Array at beginning: [0.0, 100.0, 200.0]\n",
      "Value at end: 144\n",
      "Array at end: [134.0, 237.0, 339.0]\n",
      "end main\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Process, Value, Array\n",
    "import time\n",
    "\n",
    "def add_100(number):\n",
    "    for _ in range(100):\n",
    "        time.sleep(0.01)\n",
    "        number.value += 1\n",
    "\n",
    "def add_100_array(numbers):\n",
    "    for _ in range(100):\n",
    "        time.sleep(0.01)\n",
    "        for i in range(len(numbers)):\n",
    "            numbers[i] += 1\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    shared_number = Value('i', 0) \n",
    "    print('Value at beginning:', shared_number.value)\n",
    "\n",
    "    shared_array = Array('d', [0.0, 100.0, 200.0])\n",
    "    print('Array at beginning:', shared_array[:])\n",
    "\n",
    "    process1 = Process(target=add_100, args=(shared_number,))\n",
    "    process2 = Process(target=add_100, args=(shared_number,))\n",
    "\n",
    "    process3 = Process(target=add_100_array, args=(shared_array,))\n",
    "    process4 = Process(target=add_100_array, args=(shared_array,))\n",
    "\n",
    "    process1.start()\n",
    "    process2.start()\n",
    "    process3.start()\n",
    "    process4.start()\n",
    "\n",
    "    process1.join()\n",
    "    process2.join()\n",
    "    process3.join()\n",
    "    process4.join()\n",
    "\n",
    "    print('Value at end:', shared_number.value)\n",
    "    print('Array at end:', shared_array[:])\n",
    "\n",
    "    print('end main')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How to use `Locks`\n",
    "Notice that in the above example, the 2 processes should increment the shared value by 1 for 100 times. This results in 200 total operations. But why is the end value not 200?\n",
    "\n",
    "#### Race condition\n",
    "A race condition happened here. A race condition occurs when two or more processes or threads can access shared data and they try to change it at the same time. In our example the two processes have to read the shared value, increase it by 1, and write it back into the shared variable. If this happens at the same time, the two processes read the same value, increase it and write it back. Thus, both processes write the same increased value back into the shared object, and the value was not increased by 2. See https://www.python-engineer.com/learn/advancedpython16_threading/ for a detailed explanation of race conditions.\n",
    "\n",
    "#### Avoid race conditions with `Locks`\n",
    "A lock (also known as mutex) is a synchronization mechanism for enforcing limits on access to a resource in an environment where there are many processes/threads of execution. A Lock has two states: **locked** and **unlocked**. \n",
    "If the state is locked, it does not allow other concurrent processes/threads to enter this code section until the state is unlocked again.\n",
    "\n",
    "Two functions are important:\n",
    "- `lock.acquire()` : This will lock the state and block\n",
    "- `lock.release()` : This will unlock the state again.\n",
    "\n",
    "Important: You should always release the block again after it was acquired!\n",
    "\n",
    "In our example the critical code section where the shared variable is read and increased is now locked. This prevents the second process from modyfing the shared object at the same time. Not much has changed in our code. All new changes are commented in the code below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Value at beginning: 0\n",
      "Array at beginning: [0.0, 100.0, 200.0]\n",
      "Value at end: 200\n",
      "Array at end: [200.0, 300.0, 400.0]\n",
      "end main\n"
     ]
    }
   ],
   "source": [
    "# import Lock\n",
    "from multiprocessing import Lock\n",
    "from multiprocessing import Process, Value, Array\n",
    "import time\n",
    "\n",
    "def add_100(number, lock):\n",
    "    for _ in range(100):\n",
    "        time.sleep(0.01)\n",
    "        # lock the state\n",
    "        lock.acquire()\n",
    "        \n",
    "        number.value += 1\n",
    "        \n",
    "        # unlock the state\n",
    "        lock.release()\n",
    "\n",
    "def add_100_array(numbers, lock):\n",
    "    for _ in range(100):\n",
    "        time.sleep(0.01)\n",
    "        for i in range(len(numbers)):\n",
    "            lock.acquire()\n",
    "            numbers[i] += 1\n",
    "            lock.release()\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    # create a lock\n",
    "    lock = Lock()\n",
    "    \n",
    "    shared_number = Value('i', 0) \n",
    "    print('Value at beginning:', shared_number.value)\n",
    "\n",
    "    shared_array = Array('d', [0.0, 100.0, 200.0])\n",
    "    print('Array at beginning:', shared_array[:])\n",
    "\n",
    "    # pass the lock to the target function\n",
    "    process1 = Process(target=add_100, args=(shared_number, lock))\n",
    "    process2 = Process(target=add_100, args=(shared_number, lock))\n",
    "\n",
    "    process3 = Process(target=add_100_array, args=(shared_array, lock))\n",
    "    process4 = Process(target=add_100_array, args=(shared_array, lock))\n",
    "\n",
    "    process1.start()\n",
    "    process2.start()\n",
    "    process3.start()\n",
    "    process4.start()\n",
    "\n",
    "    process1.join()\n",
    "    process2.join()\n",
    "    process3.join()\n",
    "    process4.join()\n",
    "\n",
    "    print('Value at end:', shared_number.value)\n",
    "    print('Array at end:', shared_array[:])\n",
    "\n",
    "    print('end main')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Use the lock as a context manager\n",
    "After `lock.acquire()` you should never forget to call `lock.release()` to unblock the code. You can also use a lock as a context manager, wich will safely lock and unlock your code. It is recommended to use a lock this way:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def add_100(number, lock):\n",
    "    for _ in range(100):\n",
    "        time.sleep(0.01)\n",
    "        with lock:\n",
    "            number.value += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using Queues in Python\n",
    "Data can also be shared between processes with a Queue. Queues can be used for thread-safe/process-safe data exchanges and data processing both in a multithreaded and a multiprocessing environment, which means you can avoid having to use any synchronization primitives like locks.\n",
    "\n",
    "### The queue\n",
    "A queue is a linear data structure that follows the First In First Out (FIFO) principle. A good example is a queue of customers that are waiting in line, where the customer that came first is served first."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Queue\n",
    "\n",
    "# create queue\n",
    "q = Queue()\n",
    "\n",
    "# add elements\n",
    "q.put(1) # 1\n",
    "q.put(2) # 2 1\n",
    "q.put(3) # 3 2 1 \n",
    "\n",
    "# now q looks like this:\n",
    "# back --> 3 2 1 --> front\n",
    "\n",
    "# get and remove first element\n",
    "first = q.get() # --> 1\n",
    "print(first) \n",
    "\n",
    "# q looks like this:\n",
    "# back --> 3 2 --> front"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using a queue in multiprocessing\n",
    "Operations with a queue are process-safe. The multiprocessing Queue implements all the methods of queue.Queue except for `task_done()` and `join()`. Important methods are:\n",
    "\n",
    "- `q.get()` : Remove and return the first item. By default, it blocks until the item is available.\n",
    "- `q.put(item)` : Puts element at the end of the queue. By default, it blocks until a free slot is available.\n",
    "- `q.empty()` : Return True if the queue is empty.\n",
    "- `q.close()` : Indicate that no more data will be put on this queue by the current process. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "4\n",
      "9\n",
      "16\n",
      "25\n",
      "-1\n",
      "-2\n",
      "-3\n",
      "-4\n",
      "-5\n",
      "end main\n"
     ]
    }
   ],
   "source": [
    "# communicate between processes with the multiprocessing Queue\n",
    "# Queues are thread and process safe\n",
    "from multiprocessing import Process, Queue\n",
    "\n",
    "def square(numbers, queue):\n",
    "    for i in numbers:\n",
    "        queue.put(i*i)\n",
    "\n",
    "def make_negative(numbers, queue):\n",
    "    for i in numbers:\n",
    "        queue.put(i*-1)\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    \n",
    "    numbers = range(1, 6)\n",
    "    q = Queue()\n",
    "\n",
    "    p1 = Process(target=square, args=(numbers,q))\n",
    "    p2 = Process(target=make_negative, args=(numbers,q))\n",
    "\n",
    "    p1.start()\n",
    "    p2.start()\n",
    "\n",
    "    p1.join()\n",
    "    p2.join()\n",
    "\n",
    "    # order might not be sequential\n",
    "    while not q.empty():\n",
    "        print(q.get())\n",
    "        \n",
    "    print('end main')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Process Pools\n",
    "A process pool object controls a pool of worker processes to which jobs can be submitted It supports asynchronous results with timeouts and callbacks and has a parallel map implementation. It can automatically manage the available processors and split data into smaller chunks which can then be processed in parallel by different processes. See https://docs.python.org/3.7/library/multiprocessing.html#multiprocessing.pool for all possible methods. Important methods are:\n",
    "\n",
    "- `map(func, iterable[, chunksize])` : This method chops the iterable into a number of chunks which it submits to the process pool as separate tasks. The (approximate) size of these chunks can be specified by setting chunksize to a positive integer. It blocks until the result is ready.\n",
    "- `close()` : Prevents any more tasks from being submitted to the pool. Once all the tasks have been completed the worker processes will exit.\n",
    "- `join()`: Wait for the worker processes to exit. One must call `close()` or `terminate()` before using `join()`.\n",
    "- `apply(func, args)`: Call func with arguments args. It blocks until the result is ready. func is only executed in ONE of the workers of the pool. \n",
    "\n",
    "Note: There are also asynchronous variants `map_async()` and `apply_async()` that will not block. They can execute callbacks when the results are ready."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0, 1, 8, 27, 64, 125, 216, 343, 512, 729]\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Pool \n",
    "\n",
    "def cube(number):\n",
    "    return number * number * number\n",
    "\n",
    "    \n",
    "if __name__ == \"__main__\":\n",
    "    numbers = range(10)\n",
    "    \n",
    "    p = Pool()\n",
    "\n",
    "    # by default this allocates the maximum number of available \n",
    "    # processors for this task --> os.cpu_count()\n",
    "    result = p.map(cube,  numbers)\n",
    "    \n",
    "    # or \n",
    "    # result = [p.apply(cube, args=(i,)) for i in numbers]\n",
    "    \n",
    "    p.close()\n",
    "    p.join()\n",
    "    \n",
    "    print(result)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [conda env:pythonengineer_env]",
   "language": "python",
   "name": "conda-env-pythonengineer_env-py"
  },
  "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}