Merge pull request #70 from sideshowdave7/master

EventMQ 0.3rc5 (We'll get 0.3 someday) Fix critical worker health check bug, Simplify scheduler run_count header (remove support for persistance)

7 files changed, 39 insertions, 39 deletions

diff --git a/bin/send_request b/bin/send_request
index aa84639..9c8696d 100755
--- a/bin/send_request
+++ b/bin/send_request
@@ -21,4 +21,5 @@ if __name__ == "__main__":
            'kwargs': {}
            }]
 
-    send_request(s, msg, guarantee=True, reply_requested=True)
+    while True:
+        send_request(s, msg, guarantee=True, reply_requested=True)
diff --git a/eventmq/__init__.py b/eventmq/__init__.py
index e1feca9..184ddfe 100644
--- a/eventmq/__init__.py
+++ b/eventmq/__init__.py
@@ -1,5 +1,5 @@
 __author__ = 'EventMQ Contributors'
-__version__ = '0.3-rc4'
+__version__ = '0.3-rc5'
 
 PROTOCOL_VERSION = 'eMQP/1.0'
 
diff --git a/eventmq/conf.py b/eventmq/conf.py
index e2d2b1c..6a0b3a4 100644
--- a/eventmq/conf.py
+++ b/eventmq/conf.py
@@ -86,4 +86,6 @@ RQ_PORT = 6379
 RQ_DB = 0
 RQ_PASSWORD = ''
 
+MAX_JOB_COUNT=1024
+
 # }}}
diff --git a/eventmq/jobmanager.py b/eventmq/jobmanager.py
index e94f16d..b6de15e 100644
--- a/eventmq/jobmanager.py
+++ b/eventmq/jobmanager.py
@@ -244,13 +244,14 @@ class JobManager(HeartbeatMixin, EMQPService):
 
     def check_worker_health(self):
         """
-        Checks for any dead processes in the pool and recreates them if necessary
+        Checks for any dead processes in the pool and recreates them if
+        necessary
         """
         self._workers = [w for w in self._workers if w.is_alive()]
 
         if len(self._workers) < conf.CONCURRENT_JOBS:
-            logger.warning("{} worker process(es) may have died...recreating")\
-                  .format(conf.CONCURRENT_JOBS - len(self._workers))
+            logger.warning("{} worker process(es) may have died...recreating"
+                           .format(conf.CONCURRENT_JOBS - len(self._workers)))
 
         for i in range(0, conf.CONCURRENT_JOBS - len(self._workers)):
             w = Worker(self.request_queue, self.finished_queue)
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index 9130694..086ba69 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -182,49 +182,30 @@ class Scheduler(HeartbeatMixin, EMQPService):
                                  % (ts_now, v[0], msg))
 
                     if v[4] != INFINITE_RUN_COUNT:
-                        # Decrement run_count
-                        v[4] -= 1
                         # If run_count was 0, we cancel the job
                         if v[4] <= 0:
                             cancel_jobs.append(k)
-                        # Otherwise we run the job
                         else:
-                            # Send job and update next schedule time
+                            # Decrement run_count
+                            v[4] -= 1
                             self.send_request(msg, queue=queue)
                             v[0] = next(v[2])
-                            # Rename redis key and save new run_count counter
-                            try:
-                                self.redis_server.rename(k,
-                                                         self.schedule_hash(v))
-                                self.redis_server.set(self.schedule_hash(v),
-                                                      serialize(v))
-                                self.redis_server.save()
-                            except redis.ConnectionError:
-                                logger.warning("Couldn't contact redis server")
-                            except Exception as e:
-                                logger.warning(
-                                    'Unable to update key in redis '
-                                    'server: {}'.format(e.message))
                     else:
                         # Scheduled job is in running infinitely
                         # Send job and update next schedule time
                         self.send_request(msg, queue=queue)
                         v[0] = next(v[2])
-                        # Persist changes to redis
-                        try:
-                            self.redis_server.set(
-                                self.schedule_hash(v), serialize(v))
-                            self.redis_server.save()
-                        except redis.ConnectionError:
-                            logger.warning("Couldn't contact redis server")
-                        except Exception as e:
-                            logger.warning(
-                                'Unable to update key in redis '
-                                'server: {}'.format(e.message))
 
             for job in cancel_jobs:
-                message = self.interval_jobs[k][1]
-                self.unschedule_job(message)
+                try:
+                    logger.debug('Cancelling job due to run_count: {}'
+                                 .format(k))
+                    self.redis_server.delete(k)
+                    self.redis_server.lrem('interval_jobs', 0, k)
+                except Exception as e:
+                    logger.warning(
+                        'Unable to update key in redis '
+                        'server: {}'.format(e))
                 del self.interval_jobs[k]
 
             if not self.maybe_send_heartbeat(events):
@@ -244,7 +225,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
 
             except Exception as e:
                 logger.warning('Unable to connect to redis server: {}'.format(
-                    e.message))
+                    e))
         else:
             return self._redis_server
 
@@ -304,6 +285,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
         # in memory
         try:
             if (self.redis_server.get(schedule_hash)):
+                self.redis_server.delete(schedule_hash)
                 self.redis_server.lrem('interval_jobs', 0, schedule_hash)
                 self.redis_server.save()
         except redis.ConnectionError:
@@ -357,6 +339,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
         headers = message[1]
         interval = int(message[2])
         cron = str(message[4])
+        run_count = self.get_run_count_from_headers(headers)
 
         schedule_hash = self.schedule_hash(message)
 
@@ -378,7 +361,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
                 message[3],
                 inter_iter,
                 queue,
-                self.get_run_count_from_headers(headers)
+                run_count
             ]
 
             if schedule_hash in self.cron_jobs:
@@ -412,8 +395,12 @@ class Scheduler(HeartbeatMixin, EMQPService):
         except Exception as e:
             logger.warning(str(e))
 
+        # Send a request in haste mode, decrement run_count if valid
         if 'nohaste' not in headers:
-            self.send_request(message[3], queue=queue)
+            # Don't allow decrement past 0
+            if run_count > 0:
+                self.interval_jobs[schedule_hash][4] -= 1
+                self.send_request(message[3], queue=queue)
 
     def get_run_count_from_headers(self, headers):
         run_count = INFINITE_RUN_COUNT
diff --git a/eventmq/worker.py b/eventmq/worker.py
index 00855e8..29814f4 100644
--- a/eventmq/worker.py
+++ b/eventmq/worker.py
@@ -20,8 +20,10 @@ Defines different short-lived workers that execute jobs
 from importlib import import_module
 from multiprocessing import Process
 from threading import Thread
+from . import conf
 
 import logging
+import os
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +37,7 @@ class MultiprocessWorker(Process):
         super(MultiprocessWorker, self).__init__()
         self.input_queue = input_queue
         self.output_queue = output_queue
+        self.job_count = 0
 
     def run(self):
         """
@@ -48,6 +51,7 @@ class MultiprocessWorker(Process):
         # Pull the payload off the queue and run it
         for payload in iter(self.input_queue.get, 'DONE'):
 
+            self.job_count += 1
             timeout = payload.get("timeout", None)
             msgid = payload.get('msgid', '')
 
@@ -73,6 +77,11 @@ class MultiprocessWorker(Process):
             resp['callback'] = payload['callback']
             self.output_queue.put(resp)
 
+            if self.job_count > conf.MAX_JOB_COUNT:
+                break
+
+        logger.debug("Worker death, PID: {}".format(os.getpid()))
+
 
 def _run(payload):
     if ":" in payload["path"]:
diff --git a/setup.py b/setup.py
index 09e3abd..276732e 100644
--- a/setup.py
+++ b/setup.py
@@ -7,7 +7,7 @@ from setuptools import setup, find_packages
 
 setup(
     name='eventmq',
-    version='0.3-rc4',
+    version='0.3-rc5',
     description='EventMQ messaging system based on ZeroMQ',
     packages=find_packages(),
     install_requires=['pyzmq==15.4.0',