03-ipyparallel-tutorial-IPython-task-interface.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "toc": "true"
   },
   "source": [
    "# Table of Contents\n",
    " <p><div class=\"lev1\"><a href=\"#The-IPython-task-interface\"><span class=\"toc-item-num\">1&nbsp;&nbsp;</span>The IPython task interface</a></div><div class=\"lev2\"><a href=\"#Starting-the-IPython-controller-and-engines\"><span class=\"toc-item-num\">1.1&nbsp;&nbsp;</span>Starting the IPython controller and engines</a></div><div class=\"lev2\"><a href=\"#Creating-a-LoadBalancedView-instance\"><span class=\"toc-item-num\">1.2&nbsp;&nbsp;</span>Creating a LoadBalancedView instance</a></div><div class=\"lev2\"><a href=\"#Quick-and-easy-parallelism\"><span class=\"toc-item-num\">1.3&nbsp;&nbsp;</span>Quick and easy parallelism</a></div><div class=\"lev3\"><a href=\"#Parallel-map\"><span class=\"toc-item-num\">1.3.1&nbsp;&nbsp;</span>Parallel map</a></div><div class=\"lev3\"><a href=\"#Parallel-function-decorator\"><span class=\"toc-item-num\">1.3.2&nbsp;&nbsp;</span>Parallel function decorator</a></div><div class=\"lev2\"><a href=\"#Dependencies\"><span class=\"toc-item-num\">1.4&nbsp;&nbsp;</span>Dependencies</a></div><div class=\"lev3\"><a href=\"#Functional-Dependencies\"><span class=\"toc-item-num\">1.4.1&nbsp;&nbsp;</span>Functional Dependencies</a></div><div class=\"lev4\"><a href=\"#@require\"><span class=\"toc-item-num\">1.4.1.1&nbsp;&nbsp;</span>@require</a></div><div class=\"lev4\"><a href=\"#@depend\"><span class=\"toc-item-num\">1.4.1.2&nbsp;&nbsp;</span>@depend</a></div><div class=\"lev3\"><a href=\"#other-dependencies\"><span class=\"toc-item-num\">1.4.2&nbsp;&nbsp;</span>other dependencies</a></div><div class=\"lev2\"><a href=\"#Retries-and-Resubmit\"><span class=\"toc-item-num\">1.5&nbsp;&nbsp;</span>Retries and Resubmit</a></div><div class=\"lev3\"><a href=\"#Retries\"><span class=\"toc-item-num\">1.5.1&nbsp;&nbsp;</span>Retries</a></div><div class=\"lev3\"><a href=\"#Resubmit\"><span class=\"toc-item-num\">1.5.2&nbsp;&nbsp;</span>Resubmit</a></div><div class=\"lev2\"><a href=\"#Schedulers\"><span class=\"toc-item-num\">1.6&nbsp;&nbsp;</span>Schedulers</a></div><div class=\"lev2\"><a href=\"#summary\"><span class=\"toc-item-num\">1.7&nbsp;&nbsp;</span>summary</a></div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# The IPython task interface\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "The task interface to the cluster presents the engines as a fault tolerant, dynamic load-balanced system of workers. Unlike the multiengine interface, in the task interface the user have no direct access to individual engines. By allowing the IPython scheduler to assign work, this interface is simultaneously simpler and more powerful.\n",
    "\n",
    "Best of all, the user can use both of these interfaces running at the same time to take advantage of their respective strengths. When the user can break up the user’s work into segments that do not depend on previous execution, the task interface is ideal. But it also has more power and flexibility, allowing the user to guide the distribution of jobs, without having to assign tasks to engines explicitly."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Starting the IPython controller and engines\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To follow along with this tutorial, you will need to start the IPython controller and four IPython engines. The simplest way of doing this is to use the ipcluster command:\n",
    "\n",
    "```\n",
    "$ ipcluster start -n 4 --profile='ipython-task-interface'\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a LoadBalancedView instance\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first step is to import the IPython ipyparallel module and then create a Client instance, and we will also be using a LoadBalancedView, here called lview:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:22.410485",
     "start_time": "2017-04-14T14:37:22.340430+02:00"
    },
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from ipyparallel import Client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:22.812436",
     "start_time": "2017-04-14T14:37:22.717836+02:00"
    },
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rc = Client(profile='ipython-task-interface')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For load-balanced execution, we will make use of a LoadBalancedView object, which can be constructed via the client’s load_balanced_view() method:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:23.780336",
     "start_time": "2017-04-14T14:37:23.776724+02:00"
    },
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# default load-balanced view\n",
    "lview = rc.load_balanced_view()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Quick and easy parallelism\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In many cases, you simply want to apply a Python function to a sequence of objects, but in parallel. Like the multiengine interface, these can be implemented via the task interface. The exact same tools can perform these actions in load-balanced ways as well as multiplexed ways: a parallel version of map() and @parallel() function decorator. If one specifies the argument balanced=True, then they are dynamically load balanced. Thus, if the execution time per item varies significantly, you should use the versions in the task interface."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Parallel map"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To load-balance map(),simply use a LoadBalancedView:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:41.207240",
     "start_time": "2017-04-14T14:37:40.703351+02:00"
    },
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lview.block = True\n",
    "serial_result = map(lambda x:x**10, range(32))\n",
    "parallel_result = lview.map(lambda x:x**10, range(32))\n",
    "list(serial_result)==parallel_result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-05T14:34:46.571467",
     "start_time": "2017-04-05T16:34:46.565315+02:00"
    }
   },
   "source": [
    "### Parallel function decorator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Parallel functions are just like normal function, but they can be called on sequences and in parallel. The multiengine interface provides a decorator that turns any Python function into a parallel function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:43.812513",
     "start_time": "2017-04-14T14:37:43.588280+02:00"
    },
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0.0,\n",
       " 10.0,\n",
       " 160.0,\n",
       " 810.0,\n",
       " 2560.0,\n",
       " 6250.0,\n",
       " 12960.0,\n",
       " 24010.0,\n",
       " 40960.0,\n",
       " 65610.0,\n",
       " 100000.0,\n",
       " 146410.0,\n",
       " 207360.0,\n",
       " 285610.0,\n",
       " 384160.0,\n",
       " 506250.0,\n",
       " 655360.0,\n",
       " 835210.0,\n",
       " 1049760.0,\n",
       " 1303210.0,\n",
       " 1600000.0,\n",
       " 1944810.0,\n",
       " 2342560.0,\n",
       " 2798410.0,\n",
       " 3317760.0,\n",
       " 3906250.0,\n",
       " 4569760.0,\n",
       " 5314410.0,\n",
       " 6146560.0,\n",
       " 7072810.0,\n",
       " 8100000.0,\n",
       " 9235210.0]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@lview.parallel()\n",
    "def f(x):\n",
    "    return 10.0*x**4\n",
    "\n",
    "# this is done in parallel\n",
    "f.map(range(32))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dependencies"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Often, pure atomic load-balancing is too primitive for your work. In these cases, you may want to associate some kind of Dependency that describes when, where, or whether a task can be run."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Functional Dependencies"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Functional dependencies are used to determine whether a given engine is capable of running a particular task. This is implemented via a special Exception class, UnmetDependency, found in ipyparallel.error. Its use is very simple: if a task fails with an UnmetDependency exception, then the scheduler, instead of relaying the error up to the client like any other error, catches the error, and submits the task to a different engine. This will repeat indefinitely, and a task will never be submitted to a given engine a second time.\n",
    "\n",
    "You can manually raise the UnmetDependency yourself, but IPython has provided some decorators for facilitating this behaviour.\n",
    "\n",
    "There are two decorators and a class used for functional dependencies:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:54.693124",
     "start_time": "2017-04-14T14:37:54.689378+02:00"
    },
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from ipyparallel import depend, require, dependent"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### @require"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The simplest sort of dependency is requiring that a Python module is available. The @require decorator lets you define a function that will only run on engines where names you specify are importable:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:55.558023",
     "start_time": "2017-04-14T14:37:55.553503+02:00"
    },
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "@require('numpy', 'zmq')\n",
    "def myfunc():\n",
    "    return dostuff()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, any time you apply myfunc(), the task will only run on a machine that has numpy and pyzmq available, and when myfunc() is called, numpy and zmq will be imported. You can also require specific objects, not just module names:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:57.483189",
     "start_time": "2017-04-14T14:37:57.473298+02:00"
    },
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def foo(a):\n",
    "    return a*a\n",
    "\n",
    "@require(foo)\n",
    "def bar(b):\n",
    "    return foo(b)\n",
    "\n",
    "@require(bar)\n",
    "def baz(c, d):\n",
    "    return bar(c) - bar(d)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:37:58.439781",
     "start_time": "2017-04-14T14:37:58.416444+02:00"
    },
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "-9"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lview.apply_sync(baz, 4, 5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### @depend"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "The @depend decorator lets you decorate any function with any other function to evaluate the dependency. The dependency function will be called at the start of the task, and if it returns False, then the dependency will be considered unmet, and the task will be assigned to another engine. If the dependency returns anything other than ``False``, the rest of the task will continue."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2017-04-14T12:38:06.063144",
     "start_time": "2017-04-14T14:38:06.056594+02:00"
    },
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def platform_specific(plat):\n",
    "    import sys\n",
    "    return sys.platform == plat\n",
    "\n",
    "@depend(platform_specific, 'darwin')\n",
    "def mactask():\n",
    "    do_mac_stuff()\n",
    "\n",
    "@depend(platform_specific, 'nt')\n",
    "def wintask(win):\n",
    "    do_windows_stuff()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### other dependencies"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Other dependencies are outside the scope of this tutorial (because you probably won't need them), these include: graph and impossible dependencies."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Retries and Resubmit"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Retries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Another flag for tasks is retries. This is an integer, specifying how many times a task should be resubmitted after failure. This is useful for tasks that should still run if their engine was shutdown, or may have some statistical chance of failing. The default is to not retry tasks."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resubmit"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sometimes you may want to re-run a task. This could be because it failed for some reason, and you have fixed the error, or because you want to restore the cluster to an interrupted state. For this, the Client has a rc.resubmit() method. This simply takes one or more msg_ids, and returns an AsyncHubResult for the result(s). You cannot resubmit a task that is pending - only those that have finished, either successful or unsuccessful."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Schedulers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are a variety of valid ways to determine where jobs should be assigned in a load-balancing situation. In IPython, we support several standard schemes, and even make it easy to define your own. The scheme can be selected via the scheme argument to ipcontroller, or in the TaskScheduler.schemename attribute of a controller config object.\n",
    "\n",
    "By default the leastload scheme is used: always assign tasks to the engine with the fewest outstanding tasks."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## summary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The LoadBalancedView has many more powerful features that allow quite a bit of flexibility in how tasks are defined and run. The next places to look are in the following classes:\n",
    "\n",
    "- LoadBalancedView\n",
    "- AsyncResult\n",
    "- apply()\n",
    "- dependency\n",
    "\n",
    "The following is an overview of how to use these classes together:\n",
    "\n",
    "- Create a Client and LoadBalancedView\n",
    "- Define some functions to be run as tasks\n",
    "- Submit your tasks to using the apply() method of your LoadBalancedView instance.\n",
    "- Use Client.get_result() to get the results of the tasks, or use the AsyncResult.get() method of the results to wait for and then receive the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "hide_input": false,
  "kernelspec": {
   "display_name": "Python [conda env:ipyparallelseminar]",
   "language": "python",
   "name": "conda-env-ipyparallelseminar-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  },
  "toc": {
   "toc_cell": true,
   "toc_number_sections": true,
   "toc_section_display": "none",
   "toc_threshold": 6,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}