diff options
Diffstat (limited to 'rq/queue.py')
| -rw-r--r-- | rq/queue.py | 56 |
1 files changed, 35 insertions, 21 deletions
diff --git a/rq/queue.py b/rq/queue.py index bdcf1f3..6b0c3bd 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -183,10 +183,11 @@ class Queue(object): timeout = timeout or self._default_timeout # TODO: job with dependency shouldn't have "queued" as status - job = self.job_class.create(func, args, kwargs, connection=self.connection, - result_ttl=result_ttl, ttl=ttl, status=JobStatus.QUEUED, - description=description, depends_on=depends_on, timeout=timeout, - id=job_id) + job = self.job_class.create( + func, args, kwargs, connection=self.connection, + result_ttl=result_ttl, ttl=ttl, status=JobStatus.QUEUED, + description=description, depends_on=depends_on, + timeout=timeout, id=job_id, origin=self.name) # If job depends on an unfinished job, register itself on it's # parent's dependents instead of enqueueing it. @@ -200,6 +201,7 @@ class Queue(object): try: pipe.watch(depends_on.key) if depends_on.get_status() != JobStatus.FINISHED: + job.set_status(JobStatus.DEFERRED) job.register_dependency(pipeline=pipe) job.save(pipeline=pipe) pipe.execute() @@ -248,24 +250,25 @@ class Queue(object): description=description, depends_on=depends_on, job_id=job_id, at_front=at_front) - def enqueue_job(self, job, set_meta_data=True, at_front=False): + def enqueue_job(self, job, at_front=False): """Enqueues a job for delayed execution. - If the `set_meta_data` argument is `True` (default), it will update - the properties `origin` and `enqueued_at`. - If Queue is instantiated with async=False, job is executed immediately. """ - # Add Queue key set - self.connection.sadd(self.redis_queues_keys, self.key) + + with self.connection._pipeline() as pipeline: + # Add Queue key set + self.connection.sadd(self.redis_queues_keys, self.key) + job.set_status(JobStatus.QUEUED, pipeline=pipeline) - if set_meta_data: job.origin = self.name job.enqueued_at = utcnow() - if job.timeout is None: - job.timeout = self.DEFAULT_TIMEOUT - job.save() + if job.timeout is None: + job.timeout = self.DEFAULT_TIMEOUT + job.save(pipeline=pipeline) + + pipeline.execute() if self._async: self.push_job_id(job.id, at_front=at_front) @@ -277,11 +280,16 @@ class Queue(object): def enqueue_dependents(self, job): """Enqueues all jobs in the given job's dependents set and clears it.""" # TODO: can probably be pipelined + from .registry import DeferredJobRegistry + + registry = DeferredJobRegistry(self.name, self.connection) + while True: job_id = as_text(self.connection.spop(job.dependents_key)) if job_id is None: break dependent = self.job_class.fetch(job_id, connection=self.connection) + registry.remove(dependent) self.enqueue_job(dependent) def pop_job_id(self): @@ -401,14 +409,20 @@ class FailedQueue(Queue): def quarantine(self, job, exc_info): """Puts the given Job in quarantine (i.e. put it on the failed queue). - - This is different from normal job enqueueing, since certain meta data - must not be overridden (e.g. `origin` or `enqueued_at`) and other meta - data must be inserted (`ended_at` and `exc_info`). """ - job.ended_at = utcnow() - job.exc_info = exc_info - return self.enqueue_job(job, set_meta_data=False) + + with self.connection._pipeline() as pipeline: + # Add Queue key set + self.connection.sadd(self.redis_queues_keys, self.key) + + job.ended_at = utcnow() + job.exc_info = exc_info + job.save(pipeline=pipeline) + + self.push_job_id(job.id, pipeline=pipeline) + pipeline.execute() + + return job def requeue(self, job_id): """Requeues the job with the given job ID.""" |