Feat worker limit #142

rubik · 2019-08-09T20:54:33Z

This PR adds a new parameter to the Worker class: queue_read_limit. It allows limiting the number of job IDs that are read from the queue at each polling interval.

Redis requires to set both offset and count, so I've added another instance attribute called queue_read_offset that sets the offset to 0 in case queue_read_limit is specified, and to None otherwise. I haven't added a corresponding parameter for this attribute because I think that in the majority of the cases it's not needed. I think it could be useful to investigate it in case multiple workers are run in parallel, but I haven't tested this so I refrained from complicating the public interface further.

This PR passes all tests except two, which also fail in the master branch. So they are most likely unrelated to these changes.

Fixes #141

…h poll

samuelcolvin

otherwise looks good, please update HISTORY.rst adding a new section for v0.17.

tests/conftest.py

codecov · 2019-08-10T11:31:31Z

Codecov Report

Merging #142 into master will increase coverage by <.01%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master     #142      +/-   ##
==========================================
+ Coverage   98.74%   98.75%   +<.01%     
==========================================
  Files           8        8              
  Lines         638      642       +4     
  Branches       90       91       +1     
==========================================
+ Hits          630      634       +4     
  Misses          6        6              
  Partials        2        2

samuelcolvin

please update history, also I guess it would be good to have a test that this limit actually works, eg. set queue_read_limit = 5, add 10 jobs to the queue, and check that on the first read only 5 items are retrieved.

samuelcolvin · 2019-08-10T16:03:19Z

arq/worker.py

@@ -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,


maybe this shouldn't be optional?

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

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

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

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

should be:

self.queue_read_limit: int = queue_read_limit or max_jobs

@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.

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.

@samuelcolvin Now I understand, thanks for the explanation.

…e tests for queue_read_limit

codecov · 2019-08-11T07:02:40Z

Codecov Report

Merging #142 into master will increase coverage by <.01%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master     #142      +/-   ##
==========================================
+ Coverage   98.74%   98.75%   +<.01%     
==========================================
  Files           8        8              
  Lines         638      642       +4     
  Branches       90       91       +1     
==========================================
+ Hits          630      634       +4     
  Misses          6        6              
  Partials        2        2

codecov · 2019-08-11T07:02:40Z

Codecov Report

Merging #142 into master will increase coverage by 0.01%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master     #142      +/-   ##
==========================================
+ Coverage   98.05%   98.06%   +0.01%     
==========================================
  Files           8        8              
  Lines         667      671       +4     
  Branches       95       95              
==========================================
+ Hits          654      658       +4     
  Misses         10       10              
  Partials        3        3

rubik · 2019-08-11T07:04:26Z

@samuelcolvin Yes, tests are definitely needed. I added them in the second-to-last commit. I also added the history note.

samuelcolvin · 2019-08-11T13:23:10Z

arq/worker.py

@@ -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,


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

samuelcolvin · 2019-08-11T13:23:45Z

arq/worker.py

@@ -280,6 +278,28 @@ def run(self) -> None:
                queued_jobs = await self.pool.zcard(self.queue_name)
                if queued_jobs == 0:
                    return
+            async with self.sem:  # don't bother with zrangebyscore until we have "space" to run the jobs


I don't understand why we're now doing zrangebyscore in two different places.

Is this really required?

@samuelcolvin This was introduced by mistake, it's absolutely not required. The only thing needed is what is inside _poll_iteration. This has been fixed now.

samuelcolvin · 2019-08-11T13:25:21Z

tests/test_worker.py

+
+    assert await arq_redis.zcard(default_queue_name) == 4
+    worker: Worker = worker(functions=[foobar], max_jobs=2)
+    worker.pool = await create_pool(worker.redis_settings)


why do you need this?

@samuelcolvin My understanding was that the pool is initialized in Worker.main, which we don't call in this test. It seems that it's initialized separately too, so I removed this line.

I think the point is that the worker fixture uses the standard redis connection, thus this isn't required.

samuelcolvin · 2019-08-11T13:25:37Z

tests/test_worker.py

+    assert worker.jobs_retried == 0
+
+    await worker._poll_iteration()
+    await asyncio.sleep(0.01)


what is this sleep doing? I doubt we need it.

@samuelcolvin It introduces a small delay because if Redis and Python are not fast enough, the assert below is not verified. In fact, if I remove it the test fails on my machine.

okay, I would guess you might need a big delay to avoid intermittent failures on CI where resources are more constrained (I've had problems like this a lot before), would increase to 0.1.

@samuelcolvin It makes sense. I increased it to 0.1.

looks like you've only updated this number in some places.

rubik · 2019-08-11T16:16:39Z

@samuelcolvin I have fixed the linting errors so that the checks can pass. The errors on Python 3.8 remain, but they are not related to this PR. I am not able to understand what causes them.

samuelcolvin · 2019-08-11T16:20:16Z

no problem, looks like aioredis is not compatible with python 3.8.

rubik added 2 commits August 9, 2019 22:28

feat: add queue_read_limit to limit the number of job IDs read at eac…

82614b2

…h poll

fix: set offset to None in case the limit is not specified

cbc5eb2

samuelcolvin reviewed Aug 10, 2019

View reviewed changes

tests/conftest.py Outdated Show resolved Hide resolved

chore: format code

dc3cd62

python-arq deleted a comment from codecov bot Aug 10, 2019

samuelcolvin reviewed Aug 10, 2019

View reviewed changes

rubik added 2 commits August 11, 2019 08:38

fix: set queue_read_limit to max_jobs if not specified

c3a4ccf

chore: refactor worker poll iteration into a separate method and writ…

10c9def

…e tests for queue_read_limit

chore: add history note

930af48

rubik added 2 commits August 11, 2019 09:35

chore: format code

ca7e3f7

chore: format docstring

5c487a0

samuelcolvin reviewed Aug 11, 2019

View reviewed changes

rubik added 6 commits August 11, 2019 16:42

fix: remove additional call to Redis introduced by mistake

b50397e

chore: remove unneeded call to create_pool in queue_read_limit tests

8328a60

chore: assign queue_read_limit in one line

443302f

chore: increase delay in queue_read_limit tests for CI systems

13ab4ed

Merge branch 'master' into feat-worker-limit

f0c6a24

chore: fix linting errors

6ede7ef

fix remaining sleeps

c1468e9

samuelcolvin merged commit 3b35919 into python-arq:master Aug 11, 2019

samuelcolvin mentioned this pull request Apr 23, 2020

fix concurrency with multiple workers #180

Merged

sondrelg mentioned this pull request Nov 8, 2021

Update argument docstring definition #278

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Feat worker limit #142

Feat worker limit #142

rubik commented Aug 9, 2019

samuelcolvin left a comment

codecov bot commented Aug 10, 2019

samuelcolvin left a comment

samuelcolvin Aug 10, 2019

samuelcolvin Aug 11, 2019

rubik Aug 11, 2019

samuelcolvin Aug 11, 2019

rubik Aug 11, 2019

samuelcolvin Aug 11, 2019

rubik Aug 11, 2019

codecov bot commented Aug 11, 2019

codecov bot commented Aug 11, 2019 •

edited

Loading

rubik commented Aug 11, 2019

samuelcolvin Aug 11, 2019

samuelcolvin Aug 11, 2019

rubik Aug 11, 2019

samuelcolvin Aug 11, 2019

rubik Aug 11, 2019

samuelcolvin Aug 11, 2019

samuelcolvin Aug 11, 2019

rubik Aug 11, 2019

samuelcolvin Aug 11, 2019

rubik Aug 11, 2019

samuelcolvin Aug 11, 2019

rubik commented Aug 11, 2019

samuelcolvin commented Aug 11, 2019

Feat worker limit #142

Feat worker limit #142

Conversation

rubik commented Aug 9, 2019

samuelcolvin left a comment

Choose a reason for hiding this comment

codecov bot commented Aug 10, 2019

Codecov Report

samuelcolvin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Aug 11, 2019

Codecov Report

codecov bot commented Aug 11, 2019 • edited Loading

Codecov Report

rubik commented Aug 11, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rubik commented Aug 11, 2019

samuelcolvin commented Aug 11, 2019

codecov bot commented Aug 11, 2019 •

edited

Loading