4 files changed, 41 insertions, 17 deletions

diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index 16cb506..ec15b3c 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -79,7 +79,7 @@ class JobManager(HeartbeatMixin, EMQPService):
 
         #: Define the name of this JobManager instance. Useful to know when
         #: referring to the logs.
-        self.name = kwargs.pop('name') or generate_device_name()
+        self.name = kwargs.pop('name', None) or generate_device_name()
         logger.info('Initializing JobManager {}...'.format(self.name))
 
         #: keep track of workers
@@ -106,9 +106,12 @@ class JobManager(HeartbeatMixin, EMQPService):
 
         self.poller = Poller()
 
+        # Stats and monitoring information
         self.jobs_in_flight = {}
         self.total_requests = 0
         self.total_ready_sent = 0
+        # Keep track of what pids are servicing our requests
+        self.pid_distribution = {}
 
         #: Setup worker queues
         self.request_queue = mp_queue()
@@ -242,6 +245,12 @@ class JobManager(HeartbeatMixin, EMQPService):
         if msgid in self.jobs_in_flight:
             del self.jobs_in_flight[msgid]
 
+        if not death:
+            if pid not in self.pid_distribution:
+                self.pid_distribution[pid] = 1
+            else:
+                self.pid_distribution[pid] += 1
+
     def on_request(self, msgid, msg):
         """
         Handles a REQUEST command
@@ -307,7 +316,6 @@ class JobManager(HeartbeatMixin, EMQPService):
         Worker died of natural causes, ensure its death and
         remove from tracking, will be replaced on next heartbeat
         """
-        self.kill_worker(pid, signal.SIGKILL)
         if pid in self._workers.keys():
             del self._workers[pid]
 
diff --git a/eventmq/sender.py b/eventmq/sender.py
index 8452abd..f0cc09c 100644
--- a/eventmq/sender.py
+++ b/eventmq/sender.py
@@ -18,6 +18,7 @@
 The sender is responsible for sending messages
 """
 import logging
+import sys
 import uuid
 
 import zmq
@@ -137,7 +138,10 @@ class Sender(ZMQSendMixin, ZMQReceiveMixin):
             self.zsocket.close()
 
         self.zsocket = kwargs.pop('socket', self.zcontext.socket(zmq.DEALER))
-        self.zsocket.setsockopt(zmq.IDENTITY, self.name)
+        if sys.version[0] == '2':
+            self.zsocket.setsockopt(zmq.IDENTITY, self.name)
+        else:
+            self.zsocket.setsockopt_string(zmq.IDENTITY, str(self.name))
 
         self.status = constants.STATUS.ready
 
diff --git a/eventmq/tests/test_sender.py b/eventmq/tests/test_sender.py
index b2570a3..f292696 100644
--- a/eventmq/tests/test_sender.py
+++ b/eventmq/tests/test_sender.py
@@ -35,7 +35,7 @@ class TestCase(unittest.TestCase):
 
         self.assert_(socket.poll() != 0)
         msg = socket.recv_multipart()
-        self.assertEqual(msg[0], self.sender.name)
+        self.assertEqual(msg[0].decode('ascii'), self.sender.name)
         self.assertEqual(msg[1], b'')
         self.assertEqual(msg[2], b'1')
         self.assertEqual(msg[3], b'Hello!')
diff --git a/eventmq/worker.py b/eventmq/worker.py
index 838be09..69567d5 100644
--- a/eventmq/worker.py
+++ b/eventmq/worker.py
@@ -51,10 +51,7 @@ class MultiprocessWorker(Process):
 
     @property
     def logger(self):
-        if not hasattr(self, '_logger'):
-            self._logger = logging.getLogger(__name__ + '.' + str(os.getpid()))
-
-        return self._logger
+        return logging.getLogger(__name__ + '.' + str(os.getpid()))
 
     def run(self):
         """
@@ -63,12 +60,14 @@ class MultiprocessWorker(Process):
         This is designed to run in a seperate process.
         """
         # Define the 2 queues for communicating with the worker thread
-        worker_queue = Queue.Queue()
-        worker_result_queue = Queue.Queue()
+        logger = self.logger
+
+        worker_queue = Queue.Queue(1)
+        worker_result_queue = Queue.Queue(1)
         worker_thread = Thread(target=_run,
                                args=(worker_queue,
                                      worker_result_queue,
-                                     self.logger))
+                                     logger))
 
         import zmq
         zmq.Context.instance().term()
@@ -107,10 +106,10 @@ class MultiprocessWorker(Process):
             try:
                 return_val = 'None'
                 self.job_count += 1
-                timeout = payload.get("timeout", conf.GLOBAL_TIMEOUT)
+                timeout = payload.get("timeout") or conf.GLOBAL_TIMEOUT
                 msgid = payload.get('msgid', '')
                 callback = payload.get('callback', '')
-                self.logger.debug("Putting on thread queue msgid: {}".format(
+                logger.debug("Putting on thread queue msgid: {}".format(
                     msgid))
 
                 worker_queue.put(payload['params'])
@@ -118,7 +117,7 @@ class MultiprocessWorker(Process):
                 try:
                     return_val = worker_result_queue.get(timeout=timeout)
 
-                    self.logger.debug("Got from result queue msgid: {}".format(
+                    logger.debug("Got from result queue msgid: {}".format(
                         msgid))
                 except Queue.Empty:
                     return_val = 'TimeoutError'
@@ -143,9 +142,12 @@ class MultiprocessWorker(Process):
                 return_val = str(e)
 
             if self.job_count >= conf.MAX_JOB_COUNT:
-                self.logger.debug("Worker reached job limit, exiting")
+                logger.debug("Worker reached job limit, exiting")
                 break
 
+        worker_queue.put('DONE')
+        worker_thread.join(timeout=5)
+
         self.output_queue.put(
             {'msgid': None,
              'return': 'DEATH',
@@ -153,7 +155,11 @@ class MultiprocessWorker(Process):
              'pid': os.getpid(),
              'callback': 'worker_death'}
             )
-        self.logger.debug("Worker death")
+
+        logger.debug("Worker death")
+
+        if worker_thread.is_alive():
+            logger.debug("Worker thread did not die gracefully")
 
 
 def _run(queue, result_queue, logger):
@@ -187,14 +193,20 @@ def _run(queue, result_queue, logger):
         # Blocking get so we don't spin cycles reading over and over
         try:
             payload = queue.get()
-        except:
+        except Exception as e:
+            logger.exception(e)
             continue
 
+        if payload == 'DONE':
+            break
+
         return_val = _run_job(payload, logger)
         # Signal that we're done with this job and put its return value on the
         # result queue
         result_queue.put(return_val)
 
+    logger.debug("Worker thread death")
+
 
 def _run_job(payload, logger):
     try: