multiprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Multiprocessing and multithreading"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Parallelism in python"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting multihello.py\n"
     ]
    }
   ],
   "source": [
    "%%file multihello.py\n",
    "'''hello from another process\n",
    "'''\n",
    "from multiprocessing import Process\n",
    "\n",
    "def f(name):\n",
    "    print 'hello', name\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    p = Process(target=f, args=('world',))\n",
    "    p.start()\n",
    "    p.join()\n",
    "    \n",
    "\n",
    "# EOF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hello world\r\n"
     ]
    }
   ],
   "source": [
    "!python2.7 multihello.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* On Windows: `multiprocessing` spawns with `subprocess.Popen`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "if __name__ == '__main__':\n",
    "    from multiprocessing import freeze_support\n",
    "    freeze_support()\n",
    "    \n",
    "    # Then, do multiprocessing stuff..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Data parallelism versus task parallelism"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Multithreading versus multiple threads"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* The global interpreter lock"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Processes versus threads"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Shared memory and shared objects"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Shared objects: `Value` and `Array`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting sharedobj.py\n"
     ]
    }
   ],
   "source": [
    "%%file sharedobj.py\n",
    "'''demonstrate shared objects in multiprocessing\n",
    "'''\n",
    "from multiprocessing import Process, Value, Array\n",
    "\n",
    "def f(n, a):\n",
    "    n.value = 3.1415927\n",
    "    for i in range(len(a)):\n",
    "        a[i] = -a[i]\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    num = Value('d', 0.0)\n",
    "    arr = Array('i', range(10))\n",
    "\n",
    "    p = Process(target=f, args=(num, arr))\n",
    "    p.start()\n",
    "    p.join()\n",
    "\n",
    "    print num.value\n",
    "    print arr[:]\n",
    "    \n",
    "\n",
    "# EOF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.1415927\r\n",
      "[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]\r\n"
     ]
    }
   ],
   "source": [
    "!python2.7 sharedobj.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* `Manager` and proxies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting sharedproxy.py\n"
     ]
    }
   ],
   "source": [
    "%%file sharedproxy.py\n",
    "'''demonstrate sharing objects by proxy through a manager\n",
    "'''\n",
    "from multiprocessing import Process, Manager\n",
    "\n",
    "def f(d, l):\n",
    "    d[1] = '1'\n",
    "    d['2'] = 2\n",
    "    d[0.25] = None\n",
    "    l.reverse()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    manager = Manager()\n",
    "\n",
    "    d = manager.dict()\n",
    "    l = manager.list(range(10))\n",
    "\n",
    "    p = Process(target=f, args=(d, l))\n",
    "    p.start()\n",
    "    p.join()\n",
    "\n",
    "    print d\n",
    "    print l\n",
    "\n",
    "\n",
    "# EOF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{0.25: None, 1: '1', '2': 2}\r\n",
      "[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]\r\n"
     ]
    }
   ],
   "source": [
    "!python2.7 sharedproxy.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "See: https://docs.python.org/2/library/multiprocessing.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Working in C with `ctypes` and `numpy`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting numpyshared.py\n"
     ]
    }
   ],
   "source": [
    "%%file numpyshared.py\n",
    "'''demonstrating shared objects using numpy and ctypes\n",
    "'''\n",
    "import multiprocessing as mp\n",
    "from multiprocessing import sharedctypes\n",
    "from numpy import ctypeslib\n",
    "\n",
    "def fill_arr(arr_view, i):\n",
    "    arr_view.fill(i)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    ra = sharedctypes.RawArray('i', 4)\n",
    "    arr = ctypeslib.as_array(ra)\n",
    "    arr.shape = (2, 2)\n",
    "    p1 = mp.Process(target=fill_arr, args=(arr[:1, :], 1))\n",
    "    p2 = mp.Process(target=fill_arr, args=(arr[1:, :], 2))\n",
    "    p1.start(); p2.start()\n",
    "    p1.join(); p2.join()\n",
    "    print arr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1 1]\r\n",
      " [2 2]]\r\n"
     ]
    }
   ],
   "source": [
    "!python2.7 numpyshared.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Issues: threading and locks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Low-level task parallelism: point to point communication"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* `Process`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mprocess.py\n"
     ]
    }
   ],
   "source": [
    "%%file mprocess.py\n",
    "'''demonstrate the process claas\n",
    "'''\n",
    "import multiprocessing as mp\n",
    "from time import sleep\n",
    "from random import random\n",
    "\n",
    "def worker(num):\n",
    "    sleep(2.0 * random())\n",
    "    name = mp.current_process().name\n",
    "    print \"worker {},name:{}\".format(num, name)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    master = mp.current_process().name\n",
    "    print \"Master name: {}\".format(master)\n",
    "    for i in range(2):\n",
    "        p = mp.Process(target=worker, args=(i,))\n",
    "        p.start()\n",
    "\n",
    "    # Close all child processes spawn\n",
    "    [p.join() for p in mp.active_children()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Master name: MainProcess\n",
      "worker 0,name:Process-1\n",
      "worker 1,name:Process-2\n"
     ]
    }
   ],
   "source": [
    "!python2.7 mprocess.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* `Queue` and `Pipe`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting queuepipe.py\n"
     ]
    }
   ],
   "source": [
    "%%file queuepipe.py\n",
    "'''demonstrate queues and pipes\n",
    "'''\n",
    "import multiprocessing as mp\n",
    "import pickle\n",
    "\n",
    "def qworker(q):\n",
    "    v = q.get() # blocking!\n",
    "    print \"queue worker got '{}' from parent\".format(v)\n",
    "    \n",
    "def pworker(p):\n",
    "    import pickle  # needed for encapsulation\n",
    "    msg = 'hello hello hello'\n",
    "    print \"pipe worker sending {!r} to parent\".format(msg)\n",
    "    p.send(msg)\n",
    "    v = p.recv()\n",
    "    print \"pipe worker got {!r} from parent\".format(v)\n",
    "    print \"unpickled to {}\".format(pickle.loads(v))\n",
    "\n",
    "    \n",
    "if __name__ == '__main__':\n",
    "    q = mp.Queue()\n",
    "    p = mp.Process(target=qworker, args=(q,))\n",
    "    p.start() # blocks at q.get()\n",
    "    v = 'python rocks!'\n",
    "    print \"putting '{}' on queue\".format(v)\n",
    "    q.put(v)\n",
    "    p.join()\n",
    "    print ''\n",
    "\n",
    "    # The two ends of the pipe: the parent and the child connections\n",
    "    p_conn, c_conn = mp.Pipe()\n",
    "    p = mp.Process(target=pworker, args=(c_conn,))\n",
    "    p.start()\n",
    "    msg = pickle.dumps([1,2,3],-1)\n",
    "    print \"got {!r} from child\".format(p_conn.recv())\n",
    "    print \"sending {!r} to child\".format(msg)\n",
    "    p_conn.send(msg)\n",
    "    import datetime\n",
    "    print \"\\nfinished: {}\".format(datetime.date.today())\n",
    "    p.join()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "putting 'python rocks!' on queue\r\n",
      "queue worker got 'python rocks!' from parent\r\n",
      "\r\n",
      "pipe worker sending 'hello hello hello' to parent\r\n",
      "got 'hello hello hello' from child\r\n",
      "sending '\\x80\\x02]q\\x00(K\\x01K\\x02K\\x03e.' to child\r\n",
      "pipe worker got '\\x80\\x02]q\\x00(K\\x01K\\x02K\\x03e.' from parent\r\n",
      "unpickled to [1, 2, 3]\r\n",
      "\r\n",
      "finished: 2015-07-04\r\n"
     ]
    }
   ],
   "source": [
    "!python2.7 queuepipe.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Synchronization with `Lock` and `Event`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting multi_sync.py\n"
     ]
    }
   ],
   "source": [
    "%%file multi_sync.py\n",
    "'''demonstrating locks\n",
    "'''\n",
    "import multiprocessing as mp\n",
    "\n",
    "def print_lock(lk, i):\n",
    "    name = mp.current_process().name\n",
    "    lk.acquire()\n",
    "    for j in range(5):\n",
    "        print i, \"from process\", name\n",
    "    lk.release()\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    lk = mp.Lock()\n",
    "    ps = [mp.Process(target=print_lock, args=(lk,i)) for i in range(5)]\n",
    "    [p.start() for p in ps]\n",
    "    [p.join() for p in ps]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 from process Process-1\r\n",
      "0 from process Process-1\r\n",
      "0 from process Process-1\r\n",
      "0 from process Process-1\r\n",
      "0 from process Process-1\r\n",
      "1 from process Process-2\r\n",
      "1 from process Process-2\r\n",
      "1 from process Process-2\r\n",
      "1 from process Process-2\r\n",
      "1 from process Process-2\r\n",
      "2 from process Process-3\r\n",
      "2 from process Process-3\r\n",
      "2 from process Process-3\r\n",
      "2 from process Process-3\r\n",
      "2 from process Process-3\r\n",
      "3 from process Process-4\r\n",
      "3 from process Process-4\r\n",
      "3 from process Process-4\r\n",
      "3 from process Process-4\r\n",
      "3 from process Process-4\r\n",
      "4 from process Process-5\r\n",
      "4 from process Process-5\r\n",
      "4 from process Process-5\r\n",
      "4 from process Process-5\r\n",
      "4 from process Process-5\r\n"
     ]
    }
   ],
   "source": [
    "!python2.7 multi_sync.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "e.is_set() False\n",
      "Process-7 finished waiting\n",
      "Process-2 finished waiting\n",
      "Process-5 finished waiting\n",
      "Process-1 finished waiting\n",
      "Process-4 finished waiting\n",
      "Process-6 finished waiting\n",
      "Process-8 finished waiting\n",
      "Process-9 finished waiting\n",
      "Process-10 finished waiting\n",
      "Process-3 finished waiting\n"
     ]
    }
   ],
   "source": [
    "'''events\n",
    "'''\n",
    "import multiprocessing as mp\n",
    "\n",
    "def wait_on_event(e):\n",
    "    name = mp.current_process().name\n",
    "    e.wait()\n",
    "    print name, \"finished waiting\"    \n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    e = mp.Event()\n",
    "    ps = [mp.Process(target=wait_on_event, args=(e,)) for i in range(10)]\n",
    "    [p.start() for p in ps]\n",
    "    print \"e.is_set()\", e.is_set()\n",
    "    #raw_input(\"press any key to set event\")\n",
    "    e.set()\n",
    "    [p.join() for p in ps]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "High-level task parallelism: collective communication"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* The task `Pool`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* pipes (`apply`) and `map`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[-0.007, -0.03, -0.005, -0.011, 0.005, 0.001, -0.056, -0.02, 0.058]\n",
      "0.173\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing as mp\n",
    "\n",
    "def random_mean(x):\n",
    "    import numpy as np\n",
    "    return round(np.mean(np.random.randint(-x,x+1,10000)), 3)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # create a pool with cpu_count() procsesses\n",
    "    p = mp.Pool()\n",
    "    results = p.map(random_mean, range(1,10))\n",
    "    print results\n",
    "    print p.apply(random_mean, [100])\n",
    "    p.close()\n",
    "    p.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Variants: blocking, iterative, unordered, and asynchronous"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[ 4.006 8.992 3.016 2.006 5.025 5.957 7.009 0.994 ...] QUIT\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing as mp\n",
    "\n",
    "def random_mean_count(x):\n",
    "    import numpy as np\n",
    "    return x + round(np.mean(np.random.randint(-x,x+1,10000)), 3)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # create a pool with cpu_count() procsesses\n",
    "    p = mp.Pool()\n",
    "    results = p.imap_unordered(random_mean_count, range(1,10))\n",
    "    print \"[\",\n",
    "    for i in results:\n",
    "        print i,\n",
    "        if abs(i) <= 1.0:\n",
    "            print \"...] QUIT\"\n",
    "            break\n",
    "    list(results)\n",
    "    p.close()\n",
    "    p.join()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Waiting . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . [0.997, 1.979, 2.994, 4.016, 5.031, 5.944, 7.114, 8.066, 9.0]\n",
      "\n",
      "99.965\n"
     ]
    }
   ],
   "source": [
    "import multiprocessing as mp\n",
    "\n",
    "def random_mean_count(x):\n",
    "    import numpy as np\n",
    "    return x + round(np.mean(np.random.randint(-x,x+1,10000)), 3)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # create a pool with cpu_count() procsesses\n",
    "    p = mp.Pool()\n",
    "    results = p.map_async(random_mean_count, range(1,10))\n",
    "    print \"Waiting .\",\n",
    "    i = 0\n",
    "    while not results.ready():\n",
    "        if not i%4000:\n",
    "            print \".\",\n",
    "        i += 1\n",
    "    print results.get()\n",
    "    print \"\\n\", p.apply_async(random_mean_count, [100]).get()\n",
    "    p.close()\n",
    "    p.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Issues: `random` number generators"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[31 22 13 24 23 31  2 60 13 35]\n",
      "[24, 6, 8, 6, 11, 41, 16, 64, 40, 17]\n",
      "[4, 4, 4, 4, 4, 4, 4, 4, 50, 50]\n"
     ]
    }
   ],
   "source": [
    "import numpy as np\n",
    "\n",
    "def walk(x, n=100, box=.5, delta=.2):\n",
    "    \"perform a random walk\"\n",
    "    w = np.cumsum(x + np.random.uniform(-delta,delta,n))\n",
    "    w = np.where(abs(w) > box)[0]\n",
    "    return w[0] if len(w) else n\n",
    "\n",
    "N = 10\n",
    "\n",
    "# run N trials, all starting from x=0\n",
    "pwalk = np.vectorize(walk)\n",
    "print pwalk(np.zeros(N))\n",
    "\n",
    "# run again, using list comprehension instead of ufunc\n",
    "print [walk(0) for i in range(N)]\n",
    "\n",
    "# run again, using multiprocessing's map\n",
    "import multiprocessing as mp\n",
    "p = mp.Pool()\n",
    "print p.map(walk, [0]*N)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting state.py\n"
     ]
    }
   ],
   "source": [
    "%%file state.py\n",
    "\"\"\"some good state utilities\n",
    "\"\"\"\n",
    "\n",
    "def check_pickle(x, dill=False):\n",
    "    \"checks the pickle across a subprocess\"\n",
    "    import pickle\n",
    "    import subprocess\n",
    "    if dill:\n",
    "        import dill as pickle\n",
    "        pik = \"dill\"\n",
    "    else:\n",
    "        pik = \"pickle\"\n",
    "    fail = True\n",
    "    try:\n",
    "        _x = pickle.dumps(x)\n",
    "        fail = False\n",
    "    finally:\n",
    "        if fail:\n",
    "            print \"DUMP FAILED\"\n",
    "    msg = \"python -c import {0}; print {0}.loads({1})\".format(pik,repr(_x))\n",
    "    print \"SUCCESS\" if not subprocess.call(msg.split(None,2)) else \"LOAD FAILED\"\n",
    "\n",
    "    \n",
    "def random_seed(s=None):\n",
    "    \"sets the seed for calls to 'random()'\"\n",
    "    import random\n",
    "    random.seed(s)\n",
    "    try:\n",
    "        from numpy import random\n",
    "        random.seed(s)\n",
    "    except:\n",
    "        pass\n",
    "    return\n",
    "\n",
    "\n",
    "def random_state(module='random', new=False, seed='!'):\n",
    "    \"\"\"return a (optionally manually seeded) random generator\n",
    "\n",
    "For a given module, return an object that has random number generation (RNG)\n",
    "methods available.  If new=False, use the global copy of the RNG object.\n",
    "If seed='!', do not reseed the RNG (using seed=None 'removes' any seeding).\n",
    "If seed='*', use a seed that depends on the process id (PID); this is useful\n",
    "for building RNGs that are different across multiple threads or processes.\n",
    "    \"\"\"\n",
    "    import random\n",
    "    if module == 'random':\n",
    "        rng = random\n",
    "    elif not isinstance(module, type(random)):\n",
    "        # convienence for passing in 'numpy'\n",
    "        if module == 'numpy': module = 'numpy.random'\n",
    "        try:\n",
    "            import importlib\n",
    "            rng = importlib.import_module(module)\n",
    "        except ImportError:\n",
    "            rng = __import__(module, fromlist=module.split('.')[-1:])\n",
    "    elif module.__name__ == 'numpy': # convienence for passing in numpy\n",
    "        from numpy import random as rng\n",
    "    else: rng = module\n",
    "\n",
    "    _rng = getattr(rng, 'RandomState', None) or \\\n",
    "           getattr(rng, 'Random') # throw error if no rng found\n",
    "    if new:\n",
    "        rng = _rng()\n",
    "\n",
    "    if seed == '!': # special case: don't reset the seed\n",
    "        return rng\n",
    "    if seed == '*': # special case: random seeding for multiprocessing\n",
    "        try:\n",
    "            try:\n",
    "                import multiprocessing as mp\n",
    "            except ImportError:\n",
    "                import processing as mp\n",
    "            try:\n",
    "                seed = mp.current_process().pid\n",
    "            except AttributeError:\n",
    "                seed = mp.currentProcess().getPid()\n",
    "        except:\n",
    "            seed = 0\n",
    "        import time\n",
    "        seed += int(time.time()*1e6)\n",
    "\n",
    "    # set the random seed (or 'reset' with None)\n",
    "    rng.seed(seed)\n",
    "    return rng\n",
    "\n",
    "\n",
    "# EOF"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Issues: serialization"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Better serialization: `multiprocess`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]\n"
     ]
    }
   ],
   "source": [
    "import multiprocess\n",
    "\n",
    "print multiprocess.Pool().map(lambda x:x**2, range(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Try several variants of looping patterns to see if you can speed up a toy password cracker."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "See: ['exercise'](kocham.ipynb)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Code-based versus object-based serialization: `pp(ft)`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting runppft.py\n"
     ]
    }
   ],
   "source": [
    "%%file runppft.py\n",
    "'''demonstrate ppft\n",
    "'''\n",
    "\n",
    "import ppft\n",
    "\n",
    "def squared(x):\n",
    "    return x*x\n",
    "\n",
    "server = ppft.Server()  # can take 'localhost:8000' or remote:port\n",
    "result = server.submit(squared, (5,))\n",
    "result.wait()\n",
    "print result.finished\n",
    "print result()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\r\n",
      "25\r\n"
     ]
    }
   ],
   "source": [
    "!python2.7 runppft.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Programming efficiency: `pathos`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Multi-argument `map` functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Unified API for `threading`, `multiprocessing`, and serial and parallel python (`pp`)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing allpool.py\n"
     ]
    }
   ],
   "source": [
    "%%file allpool.py\n",
    "'''demonstrate pool API\n",
    "'''\n",
    "import pathos\n",
    "\n",
    "def sum_squared(x,y):\n",
    "    return (x+y)**2\n",
    "\n",
    "x = range(5)                  \n",
    "y = range(0,10,2)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    sp = pathos.pools.SerialPool()                                          \n",
    "    pp = pathos.pools.ParallelPool()\n",
    "    mp = pathos.pools.ProcessPool()\n",
    "    tp = pathos.pools.ThreadPool()\n",
    "\n",
    "    for pool in [sp,pp,mp,tp]:\n",
    "        print pool.map(sum_squared, x, y)\n",
    "        pool.close()\n",
    "        pool.join()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0, 9, 36, 81, 144]\n",
      "[0, 9, 36, 81, 144]\n",
      "[0, 9, 36, 81, 144]\n",
      "[0, 9, 36, 81, 144]\n"
     ]
    }
   ],
   "source": [
    "!python2.7 allpool.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Strives for natural programming constructs in parallel code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "OK\n"
     ]
    }
   ],
   "source": [
    "from itertools import izip\n",
    "\n",
    "\n",
    "PRIMES = [\n",
    "    112272535095293,\n",
    "    112582705942171,\n",
    "    112272535095293,\n",
    "    115280095190773,\n",
    "    115797848077099,\n",
    "    1099726899285419]\n",
    "\n",
    "def is_prime(n):\n",
    "    if n % 2 == 0:\n",
    "        return False\n",
    "\n",
    "    import math\n",
    "    sqrt_n = int(math.floor(math.sqrt(n)))\n",
    "    for i in range(3, sqrt_n + 1, 2):\n",
    "        if n % i == 0:\n",
    "            return False\n",
    "    return True\n",
    "\n",
    "def sleep_add1(x):\n",
    "    from time import sleep\n",
    "    if x < 4: sleep(x/10.0)\n",
    "    return x+1\n",
    "\n",
    "def sleep_add2(x):\n",
    "    from time import sleep\n",
    "    if x < 4: sleep(x/10.0)\n",
    "    return x+2\n",
    "\n",
    "def test_with_multipool(Pool):\n",
    "    inputs = range(10)\n",
    "    with Pool() as pool1:\n",
    "        res1 = pool1.amap(sleep_add1, inputs)\n",
    "    with Pool() as pool2:\n",
    "        res2 = pool2.amap(sleep_add2, inputs)\n",
    "\n",
    "    with Pool() as pool3:\n",
    "        for number, prime in izip(PRIMES, pool3.imap(is_prime, PRIMES)):\n",
    "            assert prime if number != PRIMES[-1] else not prime\n",
    "\n",
    "    assert res1.get() == [i+1 for i in inputs]\n",
    "    assert res2.get() == [i+2 for i in inputs]\n",
    "    print \"OK\"\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    from pathos.pools import ProcessPool\n",
    "    test_with_multipool(ProcessPool)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Programming models and hierarchical computing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[0.0, 0.8414709848078965, 0.9092974268256817], [1.0, 0.5403023058681398, -0.4161468365471424]]\n"
     ]
    }
   ],
   "source": [
    "import pathos\n",
    "from math import sin, cos\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    mp = pathos.pools.ProcessPool()\n",
    "    tp = pathos.pools.ThreadPool()\n",
    "    \n",
    "    print mp.amap(tp.map, [sin, cos], [range(3),range(3)]).get()\n",
    "    mp.close(); tp.close()\n",
    "    mp.join(); tp.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* `Pool` caching"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Not covered: `IPython.parallel` and `scoop`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**EXERCISE:** Let's take another swing at Monte Carlo betting. You'll want to focus on `roll.py`, `trials.py` and `optimize.py`. Can you speed things up with careful placement of a `Pool`?  Are there small modifications to the code that would allow hierarchical parallelism?  Can we speed up the calculation, or does parallel computing lose to spin-up overhead? Where are we now hitting the wall?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "See: ['solution'](parallelMC.ipynb)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Remote execution"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Easy: the `pp.Server`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "localhost>$ ppserver.py -p 8000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Even easier: `Pool().server` in `pathos`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    ">>> def squared(x):\n",
    "...     return x**2\n",
    "... \n",
    ">>> import pathos\n",
    ">>> pool = pathos.pools.ParallelPool(nodes=1, servers=('localhost:8000',))\n",
    ">>> results = pool.map(squared, range(100))\n",
    ">>> print pathos.pp.stats()\n",
    "Job execution statistics:\n",
    " job count | % of all jobs | job time sum | time per job | job server\n",
    "        65 |         65.00 |       0.2004 |     0.003083 | localhost:8000\n",
    "        35 |         35.00 |       0.0538 |     0.001538 | local\n",
    "Time elapsed since server creation 21.2711749077\n",
    "0 active tasks, 1 cores\n",
    "\n",
    ">>> pool.close()\n",
    ">>> pool.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Not covered: `rpyc`, `pyro`, and `zmq`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Related: secure authentication with `ssh`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* `pathos.secure`: `connection` and `tunnel`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "handling pipe response\n",
      "SSH Tunnel to: localhost\n",
      "Remote port: 23\n",
      "Local port: 42376\n",
      "Press <Enter> to disconnect\n",
      "Kill ssh pid=63720\n"
     ]
    }
   ],
   "source": [
    "import pathos\n",
    "import sys\n",
    "\n",
    "rhost = 'localhost'\n",
    "rport = 23\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    tunnel = pathos.secure.Tunnel()\n",
    "    lport = tunnel.connect(rhost, rport)\n",
    "    print 'SSH Tunnel to:', rhost\n",
    "    print 'Remote port:', rport\n",
    "    print 'Local port:', lport\n",
    "    print 'Press <Enter> to disconnect'\n",
    "    sys.stdin.readline()\n",
    "    tunnel.disconnect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pathos\n",
    "\n",
    "launcher = pathos.secure.Pipe()\n",
    "config = launcher(command='hostname', rhost='localhost', background=False)\n",
    "launcher.launch()\n",
    "print launcher.response()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Not oovered: `paramiko`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "What about [large-scale cluster](mpi.ipynb) computing?"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}