Comments (6)
Since the queue operates directly out of the database, honeydew wouldn't load jobs into a local queue, it'd keep all the state in the database itself, and rely on transactional queries to change it. So the "queue" processes themselves become essentially dumb connectors to the database. When honeydew wants to give a job to a worker, it'll ask the database to "reserve" (mark as in-progress) and return a single row.
Honeydew has two different kinds of queues:
-
A fixed number of different kinds of jobs per row. You don't manually enqueue jobs for this kind of queue, the insertion of a row implies that the specified jobs need to be run. (For example, when a User is added, sending a welcome email and charging their credit card). This is the EctoQueue, it writes directly to the domain model.
-
Generic queues that require jobs be enqueued by the user (the Mnesia and ErlangQueue queues are of this type).
My guess is that you're looking for the latter, but with state stored in mongo.
If that's the case, you just need to implement the PollQueue
callbacks, https://github.com/koudelka/honeydew/blob/master/lib/honeydew/poll_queue.ex#L16-L22 as well as adding a new "source" to connect to mongo:
It'll probably end up looking something like this:
defmodule MongoSource do
require Logger
alias Honeydew.Job
alias Honeydew.Queue
@behaviour Queue
@impl true
def validate_args!([host: host, port: port, db: db]) when is_binary(host) and is_integer(port) and is_binary(db) do: :ok
def validate_args!(_), do: raise "bad args"
@impl true
def init(name, [host: host, port: port, db: db]) do
connection = Mongo.connect(host, port, db)
collection = Mongo.collection(connection, name)
{:ok, collection}
end
#
# Enqueue/Reservee
#
@impl true
def enqueue(job, collection) do
Mongo.put(collection, to_json(job))
collection
end
#
# Reserve a job by transactionally selecting the oldest job and locking the document so others don't try to run it.
# You'll probably want to use the current time for the lock, so you have a some way to know if a job was reserved,
# but the reserving node completely died without being able to release the lock.
#
# If you can, try to let mongo handle the concept of "now", that'll reduce the chance of problems from clock desync between your nodes.
#
# See the following for more details: https://github.com/koudelka/honeydew/blob/master/lib/honeydew/sources/ecto_source.ex#L5
#
@impl true
def reserve({pending, in_progress} = state) do
case Mongo.get_and_update(collection, where: %{state: nil}, sort_by: :enqueued_at, set: %{reserved_at: Mongo.now(), state: "reserved"}) do
:not_found ->
{:empty, collection}
job ->
{job, collection}
end
end
#
# Ack/Nack
#
# Job completed successfully
@impl true
def ack(%Job{private: id}, collection) do
Mongo.update(collection, id, state: "completed")
# or, if you don't want to keep finished jobs in the database, you can just delete the document instead
collection
end
# Job needs to be re-run
@impl true
def nack(%Job{private: id}, collection) do
Mongo.update(collection, id, state: nil)
collection
end
#
# Helpers
#
@impl true
def status(collection) do
%{count: Mongo.count(collection),
in_progress: Mongo.count(collection, where: %{state: "reserved"})}}
end
@impl true
def filter(collection, :stale) do
Mongo.get(collection, where: %{reserved_at: %{"$gt" => Mongo.now() + "1 hour"}})
end
@impl true
def cancel(%Job{private: id}, collection) do
reply =
case Mongo.get_and_update(collection, where: %{id: id}, set: %{state: "cancelled"}) do
:not_found ->
{:error, :not_found}
%Job{} ->
:ok
end
# or, you can just remove the job from the db instead of updating its status
{reply, collection}
end
end
Since the possibility of "stale" jobs exists, you'll probably want to borrow the EctoSource's :__reset_stale__
functionality, to find jobs where the entire that the worker was on crashed and left the job in a "reserved" state.
Hope that helps!
from honeydew.
You might also be able to use mongo's "change streams" to listen for jobs, rather than polling the collection, that'll probably lessen the load on the db.
from honeydew.
Hey there,
The ecto queues aren't just for persistence, the actual queue mechanics are run by the database itself. The purpose of that is to keep the database as the sole authority of queue state, and hence to be able to use the distribution/replication properties of the database rather than trying to implement it in Elixir.
For example, here's where the postgres version atomically reserves a job: https://github.com/koudelka/honeydew/blob/master/lib/honeydew/sources/ecto/sql/postgres.ex#L49-L59
Which database did you want to use?
from honeydew.
I want to use Mongo
The purpose of that is to keep the database as the sole authority of queue state, and hence to be able to use the distribution/replication properties of the database rather than trying to implement it in Elixir.
Sounds exactly what I want.
Could you please expand on cold starts question too?
from honeydew.
Oh this definitely does help! Also startAfter
from 4.2 looks so promising. I'll close this one now. Will open a new one if needed.
from honeydew.
@deadtrickster Did you ever get get Honeydew working with Mongo? We're looking to do the same thing and would love to use your library if it's public!
from honeydew.
Related Issues (20)
- Clear an entire queue HOT 1
- Disabling clock freeze warning in test environment HOT 1
- Control retry time within job HOT 2
- Support Ecto.Adapters.MyXQL for MySQL HOT 6
- Mnesia as default queue not mentioned in start_queue/2 documentation HOT 1
- Custom queue makes Honeydew crashs by timeout HOT 4
- warning: function equal?/2 required by behaviour Ecto.Type
- How to rerun jobs if the system crashes? HOT 1
- pg2 scheduled for removal HOT 10
- How to cancel Jobs in Ecto Queue? HOT 1
- Doc links to code broken HOT 2
- Creates a behaviour for `Honeydew.Logger` module
- Unexpected behavior: bad_return application start failure using Mnesia outside of Honeydew HOT 5
- no match of right hand side value: [] HOT 2
- API for accessing in_progress jobs
- Redesigning API for concurrency
- Timeout error on application start HOT 1
- ErlangQueue doesn't respect delay_secs option HOT 1
- Shutdown hangs HOT 11
- Error when listing mnesia in mix extra_applications.
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from honeydew.