{
 "metadata": {
  "name": "Part 6 - Example - Remote Iteration"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "N-Way merge with remote data"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "How to merge multiple *sorted* remote data streams using the `heapq.merge` function that ships with Python."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from __future__ import print_function\n",
      "import heapq\n",
      "\n",
      "from IPython.display import display\n",
      "from IPython import parallel\n",
      "\n",
      "rc = parallel.Client()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Imagine we have some routine that is capable of loading/creating a sorted subset of our data in an engine, based on a parameter (such as the indes of which part of the data to read):"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def load_data(arg):\n",
      "    \"\"\"Load a dataset in the global namespace. The dataset *must* be sorted.\n",
      "\n",
      "    Return the *name* of the variable in which the dataset was loaded.\"\"\"\n",
      "    global data\n",
      "    # Here, real data loading would occur\n",
      "    s =  4-arg\n",
      "    step = arg+1\n",
      "    data = range(s, s+4*step**2, step)\n",
      "    return 'data'"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Exercise"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We want a function that takes a given single-engine View and a variable name,\n",
      "and returns a local iterator on the remote object.\n",
      "It should look something like this skeleton function:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def remote_iterator(view, name):\n",
      "    \"\"\"Return an iterator on an object living on a remote engine.\"\"\"\n",
      "    # TODO: create an iterator remotely\n",
      "    while True:\n",
      "        pass\n",
      "        # TODO: yield the next item\n",
      "        # TODO: turn remote StopIteration into local StopIteration"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "#### Relevant Aside:\n",
      "\n",
      "Errors raised on engines will show up in the Client as a RemoteError.\n",
      "This means you have to be a little careful when trying to catch remote errors:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "try:\n",
      "    rc[-1].execute(\"foo = barbarbar\", block=True)\n",
      "except NameError:\n",
      "    print(\"caught NameError\")\n",
      "except Exception as e:\n",
      "    print(\"Oops! Didn't catch %r\" % e)\n",
      "    raise e\n",
      "print(\"safe and sound\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "ename": "RemoteError",
       "evalue": "NameError(name 'barbarbar' is not defined)",
       "output_type": "pyerr",
       "traceback": [
        "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
        "\u001b[1;31mNameError\u001b[0m                                 Traceback (most recent call last)\u001b[1;32m<ipython-input-7-0b69f8403738>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m",
        "\u001b[1;32m----> 1\u001b[1;33m \u001b[0mfoo\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mbarbarbar\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m",
        "\u001b[0m\u001b[1;31mNameError\u001b[0m: name 'barbarbar' is not defined"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Oops! Didn't catch <Remote[3]:NameError(name 'barbarbar' is not defined)>\n"
       ]
      }
     ],
     "prompt_number": 4
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "A RemoteError has three attributes:\n",
      "\n",
      "* `err.ename` - the class name of the remote error (e.g. `NameError`, `ValueError`)\n",
      "* `err.evalue` - the string value of the error message\n",
      "* `err.traceback` - the remote traceback as a list of strings"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "For simple builtin exceptions,\n",
      "you can re-raise remote errors as the original exception class with a case like the following:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def assign_foo():\n",
      "    try:\n",
      "        rc[-1].execute(\"foo = barbarbar\", block=True)\n",
      "    except parallel.RemoteError as e:\n",
      "        if e.ename == 'NameError':\n",
      "            raise NameError(e.evalue)\n",
      "        else:\n",
      "            raise e"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 5
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "By doing this re-cast, any exception handling outside will handle remote exceptions as if they were local."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "try:\n",
      "    assign_foo()\n",
      "except NameError:\n",
      "    print(\"caught NameError\")\n",
      "except Exception as e:\n",
      "    print(\"Oops! Didn't catch %r\" % e)\n",
      "    raise e\n",
      "print(\"safe and sound\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "caught NameError\n",
        "safe and sound\n"
       ]
      }
     ],
     "prompt_number": 48
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Can you fill out this remote_iterator function?\n",
      "\n",
      "Potentially useful:\n",
      "\n",
      "* catching RemoteErrors\n",
      "* parallel.Reference\n",
      "* yield"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def remote_iterator(view, name):\n",
      "    \"\"\"Return an iterator on an object living on a remote engine.\"\"\"\n",
      "    # TODO: create an iterator remotely\n",
      "    while True:\n",
      "        pass\n",
      "        # TODO: yield the next item\n",
      "        # TODO: turn remote StopIteration into local StopIteration"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "A local example that should be a good guideline for the remote version:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%load soln/remote_iter_hint.py"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 49
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "And the solution:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%load soln/remote_iter.py"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 50
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "And an ever-so-slightly fancier solution:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%load soln/remote_iter_slightly_better.py"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 6
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now, we bring `IPython.parallel` into action:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "dview = rc.direct_view()\n",
      "print('Engine IDs:', rc.ids)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Engine IDs: [0, 1, 2, 3]\n"
       ]
      }
     ],
     "prompt_number": 8
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Load the data on the engines\n",
      "data_refs = dview.map(load_data, rc.ids)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "data_refs"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "pyout",
       "prompt_number": 10,
       "text": [
        "<AsyncMapResult: load_data>"
       ]
      }
     ],
     "prompt_number": 10
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "list(data_refs)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "pyout",
       "prompt_number": 11,
       "text": [
        "['data', 'data', 'data', 'data']"
       ]
      }
     ],
     "prompt_number": 11
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# And we now make a local object which represents the remote iterator\n",
      "iterators = [remote_iterator(rc[e], ref) for e,ref in zip(rc.ids, data_refs)]\n",
      "for it in iterators:\n",
      "    print(list(it))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "[4, 5, 6, 7]\n",
        "[3, 5, 7, 9, 11, 13, 15, 17]"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "[2, 5, 8, 11, 14, 17, 20, 23, 26, 29, 32, 35]"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "[1, 5, 9, 13, 17, 21, 25, 29, 33, 37, 41, 45, 49, 53, 57, 61]"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n"
       ]
      }
     ],
     "prompt_number": 12
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now, let's merge those datasets into a single sorted one:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "print('Locally merge the remote sets:')\n",
      "iterators = [remote_iterator(rc[e], ref) for e,ref in zip(rc.ids, data_refs)] \n",
      "remote = list(heapq.merge(*iterators))\n",
      "print(remote)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Locally merge the remote sets:\n",
        "[1, 2, 3, 4, 5, 5, 5, 5, 6, 7, 7, 8, 9, 9, 11, 11, 13, 13, 14, 15, 17, 17, 17, 20, 21, 23, 25, 26, 29, 29, 32, 33, 35, 37, 41, 45, 49, 53, 57, 61]"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n"
       ]
      }
     ],
     "prompt_number": 13
    },
    {
     "cell_type": "heading",
     "level": 4,
     "metadata": {},
     "source": [
      "Validation"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "repeat the operation by copying the data from the engines to our local namespace and doing a regular merge here:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Key step here: pull data from each engine:\n",
      "local_data = [rc[e][ref] for e,ref in zip(rc.ids, data_refs)]\n",
      "print('Local data:')\n",
      "for subset in local_data:\n",
      "    print(subset)\n",
      "print('Sorted:')\n",
      "local = list(heapq.merge(*local_data))\n",
      "print(local)\n",
      "print(\"local == remote: %s\" % (local==remote))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Local data:\n",
        "[4, 5, 6, 7]\n",
        "[3, 5, 7, 9, 11, 13, 15, 17]\n",
        "[2, 5, 8, 11, 14, 17, 20, 23, 26, 29, 32, 35]\n",
        "[1, 5, 9, 13, 17, 21, 25, 29, 33, 37, 41, 45, 49, 53, 57, 61]\n",
        "Sorted:\n",
        "[1, 2, 3, 4, 5, 5, 5, 5, 6, 7, 7, 8, 9, 9, 11, 11, 13, 13, 14, 15, 17, 17, 17, 20, 21, 23, 25, 26, 29, 29, 32, 33, 35, 37, 41, 45, 49, 53, 57, 61]\n",
        "local == remote: True\n"
       ]
      }
     ],
     "prompt_number": 14
    }
   ],
   "metadata": {}
  }
 ]
}