Conversation
| for<'t> F: FnOnce(Conn<Self::Writable, Tx<'t, '_>>) -> BoxFuture<'t, Result<O, E>> | ||
| + Send | ||
| + Sync | ||
| + 't, |
There was a problem hiding this comment.
The +'t was useless, as it was already implied by BoxFuture itself having it as a parameter. Only a nit.
| }; | ||
| use futures::{StreamExt, TryStreamExt}; | ||
| let mut conn = state.pool.reader_conn().await?; | ||
| let mut out_of_date_tasks = select_out_of_date_tasks(&mut conn).await.try_chunks(2048); |
There was a problem hiding this comment.
Here is one big change: Instead of returning all submissions in the full database that are out of date (whose number can be unbounded, and certainly might reach millions), the code now streams results from SQLite one-by-one.
After that, we chunk the results just like the original script did (ending up with a stream whose items are vecs containing 2048 tasks each).
Side note: I think the chunk size of 2048 rows might be quite high. Maybe 256 would be big enough? ('big enough' meaning still giving the same benefits w.r.t. sending less HTTP requests to the external delegation system while reducing memory usage and SQLite transaction size / continuous time the single write connection is hogged.)
| if out_of_date_tasks.is_empty() { | ||
| return Ok(()); | ||
| } | ||
| // TODO: triggered_by_timeout early-return ? |
There was a problem hiding this comment.
I don't know exactly what this did in the original code; we'll need to handle it still. If it is really necessary to trigger this only when there are no elements and the timeout is set, then making the stream .peekable(); would allow an empty-check.
There was a problem hiding this comment.
This was only there to guard the warning below. Which I added to give devs a signal when they miss a call to notify_on_submission_change.notify_one().
| let batch = batch_or_error?; | ||
|
|
||
| send_status_update_events(state, batch.iter()).await?; | ||
| update_statuses_in_db(&state.pool, batch).await?; |
There was a problem hiding this comment.
As you can see I split off the individual parts of what was happening in its own helper functions. I think this helps with readability.
| // (swapping the nesting of `conn.transaction` and the `for task in batch`) | ||
| conn.transaction(|mut tx| { | ||
| async move { | ||
| for task in tasks { |
There was a problem hiding this comment.
The interesting bit here is that the for task in tasks was lifted outside of the two update_... and delete_... functions that took a vector-of-references before. Already they did not need a vec but would have been happy with any kind of iterator, but moving the iteration out simplifies the code further and means we don't need to build any intermediate structure.
It also makes it more obvious that is it totally fine to have the transaction open while looping over all of the elements in tasks, because we run a SQL statement for each element in there; there's no extra computational work we do besides interacting with the transaction.
And as for "oh no the transaction is open too long": no other writer can do anything as long as we have the single pool.writer_conn(). That's the real resource we need to be mindful of (to be a 'nice citizen' for the other parts of opsqueue, no writer needs to keep it for themselves for too long, as it will block those from making progress). The easiest way to accomplish this would be to make the chunk size smaller (so tasks has less elements when this function is called).
Side note: I believe there is actually no desire for atomicity between the task-statuses that we want to update/delete, so it would also be possible to swap the nesting of the transaction and the for-loop. Setting up and finishing a transaction of course has a little overhead, but on the other it means that the transactions remain small, so there is less memory being built up on the side of SQLite for it. But likely that would be overkill.
There was a problem hiding this comment.
Side note: I believe there is actually no desire for atomicity between the task-statuses that we want to update/delete, so it would also be possible to swap the nesting of the transaction and the for-loop.
I think we don't, indeed. But then, since we would only execute a single query for each item: we don't need a transaction here at all :).
I think I added the transaction because it is required for the /delegate/submit endpoint, which needs to be atomic, for correctness. And for symmetry reasons assumed it should be here as well. But because send_status_update_events is idempotent, we don't need it here at all. Problem solved!
Thanks for the help! I will also copy over some of your other clean ups!
|
|
||
| fn status_update( | ||
| task: &OutOfDateTaskRow, | ||
| ) -> Either<DelegatedJobUpdate<'_>, DelegatedJobCompletion<'_>> { |
There was a problem hiding this comment.
Splitting off this code to its own helper function aids in readability.
Note the use of Either so we get to return two different datatypes. And it makes the function immediately suitable to be passed to partition_map
| state: &ServerState, | ||
| tasks: impl Iterator<Item = &OutOfDateTaskRow>, | ||
| ) -> anyhow::Result<()> { | ||
| let (updates, completions): (Vec<_>, Vec<_>) = tasks.partition_map(status_update); |
0b49077 to
7902330
Compare
2 participants
This is a draft PR, not intended to ever be merged, but intended to be used as inspiration for @SemMulder 's continuing work to implement delegation into opsqueue.
The implementation in
sm/delegation-2hit a snag, where usingconn.transaction(...)inside afor-loop caused the compiler to complain that the vector the loop was iterating over had to be borrowed for'staticto be able to use the borrowed data inside the closure passed toconn.transaction()Dropping down to the underlying sqlx::Connection.transaction results in exactly the same problem, so there is no bug in our 'type-safe transaction pool' wrapper.
Going one level deeper, directly using begin() and commit(), does compile without problems. That is why I have the strong suspicion that we are hitting a limitation of how HRTBs are (currently) implemented in the Rust compiler.
But after that I took a step back and looked at what we were actually trying to accomplish in this piece of code: We don't need to do nearly as much copying or
Vec-manipulation as was happening in the original version. See the inline code-comments I'll be writing for more detail.