celeryd: Added --scheduler option to be used in combination with -B. Closes #229

celery/apps/worker.py

@@ -46,7 +46,8 @@ class Worker(object):
             schedule=None, task_time_limit=None, task_soft_time_limit=None,
             max_tasks_per_child=None, queues=None, events=False, db=None,
             include=None, defaults=None, pidfile=None,
-            redirect_stdouts=None, redirect_stdouts_level=None, **kwargs):
+            redirect_stdouts=None, redirect_stdouts_level=None,
+            scheduler_cls=None, **kwargs):
         if defaults is None:
             from celery import conf
             defaults = conf
@@ -60,6 +61,7 @@ class Worker(object):
         self.discard = discard
         self.run_clockservice = run_clockservice
         self.schedule = schedule or defaults.CELERYBEAT_SCHEDULE_FILENAME
+        self.scheduler_cls = scheduler_cls or defaults.CELERYBEAT_SCHEDULER
         self.events = events
         self.task_time_limit = (task_time_limit or
                                 defaults.CELERYD_TASK_TIME_LIMIT)
@@ -207,6 +209,7 @@ class Worker(object):
                                 ready_callback=self.on_listener_ready,
                                 embed_clockservice=self.run_clockservice,
                                 schedule_filename=self.schedule,
+                                scheduler_cls=self.scheduler_cls,
                                 send_events=self.events,
                                 db=self.db,
                                 max_tasks_per_child=self.max_tasks_per_child,

celery/bin/celeryd.py

@@ -43,6 +43,10 @@
     Defaults to ``celerybeat-schedule``. The extension ".db" will be
     appended to the filename.
 
+.. cmdoption:: --scheduler
+
+    Scheduler class to use. Default is celery.beat.PersistentScheduler
+
 .. cmdoption:: -E, --events
 
     Send events that can be captured by monitors like ``celerymon``.
@@ -113,7 +117,11 @@ class WorkerCommand(Command):
                      "option. The extension '.db' will be appended to the "
                     "filename. Default: %s" % (
                         conf.CELERYBEAT_SCHEDULE_FILENAME, )),
-
+            Option('--scheduler',
+                default=None,
+                action="store", dest="scheduler_cls",
+                help="Scheduler class. Default is "
+                     "celery.beat.PersistentScheduler"),
             Option('-S', '--statedb', default=conf.CELERYD_STATE_DB,
                 action="store", dest="db",
                 help="Path to the state database. The extension '.db' will "

celery/worker/__init__.py

@@ -126,7 +126,8 @@ class WorkController(object):
             max_tasks_per_child=conf.CELERYD_MAX_TASKS_PER_CHILD,
             pool_putlocks=conf.CELERYD_POOL_PUTLOCKS,
             disable_rate_limits=conf.DISABLE_RATE_LIMITS,
-            db=conf.CELERYD_STATE_DB):
+            db=conf.CELERYD_STATE_DB,
+            scheduler_cls=conf.CELERYBEAT_SCHEDULER):
 
         # Options
         self.loglevel = loglevel or self.loglevel
@@ -182,7 +183,8 @@ class WorkController(object):
         self.beat = None
         if self.embed_clockservice:
             self.beat = beat.EmbeddedService(logger=self.logger,
-                                    schedule_filename=schedule_filename)
+                                    schedule_filename=schedule_filename,
+                                    scheduler_cls=scheduler_cls)
 
         prefetch_count = self.concurrency * conf.CELERYD_PREFETCH_MULTIPLIER
         self.listener = instantiate(listener_cls,