So is the poller sending the jobs to the dispatcher? How does it know if the dispatcher has capacity to accept any new jobs?
samphilipd
sasajuric
Author of Elixir In Action
Yes.
There are various ways to handle this
The approach I used recently for a low job rate was to always start a new task (so no limit), and then from the task process use jobs to ensure that most N of them are actually performing the work. That solution is in my opinion quite simple, though it’s probably not optimal for a large incoming rate with a possible periods of large backlog, because you might end up with a very large number of processes, which might end up consuming too much memory.
Another approach is to have the dispatcher ask the poller for more work when it’s ready. I’d have the poller preload at least one batch upfront, so it can immediately send it when the dispatcher asks for more work. This would keep the memory usage constant (you’d have at most one extra batch loaded), while giving you the same properties as I’ve mentioned in my previous post (separation of latencies and failures). You should be able to do this with GenStage (I think - I never used it so far), or you could hand-code it with GenServer.
The first approach is IMO fairly lightweight, though it doesn’t necessarily work well with larger bursts. The GenStage approach is more resilient with respect to overloads, and should keep your memory usage stable, but it might be a bit more complex.
1
mbuhot
Not quite. Ecto_job uses select for update skip locked only for a worker to claim a batch of jobs.
The work of executing the job done using optimistic concurrency - the transaction is only open for the final call to Repo.transaction Where updates are submitted. All jobs in the batch execute concurrently.
3
samphilipd
I’ve just released Rihanna as an initial implementation of the concepts discussed here. It uses a very simple single dispatcher with a poll-based approach, which nonetheless has a pretty high throughput of around 1.5k jobs/s due to it’s use of advisory locks.
The supervision subtree looks like this:
Supervisor
(one_for_one supervisor)
|
|
JobDispatcher (polls and fires off jobs)
(GenServer)
/\
/ \
Task1 ... TaskN
It also has a GUI that can be used to see work in progress and retry failed jobs:
This first version is well-tested and stable at high concurrency levels and should be good enough for most use-cases of a simple, durable, database-backed distributed queueing system.
I believe that significantly higher performance with less load on the DB is possible by taking advantage of pg_notify and communication between nodes of an Erlang cluster to avoid needless contention on the DB, but this is not ready for release yet.
Thoughts/feedback/questions/bug reports are welcome!
8
mbuhot
Congrats! Isn’t Elixir a wonderful language for implementing concurrent job queues.
I didn’t see a LICENSE file in the repo, is it MIT?
3
samphilipd
jordiee
If something goes wrong in pulling and the genserver crashes you also lose all currently being executed jobs correct?
0
samphilipd
Jobs are spawned underneath a Task.Supervisor so they will run to completion regardless of the state of the Dispatcher that spawned them. Since the Dispatcher handles marking jobs as failed/completed, these jobs will never register their status with the database so the lock will be lost and the respawned dispatcher will try them over again.
The result will be that some jobs are run twice.
3
jordiee
Ah I see. admittedly I did not look at the code because I was on my phone and your diagram above looks like the job tasks are linked to the JobDispatcher. Pretty cool though I will take a look later!
1
Qqwy
TypeCheck Core Team
Wonderful! 
A great way to handle this edge case. Writing jobs to ensure that they will result in a no-op when ran again if required is usually not hard, and a great solution.
1