{
 "metadata": {
  "name": "parallel python"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#Parallel Python - Interactive parallel programming"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#IPython\n",
      "\n",
      "##Supports different styles of parallelism\n",
      "\n",
      "    Single program, multiple data (SPMD) parallelism\n",
      "    \n",
      "    Multiple program, multiple data (MPMD) parallelism\n",
      "    \n",
      "    Message passing using MPI\n",
      "    \n",
      "    Task farming\n",
      "    \n",
      "    Combinations of above approaches\n",
      "    \n",
      "    Custom user de\ufb01ned approaches\n",
      "\n",
      "##Interactive approach to\n",
      "\n",
      "    Development\n",
      "    \n",
      "    Execution\n",
      "    \n",
      "    Debugging\n",
      "    \n",
      "    Monitoring"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#ARCITECTURE\n",
      "IPython is a tool for interactive and exploratory computing.\n",
      "We have seen that IPython's kernel provides a mechanism for interactive\n",
      "*remote* computation, and we have extended this same mechanism for\n",
      "interactive remote *parallel* computation, simply by having multiple\n",
      "kernels.\n",
      "<img src=\"files\\wideView400.png\">\n",
      "\n",
      "\n",
      "##Engine\n",
      "    *Takes Python commands over a network connection\n",
      "    *One engine per core\n",
      "    *Blocks code: controller has an asynchronous API\n",
      "##Controller client \n",
      "    *Interface for working with a set of engines\n",
      "    *Hub\n",
      "    *Collection of schedulers\n",
      "    *View objects represent a subset of engines\n",
      "    *A Direct interface, where engines are addressed explicitly.\n",
      "    *A LoadBalanced interface, where the Scheduler is trusted \n",
      "        with assigning work to appropriate engines."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "###The client: lightweight handle on all engines of a cluster\n",
      "            from IPython.parallel import Client\n",
      "            rc = Client(profile='mpi')\n",
      "\n",
      "###The views: \u201cslice\u201d the client with speci\ufb01c execution\n",
      "semantics \n",
      "\n",
      "        *DirectView: direct execution on all engines\n",
      "\n",
      "                dview = rc.direct_view()\n",
      "        *LoadBalancedView: run on any one engine\n",
      "\n",
      "                lview = rc.load_balanced_view() "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "##Hub\n",
      "\n",
      "    keeps track of engine\n",
      "    connections, schedulers,\n",
      "    clients\n",
      "    \n",
      "##Cluster state Scheduler\n",
      "\n",
      "    All actions that can be\n",
      "    performed on the engine go\n",
      "    through a scheduler"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import os\n",
      "pid = os.getpid()\n",
      "print pid"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "13120\n"
       ]
      }
     ],
     "prompt_number": 1
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#Start your Engines\n",
      "\n",
      "     *Simplest method\n",
      "        ipcluster start --n=4"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#Direct Interface\n",
      "\n",
      "    * capabilities of each engine are directly and explicitly exposed to the user\n",
      "    * Blocking and non-blocking\n",
      "    * AsyncResult\n",
      "    * Remote execution\n",
      "       * apply\n",
      "       * execute\n",
      "       * map\n",
      "       * Remote function decorators\n",
      "\n",
      "    * Managing data\n",
      "       * push, pull\n",
      "       * scatter, gather"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "*Remote functions* are just like normal functions, but when they are called, they execute on one or more engines, rather than locally. \n",
      "\n",
      "IPython provides two decorators:\n",
      "\n",
      "    In [10]: @dview.remote(block=True)\n",
      "       ....: def getpid():\n",
      "       ....:     import os\n",
      "       ....:     return os.getpid()\n",
      "       ....:\n",
      "\n",
      "    In [11]: getpid()\n",
      "    Out[11]: [12345, 12346, 12347, 12348]'"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#Blocking vs. Non-blocking\n",
      "\n",
      "##Blocking\n",
      "waits until all engines are done executing the command\n",
      "\n",
      "                 *  dview.block=True\n",
      "\n",
      "##Non-blocking\n",
      "\n",
      "   * returns an AsyncResult immediate\n",
      "   * Get data later with get method\n",
      "   * Check the status with ready\n",
      "   * Wait\n",
      "\n",
      "                   * dview.block=False"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from IPython.parallel import Client\n",
      "rc = Client(profile='default')\n",
      "dview = rc.direct_view()\n",
      "with dview.sync_imports():\n",
      "    import os\n",
      "dview.block=True\n",
      "pids = dview.apply(os.getpid)\n",
      "   \n",
      "dview.block=False\n",
      "ar = dview.apply(os.getpid)  \n",
      "dview.wait(ar)\n",
      "pids = ar.get()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "importing os on engine(s)\n"
       ]
      }
     ],
     "prompt_number": 4
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "##Apply\n",
      "\n",
      "   * apply_sync (block=True)\n",
      "   *  apply_async (block=False)"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def wait(t):\n",
      "    import time\n",
      "    tic = time.time()\n",
      "    time.sleep(t)\n",
      "    return time.time()-tic\n",
      "ar = dview.apply_async(wait,1)   \n",
      "if ar.ready() == True:\n",
      "    print \"ready\"\n",
      "else:\n",
      "    print \"not-ready\"\n",
      "   \n",
      "dview.wait(ar)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "not-ready\n"
       ]
      },
      {
       "output_type": "pyout",
       "prompt_number": 5,
       "text": [
        "True"
       ]
      }
     ],
     "prompt_number": 5
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "##Execute\n",
      "Commands can be executed as strings on speci\ufb01c\n",
      "engines"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "dview['a'] = 10\n",
      "dview['b'] = 20\n",
      "dview.execute('c=a*b')\n",
      "rc[1].execute('c=a+b')\n",
      "print dview['c']"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[200, 30, 200, 200]\n"
       ]
      }
     ],
     "prompt_number": 6
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "##Map\n",
      "\n",
      "       * Applies a sequence of to a function element-by-element\n",
      "\n",
      "       * Does not do dynamic load balancing\n",
      "\n",
      "                    map_sync, map_async"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def fsquare(x):\n",
      "    return x**2\n",
      "xvals = range(32)    \n",
      "s_result = map(fsquare, xvals)\n",
      "dview.block=True\n",
      "p_result = dview.map(fsquare, xvals)\n",
      "print p_result"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361, 400, 441, 484, 529, 576, 625, 676, 729, 784, 841, 900, 961]\n"
       ]
      }
     ],
     "prompt_number": 9
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#Push and Pull\n",
      "\n",
      "       Limited to dictionary types"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "dview.block=True\n",
      "values = dict(a=1.03234,b=3453)\n",
      "dview.push(values)\n",
      "a_values = dview.pull('a')\n",
      "print a_values\n",
      "dview.push(dict(c='speed'))\n",
      "print dview.pull('c')\n",
      "dview['c'] = 'speed'\n",
      "print dview.pull('c')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[1.03234, 1.03234, 1.03234, 1.03234]\n",
        "['speed', 'speed', 'speed', 'speed']\n",
        "['speed', 'speed', 'speed', 'speed']"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n"
       ]
      }
     ],
     "prompt_number": 10
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "##Scatter and Gather\n",
      "\n",
      "           Partition a sequence and push/pull"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%px\n",
      "def fsquare(x):\n",
      "    return x**2"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 22
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "dview.scatter('a',range(24))   \n",
      "print dview['a']\n",
      "all_a = dview.gather('a')\n",
      "print all_a\n",
      "# Use\n",
      "dview.scatter('x',range(64))\n",
      "dview.execute('y=map(fsquare, x)')\n",
      "y = dview.gather('y')\n",
      "print y"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[[0, 1, 2, 3, 4, 5], [6, 7, 8, 9, 10, 11], [12, 13, 14, 15, 16, 17], [18, 19, 20, 21, 22, 23]]\n",
        "[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23]\n",
        "[0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361, 400, 441, 484, 529, 576, 625, 676, 729, 784, 841, 900, 961, 1024, 1089, 1156, 1225, 1296, 1369, 1444, 1521, 1600, 1681, 1764, 1849, 1936, 2025, 2116, 2209, 2304, 2401, 2500, 2601, 2704, 2809, 2916, 3025, 3136, 3249, 3364, 3481, 3600, 3721, 3844, 3969]"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n"
       ]
      }
     ],
     "prompt_number": 23
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#Task interface\n",
      "\n",
      "    *Presents the engines as a fault tolerant, dynamic loadbalanced\n",
      "            system of workers\n",
      "\n",
      "    *No direct access to individual engines\n",
      "\n",
      "    *Simple and powerful\n",
      "\n",
      "            lview = rc.load_balanced_view()\n",
      "\n",
      "##Components:\n",
      "\n",
      "        map\n",
      "\n",
      "        Parallel function decoders\n",
      "\n",
      "        Dependencies\n",
      "                functional \n",
      "                graph"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%px\n",
      "def simulator(x):\n",
      "    import subprocess\n",
      "    cmd = \"./simulator \" + str(x) + \" 2 4 > trial.\" + str(x)\n",
      "    p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)\n",
      "    (err,out)=p.communicate()\n",
      "    status = p.returncode\n",
      "    return \"%s, %s\" % (out,err)  "
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 24
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "   \n",
      "print simulator(1)    "
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "None, \n"
       ]
      }
     ],
     "prompt_number": 27
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from IPython.parallel import Client\n",
      "rc = Client(profile='default')\n",
      "lview = rc.load_balanced_view()\n",
      "r = lview.map_async(simulator, range(100))\n",
      "print r.get()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "['None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ', 'None, ']\n"
       ]
      }
     ],
     "prompt_number": 30
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [],
     "language": "python",
     "metadata": {},
     "outputs": []
    }
   ],
   "metadata": {}
  }
 ]
}