From fc5295edfcfd589489a56d4f4e4849edc5e063d0 Mon Sep 17 00:00:00 2001 From: Javier Lopez Date: Tue, 2 Dec 2014 11:02:14 +0100 Subject: [PATCH 01/24] Set busy state on parent while waiting for horse --- rq/worker.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rq/worker.py b/rq/worker.py index bf40a65..51411b2 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -435,7 +435,9 @@ class Worker(object): self.procline('Forked %d at %d' % (child_pid, time.time())) while True: try: + self.set_state('busy') os.waitpid(child_pid, 0) + self.set_state('idle') break except OSError as e: # In case we encountered an OSError due to EINTR (which is From d4b72d330d05d2a988de19056e3ec2d5794c1344 Mon Sep 17 00:00:00 2001 From: Travis Johnson Date: Tue, 9 Dec 2014 12:04:57 -0500 Subject: [PATCH 02/24] test for skip_queue mechanics --- tests/test_queue.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/tests/test_queue.py b/tests/test_queue.py index ed42204..8bfde6d 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -459,3 +459,13 @@ class TestFailedQueue(RQTestCase): """Ensure custom job class assignment works as expected.""" q = Queue(job_class=CustomJob) self.assertEqual(q.job_class, CustomJob) + + def test_skip_queue(self): + """Ensure the skip_queue option functions""" + q = Queue('foo') + job1 = q.enqueue(say_hello) + job2 = q.enqueue(say_hello) + assert q.dequeue() == job1 + skip_job = q.enqueue(say_hello, skip_queue=True) + assert q.dequeue() == skip_job + assert q.dequeue() == job2 From f60e4884df50db2f65e5a35bd602c4074981cbab Mon Sep 17 00:00:00 2001 From: Travis Johnson Date: Tue, 9 Dec 2014 12:05:12 -0500 Subject: [PATCH 03/24] a comment typo --- rq/queue.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rq/queue.py b/rq/queue.py index 7dfb0f8..a9049dd 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -149,7 +149,7 @@ class Queue(object): def compact(self): """Removes all "dead" jobs from the queue by cycling through it, while - guarantueeing FIFO semantics. + guaranteeing FIFO semantics. """ COMPACT_QUEUE = 'rq:queue:_compact:{0}'.format(uuid.uuid4()) From ac61f502a1ddb83aa56c761a9ef4adddcd222ee6 Mon Sep 17 00:00:00 2001 From: Travis Johnson Date: Tue, 9 Dec 2014 12:05:19 -0500 Subject: [PATCH 04/24] skip_queue functionality --- rq/queue.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/rq/queue.py b/rq/queue.py index a9049dd..ddb31e6 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -161,14 +161,18 @@ class Queue(object): if self.job_class.exists(job_id, self.connection): self.connection.rpush(self.key, job_id) - def push_job_id(self, job_id, pipeline=None): - """Pushes a job ID on the corresponding Redis queue.""" + def push_job_id(self, job_id, pipeline=None, skip_queue=False): + """Pushes a job ID on the corresponding Redis queue. + 'skip_queue' allows you to push the job onto the front instead of the back of the queue""" connection = pipeline if pipeline is not None else self.connection - connection.rpush(self.key, job_id) + if skip_queue: + connection.lpush(self.key, job_id) + else: + connection.rpush(self.key, job_id) def enqueue_call(self, func, args=None, kwargs=None, timeout=None, result_ttl=None, description=None, depends_on=None, - job_id=None): + job_id=None, skip_queue=False): """Creates a job to represent the delayed function call and enqueues it. @@ -204,7 +208,7 @@ class Queue(object): except WatchError: continue - return self.enqueue_job(job) + return self.enqueue_job(job, skip_queue=skip_queue) def enqueue(self, f, *args, **kwargs): """Creates a job to represent the delayed function call and enqueues @@ -231,6 +235,7 @@ class Queue(object): result_ttl = kwargs.pop('result_ttl', None) depends_on = kwargs.pop('depends_on', None) job_id = kwargs.pop('job_id', None) + skip_queue = kwargs.pop('skip_queue', False) if 'args' in kwargs or 'kwargs' in kwargs: assert args == (), 'Extra positional arguments cannot be used when using explicit args and kwargs.' # noqa @@ -240,9 +245,9 @@ class Queue(object): return self.enqueue_call(func=f, args=args, kwargs=kwargs, timeout=timeout, result_ttl=result_ttl, description=description, depends_on=depends_on, - job_id=job_id) + job_id=job_id, skip_queue=skip_queue) - def enqueue_job(self, job, set_meta_data=True): + def enqueue_job(self, job, set_meta_data=True, skip_queue=False): """Enqueues a job for delayed execution. If the `set_meta_data` argument is `True` (default), it will update @@ -262,7 +267,7 @@ class Queue(object): job.save() if self._async: - self.push_job_id(job.id) + self.push_job_id(job.id, skip_queue=skip_queue) else: job.perform() job.save() From db75958ad23d1ae16a028210a04bcd03e3a540b2 Mon Sep 17 00:00:00 2001 From: Travis Johnson Date: Fri, 12 Dec 2014 12:25:33 -0500 Subject: [PATCH 05/24] use 'at_front' instead of 'skip_queue' --- rq/queue.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/rq/queue.py b/rq/queue.py index ddb31e6..847319f 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -161,18 +161,18 @@ class Queue(object): if self.job_class.exists(job_id, self.connection): self.connection.rpush(self.key, job_id) - def push_job_id(self, job_id, pipeline=None, skip_queue=False): + def push_job_id(self, job_id, pipeline=None, at_front=False): """Pushes a job ID on the corresponding Redis queue. - 'skip_queue' allows you to push the job onto the front instead of the back of the queue""" + 'at_front' allows you to push the job onto the front instead of the back of the queue""" connection = pipeline if pipeline is not None else self.connection - if skip_queue: + if at_front: connection.lpush(self.key, job_id) else: connection.rpush(self.key, job_id) def enqueue_call(self, func, args=None, kwargs=None, timeout=None, result_ttl=None, description=None, depends_on=None, - job_id=None, skip_queue=False): + job_id=None, at_front=False): """Creates a job to represent the delayed function call and enqueues it. @@ -208,7 +208,7 @@ class Queue(object): except WatchError: continue - return self.enqueue_job(job, skip_queue=skip_queue) + return self.enqueue_job(job, at_front=at_front) def enqueue(self, f, *args, **kwargs): """Creates a job to represent the delayed function call and enqueues @@ -235,7 +235,7 @@ class Queue(object): result_ttl = kwargs.pop('result_ttl', None) depends_on = kwargs.pop('depends_on', None) job_id = kwargs.pop('job_id', None) - skip_queue = kwargs.pop('skip_queue', False) + at_front = kwargs.pop('at_front', False) if 'args' in kwargs or 'kwargs' in kwargs: assert args == (), 'Extra positional arguments cannot be used when using explicit args and kwargs.' # noqa @@ -245,9 +245,9 @@ class Queue(object): return self.enqueue_call(func=f, args=args, kwargs=kwargs, timeout=timeout, result_ttl=result_ttl, description=description, depends_on=depends_on, - job_id=job_id, skip_queue=skip_queue) + job_id=job_id, at_front=at_front) - def enqueue_job(self, job, set_meta_data=True, skip_queue=False): + def enqueue_job(self, job, set_meta_data=True, at_front=False): """Enqueues a job for delayed execution. If the `set_meta_data` argument is `True` (default), it will update @@ -267,7 +267,7 @@ class Queue(object): job.save() if self._async: - self.push_job_id(job.id, skip_queue=skip_queue) + self.push_job_id(job.id, at_front=at_front) else: job.perform() job.save() From 82333d2ad5053876004ff57af209ed7ab447098a Mon Sep 17 00:00:00 2001 From: Jonathan Tushman Date: Thu, 9 Oct 2014 14:09:52 -0400 Subject: [PATCH 06/24] triggering shutdown by setting a redis flag --- .gitignore | 3 -- requirements.txt | 4 +-- rq/cli/cli.py | 45 +++++++++++++++++++++++++-- rq/cli/helpers.py | 7 +++-- rq/job.py | 21 ++++--------- rq/queue.py | 10 +++--- rq/suspension.py | 18 +++++++++++ rq/utils.py | 10 ++++++ rq/worker.py | 74 +++++++++++++++++++++++++++++++++----------- tests/test_cli.py | 52 ++++++++++++++++++++++++++++--- tests/test_queue.py | 12 +++---- tests/test_worker.py | 43 +++++++++++++++++++++---- 12 files changed, 236 insertions(+), 63 deletions(-) create mode 100644 rq/suspension.py diff --git a/.gitignore b/.gitignore index 25a046e..f95cc7b 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,3 @@ .tox .vagrant Vagrantfile - -# PyCharm -.idea diff --git a/requirements.txt b/requirements.txt index 539b9a4..8da152d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ -redis -click +redis==2.7.0 +click>=3.0.0 diff --git a/rq/cli/cli.py b/rq/cli/cli.py index 18979dc..afb32ea 100755 --- a/rq/cli/cli.py +++ b/rq/cli/cli.py @@ -16,6 +16,7 @@ from rq import Connection, get_failed_queue, Queue from rq.contrib.legacy import cleanup_ghosts from rq.exceptions import InvalidJobOperationError from rq.utils import import_attribute +from rq.suspension import suspend as connection_suspend, resume as connection_resume, is_suspended from .helpers import (read_config_file, refresh, setup_loghandlers_from_args, show_both, show_queues, show_workers) @@ -24,8 +25,12 @@ from .helpers import (read_config_file, refresh, setup_loghandlers_from_args, url_option = click.option('--url', '-u', envvar='RQ_REDIS_URL', help='URL describing Redis connection details.') +config_option = click.option('--config', '-c', help='Module containing RQ settings.') -def connect(url): + +def connect(url, config=None): + settings = read_config_file(config) if config else {} + url = url or settings.get('REDIS_URL') return StrictRedis.from_url(url or 'redis://localhost:6379/0') @@ -120,7 +125,7 @@ def info(url, path, interval, raw, only_queues, only_workers, by_queue, queues): @main.command() @url_option -@click.option('--config', '-c', help='Module containing RQ settings.') +@config_option @click.option('--burst', '-b', is_flag=True, help='Run in burst mode (quit after all work is done)') @click.option('--name', '-n', help='Specify a different name') @click.option('--worker-class', '-w', default='rq.Worker', help='RQ Worker class to use') @@ -158,7 +163,12 @@ def worker(url, config, burst, name, worker_class, job_class, queue_class, path, worker_class = import_attribute(worker_class) queue_class = import_attribute(queue_class) + if is_suspended(conn): + click.secho("The worker has been paused, run reset_paused", fg='red') + sys.exit(1) + try: + queues = [queue_class(queue, connection=conn) for queue in queues] w = worker_class(queues, name=name, @@ -178,3 +188,34 @@ def worker(url, config, burst, name, worker_class, job_class, queue_class, path, except ConnectionError as e: print(e) sys.exit(1) + + +@main.command() +@url_option +@config_option +@click.option('--duration', help='Seconds you want the workers to be suspended. Default is forever.', type=int) +def suspend(url, config, duration): + """Suspends all workers, to resume run `rq resume`""" + if duration is not None and duration < 1: + click.echo("Duration must be an integer greater than 1") + sys.exit(1) + + connection = connect(url, config) + connection_suspend(connection, duration) + + if duration: + msg = """Suspending workers for {0} seconds. No new jobs will be started during that time, but then will + automatically resume""".format(duration) + click.echo(msg) + else: + click.echo("Suspending workers. No new jobs will be started. But current jobs will be completed") + + +@main.command() +@url_option +@config_option +def resume(url, config): + """Resumes processing of queues, that where suspended with `rq suspend`""" + connection = connect(url, config) + connection_resume(connection) + click.echo("Resuming workers.") diff --git a/rq/cli/helpers.py b/rq/cli/helpers.py index f25cc81..9ea4b58 100644 --- a/rq/cli/helpers.py +++ b/rq/cli/helpers.py @@ -8,7 +8,9 @@ from functools import partial import click from rq import Queue, Worker +from rq.worker import WorkerStatus from rq.logutils import setup_loghandlers +from rq.suspension import is_suspended red = partial(click.style, fg='red') green = partial(click.style, fg='green') @@ -39,8 +41,9 @@ def get_scale(x): def state_symbol(state): symbols = { - 'busy': red('busy'), - 'idle': green('idle'), + WorkerStatus.BUSY: red('busy'), + WorkerStatus.IDLE: green('idle'), + WorkerStatus.SUSPENDED: yellow('suspended'), } try: return symbols[state] diff --git a/rq/job.py b/rq/job.py index e8080f8..7b17492 100644 --- a/rq/job.py +++ b/rq/job.py @@ -12,7 +12,7 @@ from rq.compat import as_text, decode_redis_hash, string_types, text_type from .connections import resolve_connection from .exceptions import NoSuchJobError, UnpickleError from .local import LocalStack -from .utils import import_attribute, utcformat, utcnow, utcparse +from .utils import import_attribute, utcformat, utcnow, utcparse, enum try: import cPickle as pickle @@ -25,16 +25,7 @@ dumps = partial(pickle.dumps, protocol=pickle.HIGHEST_PROTOCOL) loads = pickle.loads -def enum(name, *sequential, **named): - values = dict(zip(sequential, range(len(sequential))), **named) - - # NOTE: Yes, we *really* want to cast using str() here. - # On Python 2 type() requires a byte string (which is str() on Python 2). - # On Python 3 it does not matter, so we'll use str(), which acts as - # a no-op. - return type(str(name), (), values) - -Status = enum('Status', +JobStatus = enum('JobStatus', QUEUED='queued', FINISHED='finished', FAILED='failed', STARTED='started') @@ -166,19 +157,19 @@ class Job(object): @property def is_finished(self): - return self.get_status() == Status.FINISHED + return self.get_status() == JobStatus.FINISHED @property def is_queued(self): - return self.get_status() == Status.QUEUED + return self.get_status() == JobStatus.QUEUED @property def is_failed(self): - return self.get_status() == Status.FAILED + return self.get_status() == JobStatus.FAILED @property def is_started(self): - return self.get_status() == Status.STARTED + return self.get_status() == JobStatus.STARTED @property def dependency(self): diff --git a/rq/queue.py b/rq/queue.py index ff5f860..61f5575 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -5,7 +5,7 @@ from __future__ import (absolute_import, division, print_function, import uuid from .connections import resolve_connection -from .job import Job, Status +from .job import Job, JobStatus from .utils import import_attribute, utcnow from .exceptions import (DequeueTimeout, InvalidJobOperationError, @@ -180,7 +180,7 @@ class Queue(object): # 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, status=Status.QUEUED, + result_ttl=result_ttl, status=JobStatus.QUEUED, description=description, depends_on=depends_on, timeout=timeout, id=job_id) @@ -195,7 +195,7 @@ class Queue(object): while True: try: pipe.watch(depends_on.key) - if depends_on.get_status() != Status.FINISHED: + if depends_on.get_status() != JobStatus.FINISHED: job.register_dependency(pipeline=pipe) job.save(pipeline=pipe) pipe.execute() @@ -390,7 +390,7 @@ class Queue(object): class FailedQueue(Queue): def __init__(self, connection=None): - super(FailedQueue, self).__init__(Status.FAILED, connection=connection) + super(FailedQueue, self).__init__(JobStatus.FAILED, connection=connection) def quarantine(self, job, exc_info): """Puts the given Job in quarantine (i.e. put it on the failed @@ -417,7 +417,7 @@ class FailedQueue(Queue): if self.remove(job) == 0: raise InvalidJobOperationError('Cannot requeue non-failed jobs.') - job.set_status(Status.QUEUED) + job.set_status(JobStatus.QUEUED) job.exc_info = None q = Queue(job.origin, connection=self.connection) q.enqueue_job(job) diff --git a/rq/suspension.py b/rq/suspension.py new file mode 100644 index 0000000..b734acd --- /dev/null +++ b/rq/suspension.py @@ -0,0 +1,18 @@ +WORKERS_SUSPENDED = 'rq:suspended' + + +def is_suspended(connection): + return connection.exists(WORKERS_SUSPENDED) + + +def suspend(connection, ttl=None): + """ttl = time to live in seconds. Default is no expiration + Note: If you pass in 0 it will invalidate right away + """ + connection.set(WORKERS_SUSPENDED, 1) + if ttl is not None: + connection.expire(WORKERS_SUSPENDED, ttl) + + +def resume(connection): + return connection.delete(WORKERS_SUSPENDED) \ No newline at end of file diff --git a/rq/utils.py b/rq/utils.py index 8233ac6..db56020 100644 --- a/rq/utils.py +++ b/rq/utils.py @@ -208,3 +208,13 @@ def first(iterable, default=None, key=None): def current_timestamp(): """Returns current UTC timestamp""" return calendar.timegm(datetime.datetime.utcnow().utctimetuple()) + + +def enum(name, *sequential, **named): + values = dict(zip(sequential, range(len(sequential))), **named) + + # NOTE: Yes, we *really* want to cast using str() here. + # On Python 2 type() requires a byte string (which is str() on Python 2). + # On Python 3 it does not matter, so we'll use str(), which acts as + # a no-op. + return type(str(name), (), values) \ No newline at end of file diff --git a/rq/worker.py b/rq/worker.py index bf40a65..72f1aa4 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -12,18 +12,20 @@ import sys import time import traceback import warnings +from datetime import datetime from rq.compat import as_text, string_types, text_type from .connections import get_current_connection from .exceptions import DequeueTimeout, NoQueueError -from .job import Job, Status +from .job import Job, JobStatus from .logutils import setup_loghandlers from .queue import get_failed_queue, Queue from .timeouts import UnixSignalDeathPenalty -from .utils import import_attribute, make_colorizer, utcformat, utcnow +from .utils import import_attribute, make_colorizer, utcformat, utcnow, enum from .version import VERSION from .registry import FinishedJobRegistry, StartedJobRegistry +from .suspension import is_suspended try: from procname import setprocname @@ -52,8 +54,8 @@ def compact(l): return [x for x in l if x is not None] _signames = dict((getattr(signal, signame), signame) - for signame in dir(signal) - if signame.startswith('SIG') and '_' not in signame) + for signame in dir(signal) + if signame.startswith('SIG') and '_' not in signame) def signal_name(signum): @@ -65,6 +67,12 @@ def signal_name(signum): return 'SIG_UNKNOWN' +WorkerStatus = enum('WorkerStatus', + STARTED='started', SUSPENDED='suspended', BUSY='busy', + IDLE='idle' + ) + + class Worker(object): redis_worker_namespace_prefix = 'rq:worker:' redis_workers_keys = 'rq:workers' @@ -333,6 +341,30 @@ class Worker(object): signal.signal(signal.SIGINT, request_stop) signal.signal(signal.SIGTERM, request_stop) + def check_for_suspension(self, burst): + """Check to see if the workers have been suspended by something like `rq suspend`""" + + before_state = None + notified = False + + while not self.stopped and is_suspended(self.connection): + + if burst: + self.log.info('Suspended in burst mode -- exiting.') + self.log.info('Note: There could still be unperformed jobs on the queue') + raise StopRequested + + if not notified: + self.log.info('Worker suspended, use "rq resume" command to resume') + before_state = self.get_state() + self.set_state(WorkerStatus.SUSPENDED) + notified = True + time.sleep(1) + + if before_state: + self.set_state(before_state) + + def work(self, burst=False): """Starts the work loop. @@ -348,15 +380,19 @@ class Worker(object): did_perform_work = False self.register_birth() self.log.info('RQ worker started, version %s' % VERSION) - self.set_state('starting') + self.set_state(WorkerStatus.STARTED) + try: while True: - if self.stopped: - self.log.info('Stopping on request.') - break - - timeout = None if burst else max(1, self.default_worker_ttl - 60) try: + self.check_for_suspension(burst) + + if self.stopped: + self.log.info('Stopping on request.') + break + + timeout = None if burst else max(1, self.default_worker_ttl - 60) + result = self.dequeue_job_and_maintain_ttl(timeout) if result is None: break @@ -367,20 +403,22 @@ class Worker(object): self.execute_job(job) self.heartbeat() - if job.get_status() == Status.FINISHED: + if job.get_status() == JobStatus.FINISHED: queue.enqueue_dependents(job) did_perform_work = True + finally: if not self.is_horse: self.register_death() return did_perform_work + def dequeue_job_and_maintain_ttl(self, timeout): result = None qnames = self.queue_names() - self.set_state('idle') + self.set_state(WorkerStatus.IDLE) self.procline('Listening on %s' % ','.join(qnames)) self.log.info('') self.log.info('*** Listening on %s...' % @@ -395,7 +433,7 @@ class Worker(object): if result is not None: job, queue = result self.log.info('%s: %s (%s)' % (green(queue.name), - blue(job.description), job.id)) + blue(job.description), job.id)) break except DequeueTimeout: @@ -477,12 +515,12 @@ class Worker(object): timeout = (job.timeout or 180) + 60 with self.connection._pipeline() as pipeline: - self.set_state('busy', pipeline=pipeline) + self.set_state(WorkerStatus.BUSY, pipeline=pipeline) self.set_current_job_id(job.id, pipeline=pipeline) self.heartbeat(timeout, pipeline=pipeline) registry = StartedJobRegistry(job.origin, self.connection) registry.add(job, timeout, pipeline=pipeline) - job.set_status(Status.STARTED, pipeline=pipeline) + job.set_status(JobStatus.STARTED, pipeline=pipeline) pipeline.execute() self.procline('Processing %s from %s since %s' % ( @@ -511,7 +549,7 @@ class Worker(object): result_ttl = job.get_ttl(self.default_result_ttl) if result_ttl != 0: job.ended_at = utcnow() - job._status = Status.FINISHED + job._status = JobStatus.FINISHED job.save(pipeline=pipeline) finished_job_registry = FinishedJobRegistry(job.origin, self.connection) @@ -523,7 +561,7 @@ class Worker(object): pipeline.execute() except Exception: - job.set_status(Status.FAILED, pipeline=pipeline) + job.set_status(JobStatus.FAILED, pipeline=pipeline) started_job_registry.remove(job, pipeline=pipeline) pipeline.execute() self.handle_exception(job, *sys.exc_info()) @@ -552,7 +590,7 @@ class Worker(object): 'arguments': job.args, 'kwargs': job.kwargs, 'queue': job.origin, - }) + }) for handler in reversed(self._exc_handlers): self.log.debug('Invoking exception handler %s' % (handler,)) diff --git a/tests/test_cli.py b/tests/test_cli.py index a92fb34..3977006 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -26,6 +26,17 @@ class TestCommandLine(TestCase): class TestRQCli(RQTestCase): + + def assert_normal_execution(self, result): + if result.exit_code == 0: + return True + else: + print("Non normal execution") + print("Exit Code: {}".format(result.exit_code)) + print("Output: {}".format(result.output)) + print("Exception: {}".format(result.exception)) + self.assertEqual(result.exit_code, 0) + """Test rq_cli script""" def setUp(self): super(TestRQCli, self).setUp() @@ -41,25 +52,58 @@ class TestRQCli(RQTestCase): """rq empty -u failed""" runner = CliRunner() result = runner.invoke(main, ['empty', '-u', self.redis_url, 'failed']) - self.assertEqual(result.exit_code, 0) + self.assert_normal_execution(result) self.assertEqual(result.output.strip(), '1 jobs removed from failed queue') def test_requeue(self): """rq requeue -u --all""" runner = CliRunner() result = runner.invoke(main, ['requeue', '-u', self.redis_url, '--all']) - self.assertEqual(result.exit_code, 0) + self.assert_normal_execution(result) self.assertEqual(result.output.strip(), 'Requeueing 1 jobs from failed queue') def test_info(self): """rq info -u """ runner = CliRunner() result = runner.invoke(main, ['info', '-u', self.redis_url]) - self.assertEqual(result.exit_code, 0) + self.assert_normal_execution(result) self.assertIn('1 queues, 1 jobs total', result.output) def test_worker(self): """rq worker -u -b""" runner = CliRunner() result = runner.invoke(main, ['worker', '-u', self.redis_url, '-b']) - self.assertEqual(result.exit_code, 0) + self.assert_normal_execution(result) + + def test_suspend_and_resume(self): + """rq suspend -u + rq resume -u + """ + runner = CliRunner() + result = runner.invoke(main, ['suspend', '-u', self.redis_url]) + self.assert_normal_execution(result) + + result = runner.invoke(main, ['resume', '-u', self.redis_url]) + self.assert_normal_execution(result) + + def test_suspend_with_ttl(self): + """rq suspend -u --duration=2 + """ + runner = CliRunner() + result = runner.invoke(main, ['suspend', '-u', self.redis_url, '--duration', 1]) + self.assert_normal_execution(result) + + def test_suspend_with_invalid_ttl(self): + """rq suspend -u --duration=0 + """ + runner = CliRunner() + result = runner.invoke(main, ['suspend', '-u', self.redis_url, '--duration', 0]) + + self.assertEqual(result.exit_code, 1) + self.assertIn("Duration must be an integer greater than 1", result.output) + + + + + + diff --git a/tests/test_queue.py b/tests/test_queue.py index 8990e35..c905f13 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -4,7 +4,7 @@ from __future__ import (absolute_import, division, print_function, from rq import get_failed_queue, Queue from rq.exceptions import InvalidJobOperationError -from rq.job import Job, Status +from rq.job import Job, JobStatus from rq.worker import Worker from tests import RQTestCase @@ -262,7 +262,7 @@ class TestQueue(RQTestCase): """Enqueueing a job sets its status to "queued".""" q = Queue() job = q.enqueue(say_hello) - self.assertEqual(job.get_status(), Status.QUEUED) + self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_enqueue_explicit_args(self): """enqueue() works for both implicit/explicit args.""" @@ -346,7 +346,7 @@ class TestQueue(RQTestCase): self.assertEqual(q.job_ids, []) # Jobs dependent on finished jobs are immediately enqueued - parent_job.set_status(Status.FINISHED) + parent_job.set_status(JobStatus.FINISHED) parent_job.save() job = q.enqueue_call(say_hello, depends_on=parent_job) self.assertEqual(q.job_ids, [job.id]) @@ -363,7 +363,7 @@ class TestQueue(RQTestCase): self.assertEqual(q.job_ids, []) # Jobs dependent on finished jobs are immediately enqueued - parent_job.set_status(Status.FINISHED) + parent_job.set_status(JobStatus.FINISHED) parent_job.save() job = q.enqueue_call(say_hello, depends_on=parent_job.id) self.assertEqual(q.job_ids, [job.id]) @@ -379,7 +379,7 @@ class TestQueue(RQTestCase): self.assertEqual(job.timeout, 123) # Jobs dependent on finished jobs are immediately enqueued - parent_job.set_status(Status.FINISHED) + parent_job.set_status(JobStatus.FINISHED) parent_job.save() job = q.enqueue_call(say_hello, depends_on=parent_job, timeout=123) self.assertEqual(q.job_ids, [job.id]) @@ -441,7 +441,7 @@ class TestFailedQueue(RQTestCase): get_failed_queue().requeue(job.id) job = Job.fetch(job.id) - self.assertEqual(job.get_status(), Status.QUEUED) + self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_enqueue_preserves_result_ttl(self): """Enqueueing persists result_ttl.""" diff --git a/tests/test_worker.py b/tests/test_worker.py index c6d85ff..75f30aa 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -6,8 +6,9 @@ import os from rq import get_failed_queue, Queue, Worker, SimpleWorker from rq.compat import as_text -from rq.job import Job, Status +from rq.job import Job, JobStatus from rq.registry import StartedJobRegistry +from rq.suspension import suspend, resume from tests import RQTestCase, slow from tests.fixtures import (create_file, create_file_after_timeout, @@ -222,14 +223,14 @@ class TestWorker(RQTestCase): w = Worker([q]) job = q.enqueue(say_hello) - self.assertEqual(job.get_status(), Status.QUEUED) + self.assertEqual(job.get_status(), JobStatus.QUEUED) self.assertEqual(job.is_queued, True) self.assertEqual(job.is_finished, False) self.assertEqual(job.is_failed, False) w.work(burst=True) job = Job.fetch(job.id) - self.assertEqual(job.get_status(), Status.FINISHED) + self.assertEqual(job.get_status(), JobStatus.FINISHED) self.assertEqual(job.is_queued, False) self.assertEqual(job.is_finished, True) self.assertEqual(job.is_failed, False) @@ -238,7 +239,7 @@ class TestWorker(RQTestCase): job = q.enqueue(div_by_zero, args=(1,)) w.work(burst=True) job = Job.fetch(job.id) - self.assertEqual(job.get_status(), Status.FAILED) + self.assertEqual(job.get_status(), JobStatus.FAILED) self.assertEqual(job.is_queued, False) self.assertEqual(job.is_finished, False) self.assertEqual(job.is_failed, True) @@ -251,13 +252,13 @@ class TestWorker(RQTestCase): job = q.enqueue_call(say_hello, depends_on=parent_job) w.work(burst=True) job = Job.fetch(job.id) - self.assertEqual(job.get_status(), Status.FINISHED) + self.assertEqual(job.get_status(), JobStatus.FINISHED) parent_job = q.enqueue(div_by_zero) job = q.enqueue_call(say_hello, depends_on=parent_job) w.work(burst=True) job = Job.fetch(job.id) - self.assertNotEqual(job.get_status(), Status.FINISHED) + self.assertNotEqual(job.get_status(), JobStatus.FINISHED) def test_get_current_job(self): """Ensure worker.get_current_job() works properly""" @@ -318,3 +319,33 @@ class TestWorker(RQTestCase): 'Expected at least some work done.') self.assertEquals(job.result, 'Hi there, Adam!') self.assertEquals(job.description, '你好 世界!') + + def test_pause_worker_execution(self): + """Test Pause Worker Execution""" + + SENTINEL_FILE = '/tmp/rq-tests.txt' + + try: + # Remove the sentinel if it is leftover from a previous test run + os.remove(SENTINEL_FILE) + except OSError as e: + if e.errno != 2: + raise + + q = Queue() + job = q.enqueue(create_file, SENTINEL_FILE) + + w = Worker([q]) + + suspend(self.testconn) + + w.work(burst=True) + assert q.count == 1 + + # Should not have created evidence of execution + self.assertEquals(os.path.exists(SENTINEL_FILE), False) + + resume(self.testconn) + w.work(burst=True) + assert q.count == 0 + self.assertEquals(os.path.exists(SENTINEL_FILE), True) From 2667088b686ff55972a146efc05374f9f846b363 Mon Sep 17 00:00:00 2001 From: Travis Johnson Date: Sun, 14 Dec 2014 23:02:47 -0500 Subject: [PATCH 07/24] changed the kwarg, but forgot to change the test --- tests/test_queue.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_queue.py b/tests/test_queue.py index 8bfde6d..85bbc00 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -466,6 +466,6 @@ class TestFailedQueue(RQTestCase): job1 = q.enqueue(say_hello) job2 = q.enqueue(say_hello) assert q.dequeue() == job1 - skip_job = q.enqueue(say_hello, skip_queue=True) + skip_job = q.enqueue(say_hello, at_front=True) assert q.dequeue() == skip_job assert q.dequeue() == job2 From ccd41396f61ba213c6b32252c1aeff57c9cda7a4 Mon Sep 17 00:00:00 2001 From: Jonathan Tushman Date: Mon, 15 Dec 2014 10:04:13 -0500 Subject: [PATCH 08/24] adding suspend with duration test --- rq/cli/cli.py | 2 +- tests/test_worker.py | 25 +++++++++++++++++++++++-- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/rq/cli/cli.py b/rq/cli/cli.py index afb32ea..8dd0b2b 100755 --- a/rq/cli/cli.py +++ b/rq/cli/cli.py @@ -164,7 +164,7 @@ def worker(url, config, burst, name, worker_class, job_class, queue_class, path, queue_class = import_attribute(queue_class) if is_suspended(conn): - click.secho("The worker has been paused, run reset_paused", fg='red') + click.secho('RQ is currently suspended, to resume job execution run "rq resume"', fg='red') sys.exit(1) try: diff --git a/tests/test_worker.py b/tests/test_worker.py index 75f30aa..cebfdc4 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -3,6 +3,7 @@ from __future__ import (absolute_import, division, print_function, unicode_literals) import os +from time import sleep from rq import get_failed_queue, Queue, Worker, SimpleWorker from rq.compat import as_text @@ -12,7 +13,7 @@ from rq.suspension import suspend, resume from tests import RQTestCase, slow from tests.fixtures import (create_file, create_file_after_timeout, - div_by_zero, say_hello, say_pid) + div_by_zero, say_hello, say_pid, do_nothing) from tests.helpers import strip_microseconds @@ -320,7 +321,7 @@ class TestWorker(RQTestCase): self.assertEquals(job.result, 'Hi there, Adam!') self.assertEquals(job.description, '你好 世界!') - def test_pause_worker_execution(self): + def test_suspend_worker_execution(self): """Test Pause Worker Execution""" SENTINEL_FILE = '/tmp/rq-tests.txt' @@ -349,3 +350,23 @@ class TestWorker(RQTestCase): w.work(burst=True) assert q.count == 0 self.assertEquals(os.path.exists(SENTINEL_FILE), True) + + def test_suspend_with_duration(self): + q = Queue() + for _ in xrange(5): + q.enqueue(do_nothing) + + w = Worker([q]) + + # This suspends workers for working for 2 second + suspend(self.testconn, 2) + + # So when this burst of work happens the queue should remain at 5 + w.work(burst=True) + assert q.count == 5 + + sleep(3) + + # The suspension should be expired now, and a burst of work should now clear the queue + w.work(burst=True) + assert q.count == 0 From 60c27d5a2768a34a42f19ab92f3278bcf17e468d Mon Sep 17 00:00:00 2001 From: Jonathan Tushman Date: Mon, 15 Dec 2014 10:35:19 -0500 Subject: [PATCH 09/24] fixing xrange for python3 compatability --- tests/test_worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_worker.py b/tests/test_worker.py index fd9d0bd..d3b8f41 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -353,7 +353,7 @@ class TestWorker(RQTestCase): def test_suspend_with_duration(self): q = Queue() - for _ in xrange(5): + for _ in range(5): q.enqueue(do_nothing) w = Worker([q]) From 543bcbc60b117e424995483cb3dcfe8dbfbbd77e Mon Sep 17 00:00:00 2001 From: Vincent Driessen Date: Wed, 21 Jan 2015 16:12:08 -0800 Subject: [PATCH 10/24] Explicitly cast map() result to a list, for Python 3 compat. --- rq/worker.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rq/worker.py b/rq/worker.py index 3ba297a..929f91d 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -166,11 +166,11 @@ class Worker(object): def queue_names(self): """Returns the queue names of this worker's queues.""" - return map(lambda q: q.name, self.queues) + return list(map(lambda q: q.name, self.queues)) def queue_keys(self): """Returns the Redis keys representing this worker's queues.""" - return map(lambda q: q.key, self.queues) + return list(map(lambda q: q.key, self.queues)) @property def name(self): From 01ab2f20dd90db569ec7ea954fb0c3341fee88d2 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 23 Jan 2015 13:58:25 +0700 Subject: [PATCH 11/24] Minor copywriting fixes. --- rq/worker.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/rq/worker.py b/rq/worker.py index 3ba297a..c0cb296 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -67,10 +67,13 @@ def signal_name(signum): return 'SIG_UNKNOWN' -WorkerStatus = enum('WorkerStatus', - STARTED='started', SUSPENDED='suspended', BUSY='busy', - IDLE='idle' - ) +WorkerStatus = enum( + 'WorkerStatus', + STARTED='started', + SUSPENDED='suspended', + BUSY='busy', + IDLE='idle' +) class Worker(object): @@ -342,7 +345,7 @@ class Worker(object): signal.signal(signal.SIGTERM, request_stop) def check_for_suspension(self, burst): - """Check to see if the workers have been suspended by something like `rq suspend`""" + """Check to see if workers have been suspended by `rq suspend`""" before_state = None notified = False @@ -350,8 +353,8 @@ class Worker(object): while not self.stopped and is_suspended(self.connection): if burst: - self.log.info('Suspended in burst mode -- exiting.') - self.log.info('Note: There could still be unperformed jobs on the queue') + self.log.info('Suspended in burst mode -- exiting.' + 'Note: There could still be unperformed jobs on the queue') raise StopRequested if not notified: From dc09676ee03b978691d5d3c0bc725ab30bc0077a Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 23 Jan 2015 13:58:25 +0700 Subject: [PATCH 12/24] Minor copywriting fixes. --- rq/worker.py | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/rq/worker.py b/rq/worker.py index 929f91d..b971a24 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -67,10 +67,13 @@ def signal_name(signum): return 'SIG_UNKNOWN' -WorkerStatus = enum('WorkerStatus', - STARTED='started', SUSPENDED='suspended', BUSY='busy', - IDLE='idle' - ) +WorkerStatus = enum( + 'WorkerStatus', + STARTED='started', + SUSPENDED='suspended', + BUSY='busy', + IDLE='idle' +) class Worker(object): @@ -342,7 +345,7 @@ class Worker(object): signal.signal(signal.SIGTERM, request_stop) def check_for_suspension(self, burst): - """Check to see if the workers have been suspended by something like `rq suspend`""" + """Check to see if workers have been suspended by `rq suspend`""" before_state = None notified = False @@ -350,8 +353,8 @@ class Worker(object): while not self.stopped and is_suspended(self.connection): if burst: - self.log.info('Suspended in burst mode -- exiting.') - self.log.info('Note: There could still be unperformed jobs on the queue') + self.log.info('Suspended in burst mode -- exiting.' + 'Note: There could still be unperformed jobs on the queue') raise StopRequested if not notified: From dc5cd514ee757b401415bd1dccdf70831c4797ef Mon Sep 17 00:00:00 2001 From: Vincent Driessen Date: Wed, 21 Jan 2015 16:12:08 -0800 Subject: [PATCH 13/24] Explicitly cast map() result to a list, for Python 3 compat. --- rq/worker.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rq/worker.py b/rq/worker.py index c0cb296..b971a24 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -169,11 +169,11 @@ class Worker(object): def queue_names(self): """Returns the queue names of this worker's queues.""" - return map(lambda q: q.name, self.queues) + return list(map(lambda q: q.name, self.queues)) def queue_keys(self): """Returns the Redis keys representing this worker's queues.""" - return map(lambda q: q.key, self.queues) + return list(map(lambda q: q.key, self.queues)) @property def name(self): From 7fd2ac8ca6bc737974d11a6f2aedd20679a5fa17 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 23 Jan 2015 15:16:32 +0700 Subject: [PATCH 14/24] Added "DEFERRED" Job status for jobs that have unsatisfied dependencies. --- rq/job.py | 11 ++++++++--- rq/queue.py | 2 +- tests/test_queue.py | 6 ++++-- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/rq/job.py b/rq/job.py index 9afe578..2337ad4 100644 --- a/rq/job.py +++ b/rq/job.py @@ -25,9 +25,14 @@ dumps = partial(pickle.dumps, protocol=pickle.HIGHEST_PROTOCOL) loads = pickle.loads -JobStatus = enum('JobStatus', - QUEUED='queued', FINISHED='finished', FAILED='failed', - STARTED='started') +JobStatus = enum( + 'JobStatus', + QUEUED='queued', + FINISHED='finished', + FAILED='failed', + STARTED='started', + DEFERRED='deferred' +) # Sentinel value to mark that some of our lazily evaluated properties have not # yet been evaluated. diff --git a/rq/queue.py b/rq/queue.py index 7352459..9cc0fa0 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -182,7 +182,6 @@ 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, status=JobStatus.QUEUED, description=description, depends_on=depends_on, timeout=timeout, @@ -200,6 +199,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() diff --git a/tests/test_queue.py b/tests/test_queue.py index 0a84e78..cddb3f6 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -342,8 +342,9 @@ class TestQueue(RQTestCase): # Job with unfinished dependency is not immediately enqueued parent_job = Job.create(func=say_hello) q = Queue() - q.enqueue_call(say_hello, depends_on=parent_job) + job = q.enqueue_call(say_hello, depends_on=parent_job) self.assertEqual(q.job_ids, []) + self.assertEqual(job.get_status(), JobStatus.DEFERRED) # Jobs dependent on finished jobs are immediately enqueued parent_job.set_status(JobStatus.FINISHED) @@ -351,6 +352,7 @@ class TestQueue(RQTestCase): job = q.enqueue_call(say_hello, depends_on=parent_job) self.assertEqual(q.job_ids, [job.id]) self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT) + self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_enqueue_job_with_dependency_by_id(self): """Enqueueing jobs should work as expected by id as well as job-objects.""" @@ -368,7 +370,7 @@ class TestQueue(RQTestCase): self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT) def test_enqueue_job_with_dependency_and_timeout(self): - """Jobs still know their specified timeout after being scheduled as a dependency.""" + """Jobs remember their timeout when enqueued as a dependency.""" # Job with unfinished dependency is not immediately enqueued parent_job = Job.create(func=say_hello) q = Queue() From 9320496402c1a92d946eaac28bfb03aae194c0b0 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 23 Jan 2015 16:05:29 +0700 Subject: [PATCH 15/24] Simplify FailedQueue.quarantine and ensure that a deferred job's status is set to Queued when enqueued. --- rq/queue.py | 41 ++++++++++++++++++++++++----------------- 1 file changed, 24 insertions(+), 17 deletions(-) diff --git a/rq/queue.py b/rq/queue.py index 9cc0fa0..eb50314 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -248,24 +248,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) @@ -401,14 +402,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.""" From 3e674fbe6aa20162c094e93fb9a01ead2dabc43d Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 23 Jan 2015 16:33:48 +0700 Subject: [PATCH 16/24] queue.enqueue() should set job.origin. --- rq/job.py | 7 +++++-- rq/queue.py | 9 +++++---- tests/test_queue.py | 3 +-- 3 files changed, 11 insertions(+), 8 deletions(-) diff --git a/rq/job.py b/rq/job.py index 2337ad4..fbf7d5a 100644 --- a/rq/job.py +++ b/rq/job.py @@ -88,8 +88,8 @@ class Job(object): # Job construction @classmethod def create(cls, func, args=None, kwargs=None, connection=None, - result_ttl=None, ttl=None, status=None, description=None, depends_on=None, timeout=None, - id=None): + result_ttl=None, ttl=None, status=None, description=None, + depends_on=None, timeout=None, id=None, origin=None): """Creates a new Job instance for the given function, arguments, and keyword arguments. """ @@ -107,6 +107,9 @@ class Job(object): if id is not None: job.set_id(id) + if origin is not None: + job.origin = origin + # Set the core job tuple properties job._instance = None if inspect.ismethod(func): diff --git a/rq/queue.py b/rq/queue.py index eb50314..d596106 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -182,10 +182,11 @@ class Queue(object): """ timeout = timeout or self._default_timeout - job = self.job_class.create(func, args, kwargs, connection=self.connection, - result_ttl=result_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, 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. diff --git a/tests/test_queue.py b/tests/test_queue.py index cddb3f6..f5edcbf 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -117,6 +117,7 @@ class TestQueue(RQTestCase): # say_hello spec holds which queue this is sent to job = q.enqueue(say_hello, 'Nick', foo='bar') job_id = job.id + self.assertEqual(job.origin, q.name) # Inspect data inside Redis q_key = 'rq:queue:default' @@ -131,14 +132,12 @@ class TestQueue(RQTestCase): job = Job.create(func=say_hello, args=('Nick',), kwargs=dict(foo='bar')) # Preconditions - self.assertIsNone(job.origin) self.assertIsNone(job.enqueued_at) # Action q.enqueue_job(job) # Postconditions - self.assertEquals(job.origin, q.name) self.assertIsNotNone(job.enqueued_at) def test_pop_job_id(self): From dac0be6cc79dd78f8039e4b5b989db425ed380b3 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 23 Jan 2015 17:53:27 +0700 Subject: [PATCH 17/24] Added DeferredJobsRegistry to keep track of deferred jobs. --- rq/job.py | 8 +++++++- rq/queue.py | 5 +++++ rq/registry.py | 18 +++++++++++++++++- tests/test_job.py | 11 +++++++++-- tests/test_queue.py | 22 ++++++++++++++-------- tests/test_registry.py | 19 ++++++++++++++++++- 6 files changed, 70 insertions(+), 13 deletions(-) diff --git a/rq/job.py b/rq/job.py index fbf7d5a..d815c5b 100644 --- a/rq/job.py +++ b/rq/job.py @@ -544,8 +544,14 @@ class Job(object): rq:job:job_id:dependents = {'job_id_1', 'job_id_2'} - This method adds the current job in its dependency's dependents set. + This method adds the job in its dependency's dependents set + and adds the job to DeferredJobRegistry. """ + from .registry import DeferredJobRegistry + + registry = DeferredJobRegistry(self.origin, connection=self.connection) + registry.add(self, pipeline=pipeline) + connection = pipeline if pipeline is not None else self.connection connection.sadd(Job.dependents_key_for(self._dependency_id), self.id) diff --git a/rq/queue.py b/rq/queue.py index d596106..e2c358c 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -279,11 +279,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): diff --git a/rq/registry.py b/rq/registry.py index b4cf43f..08798eb 100644 --- a/rq/registry.py +++ b/rq/registry.py @@ -24,7 +24,7 @@ class BaseRegistry(object): self.cleanup() return self.connection.zcard(self.key) - def add(self, job, ttl, pipeline=None): + def add(self, job, ttl=0, pipeline=None): """Adds a job to a registry with expiry time of now + ttl.""" score = ttl if ttl < 0 else current_timestamp() + ttl if pipeline is not None: @@ -108,3 +108,19 @@ class FinishedJobRegistry(BaseRegistry): """ score = timestamp if timestamp is not None else current_timestamp() self.connection.zremrangebyscore(self.key, 0, score) + + +class DeferredJobRegistry(BaseRegistry): + """ + Registry of deferred jobs (waiting for another job to finish). + """ + + def __init__(self, name='default', connection=None): + super(DeferredJobRegistry, self).__init__(name, connection) + self.key = 'rq:deferred:%s' % name + + def cleanup(self): + """This method is only here to prevent errors because this method is + automatically called by `count()` and `get_job_ids()` methods + implemented in BaseRegistry.""" + pass diff --git a/tests/test_job.py b/tests/test_job.py index 34859a7..b7854b3 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -7,6 +7,7 @@ from datetime import datetime from rq.compat import as_text, PY2 from rq.exceptions import NoSuchJobError, UnpickleError from rq.job import get_current_job, Job +from rq.registry import DeferredJobRegistry from rq.queue import Queue from rq.utils import utcformat @@ -331,12 +332,18 @@ class TestJob(RQTestCase): self.assertRaises(NoSuchJobError, Job.fetch, job.id, self.testconn) def test_register_dependency(self): - """Test that jobs updates the correct job dependents.""" - job = Job.create(func=say_hello) + """Ensure dependency registration works properly.""" + origin = 'some_queue' + registry = DeferredJobRegistry(origin, self.testconn) + + job = Job.create(func=say_hello, origin=origin) job._dependency_id = 'id' job.save() + + self.assertEqual(registry.get_job_ids(), []) job.register_dependency() self.assertEqual(as_text(self.testconn.spop('rq:job:id:dependents')), job.id) + self.assertEqual(registry.get_job_ids(), [job.id]) def test_cancel(self): """job.cancel() deletes itself & dependents mapping from Redis.""" diff --git a/tests/test_queue.py b/tests/test_queue.py index f5edcbf..369b3f1 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -5,6 +5,7 @@ from __future__ import (absolute_import, division, print_function, from rq import get_failed_queue, Queue from rq.exceptions import InvalidJobOperationError from rq.job import Job, JobStatus +from rq.registry import DeferredJobRegistry from rq.worker import Worker from tests import RQTestCase @@ -319,23 +320,28 @@ class TestQueue(RQTestCase): self.assertEquals(len(Queue.all()), 3) def test_enqueue_dependents(self): - """Enqueueing the dependent jobs pushes all jobs in the depends set to the queue.""" + """Enqueueing dependent jobs pushes all jobs in the depends set to the queue + and removes them from DeferredJobQueue.""" q = Queue() parent_job = Job.create(func=say_hello) parent_job.save() - job_1 = Job.create(func=say_hello, depends_on=parent_job) - job_1.save() - job_1.register_dependency() - job_2 = Job.create(func=say_hello, depends_on=parent_job) - job_2.save() - job_2.register_dependency() + job_1 = q.enqueue(say_hello, depends_on=parent_job) + job_2 = q.enqueue(say_hello, depends_on=parent_job) + registry = DeferredJobRegistry(q.name, connection=self.testconn) + self.assertEqual( + set(registry.get_job_ids()), + set([job_1.id, job_2.id]) + ) # After dependents is enqueued, job_1 and job_2 should be in queue self.assertEqual(q.job_ids, []) q.enqueue_dependents(parent_job) - self.assertEqual(set(q.job_ids), set([job_1.id, job_2.id])) + self.assertEqual(set(q.job_ids), set([job_2.id, job_1.id])) self.assertFalse(self.testconn.exists(parent_job.dependents_key)) + # DeferredJobRegistry should also be empty + self.assertEqual(registry.get_job_ids(), []) + def test_enqueue_job_with_dependency(self): """Jobs are enqueued only when their dependencies are finished.""" # Job with unfinished dependency is not immediately enqueued diff --git a/tests/test_registry.py b/tests/test_registry.py index 26470e3..3798a8f 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -5,7 +5,8 @@ from rq.job import Job from rq.queue import FailedQueue, Queue from rq.utils import current_timestamp from rq.worker import Worker -from rq.registry import FinishedJobRegistry, StartedJobRegistry +from rq.registry import (DeferredJobRegistry, FinishedJobRegistry, + StartedJobRegistry) from tests import RQTestCase from tests.fixtures import div_by_zero, say_hello @@ -135,3 +136,19 @@ class TestFinishedJobRegistry(RQTestCase): failed_job = queue.enqueue(div_by_zero) worker.perform_job(failed_job) self.assertEqual(self.registry.get_job_ids(), [job.id]) + + +class TestRegistry(RQTestCase): + + def setUp(self): + super(TestRegistry, self).setUp() + self.registry = DeferredJobRegistry(connection=self.testconn) + + def test_add(self): + """Adding a job to DeferredJobsRegistry.""" + job = Job() + self.registry.add(job) + self.assertEqual( + self.testconn.zrange(self.registry.key, 0, -1), + [job.id] + ) \ No newline at end of file From de1cd8a83c8f1da732921d7b4536371f8b6e06b9 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Mon, 26 Jan 2015 08:31:58 +0700 Subject: [PATCH 18/24] Fixed test error in Python 3. --- tests/test_registry.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/tests/test_registry.py b/tests/test_registry.py index 3798a8f..5f5cf75 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- from __future__ import absolute_import +from rq.compat import as_text from rq.job import Job from rq.queue import FailedQueue, Queue from rq.utils import current_timestamp @@ -120,7 +121,6 @@ class TestFinishedJobRegistry(RQTestCase): self.registry.cleanup(timestamp + 20) self.assertEqual(self.registry.get_job_ids(), ['baz']) - def test_jobs_are_put_in_registry(self): """Completed jobs are added to FinishedJobRegistry.""" self.assertEqual(self.registry.get_job_ids(), []) @@ -138,7 +138,7 @@ class TestFinishedJobRegistry(RQTestCase): self.assertEqual(self.registry.get_job_ids(), [job.id]) -class TestRegistry(RQTestCase): +class TestDeferredRegistry(RQTestCase): def setUp(self): super(TestRegistry, self).setUp() @@ -148,7 +148,6 @@ class TestRegistry(RQTestCase): """Adding a job to DeferredJobsRegistry.""" job = Job() self.registry.add(job) - self.assertEqual( - self.testconn.zrange(self.registry.key, 0, -1), - [job.id] - ) \ No newline at end of file + job_ids = [as_text(job_id) for job_id in + self.testconn.zrange(self.registry.key, 0, -1)] + self.assertEqual(job_ids, [job.id]) From 105b95e9b8eaa20bbd9b1a47742ffef71d2bd694 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Mon, 26 Jan 2015 08:37:03 +0700 Subject: [PATCH 19/24] Fixed an error in super call. --- tests/test_registry.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_registry.py b/tests/test_registry.py index 5f5cf75..f54b315 100644 --- a/tests/test_registry.py +++ b/tests/test_registry.py @@ -141,7 +141,7 @@ class TestFinishedJobRegistry(RQTestCase): class TestDeferredRegistry(RQTestCase): def setUp(self): - super(TestRegistry, self).setUp() + super(TestDeferredRegistry, self).setUp() self.registry = DeferredJobRegistry(connection=self.testconn) def test_add(self): From 008d72ff09e77e8186b81955ebac9509fd85b701 Mon Sep 17 00:00:00 2001 From: Selwin Ong Date: Fri, 30 Jan 2015 14:39:42 +0700 Subject: [PATCH 20/24] Documented new features in 0.5.0. --- CHANGES.md | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/CHANGES.md b/CHANGES.md index 9097cb9..12f273a 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,3 +1,25 @@ +### 0.5.0 +(Jan 30th, 2015) + +- RQ workers can now be paused and resumed using `rq suspend` and + `rq resume` commands. Thanks Jonathan Tushman! +- Jobs that are being performed are now stored in `StartedJobRegistry` + for monitoring purposes. This also prevents currently active jobs from + being orphaned/lost in the case of hard shutdowns. +- You can now monitor finished jobs by checking `FinishedJobRegistry`. + Thanks Nic Cope for helping! +- Jobs with unmet dependencies are now created with `deferred` as their + status. You can monitor deferred jobs by checking `DeferredJobRegistry`. +- It is now possible to enqueue a job at the beginning of queue using + `queue.enqueue(func, at_front=True)`. Thanks Travis Johnson! +- Command line scripts have all been refactored to use `click`. Thanks Lyon Zhang! +- Added a new `SimpleWorker` that does not fork when executing jobs. + Useful for testing purposes. Thanks Cal Leeming! +- Added `--queue-class` and `--job-class` arguments to `rqworker` script. + Thanks David Bonner! +- Many other minor bug fixes and enhancements. + + ### 0.4.6 (May 21st, 2014) From e8c3b96a5a9f4f28a3fc8a42eb330b91af627b7f Mon Sep 17 00:00:00 2001 From: Vincent Driessen Date: Fri, 30 Jan 2015 09:17:34 +0100 Subject: [PATCH 21/24] Bump to 0.5.0. --- rq/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rq/version.py b/rq/version.py index 28d738f..30d837a 100644 --- a/rq/version.py +++ b/rq/version.py @@ -1,4 +1,4 @@ # -*- coding: utf-8 -*- from __future__ import (absolute_import, division, print_function, unicode_literals) -VERSION = '0.4.6' +VERSION = '0.5.0' From fac2b10309a7148b13cd259e6770f2695c126628 Mon Sep 17 00:00:00 2001 From: Vincent Driessen Date: Fri, 30 Jan 2015 09:22:53 +0100 Subject: [PATCH 22/24] Auto-sort imports using isort. --- rq/cli/helpers.py | 3 +-- rq/job.py | 2 +- rq/queue.py | 11 +++++------ rq/utils.py | 4 ++-- rq/worker.py | 6 +++--- tests/test_job.py | 12 ++++++------ tests/test_queue.py | 8 ++++---- tests/test_worker.py | 14 +++++++------- 8 files changed, 29 insertions(+), 31 deletions(-) diff --git a/rq/cli/helpers.py b/rq/cli/helpers.py index 9ea4b58..0730d17 100644 --- a/rq/cli/helpers.py +++ b/rq/cli/helpers.py @@ -8,9 +8,8 @@ from functools import partial import click from rq import Queue, Worker -from rq.worker import WorkerStatus from rq.logutils import setup_loghandlers -from rq.suspension import is_suspended +from rq.worker import WorkerStatus red = partial(click.style, fg='red') green = partial(click.style, fg='green') diff --git a/rq/job.py b/rq/job.py index d815c5b..981694c 100644 --- a/rq/job.py +++ b/rq/job.py @@ -12,7 +12,7 @@ from rq.compat import as_text, decode_redis_hash, string_types, text_type from .connections import resolve_connection from .exceptions import NoSuchJobError, UnpickleError from .local import LocalStack -from .utils import import_attribute, utcformat, utcnow, utcparse, enum +from .utils import enum, import_attribute, utcformat, utcnow, utcparse try: import cPickle as pickle diff --git a/rq/queue.py b/rq/queue.py index e2c358c..9815f7e 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -4,15 +4,14 @@ from __future__ import (absolute_import, division, print_function, import uuid -from .connections import resolve_connection -from .job import Job, JobStatus -from .utils import import_attribute, utcnow +from redis import WatchError +from .compat import as_text, string_types, total_ordering +from .connections import resolve_connection from .exceptions import (DequeueTimeout, InvalidJobOperationError, NoSuchJobError, UnpickleError) -from .compat import total_ordering, string_types, as_text - -from redis import WatchError +from .job import Job, JobStatus +from .utils import import_attribute, utcnow def get_failed_queue(connection=None): diff --git a/rq/utils.py b/rq/utils.py index db56020..7bcb392 100644 --- a/rq/utils.py +++ b/rq/utils.py @@ -9,12 +9,12 @@ from __future__ import (absolute_import, division, print_function, unicode_literals) import calendar -import importlib import datetime +import importlib import logging import sys -from .compat import is_python_version, as_text +from .compat import as_text, is_python_version class _Colorizer(object): diff --git a/rq/worker.py b/rq/worker.py index b971a24..0c53409 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -21,11 +21,11 @@ from .exceptions import DequeueTimeout, NoQueueError from .job import Job, JobStatus from .logutils import setup_loghandlers from .queue import get_failed_queue, Queue -from .timeouts import UnixSignalDeathPenalty -from .utils import import_attribute, make_colorizer, utcformat, utcnow, enum -from .version import VERSION from .registry import FinishedJobRegistry, StartedJobRegistry from .suspension import is_suspended +from .timeouts import UnixSignalDeathPenalty +from .utils import enum, import_attribute, make_colorizer, utcformat, utcnow +from .version import VERSION try: from procname import setprocname diff --git a/tests/test_job.py b/tests/test_job.py index b7854b3..89adf6d 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -4,18 +4,18 @@ from __future__ import (absolute_import, division, print_function, from datetime import datetime +from tests import RQTestCase +from tests.fixtures import (access_self, CallableObject, Number, say_hello, + some_calculation) +from tests.helpers import strip_microseconds + from rq.compat import as_text, PY2 from rq.exceptions import NoSuchJobError, UnpickleError from rq.job import get_current_job, Job -from rq.registry import DeferredJobRegistry from rq.queue import Queue +from rq.registry import DeferredJobRegistry from rq.utils import utcformat -from tests import RQTestCase -from tests.fixtures import (access_self, CallableObject, Number, say_hello, - some_calculation) -from tests.helpers import strip_microseconds - try: from cPickle import loads, dumps except ImportError: diff --git a/tests/test_queue.py b/tests/test_queue.py index 369b3f1..175a1c0 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -2,16 +2,16 @@ from __future__ import (absolute_import, division, print_function, unicode_literals) +from tests import RQTestCase +from tests.fixtures import (div_by_zero, echo, Number, say_hello, + some_calculation) + from rq import get_failed_queue, Queue from rq.exceptions import InvalidJobOperationError from rq.job import Job, JobStatus from rq.registry import DeferredJobRegistry from rq.worker import Worker -from tests import RQTestCase -from tests.fixtures import (div_by_zero, echo, Number, say_hello, - some_calculation) - class CustomJob(Job): pass diff --git a/tests/test_worker.py b/tests/test_worker.py index d3b8f41..8a35d03 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -5,17 +5,17 @@ from __future__ import (absolute_import, division, print_function, import os from time import sleep -from rq import get_failed_queue, Queue, Worker, SimpleWorker -from rq.compat import as_text -from rq.job import Job, JobStatus -from rq.registry import StartedJobRegistry -from rq.suspension import suspend, resume - from tests import RQTestCase, slow from tests.fixtures import (create_file, create_file_after_timeout, - div_by_zero, say_hello, say_pid, do_nothing) + div_by_zero, do_nothing, say_hello, say_pid) from tests.helpers import strip_microseconds +from rq import get_failed_queue, Queue, SimpleWorker, Worker +from rq.compat import as_text +from rq.job import Job, JobStatus +from rq.registry import StartedJobRegistry +from rq.suspension import resume, suspend + class CustomJob(Job): pass From b8d425b318567073a8343b8f884547b62ced810d Mon Sep 17 00:00:00 2001 From: Vincent Driessen Date: Fri, 30 Jan 2015 09:23:07 +0100 Subject: [PATCH 23/24] Various PEP8 fixes. --- rq/queue.py | 1 - rq/suspension.py | 2 +- rq/utils.py | 2 +- rq/version.py | 1 + rq/worker.py | 7 ++----- tests/test_cli.py | 6 ------ tests/test_job.py | 2 +- tests/test_queue.py | 2 +- tests/test_worker.py | 2 +- 9 files changed, 8 insertions(+), 17 deletions(-) diff --git a/rq/queue.py b/rq/queue.py index 9815f7e..3f59a50 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -253,7 +253,6 @@ class Queue(object): If Queue is instantiated with async=False, job is executed immediately. """ - with self.connection._pipeline() as pipeline: # Add Queue key set self.connection.sadd(self.redis_queues_keys, self.key) diff --git a/rq/suspension.py b/rq/suspension.py index b734acd..93152b9 100644 --- a/rq/suspension.py +++ b/rq/suspension.py @@ -15,4 +15,4 @@ def suspend(connection, ttl=None): def resume(connection): - return connection.delete(WORKERS_SUSPENDED) \ No newline at end of file + return connection.delete(WORKERS_SUSPENDED) diff --git a/rq/utils.py b/rq/utils.py index 7bcb392..3e44a98 100644 --- a/rq/utils.py +++ b/rq/utils.py @@ -217,4 +217,4 @@ def enum(name, *sequential, **named): # On Python 2 type() requires a byte string (which is str() on Python 2). # On Python 3 it does not matter, so we'll use str(), which acts as # a no-op. - return type(str(name), (), values) \ No newline at end of file + return type(str(name), (), values) diff --git a/rq/version.py b/rq/version.py index 30d837a..0d96ce8 100644 --- a/rq/version.py +++ b/rq/version.py @@ -1,4 +1,5 @@ # -*- coding: utf-8 -*- from __future__ import (absolute_import, division, print_function, unicode_literals) + VERSION = '0.5.0' diff --git a/rq/worker.py b/rq/worker.py index 0c53409..201dfed 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -12,7 +12,6 @@ import sys import time import traceback import warnings -from datetime import datetime from rq.compat import as_text, string_types, text_type @@ -54,8 +53,8 @@ def compact(l): return [x for x in l if x is not None] _signames = dict((getattr(signal, signame), signame) - for signame in dir(signal) - if signame.startswith('SIG') and '_' not in signame) + for signame in dir(signal) + if signame.startswith('SIG') and '_' not in signame) def signal_name(signum): @@ -367,7 +366,6 @@ class Worker(object): if before_state: self.set_state(before_state) - def work(self, burst=False): """Starts the work loop. @@ -416,7 +414,6 @@ class Worker(object): self.register_death() return did_perform_work - def dequeue_job_and_maintain_ttl(self, timeout): result = None qnames = self.queue_names() diff --git a/tests/test_cli.py b/tests/test_cli.py index 3977006..f1b7fd4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -101,9 +101,3 @@ class TestRQCli(RQTestCase): self.assertEqual(result.exit_code, 1) self.assertIn("Duration must be an integer greater than 1", result.output) - - - - - - diff --git a/tests/test_job.py b/tests/test_job.py index 89adf6d..e117ea9 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -339,7 +339,7 @@ class TestJob(RQTestCase): job = Job.create(func=say_hello, origin=origin) job._dependency_id = 'id' job.save() - + self.assertEqual(registry.get_job_ids(), []) job.register_dependency() self.assertEqual(as_text(self.testconn.spop('rq:job:id:dependents')), job.id) diff --git a/tests/test_queue.py b/tests/test_queue.py index 175a1c0..e4e9253 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -340,7 +340,7 @@ class TestQueue(RQTestCase): self.assertFalse(self.testconn.exists(parent_job.dependents_key)) # DeferredJobRegistry should also be empty - self.assertEqual(registry.get_job_ids(), []) + self.assertEqual(registry.get_job_ids(), []) def test_enqueue_job_with_dependency(self): """Jobs are enqueued only when their dependencies are finished.""" diff --git a/tests/test_worker.py b/tests/test_worker.py index 8a35d03..c47a62e 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -334,7 +334,7 @@ class TestWorker(RQTestCase): raise q = Queue() - job = q.enqueue(create_file, SENTINEL_FILE) + q.enqueue(create_file, SENTINEL_FILE) w = Worker([q]) From 1018330cea8ac5e6fe407860f012c6c90e279e62 Mon Sep 17 00:00:00 2001 From: Leonid Shvechikov Date: Sun, 1 Feb 2015 10:37:21 +0300 Subject: [PATCH 24/24] Tiny fix of the indefinite article before "RQ" "An" should be used here because "RQ" sounds like "ar-kiu". --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 8b83f68..2b1fa71 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,7 @@ def count_words_at_url(url): You do use the excellent [requests][r] package, don't you? -Then, create a RQ queue: +Then, create an RQ queue: ```python from rq import Queue, use_connection