futures: new sui-futures crate
#24502
Conversation
amnn
requested review from
emmazzz,
evan-wall-mysten,
henryachen,
nickvikeras,
tpham-mysten and
wlmyng
amnn
temporarily deployed
to
sui-typescript-aws-kms-test-env
GitHub Actions
Inactive
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
## Description Pull the utilities related to futures/tasks/streams from the indexing framework and into their own crate. ## Test plan CI
## Description An abstraction for coordinating the lifecycle of tokio tasks, looking after: - Building up services by adding tasks or through composition. - Waiting for multiple tasks to complete. - Running secondary tasks that need to be wound down when the service winds down, but otherwise don't affect the lifecycle of the service (the service doesn't wait for these tasks to complete to shutdown). - Graceful shutdown, with a grace period, because the service has wrapped up, or because of a termination request, or an error. - Panic unwinding. - Cascading termination (errors from multiple components, multiple termination requests, timeout during shutdown). - Signal handling from the operating system. ## Test plan New unit tests: ``` sui-futures$ cargo nextest run -- service ```
## Description An abstraction over `JoinHandle<T>` that aborts the task when it is dropped. ## Test plan New unit tests: ``` sui-futures$ cargo nextest run ```
amnn
temporarily deployed
to
sui-typescript-aws-kms-test-env
GitHub Actions
Inactive
| exits: Vec<BoxFuture<'static, ()>>, | ||
|
|
||
| /// Tasks that control the lifetime of the service in the happy path. | ||
| fsts: JoinSet<anyhow::Result<()>>, |
There was a problem hiding this comment.
What do fsts and snds stand for? I don't see these mentioned anywhere else.
There was a problem hiding this comment.
"firsts" and "seconds", comes from a past functional programming life.
| /// | ||
| /// After shutdown signals are sent, tasks have this duration to complete gracefully before being | ||
| /// forcefully aborted. | ||
| pub const GRACE: Duration = Duration::from_secs(30); |
There was a problem hiding this comment.
How do the tasks know to try to abort without a CancellationToken? Or do we just assume that all tasks should take less than GRACE to complete?
There was a problem hiding this comment.
Tasks that are running inside the tokio runtime as a task, can be aborted at any await point, so we are using that aborting behaviour instead of an explicit cancellation token.
| /// | ||
| /// After shutdown signals are sent, tasks have this duration to complete gracefully before being | ||
| /// forcefully aborted. | ||
| pub const GRACE: Duration = Duration::from_secs(30); |
There was a problem hiding this comment.
not to be confused with Eve Frontier Grace Points
| let (btx, brx) = oneshot::channel::<()>(); | ||
|
|
||
| // A secondary task and a primary task. | ||
| let snd = Service::new().spawn_aborting(async move { Ok(brx.await?) }); |
There was a problem hiding this comment.
interesting, now that i think about it, we have methods on Service for adding primary tasks, but not secondaries, except through fn attach. Works for me
|
|
||
| /// Runs the service, waiting for interrupt signals from the operating system to trigger | ||
| /// graceful shutdown, within the defualt grace period. | ||
| pub async fn main(self) -> Result<(), Error> { |
There was a problem hiding this comment.
I suppose when I see main and run, and the doc comment, I'm led to think that the tasks start running when the Service starts, which is not the case given fn spawn starts the task in the background immediately, and Service coordinates shutdown. One thing I was a bit stuck on is, say I add a task that fails immediately when .spawn is called.
From the perspective of Service, that failure is not observed until run()/main() or shutdown(). Kind of like a JoinHandle... I think the naming feels a bit off for me because run implies "start running" rather than "observe the running tasks until they complete, intercept around shutdown"
There was a problem hiding this comment.
main is supposed to hint that you should call this in your own main function, but I'm not attached to these names -- I'll happily accept suggestions.
| let _ = handle.await.unwrap(); | ||
| } | ||
|
|
||
| #[tokio::test(start_paused = true)] |
Description
Factor out futures/streaming/tasks extensions from the indexer-alt codebase into their own crate. This was done to avoid a circular dependency between the metrics service and the indexing framework.
This PR also introduces the
Servicesabstraction which allows us to combine tokio tasks into a single bundle, taking into account lifecycles, graceful shutdown, error and interrupt handling.Test plan
CI
Stack
Release notes
Check each box that your changes affect. If none of the boxes relate to your changes, release notes aren't required.
For each box you select, include information after the relevant heading that describes the impact of your changes that a user might notice and any actions they must take to implement updates.