Job queue #6
Open
Job queue #6
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
jagerman
reviewed
Jan 27, 2026
include/oxen/quic/loop.hpp
Outdated
| // | ||
| // Do not use this unless you know you need it. | ||
| // | ||
| // The interface is the same as `Loop::call` et al, but when this JobQueue is |
Member
There was a problem hiding this comment.
Grammar nitpick: et al. refers to other people, not other things. etc. is what you want here.
include/oxen/quic/loop.hpp
Outdated
| template <typename T, typename... Args> | ||
| std::shared_ptr<T> make_shared(Args&&... args) | ||
| { | ||
| return main_queue->make_shared<T>(std::forward<decltype(args)>(args)...); |
Member
There was a problem hiding this comment.
No need for decltype when you have Args...
Suggested change
| return main_queue->make_shared<T>(std::forward<decltype(args)>(args)...); | |
| return main_queue->make_shared<T>(std::forward<Args>(args)...); |
include/oxen/quic/loop.hpp
Outdated
Comment on lines
324
to
325
| // FIXME: this *may* be superfluous with the addition of JobQueue, but since it's | ||
| // public I'm not sure if we've used it outside this class... |
Member
There was a problem hiding this comment.
Delete FIXME. It seems fine to leave this here.
include/oxen/quic/loop.hpp
Outdated
| ::event_base* loop, std::chrono::microseconds _t, std::function<void()> task, bool start_immediately = true); | ||
|
|
||
| Ticker() = default; | ||
| Ticker(std::shared_ptr<bool> keepalive) : alive{keepalive} {} |
Member
There was a problem hiding this comment.
Avoid an atomic ref/deref:
Suggested change
| Ticker(std::shared_ptr<bool> keepalive) : alive{keepalive} {} | |
| Ticker(std::shared_ptr<bool> keepalive) : alive{std::move(keepalive)} {} |
Member
There was a problem hiding this comment.
Also keepalive seems misleading as a parameter name?
include/oxen/quic/loop.hpp
Outdated
| std::shared_ptr<bool> alive; | ||
|
|
||
| Wakeable() = default; | ||
| Wakeable(std::shared_ptr<bool> keepalive) : alive{keepalive} {} |
include/oxen/quic/loop.hpp
Outdated
| ~JobQueue(); | ||
|
|
||
| // Returns a pointer deleter that defers the actual destruction call to this network | ||
| // object's event loop. |
Member
There was a problem hiding this comment.
This comment is now 2 generations out of date:
networkis no longer involvedthis object's event loopshould now bethis job queue
Member
There was a problem hiding this comment.
And similar for the functions that follow this
include/oxen/quic/loop.hpp
Outdated
|
|
||
| private: | ||
| std::list<std::weak_ptr<Ticker>> tickers; | ||
| std::list<std::weak_ptr<Wakeable>> wakeables; |
Member
There was a problem hiding this comment.
I think these should stay in loop rather than being under the job queue: they already have their own event on the loop, and so feel like that should be parallel to rather than under JobQueue.
It is useful to be able to have a job queue separate from the Loop's main job queue so that jobs depending on something which will be destroyed can be cancelled without affecting other classes using the Loop. For example, A and B both use the Loop and B will be destroyed, with its destructor happening on the Loop. If some other class can queue jobs which reference B (or any of its members) *after* B's destructor is queued, but before its destructor finishes (and those other classes are informed of its destruction), those jobs would execute after B's destructor and segfault. Instead, if B owns a JobQueue and jobs which would reference it are queued onto *that* queue, then when B is destroyed, so is that queue, but any jobs in the Loop's main queue (or any other separate JobQueues) are unaffected.
Adds a test that jobs queued onto a JobQueue from a Loop after that JobQueue is destroyed indeed do not execute, and that jobs queued onto that loop's main JobQueue which would execute *after* that job *do* execute.
Adds some tests for the event queue behaving correctly when multiple job queues are created/destroyed -- in this case, "ticker" tests were added. There are also bug fixes related to said tickers, exposed by said tests.
- add JobQueue::stop() to be able to stop without destruction - make JobQueue::stop()/destruction synchronize itself rather than require the caller to worry about synchronization. - refactor Loop construction a little to put all the initialization into a static function rather than doing it inline in the constructor itself: this allows the next point: - Hold a plain JobQueue rather than a unique_ptr<JobQueue> inside Loop - Delete `loop.make_job_queue()` -- the caller can just go `std::make_unique<JobQueue>(loop)` or use some other smart pointer or even just a plain object. - fix some forwarding std::move's that should be std::forwards instead.
Removes some unnecessary containers/checks on Ticker and Wakeable, as it is the owner's responsibility to ensure these are destroyed before the Loop from which they were created stops.
Since Ticker and Wakeable create their own events and their lifetime is managed by the owner, they do not need to be creatable via both JobQueue and Loop. Since their lifetime is no longer cleaned up automatically by a JobQueue, that test section has been removed. "one-shot" jobs created via `call_later` remain on JobQueue, however, as the creator of that job does not keep a reference to/own it (so it needs to die with the JobQueue when it dies).
It makes sense for JobQueue to have its own `call_later`, so the user doesn't have to manage a bunch of one-off jobs. As such, JobQueue needs to manage those jobs and cancel them when it is stopped/destroyed, so event_base_once can't work here. I don't *think* this has any performance implications.
Where applicable, its children (Connections, Streams, Datagrams, etc.) will use this JobQueue instead of the main Loop one. There are a few places with timers and such that were not changed, but they clean up their own events on destruction in those cases so it should not be an issue.
When a JobQueue is destroyed with a pending call_get on its queue, the destructor of that callable will now release the promise with a future error; otherwise a race condition could cause another thread to hang permanently.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Adds a separable job queue one can request from the Loop.
See the comment for
Loop::make_job_queuefor detailsAdds not-quite-comprehensive tests for said job queues.
(I did not yet test Wakeable, but it is derivative of Ticker so I think it is ok. I will add a test for it soon and update this accordingly.)