[Feature] Job dependencies

[guewen]

Some details may be inaccurate as I didn't verify all my assumptions.

Implementation Draft

First let's state what we would expect from dependencies

Design assumptions

1. We can delay a job A with a dependency on another job Z: A will start only when Z is in state 'done'
2. We can delay jobs A and B with a dependency on another job Z: both A and B will start only when Z is in state 'done'
3. We can delay a job A with a dependency on other jobs Z and Y: A will start only when Z and Y are in state 'done'
4. We can delay jobs A and B with a dependency on other jobs Z and Y: A will start only when Z and Y are in state 'done'
5. When job Z finishes and A depends on it, A will get the result of Z
6. When job Z and Y finishes and A depends on them, A will get a list with the results of Z and Y
7. When job Z and Y finishes and A and B depends on them, A and B will get a list with the results of Z and Y

Assumptions 1. and 2. are probably the easiest to implement, and we could start with them, but we have to think the datamodel and API to be sure we can do 3. and 4. as well.

Datamodel and workflow

• Add a new field on queue_job: dependencies. This is a JSON field with a dict expressing the dependencies, with 2 keys: depends_on, reverse_depends_on. We don't want FKs, jobs are not supposed to be deleted and if they are, we just have to ignore the irrelevant references.
• Add a new field for the result that is passed to the dependencies, JSON serialized (we can pass recordsets, ...). As we currently store the result as a message for the end-user, not sure how we should handle it
• We add a new state wait_dependency, these jobs will be ignored by the jobrunner out of the box
• When a job finishes, the job runner looks if the job has a non-empty reverse_depends_on list. If it finds dependent jobs, for each job in reverse_depends_on, it has to read the depends_on list, check if all the jobs in this list are done and in such case, change the state of the dependent job to 'pending'.

Using the denormalization with a reverse_depends_on field means that we only never have to run a SELECT ... WHERE depends_on = ? to find the dependencies. We have a cost penalty (besides the cost to read reverse_depends_on) only when we actually have dependencies.

At first, it does looks not too difficult, the difficulty here lies in the transactions. If a job A depends on Z, when Z is done, it will check the state of the dependencies, Z being the only dependency and being done, we are good. Now if A depends on Z and Y, if Z finishes, it has to check if Y is done as well. Y may or may not be dead done, but Z being isolated in a transaction will not know the real state of Y. Changing the isolation level would not help as Y can be 'done' then rollbacked. I thought about using postgres 2-phases commit - not sure we can integration it properly in odoo - or opening a new transaction after Z is committed (and take the risk that the second transaction fails, then handle retries, ...).

API

On the API side, I thought about different options too, I remember I had a preference and some ideas to actually implement them but can't remember ;)
There is much more thoughts to put here, also considering that dependent jobs may receive a result from the jobs they depend on.

Dependencies API

There is no support of then (executed after done or failure), only done (executed after done).
The "failure" callback is handled synchronously in Celery because you may want to use the traceback from it. This is not drafted here neither. Both of these use cases can be added later, we only have to keep in mind the design to support them (keep this info in the dependencies json field.

Comments in the code explain the additions.

from odoo import _, api, fields, models
from odoo.addons.queue_job.job import job, chain, group


class ProductProduct(models.Model):
    _inherit = 'product.product'

    @api.job
    def generate_thumbnail(self, size):
        return self._gen(size)

    @api.multi
    def button_generate_usual(self):
        # This is the current implementation, which will not change
        self.ensure_one()
        self.with_delay(
            priority=30,
            description=_("generate xxx")
        ).generate_thumbnail((50, 50))

    @api.multi
    def button_generate_simple_with_delayable(self):
        self.ensure_one()
        # Introduction of a delayable object, using a builder pattern
        # allowing to chain jobs or set properties. The delay() method
        # on the delayable object actually stores the delayable objects
        # in the queue_job table
        (
            self.delayable()
            .generate_thumbnail((50, 50))
            .set(priority=30)
            .set(description=_("generate xxx"))
            .delay()
        )

    @api.multi
    def button_generate_chain(self):
        product1 = self.browse(1)
        product2 = self.browse(2)
        product3 = self.browse(3)
        self.ensure_one()
        # the delayable object can be chained
        job1 = (
            product1.delayable().generate_thumbnail((50, 50))
            .set(description=_("generate xxx"))
        )
        job2 = (
            product2.delayable().generate_thumbnail((50, 50))
            .set(description=_("generate xxx")),
        )
        job3 = (
            product3.delayable().generate_thumbnail((50, 50))
            .set(description=_("generate xxx")),
        )
        # the builder pattern allow to set properties at once
        for job_ in (job1, job2, job3):
            # job.set(xxx) is  better than job.priority = 10
            # or than job.priority(10), because it allows to pass
            # a splat dictionary set(**properties) or custom properties
            job.set(priority=30)
        # Or self.delay_chain() as syntactic sugar, calling chain
        # under the hood.
        # .delay() setup the links in the chain and stores the job
        chain(
            job1,
            # job2 is executed after product1
            job2,
            # job3 is executed after product2
            job3
        ).delay()

    @api.multi
    def button_generate_chain_done(self):
        product1 = self.browse(1)
        product2 = self.browse(2)
        product3 = self.browse(3)
        self.ensure_one()
        job1 = product1.delayable().generate_thumbnail((50, 50))
        job2 = product2.delayable().generate_thumbnail((50, 50))
        job3 = product3.delayable().generate_thumbnail((50, 50))
        # similar than the chain
        job1.on_done(job2.on_done(job3)).delay()

    @api.multi
    def button_generate_done_multi(self):
        product1 = self.browse(1)
        product2 = self.browse(2)
        product3 = self.browse(3)
        self.ensure_one()
        job1 = product1.delayable().generate_thumbnail((50, 50))
        job2 = product2.delayable().generate_thumbnail((50, 50))
        job3 = product3.delayable().generate_thumbnail((50, 50))
        # job2 and job3 executed only when job1 is done
        job1.on_done(job2, job3).delay()


class Album(models.Model):
    _name = 'album'

    @api.job
    def generate_page(self, results, color, images_per_page=10):
        return self._gen(results, images_per_page)

    @api.multi
    def button_generate(self):
        self.ensure_one()
        product1 = self.env['product.product'].browse(1)
        product2 = self.env['product.product'].browse(2)
        product3 = self.env['product.product'].browse(3)
        job1 = (
            product1.delayable().generate_thumbnail((50, 50))
            .set(priority=30)
            .set(description=_("generate xxx"))
        )
        job2 = (
            product2.delayable().generate_thumbnail((50, 50))
            .set(priority=30)
            .set(description=_("generate xxx"))
        )
        job3 = (
            product3.delayable().generate_thumbnail((50, 50))
            .set(priority=30)
            .set(description=_("generate xxx"))
        )
        # could have self.delay_group() in base.model as syntactic sugar
        group(job1, job2, job3).on_done(
            # executed only when job1, job2 and job3 are done
            self.delayable().generate_page('#000000', images_per_page=12)
            .set(priority=12)
            .set(description=_("generate page x"))
            # delay makes the links and store in db
        ).delay()

Results API

Still do define

# jobs definitions
from odoo.addons.queue_job import Result

@job
def render_thumbnail(self):
    # could use a class to differentiate the current user message and 
    # the result for the dependent job(s)
    thumbnail = self.env['thumbnail'].do_stuff()
    return Result(message='thumbnail generation is a success',
                          results=thumbnail)

[simahawk]

+1 for having a Result object.

Regarding the api: why not using a notation similar to JS promise? Like:

# same model, execute in any case
model1_job.then(model1_job)
# diff model, execute in any case
model1_job.then(model2_job)
# execute only success or failure
model1_job.done(model2_job1).fail(model2_job2)
# similar to Group -> `$.when`
self.delay_when(model1_job, model2_job, ...).then(model1_job2)

This might imply that we can init a job w/out scheduling it straight. Something like this maybe?

model.with_delay(init_only=1).do_this

[guewen]

Regarding the api: why not using a notation similar to JS promise?

In your example, could you clarify what is passed to then, done, ...?

Something like this?

model1_job.then(
    model.with_delay(init_only=1).do_this
)

If so, it seems kind of artificial to have with_delay, at the same time we may need to pass job properties like the priority etc. The other thing is, how would you pass arguments to do_this?

[guewen]

@simahawk I started with something similar (besides the fail case that I didn't considered but makes sense, to be thorough) but couldn't figure out out to pass elegantly both the job properties and the arguments for the job method. Then I came up with the design above which is almost similar when you delay a job (with_delay) and delay a job with a dependency (when_done, note that we can rename it to then or anything). I find it rather straightforward since it's the same way to pass arguments around.

Still I'm curious to see you elaborate your design because I certainly did not consider all possibilities :)

[simahawk]

at the same time we may need to pass job properties like the priority etc. The other thing is, how would you pass arguments to do_this?

Can't we do it as this?

model.with_delay(init_only=True, priority=1).do_this(random_arg)

Even tho all in all we can even avoid init_only arg as long as we are able to have back a deferred/delayed object. I mean, is the same thing that you'd get here

group = Group(
    record1.with_delay().generate_thumbnails(),
    record2.with_delay().generate_thumbnails()
)

what do you have as args for the group? A pre-configured DelayedRecordset?

In the end having the possibility to init a job as a record just helps organizing our code and maybe pass jobs around. If with_delay gives us back already an object to play with properly... probably we don't need another arg.

On one hand having a job scheduled only if another one passed or failed, well, would keep the queue clean. OTOH you may want to schedule them and be able to check if you've already a job scheduled for a specific purpose.

[guewen]

I updated the description with a new API draft discussed with @simahawk.
Still need to think a bit more about the Result. As a result can contain the result of one or many jobs, we should implement the same accessor than recordsets, when you have one result, you can access it directly (so it is consistent with odoo API's experience).

[lmignon]

@guewen and @simahawk The API looks very promising... I'm not sure how to understand what's the UC behind the Result. Since the method execution is async, the caller is useless for the caller...

[guewen]

The API looks very promising... I'm not sure how to understand what's the UC behind the Result. Since the method execution is async, the caller is useless for the caller...

Let's take this use case of a generation of several thumbnails (job A,B,C), then the generation of a page joining these thumbnails (job Z). The jobs A, B and C returns "transient" images which will only serve the purpose of the page. Of course, you can create a table or a column in an existing column to hold the thumbnails, but it may be more convenient to store the Result and propagate it directly to the job Z. All in all, there is nothing we shouldn't be able to do storing data by ourselves in DB, it is only more convenient if you don't have to deal with all the nitty-gritty details of the schema, storing and purging the transient data.
That being said, storing the result means more data in the queue_job table: we have to ensure of the cost a result may have, but as long as we don't read the field and have no index on it, it shouldn't really increase?

[lmignon]

@guewen Thank you for the explanation. I'm still wondering how the Result(s) of job A,B,C will be available in job z.

[guewen]

Also, regarding the use case, I have to say this one is out of my imagination and I have no true life use case now, I'd say the Result may be implemented later, but we have to decide how to propagate them to the "children" jobs at the beginning to be able to add them later.

[guewen]

I'm still wondering how the Result(s) of job A,B,C will be available in job z.

That's the trickiest part to decide. Celery defines that tasks being used as dependencies must have a results positional argument as first argument. That could be a way, as well as having an optional results keyword argument.

[guewen]

Another option. We have the job_uuid in the context, we already spoke about adding a generic method to get the job object from a job method, such as self.running_job(). If we don't want to modify the signature of the children jobs, we could access to the results using something like self.running_job().predecessor_results(). It would be less explicit than an argument though.

[guewen]

Here's a WIP branch: https://github.com/guewen/queue/tree/12.0-dependencies