Merge remote branch 'themonkey/master'

Author: binarymatt

CHANGES.txt

@@ -0,0 +1,5 @@
+2011-03-01  whit  <[email protected]>
+
+	* Added hooks for the worker to allow worker subclasses to insert
+  	  code before and after forking
+

pyres/__init__.py

@@ -105,9 +105,11 @@ def push(self, queue, item):
         self.watch_queue(queue)
         self.redis.rpush("resque:queue:%s" % queue, ResQ.encode(item))
 
-    def pop(self, queue):
-        ret = self.redis.lpop("resque:queue:%s" % queue)
+    def pop(self, queue, timeout=10):
+        ret = self.redis.blpop("resque:queue:%s" % queue, timeout=timeout)
         if ret:
+            if isinstance(ret, tuple):
+                q, ret = ret
             return ResQ.decode(ret)
         return ret
 

pyres/extensions.py

@@ -1,11 +1,16 @@
+import os
+import datetime
+import time
+import signal
+
 try:
     import multiprocessing
 except:
     import sys
     sys.exit("multiprocessing was not available")
-import os, datetime, time, signal
+
 from pyres import ResQ
-    
+
 from pyres.exceptions import NoQueueError
 from pyres.worker import Worker
 
@@ -18,20 +23,23 @@ def work(self, interval=5):
             job = self.reserve()
             if job:
                 print "got: %s" % job
+
                 self.child = os.fork()
+
                 if self.child:
-                    print 'Forked %s at %s' % (self.child, datetime.datetime.now())
+                    print 'Forked %s at %s' % (self.child,
+                                               datetime.datetime.now())
                     os.waitpid(self.child, 0)
                 else:
-                    print 'Processing %s since %s' % (job._queue, datetime.datetime.now())
+                    print 'Processing %s since %s' % (job._queue,
+                                                      datetime.datetime.now())
                     self.process(job)
                     os._exit(0)
                 self.child = None
             else:
                 break
-                
+
         self.unregister_worker()
-    
 
 class Manager(object):
     def __init__(self, queues, host, max_children=10):
@@ -43,28 +51,27 @@ def __init__(self, queues, host, max_children=10):
         self.resq = ResQ(host)
         self.validate_queues()
         self.reports = {}
-    
+
     def __str__(self):
-        import os; 
         hostname = os.uname()[1]
         pid = os.getpid()
         return 'Manager:%s:%s:%s' % (hostname, pid, ','.join(self.queues))
-    
+
     def validate_queues(self):
         if not self.queues:
             raise NoQueueError("Please give each worker at least one queue.")
-    
+
     def check_rising(self, queue, size):
         if queue in self.reports:
-            time = time.time()
+            new_time = time.time()
             old_size = self.reports[queue][0]
             old_time = self.reports[queue][1]
-            if time > old_time + 5 and size > old_size + 20:
+            if new_time > old_time + 5 and size > old_size + 20:
                 return True
         else:
             self.reports[queue] = (size, time.time())
             return False
-    
+
     def work(self):
         self.startup()
         while True:
@@ -74,44 +81,46 @@ def work(self):
             for queue in self.queues:
                 #check queue size
                 size = self.resq.size(queue)
-                if check_rising(queue,size):
+
+                if self.check_rising(queue, size):
                     if len(self.children) < self.max_children:
                         self.start_child(queue)
-    
+
     def startup(self):
         self.register_manager()
         self.register_signals()
-    
+
     def register_manager(self):
-        self.resq.redis.sadd('managers',str(self))
-    
+        self.resq.redis.sadd('managers', str(self))
+
     def unregister_manager(self):
-        self.resq.redis.srem('managers',str(self))
-    
+        self.resq.redis.srem('managers', str(self))
+
     def register_signals(self):
         signal.signal(signal.SIGTERM, self.shutdown_all)
         signal.signal(signal.SIGINT, self.shutdown_all)
         signal.signal(signal.SIGQUIT, self.schedule_shutdown)
         signal.signal(signal.SIGUSR1, self.kill_children)
-    
+
     def shutdown_all(self, signum, frame):
         self.schedule_shutdown(signum, frame)
         self.kill_children(signum, frame)
-    
+
     def schedule_shutdown(self, signum, frame):
         self._shutdown = True
-    
+
     def kill_children(self):
         for child in self.children:
             child.terminate()
-    
+
     def start_child(self, queue):
-        p = multiprocessing.Process(target=JuniorWorker.run, args=([queue], self._host))
+        p = multiprocessing.Process(target=JuniorWorker.run, args=([queue],
+                                                                   self._host))
         self.children.append(p)
         p.start()
         return True
-    
+
     @classmethod
-    def run(cls, queues=[], host="localhost:6379"):
+    def run(cls, queues=(), host="localhost:6379"):
         manager = cls(queues, host)
-        manager.work()
+        manager.work()

pyres/failure/redis.py

@@ -1,10 +1,12 @@
 import datetime, time
+from base64 import b64encode
+
 from base import BaseBackend
 from pyres import ResQ
 
 class RedisBackend(BaseBackend):
     """Extends the ``BaseBackend`` to provide a Redis backend for failed jobs."""
-    
+
     def save(self, resq=None):
         """Saves the failed Job into a "failed" Redis queue preserving all its original enqueud info."""
         if not resq:
@@ -20,15 +22,15 @@ def save(self, resq=None):
             data['worker'] = self._worker
         data = ResQ.encode(data)
         resq.redis.rpush('resque:failed', data)
-    
+
     @classmethod
     def count(cls, resq):
         return int(resq.redis.llen('resque:failed'))
 
     @classmethod
     def all(cls, resq, start=0, count=1):
-        from base64 import b64encode
         items = resq.redis.lrange('resque:failed', start, count) or []
+
         ret_list = []
         for i in items:
             failure = ResQ.decode(i)
@@ -39,4 +41,4 @@ def all(cls, resq, start=0, count=1):
     @classmethod
     def clear(cls, resq):
         return resq.redis.delete('resque:failed')
-    
+

pyres/job.py

@@ -1,23 +1,30 @@
-from datetime import datetime, timedelta
-from pyres import ResQ, str_to_class, safe_str_to_class
+from datetime import timedelta
+from pyres import ResQ, safe_str_to_class
 from pyres import failure
 from pyres.failure.redis import RedisBackend
 
+
 class Job(object):
     """Every job on the ResQ is an instance of the *Job* class.
 
     The ``__init__`` takes these keyword arguments:
 
-        ``queue`` -- A string defining the queue to which this Job will be added.
+        ``queue`` -- A string defining the queue to which this Job will be
+                     added.
 
-        ``payload`` -- A dictionary which contains the string name of a class which extends this Job and
-        a list of args which will be passed to that class.
+        ``payload`` -- A dictionary which contains the string name of a class
+                       which extends this Job and a list of args which will be
+                       passed to that class.
 
         ``resq`` -- An instance of the ResQ class.
 
-        ``worker`` -- The name of a specific worker if you'd like this Job to be done by that worker. Default is "None".
+        ``worker`` -- The name of a specific worker if you'd like this Job to be
+                      done by that worker. Default is "None".
 
     """
+
+    safe_str_to_class = staticmethod(safe_str_to_class)
+    
     def __init__(self, queue, payload, resq, worker=None):
         self._queue = queue
         self._payload = payload
@@ -33,26 +40,30 @@ def __str__(self):
             self._queue, self._payload['class'], repr(self._payload['args']))
 
     def perform(self):
-        """This method converts payload into args and calls the ``perform`` method
-        on the payload class.
+        """This method converts payload into args and calls the ``perform``
+        method on the payload class.
+
+        #@ add entry_point loading
 
         """
         payload_class_str = self._payload["class"]
-        payload_class = safe_str_to_class(payload_class_str)
+        payload_class = self.safe_str_to_class(payload_class_str)
         payload_class.resq = self.resq
-        args = self._payload.get("args", None)
+        args = self._payload.get("args")
+
         try:
             return payload_class.perform(*args)
         except:
             if not self.retry(payload_class, args):
                 raise
 
     def fail(self, exception):
-        """This method provides a way to fail a job and will use whatever failure backend
-        you've provided. The default is the ``RedisBackend``.
+        """This method provides a way to fail a job and will use whatever
+        failure backend you've provided. The default is the ``RedisBackend``.
 
         """
-        fail = failure.create(exception, self._queue, self._payload, self._worker)
+        fail = failure.create(exception, self._queue, self._payload,
+                              self._worker)
         fail.save(self.resq)
         return fail
 
@@ -72,11 +83,11 @@ def retry(self, payload_class, args):
         return False
 
     @classmethod
-    def reserve(cls, queue, res, worker=None):
+    def reserve(cls, queue, res, worker=None, timeout=10):
         """Reserve a job on the queue. This marks this job so that other workers
         will not pick it up.
 
         """
-        payload = res.pop(queue)
+        payload = res.pop(queue, timeout=timeout)
         if payload:
             return cls(queue, payload, res, worker)

pyres/json_parser.py

@@ -1,15 +1,14 @@
 from datetime import datetime
+
 try:
     #import simplejson as json
     import json
 except ImportError:
     import simplejson as json
 
-
 DATE_FORMAT = '%Y-%m-%dT%H:%M:%S'
 DATE_PREFIX = '@D:'
 
-
 class CustomJSONEncoder(json.JSONEncoder):
 
     def default(self, o):