first working commit of horde work.

With this commit, the horde is working. However,
there are no tests. On a positive note, completely 
un-scientific benchmarks show that the horde much 
faster than the older worker model.

Author: Matt George

pyres/horde.py

@@ -3,37 +3,39 @@
 except:
     import sys
     sys.exit("multiprocessing was not available")
-import os, datetime, time, signal, sys
-from pyres import ResQ
-from pyres.exceptions import NoQueueError
+
+import time, os, signal
 from pyres.worker import Worker
+from pyres import ResQ
+from pyres.utils import OrderedDict
+import datetime
+
 class Minion(multiprocessing.Process, Worker):
-    def __init__(self, conn, queues=[]):
-        self.conn = conn
+    def __init__(self, queues, server, password):
+        multiprocessing.Process.__init__(self, name='Minion')
         self.queues = queues
-        super(Minion,self).__init__(name='Minion')
-    
-    def _check_message(self):
-        if self.conn.poll():
-            message = self.conn.recv()
-            self.process_message(message)
-    
-    def process_message(self, message):
-        pass
+        self.validate_queues()
+        self._shutdown = False
+        self.child = None
+        if isinstance(server,basestring):
+            self.resq = ResQ(server=server, password=password)
+        elif isinstance(server, ResQ):
+            self.resq = server
+        else:
+            raise Exception("Bad server argument")
+        #Worker.__init__(self, queues=queues, server="localhost:6379", password=None)
+        #super(Minion,self).__init__(name='Minion')
 
     def work(self, interval=5):
         self.startup()
         while True:
-            self._check_messages()
             if self._shutdown:
                 print 'shutdown scheduled'
                 break
             job = self.reserve()
             if job:
                 self.process(job)
             else:
-                if interval == 0:
-                    break
                 time.sleep(interval)
         self.unregister_worker()
 
@@ -43,54 +45,141 @@ def run(self):
         #    job = self.q.get()
         #    print 'pid: %s is running %s ' % (self.pid,job)
 
-class Khan(Worker):
+
+class Khan(object):
     _command_map = {
-        'ADD': '_add_minion',
+        'ADD': 'add_minion',
         'REMOVE': '_remove_minion',
         'SHUTDOWN': '_schedule_shutdown'
     }
-    _workers = {}
-    def __init__(self, pool_size=5, queue_list=[], server='localhost:6379', password=None):
+    _workers = OrderedDict()
+    def __init__(self, pool_size=5, queues=[], server='localhost:6379', password=None):
+        #super(Khan,self).__init__(queues=queues,server=server,password=password)
+        self._shutdown = False
         self.pool_size = pool_size
-        self.resq = ResQ(server, password=password)
-        self._workers = list()
+        self.queues = queues
+        self.server = server
+        self.password = password
+        self.pid = os.getpid()
+        if isinstance(server,basestring):
+            self.resq = ResQ(server=server, password=password)
+        elif isinstance(server, ResQ):
+            self.resq = server
+        else:
+            raise Exception("Bad server argument")
+        #self._workers = list()
+    
+    def startup(self):
+        self.register_signal_handlers()
+        self.register_worker()
+    
+    def register_signal_handlers(self):
+        signal.signal(signal.SIGTERM, self.schedule_shutdown)
+        signal.signal(signal.SIGINT, self.schedule_shutdown)
+        signal.signal(signal.SIGQUIT, self.schedule_shutdown)
+        signal.signal(signal.SIGUSR1, self.kill_child)
+        signal.signal(signal.SIGUSR1, self.add_child)
+    
+    def _schedule_shutdown(self):
+        self.schedule_shutdown(None, None)
+    
+    def schedule_shutdown(self, signum, frame):
+        print 'Shutdown scheduled'
+        self._shutdown = True
+    
+    def kill_child(self, signum, frame):
+        self._remove_minion()
+    
+    def add_child(self, signum, frame):
+        self.add_minion()
 
-    def _check_command(self):
+    def register_khan(self):
+        self.resq.redis.sadd('resque:khans',str(self))
+        self.started = datetime.datetime.now()
+    
+    def _check_commands(self):
         if not self._shutdown:
-            command = self.resq.redis.pop('resque:khan:%s' % self)
+            print 'Checking commands'
+            command_key = 'resque:khan:%s' % self
+            command = self.resq.redis.pop(command_key)
+            print 'COMMAND', command
             if command:
                 self.process_command(command)
-                self._check_command()
+                self._check_commands()
 
     def process_command(self, command):
+        print 'Processing Command'
         #available commands, shutdown, add 1, remove 1
         command = self._command_map.get(command, None)
         if command:
             fn = getattr(self, command)
             if fn:
                 fn()
 
-    def _add_minion(self):
-        parent_conn, child_conn = multiprocessing.Pipe()
-        m = Minion(child_conn, q)
+    def add_minion(self):
+        print 'Adding minion'
+        m = self._add_minion()
         m.start()
-        self._workers.append((parent_conn, m))
 
-    def _remove_minion(self):
-        self._workers
-    def run():
+    def _add_minion(self):
+        print 'Adding mminion'
+        #parent_conn, child_conn = multiprocessing.Pipe()
+        m = Minion(self.queues, self.server, self.password)
+        #m.start()
+        self._workers[m.pid] = m
+        return m
+        #self._workers.append(m)
+    
+    def _shutdown_minions(self):
+        """
+        send the SIGNINT signal to each worker in the pool.
+        """
+        for minion in self._workers.values():
+            minion.terminate()
+            minion.join()
+    
+    def _remove_minion(self, pid=None):
+        #if pid:
+        #    m = self._workers.pop(pid)
+        pid, m = self._workers.popitem(False)
+        m.terminate()
+        return m
+    
+    def register_worker(self):
+        self.resq.redis.sadd('resque:khans',str(self))
+        #self.resq._redis.add("worker:#{self}:started", Time.now.to_s)
+        self.started = datetime.datetime.now()
+    
+    def unregister_worker(self):
+        print 'Unregistering'
+        self.resq.redis.srem('resque:khans',str(self))
+        self.started = None
+    
+    def work(self, interval=5):
         self.startup()
-        q = multiprocessing.Queue()
-        for i in range(pool_size):
-            self._add_minion()
+        for i in range(self.pool_size):
+            m = self._add_minion()
+            m.start()
+        
         while True:
             self._check_commands()
             if self._shutdown:
                 #send signals to each child
+                self._shutdown_minions()
                 break
-            time.sleep(5)
+            #get job
+            else:
+                time.sleep(interval)
         self.unregister_worker()
-
+    
+    def __str__(self):
+        hostname = os.uname()[1]
+        return '%s:%s' % (hostname, self.pid)
+        
+    @classmethod
+    def run(cls, pool_size=5, queues=[], server='localhost:6379'):
+        worker = cls(pool_size=pool_size, queues=queues, server=server)
+        worker.work()
 
 #if __name__ == "__main__":
 #    k = Khan()
@@ -104,6 +193,6 @@ def run():
     if len(args) < 1:
         parser.print_help()
         parser.error("Please give the horde at least one queue.")
-    khan = Khan(queue_list=args, server=options.dest)
-    khan.run()
-    #Worker.run(queues, options.server)
+    Khan.run(pool_size=2, queues=args, server=options.server)
+    #khan.run()
+    #Worker.run(queues, options.server)

pyres/utils.py

@@ -0,0 +1,98 @@
+from UserDict import DictMixin
+
+class OrderedDict(dict, DictMixin):
+
+    def __init__(self, *args, **kwds):
+        if len(args) > 1:
+            raise TypeError('expected at most 1 arguments, got %d' % len(args))
+        try:
+            self.__end
+        except AttributeError:
+            self.clear()
+        self.update(*args, **kwds)
+
+    def clear(self):
+        self.__end = end = []
+        end += [None, end, end]         # sentinel node for doubly linked list
+        self.__map = {}                 # key --> [key, prev, next]
+        dict.clear(self)
+
+    def __setitem__(self, key, value):
+        if key not in self:
+            end = self.__end
+            curr = end[1]
+            curr[2] = end[1] = self.__map[key] = [key, curr, end]
+        dict.__setitem__(self, key, value)
+
+    def __delitem__(self, key):
+        dict.__delitem__(self, key)
+        key, prev, next = self.__map.pop(key)
+        prev[2] = next
+        next[1] = prev
+
+    def __iter__(self):
+        end = self.__end
+        curr = end[2]
+        while curr is not end:
+            yield curr[0]
+            curr = curr[2]
+
+    def __reversed__(self):
+        end = self.__end
+        curr = end[1]
+        while curr is not end:
+            yield curr[0]
+            curr = curr[1]
+
+    def popitem(self, last=True):
+        if not self:
+            raise KeyError('dictionary is empty')
+        key = reversed(self).next() if last else iter(self).next()
+        value = self.pop(key)
+        return key, value
+
+    def __reduce__(self):
+        items = [[k, self[k]] for k in self]
+        tmp = self.__map, self.__end
+        del self.__map, self.__end
+        inst_dict = vars(self).copy()
+        self.__map, self.__end = tmp
+        if inst_dict:
+            return (self.__class__, (items,), inst_dict)
+        return self.__class__, (items,)
+
+    def keys(self):
+        return list(self)
+
+    setdefault = DictMixin.setdefault
+    update = DictMixin.update
+    pop = DictMixin.pop
+    values = DictMixin.values
+    items = DictMixin.items
+    iterkeys = DictMixin.iterkeys
+    itervalues = DictMixin.itervalues
+    iteritems = DictMixin.iteritems
+
+    def __repr__(self):
+        if not self:
+            return '%s()' % (self.__class__.__name__,)
+        return '%s(%r)' % (self.__class__.__name__, self.items())
+
+    def copy(self):
+        return self.__class__(self)
+
+    @classmethod
+    def fromkeys(cls, iterable, value=None):
+        d = cls()
+        for key in iterable:
+            d[key] = value
+        return d
+
+    def __eq__(self, other):
+        if isinstance(other, OrderedDict):
+            return len(self)==len(other) and \
+                   all(p==q for p, q in  zip(self.items(), other.items()))
+        return dict.__eq__(self, other)
+
+    def __ne__(self, other):
+        return not self == other

pyres/worker.py

@@ -41,7 +41,6 @@ def register_worker(self):
         self.resq.redis.sadd('resque:workers',str(self))
         #self.resq._redis.add("worker:#{self}:started", Time.now.to_s)
         self.started = datetime.datetime.now()
-        
 
     def _set_started(self, dt):
         if dt:
@@ -228,7 +227,7 @@ def worker_pids(self):
                                        grep pyres_worker").split("\n"))
 
     @classmethod
-    def run(cls, queues, server, interval):
+    def run(cls, queues, server="localhost:6379", interval):
         worker = cls(queues=queues, server=server)
         if interval is not None:
             worker.work(interval)