使用jupyter(ipython)运行后台任务

Background Jobs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Simple interactive bacgkround jobs with IPython\n",
    "\n",
    "We start by loading the `backgroundjobs` library and defining a few trivial functions to illustrate things with."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from IPython.lib import backgroundjobs as bg\n",
    "\n",
    "import sys\n",
    "import time\n",
    "\n",
    "def sleepfunc(interval=2, *a, **kw):\n",
    "    args = dict(interval=interval,\n",
    "                args=a,\n",
    "                kwargs=kw)\n",
    "    time.sleep(interval)\n",
    "    return args\n",
    "\n",
    "def diefunc(interval=2, *a, **kw):\n",
    "    time.sleep(interval)\n",
    "    raise Exception(\"Dead job with interval %s\" % interval)\n",
    "\n",
    "def printfunc(interval=1, reps=5):\n",
    "    for n in range(reps):\n",
    "        time.sleep(interval)\n",
    "        print('In the background... %i' % n)\n",
    "        sys.stdout.flush()\n",
    "    print('All done!')\n",
    "    sys.stdout.flush()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we can create a job manager (called simply `jobs`) and use it to submit new jobs.\n",
    "\n",
    "Run the cell below, it will show when the jobs start. Wait a few seconds until you see the 'all done' completion message:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting job # 0 in a separate thread.\n",
      "Starting job # 2 in a separate thread.\n",
      "Starting job # 3 in a separate thread.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<BackgroundJob #3: printfunc(1,3)>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "In the background... 0\n",
      "In the background... 1\n",
      "In the background... 2\n",
      "All done!\n"
     ]
    }
   ],
   "source": [
    "jobs = bg.BackgroundJobManager()\n",
    "\n",
    "# Start a few jobs, the first one will have ID # 0\n",
    "jobs.new(sleepfunc, 4)\n",
    "jobs.new(sleepfunc, kw={'reps':2})\n",
    "jobs.new('printfunc(1,3)')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can check the status of your jobs at any time:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Completed jobs:\n",
      "0 : <function sleepfunc at 0x10521f2f0>\n",
      "2 : <function sleepfunc at 0x10521f2f0>\n",
      "3 : printfunc(1,3)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "jobs.status()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For any completed job, you can get its result easily:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'args': (), 'interval': 4, 'kwargs': {}}"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "jobs[0].result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Errors and tracebacks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The jobs manager tries to help you with debugging:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting job # 4 in a separate thread.\n",
      "Starting job # 5 in a separate thread.\n"
     ]
    }
   ],
   "source": [
    "# This makes a couple of jobs which will die.  Let's keep a reference to\n",
    "# them for easier traceback reporting later\n",
    "diejob1 = jobs.new(diefunc, 1)\n",
    "diejob2 = jobs.new(diefunc, 2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can get the traceback of any dead job.  Run the line\n",
    "below again interactively until it prints a traceback (check the status\n",
    "of the job):\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Status of diejob1: Dead (Exception), call jobs.traceback() for details\n",
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n",
      "\u001b[1;31mException\u001b[0m                                 Traceback (most recent call last)\n",
      "\u001b[1;32m/Users/minrk/dev/ip/mine/IPython/lib/backgroundjobs.py\u001b[0m in \u001b[0;36mcall\u001b[1;34m(self)\u001b[0m\n",
      "\u001b[0;32m    489\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m    490\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mcall\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m--> 491\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0margs\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0m\n",
      "\u001b[1;32m<ipython-input-1-169e49434ce0>\u001b[0m in \u001b[0;36mdiefunc\u001b[1;34m(interval, *a, **kw)\u001b[0m\n",
      "\u001b[0;32m     13\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mdiefunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m2\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m     14\u001b[0m     \u001b[0mtime\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msleep\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m---> 15\u001b[1;33m     \u001b[1;32mraise\u001b[0m \u001b[0mException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"Dead job with interval %s\"\u001b[0m \u001b[1;33m%\u001b[0m \u001b[0minterval\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0m\u001b[0;32m     16\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m     17\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mprintfunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mreps\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m5\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\n",
      "\u001b[1;31mException\u001b[0m: Dead job with interval 1\n"
     ]
    }
   ],
   "source": [
    "print(\"Status of diejob1: %s\" % diejob1.status)\n",
    "diejob1.traceback()  # jobs.traceback(4) would also work here, with the job number"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This will print all tracebacks for all dead jobs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Traceback for: <BackgroundJob #4: <function diefunc at 0x10521f7b8>>\n",
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n",
      "\u001b[1;31mException\u001b[0m                                 Traceback (most recent call last)\n",
      "\u001b[1;32m/Users/minrk/dev/ip/mine/IPython/lib/backgroundjobs.py\u001b[0m in \u001b[0;36mcall\u001b[1;34m(self)\u001b[0m\n",
      "\u001b[0;32m    489\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m    490\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mcall\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m--> 491\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0margs\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0m\n",
      "\u001b[1;32m<ipython-input-1-169e49434ce0>\u001b[0m in \u001b[0;36mdiefunc\u001b[1;34m(interval, *a, **kw)\u001b[0m\n",
      "\u001b[0;32m     13\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mdiefunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m2\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m     14\u001b[0m     \u001b[0mtime\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msleep\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m---> 15\u001b[1;33m     \u001b[1;32mraise\u001b[0m \u001b[0mException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"Dead job with interval %s\"\u001b[0m \u001b[1;33m%\u001b[0m \u001b[0minterval\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0m\u001b[0;32m     16\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m     17\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mprintfunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mreps\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m5\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\n",
      "\u001b[1;31mException\u001b[0m: Dead job with interval 1\n",
      "\n",
      "Traceback for: <BackgroundJob #5: <function diefunc at 0x10521f7b8>>\n",
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m\n",
      "\u001b[1;31mException\u001b[0m                                 Traceback (most recent call last)\n",
      "\u001b[1;32m/Users/minrk/dev/ip/mine/IPython/lib/backgroundjobs.py\u001b[0m in \u001b[0;36mcall\u001b[1;34m(self)\u001b[0m\n",
      "\u001b[0;32m    489\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m    490\u001b[0m     \u001b[1;32mdef\u001b[0m \u001b[0mcall\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m--> 491\u001b[1;33m         \u001b[1;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mfunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0margs\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0m\n",
      "\u001b[1;32m<ipython-input-1-169e49434ce0>\u001b[0m in \u001b[0;36mdiefunc\u001b[1;34m(interval, *a, **kw)\u001b[0m\n",
      "\u001b[0;32m     13\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mdiefunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m2\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m     14\u001b[0m     \u001b[0mtime\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msleep\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m---> 15\u001b[1;33m     \u001b[1;32mraise\u001b[0m \u001b[0mException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"Dead job with interval %s\"\u001b[0m \u001b[1;33m%\u001b[0m \u001b[0minterval\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0m\u001b[0;32m     16\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m     17\u001b[0m \u001b[1;32mdef\u001b[0m \u001b[0mprintfunc\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0minterval\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mreps\u001b[0m\u001b[1;33m=\u001b[0m\u001b[1;36m5\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\n",
      "\u001b[1;31mException\u001b[0m: Dead job with interval 2\n",
      "\n"
     ]
    }
   ],
   "source": [
    "jobs.traceback()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The job manager can be flushed of all completed jobs at any time:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Flushing 3 Completed jobs.\n",
      "Flushing 2 Dead jobs.\n"
     ]
    }
   ],
   "source": [
    "jobs.flush()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After that, the status is simply empty:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "jobs.status()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Jobs have a `.join` method that lets you wait on their thread for completion:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting job # 0 in a separate thread.\n"
     ]
    }
   ],
   "source": [
    "j = jobs.new(sleepfunc, 2)\n",
    "j.join?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise\n",
    "\n",
    "1. Start a new job that calls `sleepfunc` with a 5-second wait\n",
    "2. Print a short message that indicates you are waiting (note: you'll need to flush stdout to see that print output appear).\n",
    "3. Wait on the job and then print its result."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.4.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}