code_examples/python_concurrency/Concurrency in Python.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:15e5ddc505dcd77fe827ac4beb6d248a9d735a4b4fcb6eb9b83dd61ec6a26f03"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "heading",
     "level": 4,
     "metadata": {},
     "source": [
      "N.B. If you want to read about some of this stuff in more detail, see [here](http://www.slideshare.net/dabeaz/an-introduction-to-python-concurrency). It's a great (albeit quite long) overview of some of the issues discussed here. I have taken a couple of pictures from this presentation and reused them below."
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "What is Concurrency?"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In practice, concurrency means doing multiple tasks at the same time. This doesn't necessarily guarantee any sort of speed-up or performance gain! For instance, on a single CPU, concurrency is technically still possible, but is only achieved by the processor switching back and forth between instruction sets. This does not result in a performance gain (in fact, it often results in a loss of performance due to the overhead associated with any checks that have to be performed every time the instruction set is switched)."
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "When is Concurrency useful?"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The above example of concurrency on a single core is sort of a fringe case; in general we're interested in systems with multiple cores, i.e. the ability to run things in parallel. Some typical use cases include...\n",
      "* High performance computing - If you study nuclear engineering, you're familiar with concurrency\n",
      "* Servers - Being able to accept and execute commands from many clients all at once, independently\n",
      "* GUI's - Separating the computationally-intensive bits of a program from the GUI updating to make sure the UI stays responsive\n",
      "* Streaming Data - Do many things to the same chunk of data before the next chunk arrives"
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "Pythonic Concurrency - Use cases"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "So what does any of this have to do with Python? There are many cases where concurrency with python are very useful; the GUI and streaming data cases for example. The most common application for scientists and engineers is wanting to speed up the execution of some code by parallelizing, taking advantage of all (or at least more) of the computing power of multi-core systems.\n",
      "\n",
      "Python has tools that allow the programmer to take full advantage of their computing system. The level of performance gain that you can get from concurrency for typical scientific python scenarios (e.g. simulating or crunching data on a personal computer with [2,16] cores) depends on the type of problem and the number of cores."
     ]
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "Is The Effort Required to Code in Parallel Worth the Potential Speedup?"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "You want to make sure you're using the right tool for the job. If you are looking for major performance gains (10x - 10000x), concurrent programming in python will not get you there. For these cases you're better off\n",
      "* Making sure your code is formulated in such a way that takes advantage of built-in high-performance libraries (see Andy's vectorization with numpy talk)\n",
      "* Creating python extensions (Python/C API, SWIG, Boost.python, cython, etc.)\n",
      "* or looking into just-in-time (JIT) compilation (PyPy, SEJIT)\n",
      "\n",
      "A good rule-of-thumb regarding when to go about parallelizing your python code is the \"cup-of-tea\" threshold. If your code currently runs on the order of 5-10 minutes (i.e. long enough for you to get up and focus on something else like making a cup of tea) but you're not ready to invest a ton of time in writing the code in a low-level language (premature optimization is the root of all evil - Knuth), then parallelizing your current code is a good approach, especially if it's naively parallel!"
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "What we'll talk about today"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "This demo will focus on two tools within the python standard library - the **threading** module, and the **multiprocessing** module. We'll cover when to use one module vs. the other, and give some examples to demonstrate how they work.\n",
      "\n",
      "There are many other tools that can be used to implement concurrent programs; most notably tools associated with distributed computing, i.e. message passing to computers connected via networking. These tools, such as sockets, 0MQ (pyzmq), or xmlrpc can be used to set up concurrent jobs on either multiple PC's or even on a single PC (using localhost as the network). In order to keep this talk a reasonable length, we won't be discussing these tools here; however this doesn't mean that they aren't incredibly useful!"
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "The Threading module"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import threading"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "There are several ways to spawn a thread..."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Subclass the threading.Thread object and overwrite it's run() method with your code\n",
      "class ExampleCountingThread(threading.Thread):\n",
      "    def __init__(self, countdown=10000, modulo=1000):\n",
      "        # If you are going to have your own constructor, make sure you call the parent constructor too!\n",
      "#        Thread.__init__(self)\n",
      "        super(ExampleCountingThread, self).__init__()\n",
      "        self.value = countdown\n",
      "        self.print_interval = modulo\n",
      "    def run(self):\n",
      "        # A simple example that counts down from 10000 and prints the value when the modulus with the print interval is 0\n",
      "        while self.value > 0:\n",
      "            if self.value % self.print_interval == 0:\n",
      "                print self.value\n",
      "            self.value -= 1\n",
      "        print '%s done processing' %self.name"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Run the thread\n",
      "t1 = ExampleCountingThread(100000, 10000)\n",
      "# When you call thread.start(), the run() method is invoked\n",
      "t1.start()\n",
      "# Wait for thread t1 to exit - you may get funky behavior \n",
      "# if you don't join your threads. Join can only be called \n",
      "# from another thread (i.e. a thread can't join itself)\n",
      "t1.join()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Or, we can pass the function we want to run in as a target function\n",
      "def countdown(value=10000, modulo=1000, pf=True):\n",
      "    '''This is the same function as implemented in the run() method above'''\n",
      "    while value > 0:\n",
      "        if value % modulo == 0:\n",
      "            print value\n",
      "        value -= 1\n",
      "    if pf: print 'Done processing!'\n",
      "    return"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Run the thread\n",
      "t2 = threading.Thread(target=countdown, args=(100000,10000))\n",
      "t2.start()\n",
      "t2.join()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Well that was easy... Let's try to get some performance gains!"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "If you have at least 2 processors, you'd expect 2 threads running the same function concurrently to run in about the same amount of time it would take one processor to run the function (although a bit slower because of Amdahl's law), albeit producing twice the amount of data. Let's try it:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%timeit\n",
      "# Sequential example: Count down from 100000 twice in a row. Note: suppress in-loop printing with larger modulo\n",
      "countdown(10000000, 1e8, pf=False) # First run\n",
      "countdown(10000000, 1e8, pf=False) # Second run"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%timeit\n",
      "# Now, let's try concurrency with threads. We expect two threads running the countdown function at once to run in about 50%\n",
      "# of the time it took the sequential version to run\n",
      "t1 = threading.Thread(target=countdown, args=(10000000, 1e8, False))\n",
      "t2 = threading.Thread(target=countdown, args=(10000000, 1e8, False))\n",
      "t1.start(); t2.start()\n",
      "t1.join(); t2.join()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Wait... What?!"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "How is it that the concurrent example took LONGER than the sequential one? Those who are parallel-saavy might say that maybe the overhead for creating a thread in python is massive. A threading library that had a thread overhead that was THAT massive would be pretty useless (on my home desktop, the timeit results would imply it takes 3 seconds to spawn each thread).\n",
      "\n",
      "That's not actually what's going on however; which brings us to the big problem with the threading module: "
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "The Dreaded GIL"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from IPython.display import YouTubeVideo\n",
      "YouTubeVideo('a1Y73sPHKxw', autoplay=True)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The Global Interpreter Lock, or GIL for short, is a construct implemented within the python interpreter itself that ensures that any piece of code that is currently running has full and **exclusive** access of the machinery of the interpreter. This means that two threads (which are indeed true threads, either posix or windows implemented depending on your system) can't run concurrently within a single interpreter. I don't want to go into any details, because one can fall down the rabbit-hole very rapidly here. Suffice it to say that the GIL prevents multiple pieces of python code from running at the same time **within the same interpreter**. This has nothing to do with hardware (i.e. you could have 7 idle cores just waiting to be used; the GIL will not let you use them), but is a limitation built into C/Python itself.\n",
      "\n",
      "Rather than talk about the merits/nitty-gritty of the GIL, let's instead move on to a solution that works around it: the multiprocessing module"
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "A Quirk of the GIL, and final thoughts on the threading module"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The GIL limitation might make it seem that the threading module is entirely useless for concurrent applications. This is not entirely so. The GIL really only causes problems for **cpu-bound** applications. The GIL has a quirk that it is actually released during blocking I/O."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from IPython.display import Image\n",
      "embed = Image('http://image.slidesharecdn.com/introconcurrent-100925150728-phpapp02/95/an-introduction-to-python-concurrency-85-728.jpg')\n",
      "embed\n"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "http://image.slidesharecdn.com/introconcurrent-100925150728-phpapp02/95/an-introduction-to-python-concurrency-85-728.jpg?cb=1285445265"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "This quirk means that despite the GIL, the threading module is still useful for I/O bound applications. One such application would be a server, where you might have hundreds or thousands of threads waiting for input. As long as the server doesn't do anything computationally intensive when it gets a request, it will do a fine job handling tons of concurrent requests.\n",
      "\n",
      "Another note: the Python interpreter does not handle thread scheduling at all: this is left up to the host operating system. Thus there is no additional performance degradation from Python when it comes to thread handling; in other words, in I/O bound applications where the GIL doesn't interfere as much, the thread accessing and scheduling shouldn't be any worse than it would be if you implemented the same application using C and pthreads for example."
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "Getting Around the GIL: Multiprocessing Module"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Recall that the GIL only affects code running within **a single interpreter**. To get around this limitation, the multiprocessing module allows you to spawn \"threads\" in separate processes. In other words, each \"thread\" or \"process\" you spawn is actually a separate instance of a python interpreter!"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from multiprocessing import Process"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Look at the system monitor to see all of the individual processes\n",
      "import time\n",
      "\n",
      "def wait_then_print(num, wait=2.0):\n",
      "    '''Sleep for \"wait\" seconds then print the given number'''\n",
      "    time.sleep(wait)\n",
      "    print 'Process %s done waiting!' %num\n",
      "    return\n",
      "\n",
      "# Create a list of processes\n",
      "processes = []\n",
      "for i in range(4): processes.append(Process(target=wait_then_print, args=(i, 5.0)))\n",
      "    \n",
      "# Start each of the processes\n",
      "for p in processes: p.start()\n",
      "    \n",
      "# Join the processes\n",
      "for p in processes: p.join()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "Let's try the countdown example again..."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%timeit\n",
      "# Sequential example: Count down from 100000 twice in a row. Note: suppress in-loop printing with larger modulo\n",
      "countdown(10000000, 1e8, pf=False) # First run\n",
      "countdown(10000000, 1e8, pf=False) # Second run"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%timeit\n",
      "# Now, let's try concurrency with threads. We expect two threads running the countdown function at once to run in about 50%\n",
      "# of the time it took the sequential version to run\n",
      "p1 = Process(target=countdown, args=(10000000, 1e8, False))\n",
      "p2 = Process(target=countdown, args=(10000000, 1e8, False))\n",
      "p1.start(); p2.start()\n",
      "p1.join(); p2.join()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Wooo! It worked!"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Another thing you might notice from the above example is that the syntax for using multiprocessing and threading are nearly identical. This is great news! You don't need to learn a whole new syntax to use the module - if you had been practicing with threading and not getting results, you can just as easily try multiprocessing by switching out thread with Process (and maybe a few other minor tweaks)"
     ]
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "The multiprocessing model"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Since each process that is spawned is running in it's own Python interpreter, the shared-memory model that we had for the threading module doesn't apply here (since the different processes don't have access to the same data). Thus the multiprocessing module uses a distributed memory model, where information is exchanged between threads via message passing.\n",
      "\n",
      "The two main tools for passing data between threads are **pipes** and **queues**\n",
      "\n",
      "Note: It is also possible to set up shared-memory that can be accessed by multiple processes; however, the multiprocessing documentation recommends against this if at all possible; I personally never do it."
     ]
    },
    {
     "cell_type": "heading",
     "level": 4,
     "metadata": {},
     "source": [
      "The [multiprocessing documentation](https://docs.python.org/2/library/multiprocessing.html) is very good and has several examples. I am not going to try to cover all of the stuff available, but rather do a couple of examples with some of the things that I use often: **Process**, **Pipe**, **Queue**, and **Pool**; and cover a quirk of multiprocessing that can cause a real headache if you're unaware of it"
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "An Example: Concurrent Image Processing"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Let's do some concurrent image processing on single image. In this example, we want to apply a edge detection filter (the canny filter) to the image with different degrees of gaussian blurring"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%pylab\n",
      "import time"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# First, import an image to process\n",
      "from skimage import data\n",
      "from skimage.filter import canny\n",
      "img = data.camera()\n",
      "imshow(img, cmap=cm.Greys_r)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from multiprocessing import Process, Queue\n",
      "\n",
      "# Let's use the object-oriented paradigm for setting up the processes\n",
      "class FilterProcess(Process):\n",
      "    def __init__(self, inq, outq):\n",
      "        '''Each process must be given an input queue, an output queue, and a sigma for the filter'''\n",
      "        Process.__init__(self)   # Make sure to call the parent constructor!!!!\n",
      "        self.input_queue = inq\n",
      "        self.output_queue = outq\n",
      "        \n",
      "    def run(self):\n",
      "        # Keep checking the input queue for images until it gets a special \"STOP\" key\n",
      "        for image, sigma in iter(self.input_queue.get, \"STOP\"):\n",
      "            # Apply filter\n",
      "            edges = canny(image, sigma)\n",
      "            # Put\n",
      "            self.output_queue.put((edges, sigma))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Now let's set up a group of processes to apply the filters. Let's look at 4 different filter levels\n",
      "\n",
      "# Setup the queues for passing data to and from the FilterProcesses\n",
      "to_workers = Queue()\n",
      "from_workers = Queue()\n",
      "\n",
      "# Choose how many workers you want to set up\n",
      "num_processes = 2\n",
      "\n",
      "# Make a list of processes\n",
      "processes = []\n",
      "for i in range(num_processes): processes.append(FilterProcess(to_workers, from_workers))\n",
      "# Start the processes\n",
      "for p in processes: p.start()\n",
      "# Now the processes are waiting for us to send them an image on the to_workers queue"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Set up the filter sigmas and the output container\n",
      "filter_widths = arange(0,10)\n",
      "edges = np.zeros((img.shape[0], img.shape[1], len(filter_widths)))\n",
      "\n",
      "# Some utiltiy functions to help us stop the processes when we're done, and plot the filtered results\n",
      "def stop_processes(process_list):\n",
      "    for i in range(num_processes): to_workers.put(\"STOP\")\n",
      "    for p in process_list: p.join()\n",
      "        \n",
      "def show_edges(edges):\n",
      "    fig, ax = subplots(2,5)\n",
      "    for i, a in enumerate(ax.ravel()):\n",
      "        a.imshow(edges[:,:,i], cmap=cm.Greys_r)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we can compare the results of running the edge filter sequentially or in parallel"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Single process\n",
      "tic = time.time()\n",
      "for i,sigma in enumerate(filter_widths):\n",
      "    edges[:,:,i] = canny(img, sigma)\n",
      "toc = time.time()\n",
      "print \"%.5f\" %(toc-tic)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "show_edges(edges)\n",
      "edges = np.zeros((img.shape[0], img.shape[1], len(filter_widths)))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Concurrent Processes\n",
      "tic = time.time()\n",
      "for i, sigma in enumerate(filter_widths):\n",
      "    to_workers.put((img, sigma))\n",
      "    \n",
      "for i in range(10):\n",
      "    edges[:,:,i] = from_workers.get()[0]\n",
      "toc = time.time()\n",
      "print \"%.5f\" %(toc-tic)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "show_edges(edges)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Note that the processes are still waiting for new input on the queue (since we haven't given them the stop signal yet) so we can analyze new images if we wanted to"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "img2 = data.moon()\n",
      "edges = np.zeros((img2.shape[0], img2.shape[1], len(filter_widths)))\n",
      "# Analyze new image\n",
      "tic = time.time()\n",
      "for i, sigma in enumerate(filter_widths):\n",
      "    to_workers.put((img2, sigma))\n",
      "    \n",
      "for i in range(10):\n",
      "    edges[:,:,i] = from_workers.get()[0]\n",
      "toc = time.time()\n",
      "print \"%.5f\" %(toc-tic)\n",
      "show_edges(edges)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "stop_processes(processes)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "Pool Example - Simplest Way to Unlock Naive Parallelism"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "There are many features in the multiprocessing library that I'm not covering, but one feature that I do think warrants explicit mention (I use it regularly) is Pool.\n",
      "\n",
      "multiprocessing.Pool is a simple, high-level interface to mapping work out to many worker threads, then collecting the results. This is very useful for speeding up problems that are \"naively parallel\".\n",
      "\n",
      "Let's go back to last week for an example. Say we want to do a monte-carlo simulation of particle diffusion, and we've already been smart about vectorizing our problem, but it still runs too slow..."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import sys\n",
      "sys.path.append('..')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from numpyVectorization.motion_gauss import simulateParticles_loop"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%timeit\n",
      "# Serial example\n",
      "num_particles = 100000\n",
      "num_steps = 100\n",
      "steps, trajectories = simulateParticles_loop(num_particles, num_steps, showPlots=False)  # Disable graphics"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Pool allows us to map the simulation out (the trajectories don't depend on each other, so the problem is \"naively parallel\") to as many processes as we'd like.\n",
      "\n",
      "Before we dive into the parallel example, let's look at the basics of Pool"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from multiprocessing import Pool, cpu_count # A helper function that determines how many cores you have\n",
      "\n",
      "# Find out how many cpus you have\n",
      "num_cpus = cpu_count()\n",
      "print 'My computer has %s processors' %(num_cpus)\n",
      "\n",
      "# Make a dummy function that we can send to the workers in the pool\n",
      "def print_num(num):\n",
      "    print 'The number is: ', num\n",
      "    return num\n",
      "\n",
      "# Make the pool object\n",
      "p = Pool(processes=num_cpus)\n",
      "# Make some input to map out to the workers\n",
      "nums_to_print = range(num_cpus)\n",
      "# Here's where we farm out the work to the workers in the pool.\n",
      "#results = p.map(print_num, nums_to_print)\n",
      "results = p.map_async(print_num, nums_to_print)\n",
      "print results"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# The asynchronous result object has some useful methods\n",
      "print results.ready()  # Return whether the results have been produced by all the workers\n",
      "print results.successful()   # Returns whether each of the calls was successful. If any of the workers raises an exception, this is false\n",
      "\n",
      "# To actually retrieve the result from the async result object, use get\n",
      "results = results.get()\n",
      "print results"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Shut down the pool\n",
      "# First, you have to close (or terminate)\n",
      "p.close()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Then you can join\n",
      "p.join()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Pool has two main methods for using the workers: apply and map. They mirror the builtin functions the [apply](https://docs.python.org/2/library/functions.html#apply) and [map](https://docs.python.org/2/library/functions.html#map), except they use the processes in the pool rather than the main process.\n",
      "\n",
      "There are synchronous and asynchronous versions of each. The only difference there is the synchronous calls are blocking (i.e. the process that created the pool will be blocked until ALL of the workers return) while the asynchronous calls aren't. The asynchronous calls (map_async or apply_async) return an AsynchronousResult object like the one we saw above. You can poll the state of this object and acquire the results using get() when they're ready."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Synchronous vs. Asynchronous\n",
      "import time\n",
      "\n",
      "# Make a dummy function that will \"compute\" for a while\n",
      "def dummy(t):\n",
      "    time.sleep(t)\n",
      "    print \"Done sleeping\"\n",
      "\n",
      "# Set up a pool\n",
      "p = Pool(processes=3)   # You don't HAVE to use the number of cores you have available"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Synchronous\n",
      "map_result = p.map(dummy, [5]*3)\n",
      "print \"I couldn't do anything until the workers were done, because this was a synchronous call\""
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Asynchronous\n",
      "amap_result = p.map_async(dummy, [5]*3)\n",
      "print \"I on the other hand, can do whatever I want!\"\n",
      "print \"Are my results ready?\", amap_result.ready()\n",
      "print \"Okay, in the mean time; Here, have some random numbers\"\n",
      "print np.random.rand(10,10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print \"Are my results ready now?\", amap_result.ready()\n",
      "print \"If so, let's get them!\"\n",
      "if amap_result.ready(): \n",
      "    results = amap_result.get()\n",
      "    print results"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Notice that there was some weirdness in the printing here: i.e. the worker threads didn't actually cough up their print statements until after the get() was called. I can't tell you exactly what's going on here, but anyone who's ever played with threading, synchronization, and stdout before has seen weird thing happen. The moral of the story: don't rely on printing from workers to notify you of anything."
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Now back to our simulation example"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "See simulationExample_multiprocessing.py"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import numpy as np                                                              \n",
      "import time                                                                     \n",
      "from multiprocessing import Pool, cpu_count                                     \n",
      "import sys                                                                      \n",
      "sys.path.append('..')                                         \n",
      "from numpyVectorization.motion_gauss import simulateParticles_loop              "
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Define a wrapper function so we can pass in a tuple of args\n",
      "def wrapper_fun(args):                                                          \n",
      "    return simulateParticles_loop(*args)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Set up simulation/Pool parameters\n",
      "num_particles = 100000                                                        \n",
      "num_steps = 100                                                               \n",
      "num_cpus = cpu_count()                                                        \n",
      "# Determine how many particles to run on each worker                          \n",
      "num_particles_per_core = int(num_particles / num_cpus)\n",
      "args = [(num_particles_per_core, num_steps, False)] * num_cpus"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Pool-itize\n",
      "tic = time.time()\n",
      "p = Pool(num_cpus)\n",
      "result = p.map_async(wrapper_fun, args)\n",
      "poolresult = result.get()\n",
      "toc = time.time()\n",
      "print \"%.5f sec to compute result with %s cores\" %((toc-tic), num_cpus)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Serial again, for comparison\n",
      "tic = time.time()\n",
      "simulateParticles_loop(num_particles, num_steps, False)\n",
      "toc = time.time()\n",
      "print \"%.5f sec to compute result without Pool\" %(toc-tic)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "Final Thoughts and Caveats"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "* While multiprocessing is relatively easy to try, it can lead you down a wormhole if it doesn't do what you expect (see Pool example)\n",
      "* Multiprocessing in windows: Special considerations\n",
      "* Only pickled objects and buffers (i.e. serialized data) on pipes/queues\n",
      "* constructor vs. run call - any thing you initialize in a Process constructor must be picklable! (i.e. no file handles or contexts)"
     ]
    }
   ],
   "metadata": {}
  }
 ]
}