7b. Standard library (2) - threading and multiprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Threading and multiprocessing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Threading: running tasks in the background of the current process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "import threading"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Method 1: defining your background task as a function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def do_in_background(sleep):\n",
    "    print('Background process starting...')\n",
    "    import time\n",
    "    time.sleep(sleep)\n",
    "    print('Background process finished!')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Background process starting...\n",
      "Background process finished!"
     ]
    }
   ],
   "source": [
    "task = threading.Thread(target=do_in_background, args=(1,))\n",
    "task.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Method 2: defining your task as a subclass of threading.Thread"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "class MyTask(threading.Thread):\n",
    "        \n",
    "    def run(self):\n",
    "        print(self.getName(), 'started.')\n",
    "        import time\n",
    "        time.sleep(10)\n",
    "        print(self.getName(), 'finished.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Thread-7 started.\n",
      "Thread-8 started.\n",
      "Tasks started.\n",
      "These statements will be executed even though the ones above have not finished!\n",
      "Thread-9 started.\n",
      "Thread-7 finished.\n",
      "Thread-8 finished.\n",
      "Thread-9 finished.\n",
      "Done!\n"
     ]
    }
   ],
   "source": [
    "tasks = []\n",
    "\n",
    "# Start some tasks and add them to our list.\n",
    "for i in range(0,3):\n",
    "    task = MyTask()\n",
    "    task.start()\n",
    "    tasks.append(task)\n",
    "\n",
    "print('Tasks started.')\n",
    "print('These statements will be executed even though the ones above have not finished!')\n",
    "\n",
    "# Wait for all tasks to complete\n",
    "for task in tasks:\n",
    "    task.join()\n",
    "print('Done!')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Multiprocessing: running tasks as separate processes (i.e. allowing you to use more than one core)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Pool of workers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]\n",
      "Results of map function are being calculated in the background (asynchronously)...\n",
      "[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Pool\n",
    "\n",
    "def f(x):\n",
    "    return x*x\n",
    "\n",
    "# Start 4 worker processes\n",
    "pool = Pool(processes=4)\n",
    "# Calculate squares of 1 to 9 in parallel\n",
    "print(pool.map(f, range(10)))\n",
    "\n",
    "result = pool.map_async(f, range(10))\n",
    "print(\"Results of map function are being calculated in the background (asynchronously)...\")\n",
    "print(result.get(timeout=1))           # waits for the results to become available"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Processes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8 cores available.\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing\n",
    "\n",
    "cpu_count = multiprocessing.cpu_count()\n",
    "print(\"{} cores available.\".format(cpu_count))\n",
    "\n",
    "def my_background_task():\n",
    "    import time\n",
    "    time.sleep(3)\n",
    "\n",
    "proc = multiprocessing.Process(target=my_background_task)\n",
    "proc.start()\n",
    "proc.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "#### N.B. Printing in a background process might crash the IPython Notebook!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Communication between threads or processes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<h3>Queues</h3>\n",
    "<ul>\n",
    "<li>queue.Queue (FIFO)</li>\n",
    "<li>queue.LifoQueue (LIFO)</li>\n",
    "<li>queue.PriorityQueue</li>\n",
    "</ul>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "42\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Process, Queue\n",
    "\n",
    "def task(q):\n",
    "    # Do some calculations\n",
    "    result = 42\n",
    "    q.put(result)\n",
    "    \n",
    "# Create a new queue\n",
    "q = Queue()\n",
    "# Create a process, pass our queue as argument\n",
    "p = Process(target=task, args=(q,))\n",
    "# Start the process\n",
    "p.start()\n",
    "# Retrieve and print the (first) value in the queue;\n",
    "# the command will wait until a value becomes available.\n",
    "print(q.get())\n",
    "# Make sure the process terminates\n",
    "p.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Pipes: direct, two-way communication channel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "11\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing import Process, Pipe\n",
    "\n",
    "def child_task(pipe):\n",
    "    # Receive input value at child's end of the pipe (sent by parent)\n",
    "    input = pipe.recv()\n",
    "    # Calculate some output value\n",
    "    output = input + 1\n",
    "    # Send output from child's end of the pipe (to parent)\n",
    "    pipe.send(output)\n",
    "    # Close this end of the pipe\n",
    "    pipe.close()\n",
    "\n",
    "# Create a Pipe, which returns two \"ends\" of the pipe\n",
    "pipe_parent, pipe_child = Pipe()\n",
    "# Create and start a child process and pass the child's end of the pipe\n",
    "p = Process(target=child_task, args=(pipe_child,))\n",
    "p.start()\n",
    "# Send an input value to the child\n",
    "pipe_parent.send(10)\n",
    "# Wait for an output value to be sent by the child\n",
    "print(pipe_parent.recv())\n",
    "# Make sure the process terminates\n",
    "p.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "#### N.B.: two different processes/threads should not try to read from or write to the same end of a pipe!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Synchronization between processes using locks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from multiprocessing import Process, Lock\n",
    "\n",
    "def print_task(lock):\n",
    "    lock.acquire()\n",
    "    # Do something which requires locking,\n",
    "    # e.g. writing to a file.\n",
    "    lock.release()\n",
    "\n",
    "\n",
    "lock = Lock()\n",
    "\n",
    "for num in range(10):\n",
    "    Process(target=print_task, args=(lock,)).start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### <h3>A lot of other, additional modules are available that allow you to exploit different kinds of parallelism. See for instance:</h3>     <a href=\"https://wiki.python.org/moin/ParallelProcessing\" target=\"_blank\">Python Wiki page about Parallel Processing</a>"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "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.4.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}