Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Feat worker limit #142

Merged
merged 15 commits into from
Aug 11, 2019
Merged

Feat worker limit #142

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
82614b2
feat: add queue_read_limit to limit the number of job IDs read at eac…
rubik Aug 9, 2019
cbc5eb2
fix: set offset to None in case the limit is not specified
rubik Aug 9, 2019
dc3cd62
chore: format code
rubik Aug 10, 2019
c3a4ccf
fix: set queue_read_limit to max_jobs if not specified
rubik Aug 11, 2019
10c9def
chore: refactor worker poll iteration into a separate method and writ…
rubik Aug 11, 2019
930af48
chore: add history note
rubik Aug 11, 2019
ca7e3f7
chore: format code
rubik Aug 11, 2019
5c487a0
chore: format docstring
rubik Aug 11, 2019
b50397e
fix: remove additional call to Redis introduced by mistake
rubik Aug 11, 2019
8328a60
chore: remove unneeded call to create_pool in queue_read_limit tests
rubik Aug 11, 2019
443302f
chore: assign queue_read_limit in one line
rubik Aug 11, 2019
13ab4ed
chore: increase delay in queue_read_limit tests for CI systems
rubik Aug 11, 2019
f0c6a24
Merge branch 'master' into feat-worker-limit
rubik Aug 11, 2019
6ede7ef
chore: fix linting errors
rubik Aug 11, 2019
c1468e9
fix remaining sleeps
samuelcolvin Aug 11, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion arq/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,9 +138,12 @@ class Worker:
:param job_timeout: default job timeout (max run time)
:param keep_result: default duration to keep job results for
:param poll_delay: duration between polling the queue for new jobs
:param queue_read_limit: the maximum number of jobs to pull from the queue each time it's polled
:param max_tries: default maximum number of times to retry a job
:param health_check_interval: how often to set the health check key
:param health_check_key: redis key under which health check is set
:param retry_jobs: whether to retry jobs on Retry or CancelledError or not
:param max_burst_jobs: the maximum number of jobs to process in burst mode (disabled with negative values)
"""

def __init__(
Expand All @@ -158,6 +161,7 @@ def __init__(
job_timeout: SecondsTimedelta = 300,
keep_result: SecondsTimedelta = 3600,
poll_delay: SecondsTimedelta = 0.5,
queue_read_limit: Optional[int] = None,
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe this shouldn't be optional?

Is there any reason not to set it to 1000 or max_jobs?

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@rubik did you see this comment? What do you think?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@samuelcolvin I agree, yes. One of the last commits sets it to max_jobs if it's None.

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

okay, so the type shouldn't be Optional, on the class

should be:

self.queue_read_limit: int = queue_read_limit or max_jobs

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@samuelcolvin I don't follow. In your line with or you allow queue_read_limit to be None. That means that we need to keep Optional in the type hint. If the user does not specify it, it's None. I changed the assignment to be a one-liner like yours.

Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What you've done is correct.

My point is that if you do:

foo: Optional[int] = 1

bar = foo

if bar is None:
    bar = 123

Mypy reads this as implicitly setting the type of bar to Optional[int], that doesn't change in the if block.

If instead you do

foo: Optional[int] = 1
bar = foo or 123

The type of bar is int, this is what we do above, except we're explicit and add a type hint.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@samuelcolvin Now I understand, thanks for the explanation.

max_tries: int = 5,
health_check_interval: SecondsTimedelta = 3600,
health_check_key: Optional[str] = None,
Expand All @@ -180,6 +184,12 @@ def __init__(
self.job_timeout_s = to_seconds(job_timeout)
self.keep_result_s = to_seconds(keep_result)
self.poll_delay_s = to_seconds(poll_delay)
self.queue_read_limit = queue_read_limit
if self.queue_read_limit is None:
# Redis requires offset and count to be both set or both unset
self.queue_read_offset = None
else:
self.queue_read_offset = 0
self.max_tries = max_tries
self.health_check_interval = to_seconds(health_check_interval)
if health_check_key is None:
Expand Down Expand Up @@ -260,7 +270,9 @@ async def main(self):
async for _ in poll(self.poll_delay_s): # noqa F841
async with self.sem: # don't bother with zrangebyscore until we have "space" to run the jobs
now = timestamp_ms()
job_ids = await self.pool.zrangebyscore(self.queue_name, max=now)
job_ids = await self.pool.zrangebyscore(
self.queue_name, offset=self.queue_read_offset, count=self.queue_read_limit, max=now
)
await self.run_jobs(job_ids)

# required to make sure errors in run_job get propagated
Expand Down
11 changes: 9 additions & 2 deletions tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,16 @@ async def arq_redis(loop):
async def worker(arq_redis):
worker_: Worker = None

def create(functions=[], burst=True, poll_delay=0, **kwargs):
def create(functions=[], burst=True, poll_delay=0, queue_read_limit=40, **kwargs):
nonlocal worker_
worker_ = Worker(functions=functions, redis_pool=arq_redis, burst=burst, poll_delay=poll_delay, **kwargs)
worker_ = Worker(
functions=functions,
redis_pool=arq_redis,
burst=burst,
poll_delay=poll_delay,
queue_read_limit=queue_read_limit,
**kwargs,
)
return worker_

yield create
Expand Down
1 change: 1 addition & 0 deletions tests/test_worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ class Settings:
functions = [func(foobar, name='foobar')]
burst = True
poll_delay = 0
queue_read_limit = 10

loop.run_until_complete(arq_redis.enqueue_job('foobar'))
worker = run_worker(Settings)
Expand Down