Add basic test suite to background jobs library.

IPython/lib/tests/test_backgroundjobs.py

@@ -0,0 +1,91 b''
+"""Tests for pylab tools module.
+"""
+#-----------------------------------------------------------------------------
+# Copyright (c) 2011, the IPython Development Team.
+#
+# Distributed under the terms of the Modified BSD License.
+#
+# The full license is in the file COPYING.txt, distributed with this software.
+#-----------------------------------------------------------------------------
+
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
+from __future__ import print_function
+
+# Stdlib imports
+import sys
+import time
+
+# Third-party imports
+import nose.tools as nt
+
+# Our own imports
+from IPython.lib import backgroundjobs as bg
+from IPython.testing import decorators as dec
+
+#-----------------------------------------------------------------------------
+# Globals and constants
+#-----------------------------------------------------------------------------
+t_short = 0.0001  # very short interval to wait on jobs
+
+#-----------------------------------------------------------------------------
+# Local utilities
+#-----------------------------------------------------------------------------
+def sleeper(interval=t_short, *a, **kw):
+    args = dict(interval=interval,
+                other_args=a,
+                kw_args=kw)
+    time.sleep(interval)
+    return args
+
+def crasher(interval=t_short, *a, **kw):
+    time.sleep(interval)
+    raise Exception("Dead job with interval %s" % interval)
+
+#-----------------------------------------------------------------------------
+# Classes and functions
+#-----------------------------------------------------------------------------
+
+def test_result():
+    """Test job submission and result retrieval"""
+    jobs = bg.BackgroundJobManager()
+    j = jobs.new(sleeper)
+    j.join()
+    nt.assert_equals(j.result['interval'], t_short)
+    
+
+def test_flush():
+    """Test job control"""
+    jobs = bg.BackgroundJobManager()
+    j = jobs.new(sleeper)
+    j.join()
+    nt.assert_equals(len(jobs.completed), 1)
+    nt.assert_equals(len(jobs.dead), 0)
+    jobs.flush()
+    nt.assert_equals(len(jobs.completed), 0)
+    
+
+def test_dead():
+    """Test control of dead jobs"""
+    jobs = bg.BackgroundJobManager()
+    j = jobs.new(crasher)
+    j.join()
+    nt.assert_equals(len(jobs.completed), 0)
+    nt.assert_equals(len(jobs.dead), 1)
+    jobs.flush()    
+    nt.assert_equals(len(jobs.dead), 0)
+
+
+def test_longer():
+    """Test control of longer-running jobs"""
+    jobs = bg.BackgroundJobManager()
+    # Sleep for long enough for the following two checks to still report the
+    # job as running, but not so long that it makes the test suite noticeably
+    # slower. 
+    j = jobs.new(sleeper, 0.1)
+    nt.assert_equals(len(jobs.running), 1)
+    nt.assert_equals(len(jobs.completed), 0)
+    j.join()
+    nt.assert_equals(len(jobs.running), 0)
+    nt.assert_equals(len(jobs.completed), 1)

IPython/lib/backgroundjobs.py

@@ -70,10 +70,11 b' class BackgroundJobManager(object):'
     """
 
     def __init__(self):
-        # Lists for job management
-        self.running  = []
-        self.completed = []
-        self.dead = []
+        # Lists for job management, accessed via a property to ensure they're
+        # up to date.x
+        self._running  = []
+        self._completed = []
+        self._dead = []
         # A dict of all jobs, so users can easily access any of them
         self.all = {}
         # For reporting
@@ -85,6 +86,21 b' class BackgroundJobManager(object):'
         self._s_completed = BackgroundJobBase.stat_completed_c
         self._s_dead      = BackgroundJobBase.stat_dead_c
 
+    @property
+    def running(self):
+        self._update_status()
+        return self._running
+
+    @property
+    def dead(self):
+        self._update_status()
+        return self._dead
+
+    @property
+    def completed(self):
+        self._update_status()
+        return self._completed
+
     def new(self,func_or_exp,*args,**kwargs):
         """Add a new background job and start it in a separate thread.
 
@@ -196,22 +212,27 b' class BackgroundJobManager(object):'
         are used to report jobs completed/dead since the last update, and are
         then cleared by the reporting function after each call."""
         
-        run,comp,dead = self._s_running,self._s_completed,self._s_dead
-        running = self.running
-        for num in range(len(running)):
-            job  = running[num]
+        # Status codes
+        srun, scomp, sdead = self._s_running, self._s_completed, self._s_dead
+        # State lists, use the actual lists b/c the public names are properties
+        # that call this very function on access
+        running, completed, dead = self._running, self._completed, self._dead
+
+        # Now, update all state lists
+        for num, job in enumerate(running):
             stat = job.stat_code
-            if stat == run:
+            if stat == srun:
                 continue
-            elif stat == comp:
-                self.completed.append(job)
+            elif stat == scomp:
+                completed.append(job)
                 self._comp_report.append(job)
                 running[num] = False
-            elif stat == dead:
-                self.dead.append(job)
+            elif stat == sdead:
+                dead.append(job)
                 self._dead_report.append(job)
                 running[num] = False
-        self.running = filter(None,self.running)
+        # Remove dead/completed jobs from running list
+        running[:] = filter(None, running)
 
     def _group_report(self,group,name):
         """Report summary for a given job group.
@@ -290,15 +311,10 b' class BackgroundJobManager(object):'
         completed since the last _status_new() call, the flush operation
         aborts."""
 
-        if self._status_new():
-            error('New jobs completed since last '\
-                  '_status_new(), aborting flush.')
-            return
-
         # Remove the finished jobs from the master dict
-        all = self.all
+        alljobs = self.all
         for job in self.completed+self.dead:
-            del(all[job.num])
+            del(alljobs[job.num])
 
         # Now flush these lists completely
         fl_comp = self._group_flush(self.completed, 'Completed')

docs/examples/lib/BackgroundJobs.ipynb

@@ -11,7 +11,7 b''
                     "language": "python", 
                     "outputs": [], 
                     "collapsed": false, 
-                    "prompt_number": 15, 
+                    "prompt_number": 35, 
                     "input": "from IPython.lib import backgroundjobs as bg\n\nimport sys\nimport time\n\ndef sleepfunc(interval=2, *a, **kw):\n    args = dict(interval=interval,\n                args=a,\n                kwargs=kw)\n    time.sleep(interval)\n    return args\n\ndef diefunc(interval=2, *a, **kw):\n    time.sleep(interval)\n    raise Exception(\"Dead job with interval %s\" % interval)\n\ndef printfunc(interval=1, reps=5):\n    for n in range(reps):\n        time.sleep(interval)\n        print 'In the background...', n\n        sys.stdout.flush()\n    print 'All done!'\n    sys.stdout.flush()"
                 }, 
                 {
@@ -49,7 +49,7 b''
                         }
                     ], 
                     "collapsed": false, 
-                    "prompt_number": 28, 
+                    "prompt_number": 36, 
                     "input": "jobs = bg.BackgroundJobManager()\n\n# Start a few jobs, the first one will have ID # 0\njobs.new(sleepfunc, 4)\njobs.new(sleepfunc, kw={'reps':2})\njobs.new('printfunc(1,3)')\n\n# This makes a couple of jobs which will die.  Let's keep a reference to\n# them for easier traceback reporting later\ndiejob1 = jobs.new(diefunc, 1)\ndiejob2 = jobs.new(diefunc, 2)"
                 }, 
                 {
@@ -63,11 +63,11 b''
                         {
                             "output_type": "stream", 
                             "stream": "stdout", 
-                            "text": "Completed jobs:\n0 : <function sleepfunc at 0x30c8500>\n2 : <function sleepfunc at 0x30c8500>\n3 : printfunc(1,3)\n\nDead jobs:\n4 : <function diefunc at 0x30df758>\n5 : <function diefunc at 0x30df758>\n\n"
+                            "text": "Completed jobs:\n0 : <function sleepfunc at 0x30e1578>\n2 : <function sleepfunc at 0x30e1578>\n3 : printfunc(1,3)\n\nDead jobs:\n4 : <function diefunc at 0x304d488>\n5 : <function diefunc at 0x304d488>\n\n"
                         }
                     ], 
                     "collapsed": false, 
-                    "prompt_number": 29, 
+                    "prompt_number": 37, 
                     "input": "jobs.status()"
                 }, 
                 {
@@ -77,16 +77,10 b''
                 {
                     "cell_type": "code", 
                     "language": "python", 
-                    "outputs": [
-                        {
-                            "output_type": "pyout", 
-                            "prompt_number": 31, 
-                            "text": "{'args': (), 'interval': 4, 'kwargs': {}}"
-                        }
-                    ], 
+                    "outputs": [], 
                     "collapsed": false, 
-                    "prompt_number": 31, 
-                    "input": "jobs[0].result"
+                    "prompt_number": 43, 
+                    "input": "jobs[0].result\nj0 = jobs[0]\nj0.join?"
                 }, 
                 {
                     "source": "You can get the traceback of any dead job.  Run the line\nbelow again interactively until it prints a traceback (check the status\nof the job):\n", 
@@ -155,6 +149,34 b''
                     "input": "jobs.status()"
                 }, 
                 {
+                    "source": "It's easy to wait on a job:", 
+                    "cell_type": "markdown"
+                }, 
+                {
+                    "cell_type": "code", 
+                    "language": "python", 
+                    "outputs": [
+                        {
+                            "output_type": "stream", 
+                            "stream": "stdout", 
+                            "text": "Starting job # 7 in a separate thread.\nWill wait for j now...\n"
+                        }, 
+                        {
+                            "output_type": "stream", 
+                            "stream": "stdout", 
+                            "text": "Result from j:\n"
+                        }, 
+                        {
+                            "output_type": "pyout", 
+                            "prompt_number": 46, 
+                            "text": "{'args': (), 'interval': 2, 'kwargs': {}}"
+                        }
+                    ], 
+                    "collapsed": false, 
+                    "prompt_number": 46, 
+                    "input": "j = jobs.new(sleepfunc, 2)\nprint \"Will wait for j now...\"\nsys.stdout.flush()\nj.join()\nprint \"Result from j:\"\nj.result"
+                }, 
+                {
                     "input": "", 
                     "cell_type": "code", 
                     "collapsed": true,