Merge pull request binarymatt#74 from joeshaw/joeshaw-setproctitle

setproctitle() changes

Author: binarymatt

pyres/horde.py

@@ -219,7 +219,7 @@ def register_signal_handlers(self):
         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)
+        signal.signal(signal.SIGUSR2, self.add_child)
 
     def _schedule_shutdown(self):
         self.schedule_shutdown(None, None)

pyres/scheduler.py

@@ -2,7 +2,7 @@
 import time
 import logging
 
-from pyres import ResQ
+from pyres import ResQ, __version__
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +32,7 @@ def schedule_shutdown(self, signal, frame):
         self._shutdown = True
 
     def __call__(self):
+        _setproctitle("Starting")
         logger.info('starting up')
         self.register_signal_handlers()
         #self.load_schedule()
@@ -40,6 +41,7 @@ def __call__(self):
             if self._shutdown:
                 break
             self.handle_delayed_items()
+            _setproctitle("Waiting")
             logger.debug('sleeping')
             time.sleep(5)
         logger.info('shutting down complete')
@@ -63,6 +65,7 @@ def next_item(self, timestamp):
 
     def handle_delayed_items(self):
         for timestamp in self.next_timestamp():
+            _setproctitle('Handling timestamp %s' % timestamp)
             logger.info('handling timestamp: %s' % timestamp)
             for item in self.next_item(timestamp):
                 logger.debug('queueing item %s' % item)
@@ -81,3 +84,11 @@ def run(cls, server, password=None):
         sched()
 
 
+try:
+    from setproctitle import setproctitle
+except ImportError:
+    def setproctitle(name):
+        pass
+
+def _setproctitle(msg):
+    setproctitle("pyres_scheduler-%s: %s" % (__version__, msg))

pyres/worker.py

@@ -111,6 +111,11 @@ def __str__(self):
             return self.id
         return '%s:%s:%s' % (self.hostname, self.pid, ','.join(self.queues))
 
+    def _setproctitle(self, msg):
+        setproctitle("pyres_worker-%s [%s]: %s" % (__version__,
+                                                   ','.join(self.queues),
+                                                   msg))
+
     def work(self, interval=5):
         """Invoked by ``run`` method. ``work`` listens on a list of queues and sleeps
         for ``interval`` time.
@@ -124,7 +129,7 @@ def work(self, interval=5):
         Finally, the ``process`` method actually processes the job by eventually calling the Job instance's ``perform`` method.
 
         """
-        setproctitle('pyres_worker-%s: Starting' % __version__)
+        self._setproctitle("Starting")
         self.startup()
 
         while True:
@@ -140,10 +145,9 @@ def work(self, interval=5):
                 self.before_fork(job)
                 self.child = os.fork() 
                 if self.child:
-                    setproctitle("pyres_worker%s: Forked %s at %s" %
-                                 (__version__,
-                                  self.child,
-                                  datetime.datetime.now()))
+                    self._setproctitle("Forked %s at %s" %
+                                       (self.child,
+                                        datetime.datetime.now()))
                     logger.info('Forked %s at %s' % (self.child,
                                                       datetime.datetime.now()))
 
@@ -157,9 +161,9 @@ def work(self, interval=5):
                     #os.wait()
                     logger.debug('done waiting')
                 else:
-                    setproctitle("pyres_worker-%s: Processing %s since %s" %
-                                 (__version__, job._queue,
-                                  datetime.datetime.now()))
+                    self._setproctitle("Processing %s since %s" %
+                                       (job._queue,
+                                        datetime.datetime.now()))
                     logger.info('Processing %s since %s' %
                                  (job._queue, datetime.datetime.now()))
                     self.after_fork(job)
@@ -170,8 +174,7 @@ def work(self, interval=5):
                 if interval == 0:
                     break
                 #procline @paused ? "Paused" : "Waiting for #{@queues.join(',')}"
-                setproctitle("pyres_worker-%s: Waiting for %s " %
-                             (__version__, ','.join(self.queues)))
+                self._setproctitle("Waiting")
                 #time.sleep(interval)
         self.unregister_worker()