allow to queue tasks with a delay #release-note

Author: lsbardel

README.rst

@@ -133,6 +133,24 @@ Launch a python shell and play with the api
  >>> task.status_string
  'SUCCESS'
 
+You can also queue tasks with a ``delay``
+
+.. code:: python
+
+ >>> task = api.queue_task('addition', a=4, b=6, callback=False, delay=2).wait()
+ >>> task.status_string
+ 'QUEUED'
+ >>> task.time_queued # timestamp
+ >>> task = task.done_callback.wait()
+ >>> task.status_string
+ 'SUCCESS'
+ >>> task.time_started - task.time_queued
+ 2.00
+
+**NOTE**: The ``wait`` method in a task future can only be used on the shell
+or when the event loop is not running. In all other cases one should ``await``
+for the task future in a coroutine.
+
 API
 =============
 

pq/mq.py

@@ -79,14 +79,19 @@ def queue(self, task, callback=True):
  called back once the task is done, otherwise return a future
  called back once the task is queued
  '''
- future = TaskFuture(task.id, self.backend, loop=self._loop)
+ future_done = TaskFuture(task.id, self.backend, loop=self._loop)
  if task.queue:
- self.callbacks[task.id] = future
+ self.callbacks[task.id] = future_done
  else: # the task is not queued instead it is executed immediately
  coro = self.backend._execute_task(task)
+ return chain_future(coro, next=future_done)
+ coro = self._queue_task(task, future_done)
+ if callback:
+ ensure_future(coro, loop=self._loop)
+ return future_done
+ else:
+ future = TaskFuture(task.id, self.backend, loop=self._loop)
  return chain_future(coro, next=future)
- result = ensure_future(self._queue_task(task, future), loop=self._loop)
- return future if callback else result
 
  @abstractmethod
  async def size(self, *queues): # pragma nocover

pq/server/consumer.py

@@ -32,17 +32,26 @@ async def _execute_task(self, task, worker=None):
  expiry = task.expiry
  if expiry and time_ended > expiry:
  raise TaskTimeout
- else:
- kwargs = task.kwargs or {}
- task.status = states.STARTED
- task.time_started = time_ended
- if worker:
- task.worker = worker.aid
- logger.info(task.lazy_info())
- await self.pubsub.publish('started', task)
- job = JobClass(self, task)
- # This may block for a while
- task.result = await self._consume(job, kwargs)
+
+ if task.delay: # Task with delay
+ start_time = task.time_queued + task.delay
+ gap = start_time - time_ended
+ if gap > 0:
+ self._loop.call_later(gap, self._queue_again, task)
+ if self._concurrent_tasks:
+ self._concurrent_tasks.discard(task_id)
+ return task
+
+ kwargs = task.kwargs or {}
+ task.status = states.STARTED
+ task.time_started = time_ended
+ if worker:
+ task.worker = worker.aid
+ logger.info(task.lazy_info())
+ await self.pubsub.publish('started', task)
+ job = JobClass(self, task)
+ # This may block for a while
+ task.result = await self._consume(job, kwargs)
  else:
  raise TaskError('Invalid status %s' % task.status_string)
  except TaskTimeout:
@@ -108,6 +117,9 @@ async def _consume_in_subprocess(self, job, kwargs):
  raise RemoteStackTrace
  return job.task.result
 
+ def _queue_again(self, task):
+ self.broker.queue(task, False)
+
  def json_params(self):
  for name, value in self.cfg.items():
  try:

pq/server/producer.py

@@ -26,22 +26,23 @@ class TaskProducer(models.RegistryMixin, ExecutorMixin, ABC):
  """
  app = None
 
- def __init__(self, cfg, logger=None, **kw):
+ def __init__(self, cfg, *, logger=None, **kw):
  self.cfg = cfg
  self.logger = logger or logging.getLogger('pulsar.queue')
  self._closing = False
- store = create_store(cfg.data_store)
+ loop = cfg.params.pop('loop', None)
+ store = create_store(cfg.data_store, loop=loop)
  if not cfg.message_broker:
  broker = store
  else:
- broker = create_store(cfg.message_broker)
+ broker = create_store(cfg.message_broker, loop=loop)
  if self.cfg.callable:
  self.app = self.cfg.callable(self)
  self.store_task = getattr(self.app, 'store_task', store_task)
  self.pubsub = PubSub(self, store)
  self.broker = brokers.get(broker.name)(self, broker)
  self.green_pool = getattr(self.app, 'green_pool', GreenPool())
- self.http = getattr(self.app, 'http', HttpClient())
+ self.http = getattr(self.app, 'http', HttpClient(loop=loop))
 
  def __str__(self):
  return repr(self)
@@ -103,7 +104,11 @@ def queue_task(self, jobname, callback=True, **kwargs):
  '''
  task = self._create_task(jobname, **kwargs)
  if task:
- return self.broker.queue(task, callback)
+ future = self.broker.queue(task, callback)
+ if self._loop.is_running():
+ return self.green_pool.wait(future)
+ else:
+ return future
 
  def queue_task_local(self, jobname, **kwargs):
  kwargs['queue'] = self.node_name
@@ -118,7 +123,7 @@ def execute_task(self, jobname, **kwargs):
 
  # INTERNALS
  def _create_task(self, jobname, meta_params=None, expiry=None, queue=True,
- **kwargs):
+ delay=None, **kwargs):
  '''Try to queue a new :ref:`Task`.
 
  This method returns a :class:`.Future` which results in the
@@ -160,6 +165,7 @@ def _create_task(self, jobname, meta_params=None, expiry=None, queue=True,
  expiry=expiry,
  kwargs=kwargs,
  status=states.QUEUED,
+ delay=delay,
  **meta_params)
  else:
  raise TaskNotAvailable(jobname)

pq/server/pubsub.py

@@ -64,7 +64,7 @@ def __call__(self, channel, message):
 
  def _channel(self, event=''):
  event = 'task_%s' % event
- prefix = self.cfg.task_queue_prefix
+ prefix = self.cfg.name
  return '%s_%s' % (prefix, event) if prefix else event
 
 

pq/tasks/task.py

@@ -37,6 +37,7 @@ class Task:
  exception = None
  stacktrace = None
  worker = None
+ delay = None
 
  def __init__(self, id=None, name=None, time_queued=None,
  expiry=None, status=None, kwargs=None, queue=None,

tests/test_app.py

@@ -46,19 +46,16 @@ def rpc_name(cls):
  @classmethod
  async def setUpClass(cls):
  # The name of the task queue application
- name = cls.name()
- queues = ['%s1' % name, '%s2' % name]
- pq = api.PulsarQueue(cls.name(),
- wsgi=True,
- config='tests.config',
- queue_callable=dummy,
- task_queues=queues,
- default_task_queue=queues[0],
- schedule_periodic=cls.schedule_periodic,
- rpc_bind='127.0.0.1:0',
- concurrency=cls.concurrency,
- rpc_concurrency=cls.concurrency,
- rpc_keep_alive=cls.rpc_timeout)
+ params = cls.params()
+ params.update(dict(
+ wsgi=True,
+ schedule_periodic=cls.schedule_periodic,
+ rpc_bind='127.0.0.1:0',
+ concurrency=cls.concurrency,
+ rpc_concurrency=cls.concurrency,
+ rpc_keep_alive=cls.rpc_timeout
+ ))
+ pq = api.PulsarQueue(**params)
  cfgs = await pq.start()
  cls.tq = cfgs[0].app()
  cls.rpc = cfgs[1].app()
@@ -67,14 +64,30 @@ async def setUpClass(cls):
  timeout=cls.rpc_timeout)
  # Now flush the task queue
  backend = await cls.tq.backend.start()
- await backend.flush_queues(*queues)
+ await backend.flush_queues(*cls.queues())
 
  @classmethod
  def tearDownClass(cls):
  coros = [send('arbiter', 'kill_actor', a.name) for a in
  (cls.tq, cls.rpc) if a is not None]
  return asyncio.gather(*coros)
 
+ @classmethod
+ def queues(cls):
+ name = cls.name()
+ return ['%s1' % name, '%s2' % name]
+
+ @classmethod
+ def params(cls):
+ queues = cls.queues()
+ return dict(
+ name=cls.name(),
+ config='tests.config',
+ queue_callable=dummy,
+ task_queues=queues,
+ default_task_queue=queues[0]
+ )
+
 
 class TestTaskQueue(TaskQueueBase, unittest.TestCase):
 
@@ -261,6 +274,35 @@ async def test_scrape(self):
  self.assertEqual(task.status_string, 'SUCCESS')
  self.assertTrue(task.result)
 
+ async def test_delay(self):
+ task = await self.tq.backend.queue_task('scrape',
+ delay=2,
+ url='https://www.bbc.co.uk/')
+
+ self.assertEqual(task.status_string, 'SUCCESS')
+ self.assertEqual(task.delay, 2)
+ self.assertTrue(task.time_started - task.time_queued > 2)
+ self.assertTrue(task.result)
+
+ def test_sync(self):
+ loop = asyncio.new_event_loop()
+ tasks = api.TaskApp(loop=loop, **self.params()).backend
+ self.assertEqual(tasks._loop, loop)
+ task = tasks.queue_task('scrape', url='https://github.com')
+ self.assertIsInstance(task, asyncio.Future)
+ self.assertTrue(task.task_id)
+ task = task.wait()
+ self.assertEqual(task.status_string, 'SUCCESS')
+ self.assertFalse(tasks._loop.is_running())
+ #
+ task = tasks.queue_task('scrape', url='https://github.com',
+ callback=False)
+ task = task.wait()
+ self.assertEqual(task.status_string, 'QUEUED')
+ self.assertTrue(task.done_callback)
+ task = task.done_callback.wait()
+ self.assertEqual(task.status_string, 'SUCCESS')
+
  # RPC
  async def test_rpc_job_list(self):
  data = await self.proxy.job_list()