Custom serializers #143

rubik · 2019-08-10T11:05:22Z

Changes

I would like to propose this change, which allows the user to specify custom serializers in alternative to pickle, which is left as the default.

The public API is changed only very slightly: there are new _serialize and _deserialize arguments to create_pool and Worker. The rest of the changes essentially reduces to passing these to the pickle functions.

For backward compatibility, I have left the pickle_jobs, etc. functions in arq.jobs with the same name, even though the name is slightly inappropriate now.

Rationale

The main reason I am proposing this change is to allow better memory usage. The default serialization method, pickle, is quite space inefficient. Since the job data is essentially JSON, we can attain much better memory usage by switching to serialization methods that are more appropriate.

In my case, by using MsgPack instead of Pickle, I am seeing an improvement of 47%-49% in memory usage which is quite significant. At tens to hundreds of thousands of tasks this means hundreds of MBs if not GBs in savings.

Usage

The usage is quite simple. Existing code works as usual, but one can use a different serialization method as follows:

import msgpack
from arq.connections import create_pool, RedisSettings

pool = await create_pool(RedisSettings(), _serialize=msgpack.packb, _deserialize=msgpack.unpackb)

WorkerSettings:
    serialize = msgpack.packb
    deserialize = msgpack.unpackb

Tests and documentation

The tests pass, except two tests that also fail on master. If this change is approved, I can write a paragraph in the documentation on custom serialization functions.

…ions

codecov · 2019-08-10T11:30:20Z

Codecov Report

Merging #143 into master will decrease coverage by 1.59%.
The diff coverage is 81.03%.

@@            Coverage Diff            @@
##           master     #143     +/-   ##
=========================================
- Coverage   98.74%   97.15%   -1.6%     
=========================================
  Files           8        8             
  Lines         638      667     +29     
  Branches       90       95      +5     
=========================================
+ Hits          630      648     +18     
- Misses          6       12      +6     
- Partials        2        7      +5

codecov · 2019-08-10T11:30:20Z

Codecov Report

Merging #143 into master will decrease coverage by 0.69%.
The diff coverage is 92.53%.

@@            Coverage Diff            @@
##           master     #143     +/-   ##
=========================================
- Coverage   98.74%   98.05%   -0.7%     
=========================================
  Files           8        8             
  Lines         638      667     +29     
  Branches       90       95      +5     
=========================================
+ Hits          630      654     +24     
- Misses          6       10      +4     
- Partials        2        3      +1

samuelcolvin

Sounds good to me, will need history and docs and a few tweaks suggested below.

samuelcolvin · 2019-08-10T11:34:03Z

arq/connections.py

+    def __init__(
+        self,
+        pool_or_conn,
+        _serialize: Optional[Callable[[Any], bytes]] = None,


Suggested change

_serialize: Optional[Callable[[Any], bytes]] = None,

_job_serializer: Optional[Callable[[Any], bytes]] = None,

Might be even clearer?

and same obviously for _deserialize -> _job_deserializer.

samuelcolvin · 2019-08-10T11:37:49Z

arq/jobs.py

+    pass
+
+
+def pickle_job(


let's rename unpickle_job to deserialize_job, better to get it right now.

Change should be pretty explicit and easy to fix.

samuelcolvin · 2019-08-10T11:37:56Z

arq/jobs.py

    pass


-def pickle_job(function_name: str, args: tuple, kwargs: dict, job_try: int, enqueue_time_ms: int):
+class PickleError(SerializationError):


samuelcolvin

and please update HISTORY.rst.

Otherwise LGTM.

samuelcolvin · 2019-08-10T16:08:24Z

arq/connections.py

+        self,
+        pool_or_conn,
+        _job_serializer: Optional[Callable[[Any], bytes]] = None,
+        _job_deserializer: Optional[Callable[[bytes], Any]] = None,


Suggested change

_job_deserializer: Optional[Callable[[bytes], Any]] = None,

_job_deserializer: Optional[Callable[[bytes], Dict[str, Any]] = None,

Or am I missing something?

samuelcolvin · 2019-08-10T16:09:13Z

arq/connections.py

+    def __init__(
+        self,
+        pool_or_conn,
+        _job_serializer: Optional[Callable[[Any], bytes]] = None,


Suggested change

_job_serializer: Optional[Callable[[Any], bytes]] = None,

_job_serializer: Optional[Callable[[Dict[str, Any]], bytes]] = None,

I think?

and obviously elsewhere where it's used, might be worth having these types defined once, eg. in jobs.py?

samuelcolvin · 2019-08-10T16:12:26Z

arq/jobs.py

+    job_try: int,
+    enqueue_time_ms: int,
+    *,
+    _serialize: Optional[Callable[[Any], bytes]] = None,


Suggested change

_serialize: Optional[Callable[[Any], bytes]] = None,

serialize: Optional[Callable[[Any], bytes]] = None,

since this is explicitly required as a keyword arg, I don't think you need to make it "private".

tests/test_worker.py

…q.jobs

…lizer

rubik · 2019-08-10T19:37:55Z

@samuelcolvin I addressed your suggestions. You said that serializer should be public and therefore starting without underscore for the functions in arq.jobs. Following that reasoning, the same correction needs to be applied to arq.connections.create_pool. That also makes the arguments equal to the parameters in arq.worker.Worker. That was done in a separate commit, 0cdaf4b, so if you don't agree I can revert it.

I don't think the HISTORY.rst change belongs to this PR and you should do it in a separate commit, also because I don't know when you will release the next version (even though I hope very soon).

samuelcolvin · 2019-08-10T19:58:37Z

The way I work every commit to code has an entry in HISTORY.rst, it's the case for all projects so I'm afraid it's a requirement.

Please add the following to history

v0.17 (unreleased)
..................
* custom serializers, eg. to use msgpack rather than pickle, #143 by @rubik

samuelcolvin

otherwise looking good.

samuelcolvin · 2019-08-10T19:59:53Z

arq/connections.py

@@ -141,29 +164,33 @@ class ArqRedis(Redis):
            password=settings.password,
            timeout=settings.conn_timeout,
            encoding='utf8',
-            commands_factory=ArqRedis,
+            commands_factory=functools.partial(
+                ArqRedis, _job_serializer=job_serializer, _job_deserializer=job_deserializer


Suggested change

ArqRedis, _job_serializer=job_serializer, _job_deserializer=job_deserializer

ArqRedis, job_serializer=job_serializer, job_deserializer=job_deserializer

I think since the change?

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

tests/test_worker.py

+    worker: Worker = worker(functions=[foobar])
+    with pytest.raises(SerializationError) as exc_info:
+        await worker.main()
+        assert exc_info.value.startswith('unable to deserialize job: ')


Suggested change

assert exc_info.value.startswith('unable to deserialize job: ')

assert exc_info.value.startswith('unable to deserialize job: ')

this is indented wrongly I think.

samuelcolvin · 2019-08-10T20:03:29Z

tests/test_worker.py

+    )
+    with pytest.raises(SerializationError) as exc_info:
+        await worker.main()
+        assert exc_info.value.startswith('unable to deserialize job: ')


Suggested change

assert exc_info.value.startswith('unable to deserialize job: ')

assert exc_info.value.startswith('unable to deserialize job: ')

rubik · 2019-08-10T20:04:24Z

@samuelcolvin Done. Then after you merge this PR I'll add it for the other one.

rubik · 2019-08-11T06:29:10Z

@samuelcolvin I fixed the code as per your latest feedback.

rubik · 2019-08-11T08:16:33Z

It seems some tests fail on Python 3.8, but I don't think the errors are related to this PR.

arq/connections.py

samuelcolvin · 2019-08-11T15:43:56Z

Great, thanks. I'll deploy as soon as the other PR is passing and I can merge it.

rubik · 2019-08-11T15:51:49Z

@samuelcolvin Thanks for taking the time to review it multiple times.

rubik added 3 commits August 10, 2019 00:25

feat: add ability to specify custom serializer and deserializer funct…

ae3483d

…ions

feat: add _serialize, _deserialize arguments to create_pool

e2fb479

fix: add back PickleError for backward compatibility

0807eb7

rubik changed the title ~~Feat custom serializer~~ Custom serializers Aug 10, 2019

chore: format code

9e69f8b

samuelcolvin reviewed Aug 10, 2019

View reviewed changes

rubik added 2 commits August 10, 2019 17:10

fix: apply code review suggestions

55f8d9a

chore: add tests for custom serializers

81a2dea

samuelcolvin reviewed Aug 10, 2019

View reviewed changes

rubik added 4 commits August 10, 2019 19:53

chore: add docs paragraph about serializers and an example

9cdc0dd

chore: make serializer type hints stricter and define constants in ar…

f5b8417

…q.jobs

chore: make serialization functions public and uniform the names to *…

55f9417

…lizer

fix: pass _job_*lizer arguments to recursive call of create_pool

245fd35

rubik force-pushed the feat-custom-serializer branch from 694116f to 084bdac Compare August 10, 2019 19:28

rubik added 2 commits August 10, 2019 21:30

chore: make job_*lizer arguments public in create_pool

0cdaf4b

chore: remove unnecessary defer in serializer example

5254e52

rubik force-pushed the feat-custom-serializer branch from 084bdac to 5254e52 Compare August 10, 2019 19:30

rubik added 2 commits August 10, 2019 21:34

chore: add precise error value when testing incompatible serializers

699d878

chore: format code

5d4f4df

chore: add history entry

a675995

samuelcolvin reviewed Aug 10, 2019

View reviewed changes

rubik added 2 commits August 11, 2019 08:26

chore: make the serializer arguments in ArqRedis public

723ea97

chore: adjust indentation in serialization tests

2c0ca80

fix: serializationerror tests

fae985a

samuelcolvin reviewed Aug 11, 2019

View reviewed changes

arq/connections.py Outdated Show resolved Hide resolved

arq/connections.py Outdated Show resolved Hide resolved

chore: fix ArqRedis' docstring

31047ef

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Custom serializers #143

Custom serializers #143

rubik commented Aug 10, 2019

codecov bot commented Aug 10, 2019

codecov bot commented Aug 10, 2019 •

edited

Loading

samuelcolvin left a comment

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

samuelcolvin left a comment

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

rubik commented Aug 10, 2019

samuelcolvin commented Aug 10, 2019

samuelcolvin left a comment

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

samuelcolvin Aug 10, 2019

rubik commented Aug 10, 2019

rubik commented Aug 11, 2019

rubik commented Aug 11, 2019

samuelcolvin commented Aug 11, 2019

rubik commented Aug 11, 2019

	_serialize: Optional[Callable[[Any], bytes]] = None,
	_job_serializer: Optional[Callable[[Any], bytes]] = None,

	_job_deserializer: Optional[Callable[[bytes], Any]] = None,
	_job_deserializer: Optional[Callable[[bytes], Dict[str, Any]] = None,

	_job_serializer: Optional[Callable[[Any], bytes]] = None,
	_job_serializer: Optional[Callable[[Dict[str, Any]], bytes]] = None,

	_serialize: Optional[Callable[[Any], bytes]] = None,
	serialize: Optional[Callable[[Any], bytes]] = None,

	ArqRedis, _job_serializer=job_serializer, _job_deserializer=job_deserializer
	ArqRedis, job_serializer=job_serializer, job_deserializer=job_deserializer

	assert exc_info.value.startswith('unable to deserialize job: ')
	assert exc_info.value.startswith('unable to deserialize job: ')

Custom serializers #143

Custom serializers #143

Conversation

rubik commented Aug 10, 2019

Changes

Rationale

Usage

Tests and documentation

codecov bot commented Aug 10, 2019

Codecov Report

codecov bot commented Aug 10, 2019 • edited Loading

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

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

rubik commented Aug 10, 2019

samuelcolvin commented Aug 10, 2019

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

rubik commented Aug 10, 2019

rubik commented Aug 11, 2019

rubik commented Aug 11, 2019

samuelcolvin commented Aug 11, 2019

rubik commented Aug 11, 2019

codecov bot commented Aug 10, 2019 •

edited

Loading