Quick demonstration of remotely calling the same function a certain number of times.

multi_apply

{
 "metadata": {
  "name": ""
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import time\n",
      "\n",
      "from itertools import cycle, izip\n",
      "\n",
      "from IPython.parallel import Client"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "rc = Client()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "dv = rc.direct_view()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "lv = rc.load_balanced_view()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 4
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def iter_async_results(async_results, timeout=1E-03):\n",
      "    \"\"\"\n",
      "    This performs the same as iterating over a map_async result but is intended\n",
      "    for a list of AsyncResult instances that were generated from calling\n",
      "    view.apply many times. (This is useful for calling a remote function\n",
      "    without arguments.)\n",
      "    \"\"\"\n",
      "    pending = set(async_results)\n",
      "    while pending:\n",
      "        ready = set(ar for ar in pending if ar.ready())\n",
      "        if not ready:\n",
      "            time.sleep(timeout)\n",
      "            continue\n",
      "        # update pending to exclude those that just finished\n",
      "        pending.difference_update(ready)\n",
      "        for ar in ready:\n",
      "            # we know these are done, so don't worry about blocking\n",
      "            yield ar.get()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 5
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def multi_apply(view, function, num, *args, **kw_args):\n",
      "    targets = view.client._build_targets(view.targets)[1]\n",
      "    views = [view.client[t] for t in targets]\n",
      "    results = [v.apply_async(function, *args, **kw_args) for (v, i) in izip(cycle(views), xrange(num))]\n",
      "    return iter_async_results(results)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 6
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def test_function():\n",
      "    time.sleep(1E-04)\n",
      "    return a * b"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 7
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "dv.execute(\"import time\", block=True);"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 8
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "dv.push(dict(a=1, b=1), block=True);"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "num = int(1E04)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 10
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%timeit\n",
      "results = multi_apply(dv, test_function, num)\n",
      "res = sum(results)\n",
      "assert num == res"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "1 loops, best of 3: 29.5 s per loop\n"
       ]
      }
     ],
     "prompt_number": 11
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "a = 1\n",
      "b = 1"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 12
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "%%timeit\n",
      "assert num == sum(test_function() for i in xrange(num))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "1 loops, best of 3: 1.96 s per loop\n"
       ]
      }
     ],
     "prompt_number": 13
    }
   ],
   "metadata": {}
  }
 ]
}