Changeset 91

Timestamp:

10/04/07 18:42:02 (1 year ago)

Author:

edsuom

Message:

Fixes, improvements, and tests for asynqueue.jobs

Files:

• projects/AsynQueue/trunk/asynqueue/jobs.py (modified) (8 diffs)
• projects/AsynQueue/trunk/asynqueue/tasks.py (modified) (2 diffs)
• projects/AsynQueue/trunk/asynqueue/test/test_base.py (modified) (1 diff)
• projects/AsynQueue/trunk/asynqueue/test/test_jobs.py (modified) (10 diffs)
• projects/AsynQueue/trunk/asynqueue/test/test_processworker.py (modified) (1 diff)
• projects/AsynQueue/trunk/asynqueue/test/test_tasks.py (modified) (1 diff)

projects/AsynQueue/trunk/asynqueue/jobs.py

@@ -30 +30 @@
-import base, workers
-
+VERBOSE = False
+def log(msgProto, *args):
+    if VERBOSE:
+        print msgProto % args
+
+
+class TrustError(Exception):
+    pass
+
-
-class ChildRoot(pb.Root):
@@ -48 +57 @@
-        return result
-
+    def oops(self, *arg):
+        """
+        Returns a C{False} status code for a remote call along with a string
+        traceback of the exception raised. You can supply your own exception or
+        L{Failure} instance. If you don't, the current exception will be used.
+        """
+        if arg and isinstance(arg[0], Failure):
+            failureObject = arg[0]
+        else:
+            failureObject = Failure(*arg)
+        return False, failureObject.getTraceback()
+
-    def remote_newJob(self, jobID, jobCode):
-        """
@@ -57 +78 @@
-        my process exits.
-
-
-
+
+
+
+
+
-        """
-        if not self.trusted:
-False
+self.oops(TrustError)
-        try:
-            namespace = {}
-            exec jobCode in namespace
-        except:
-False
+self.oops()
-        self.jobs[jobID] = namespace
-
+, [x[0] for x in namespace.iteritems() if callable(x[1])]
-    
-    def remote_runJob(self, jobID, callName, *args, **kw):
@@ -74 +98 @@
-        Calls the object that is present in the namespace of I{jobID} under the
-        specified I{callName}, with any supplied arguments and
-
+
+
+
+
+
+
+
-        """
-        if not self.trusted:
-aise RuntimeError("Caller is not trusted at this time"
+eturn self.oops(TrustError
-        calledObject = self.jobs[jobID].get(callName, None)
-        if callable(calledObject):
-
-
+
+
+
+
+
+
-            "No callable object '%s' defined in namespace for job %d" \
-
+)
-
-    def remote_exit(self):
@@ -109 +143 @@
-            kw['noTypeCheck'] = True
-        workers.RemoteCallWorker.__init__(self, *args, **kw)
-
+
-    def _runNow(self, null, task):
-        funcName, args, kw = task.callTuple
@@ -153 +187 @@
-        return self.queue.shutdown()
-
+    def jobTried(self, result, jobID, worker):
+        """
+        Callback for loading a new job or running a call on an existing one.
+
+        If the worker's root reference raised an unexpected failure, returns
+        C{False}. If everything went OK, returns C{True}. If there was a
+        failure that may not have been the worker's fault, returns C{None}.
+        """
+        if hasattr(result, 'check'):
+            # Oops, failure from a bogus root reference.
+            log("Worker %d supplied nonconforming root reference", worker.ID)
+            return False
+        if result[0]:
+            msg = "Callable objects: %s" % ", ".join(result[1])
+            log("Job %d loaded OK on worker %s\n%s", jobID, worker.ID, msg)
+            self.queue.qualifyWorker(worker, jobID)
+            return True
+        log("Job %d failed on worker %d:\n%s", jobID, worker.ID, result[1])
+        return None
+
-    def attachChild(self, childRoot, N=3):
-        """
@@ -158 +212 @@
-        PB root reference.
-
+        Tries to load all of the currently registered jobs on the worker. If an
+        unexpected failure (not a simple job-loading exception) arises, the
+        worker is not hired.
+
-        The default number (3) of job runs that the worker is willing to queue
-        up on its end can be overridden with the I{N} keyword.
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-        worker = ChildWorker(childRoot, N)
-result = [self.queue.attachWorker(worker)]
+self.queue.attachWorker(worker)
-        dList = []
-        for jobID, jobInfo in self.jobs.iteritems():
-            jobCode = jobInfo[0]
-            d = childRoot.callRemote('newJob', jobID, jobCode)
-Callback(tried, jobID
-Errback(oops
+Both(self.jobTried, jobID, worker
+Callback(jobTried
-            dList.append(d)
-lambda _: result[0]
+allDone
-    
-    def detachChild(self, childID):
@@ -206 +267 @@
-    
-        """
-        def tried(success, worker):
-            if success:
-                self.queue.qualifyWorker(worker, jobID)
-
-        jobID = self._jobCounter = getattr(self, '_jobCounter', 0) + 1
-(jobCode, niceness)
+[jobCode, niceness]
-        
-        dList = []
-        for worker in self.queue.workers():
-            d = worker.remoteCaller('newJob', jobID, jobCode)
-Callback(tried
+Both(self.jobTried, jobID
-            dList.append(d)
-        d = defer.DeferredList(dList)

projects/AsynQueue/trunk/asynqueue/tasks.py

@@ -290 +290 @@
-        request an initial assignment from each queue.
-
-
-
+
+
+
-        """
-        if not IWorker.providedBy(worker):
@@ -305 +306 @@
-        for series in qualifications:
-            self.assignmentFactory.request(worker, series)
-
+worker.ID = 
-        self._workerCounter = workerID
-        self.workers[workerID] = worker

projects/AsynQueue/trunk/asynqueue/test/test_base.py

@@ -35 +35 @@
-class Test_Priority(TestCase):
-    def setUp(self):
-
+base.
-
+    def test_getInOrder(self):
+        dList = []
+        for item in (2,1,4,0,3):
+            self.heap.put(item)
+        for item in xrange(5):
+            d = self.heap.get()
+            d.addCallback(self.failUnlessEqual, item)
+            dList.append(d)
+        return defer.DeferredList(dList)
+
+    def test_getBeforePut(self):
+        dList, items = [], []
+        for item in xrange(5):
+            self.heap.put(item)
+        for item in xrange(5):
+            d = self.heap.get()
+            d.addCallback(items.append)
+            dList.append(d)
+        return defer.DeferredList(dList).addCallback(
+            lambda _: self.failUnlessEqual(sum(items), 10))
+    
-    def test_cancel(self):
-
+
+
+
+
+
+
+
+
+
+
-
-

projects/AsynQueue/trunk/asynqueue/test/test_jobs.py

@@ -33 +33 @@
-def test(a, b, c=0):
-    return a + 2*b + 3*c
+
+def bogusable(x):
+    return 1.0 / x
+
-"""
+
+
+class Test_ChildRoot(mock.TestCase):
+    def setUp(self):
+        self.root = jobs.ChildRoot()
+        self.root.trusted = True
+
+    def test_newJob_OK(self):
+        result = self.root.remote_newJob(JOB_ID, JOB_CODE)
+        self.failUnlessEqual(result[0], True)
+        self.failUnlessElementsEqual(result[1], ['test', 'bogusable'])
+    
+    def test_newJob_bogus(self):
+        bogusJobCode = JOB_CODE + "\nbogus\n"
+        result = self.root.remote_newJob(JOB_ID, bogusJobCode)
+        self.failUnlessEqual(result[0], False)
+        self.failUnless('bogus' in result[1])
+
+    def test_runJob_OK(self):
+        def check(result):
+            self.failUnlessEqual(result[0], True)
+            self.failUnlessEqual(result[1], 600)
+
+        self.root.remote_newJob(JOB_ID, JOB_CODE)
+        d = self.root.remote_runJob(JOB_ID, 'test', 100, 100, 100)
+        d.addCallback(check)
+        return d
+
+    def test_runJob_bogus(self):
+        def check(result):
+            self.failUnlessEqual(result[0], False)
+            self.failUnless('ZeroDivisionError' in result[1])
+
+        self.root.remote_newJob(JOB_ID, JOB_CODE)
+        d = self.root.remote_runJob(JOB_ID, 'bogusable', 0)
+        d.addCallback(check)
+        return d
-
-
@@ -51 +92 @@
-            exec jobCode in namespace
-            self.jobs[jobID] = namespace
-
+
+
-        elif not self.bogus and called == 'runJob':
-            namespace = self.jobs[args[0]]
-            calledObject = namespace[args[1]]
-
+
+
+
+
-        elif called == 'exit':
-            result = None
@@ -83 +128 @@
-    def setUp(self):
-        self.root = Mock_Root()
+        self.root.clearCalls()
-        self.worker = jobs.ChildWorker(self.root, noTypeCheck=True)
-        return self.root.callRemote('newJob', JOB_ID, JOB_CODE)
@@ -89 +135 @@
-        return self.worker.stop()
-
-
+_OK
-        def checkResult(result):
-1+2*2+3*3
+(True, 1+2*2+3*3)
-            self.nextCall(self.root.calls)
-            self.nextCall('Root.callRemote')
@@ -103 +149 @@
-        return task.d
-
+    def test_SingleTask_Fail(self):
+        def checkResult(result):
+            self.failUnlessEqual(result[0], False)
+            self.failUnless('ZeroDivisionError' in result[1])
+
+        task = Mock_Task('bogusable', (0,), {}, 0, JOB_ID)
+        self.worker.run(task)
+        task.d.addCallback(checkResult)
+        return task.d
+
-
-class Test_JobManager(mock.TestCase):
@@ -116 +172 @@
-        def attached(childID):
-            self.childID = childID
+            return childID
-        
-        self.root = Mock_Root(bogus)
@@ -129 +186 @@
-
-    def test_attachChild_bogus(self):
-        def check(result):
-            self.failUnless(isinstance(result, failure.Failure))
-
-        self.mgr.jobs[JOB_ID] = (JOB_CODE, 0)
-Both(check
+Callback(self.failUnlessEqual, None
-
-    def test_new(self):
@@ -148 +202 @@
-    def test_run_one(self):
-        def check(result):
-50
+(True, 50)
-        
-        d = self._attach()
@@ -167 +221 @@
-            yield wfd
-            results.append(wfd.getResult())
-results
+[x[1] for x in results]
-
-    @defer.deferredGenerator
@@ -182 +236 @@
-            dList.append(d)
-        yield defer.waitForDeferred(defer.DeferredList(dList))
-results
+[x[1] for x in results]

projects/AsynQueue/trunk/asynqueue/test/test_processworker.py

@@ -45 +45 @@
-            self.mgr.children = {1:process}
-            d = root.callRemote('newJob', JOB_ID, JOB_CODE)
-True
+(True, ['test'])
-            return d
-        

projects/AsynQueue/trunk/asynqueue/test/test_tasks.py

@@ -172 +172 @@
-                 if isinstance(x, defer.Deferred)], [True])
-
+    def testHireSetWorkerID(self):
+        worker = MockWorker()
+        workerID = self.mgr.hire(worker)
+        self.failUnlessEqual(getattr(worker, 'ID', None), workerID)
+
-    def testHireClassQualifications(self):
-        class CQWorker(MockWorker):