Pooling and queue libraries in Elixir

for the simple queue there is backy GitHub - kuon/backy: Simple background jobs backed by PostgreSQL for Elixir. · GitHub which persist to a postgres table.. not sure it does distributed, so maybe no bonus points.. the readme has links to other queue libs, but they all seem to use redis

fyi I’ve only used it for small stuff and not in production.

I don’t really need Postgres either, but I’ll look at that too, thank you.

In the meantime I found GitHub - sheharyarn/que: Simple Job Processing in Elixir with Mnesia ⚡ · GitHub any thoughts?

Pooling

Elasticsearch communication occurs over HTTP. Hackney is, for better or worse, the most supported HTTP client for Elixir right now. It comes with pooling built-in. By using a specific pool for your Elasticsearch HTTP requests, your connections will automatically be reused until they are closed.

Asynchronous Worker Queue

For an asynchronous worker queue, I would recommend using a library like Verk. Verk is based on Sidekiq and stores its queues in Redis. This makes it tolerant of restarts and failures in your own application.

As I wrote, if I would have to use Redis for the medium of the queue, I can just skip Elixir altogether. I don’t need a Sidekiq/Resque clone, because then I can simply use Sidekiq. I want or rather need this app to be self contained, without any third party software as dependency. And if the data it will be keeping can be synchronized across nodes out of the box then it’s even better.

The queries for Elasticsearch are rather long, it takes 10s for more for Elasticsearch cluster to respond. And each job will need to make many queries to elastic so I need some pooling library so I can keep sane number of jobs. So yes, I I’ll use hackney I will use it’s connections pool to reuse connections, but I also need a poll of workers for jobs that do a little bit more than just querying Elasticsearch.

The problem is that Elasticsearch alone is not fast enough. I need to pre-cache complicated queries, do a map-reduce on it and then some

For a simple queue take a look at queue — OTP 29.0.2 (stdlib 8.0.1) you can persist it into anything with erlang — OTP 29.0.2 (erts 17.0.2)

I’m not aware of an embedded solution for a distributed queue. I’d likely try RabbitMQ in that scenario.

However, if you want a local persistent queue with absolutely no external dependencies, then the solution would not be overly complex. I don’t know if there are libraries which do this out of the box, but rolling your own simple solution would IMO not be too complicated, and is at least a nice exercise of a bit of OTP

The idea is to have an in-memory queue, and persist the queue state to disk every time an item is added, and every time the job succeeds.

Persisting without an external component can be done in a couple of ways:

• File.write + :erlang.term_to_binary
• dets
• mnesia

In a simplest solution, you’d have one process, a GenServer, which would act as a scheduler. The scheduler reads from the queue file, and starts the first job in a separate process. If the job succeeds, the scheduler removes it, persists the new queue state, and starts the next job, if there is one. The scheduler should also monitor the job process (or link to it), to detect if it crashed. Then you need to handle the crash by retrying, deferring the job, giving up.

Every time you want to add another item, you send a request to the scheduler, which adds it to the end of the in memory structure (for example :queue), stores to disk, and starts the job if nothing is running.

I’d suggest to try rolling a simple implementation, perhaps starting with a plain in-memory queue, and then adding naive persistence with File.write(:erlang.term_to_binary(queue)).

If you have more elaborate needs, such as pooled or rate-limited queues, you could take a look at jobs. I’m actually using it myself for these particular cases, and I’m quite happy with it.

The author of Sidekiq has a new project for async work queues called Faktory which is language agnostic and addresses some of the main design short comings of Sidekiq.

There is a fully featured Elixir client for it that is actively being developed:

Hi Sasa,

I would love to use this lib jobs from within elixir but I have no clue how to go about it after I have installed it. Is there an elixir wrapper for it kindly?

I wrote Stagger as a simple persistent queue that acts as a GenStage producer. It’s usage is targeted more at capturing data when backend processing might not be available but can be resumed later.

I don’t know of any. But you don’t need a wrapper, you can use jobs directly. The examples in Erlang are here, and the comparison of syntax differences can be found here. So e.g. if we take this example:

jobs:add_queue(q, [{standard_rate,1}]).
...
jobs:run(q, fun() -> io:fwrite("job: ~p~n", [time()]) end).

It could be roughly translated as e.g.:

:jobs.add_queue(:q, [standard_rate: 1])
...
:jobs.run(:q, fn -> IO.inspect(DateTime.utc_now()) end)

More generally, to use jobs, you need to first create a named queue with :jobs.add_queue(queue_name, opts), where queue_name is an atom. Typically this can be done in the application start callback.

Then you can run a queued operation from any process as :jobs.run(queue_name, fn -> ... end).

Take a look ad add_queue docs, in especially the “regulators” section. To create a rate-limited queue, you can pass rate: [...]. To create a counter queue (limit max simultaneous jobs), you can pass counter: [...].