Structured Concurrency: Part 2

[rakanalh]

Description

Related to #1214 so this is to be merged AFTER.

This PR:

• Primitives:
  • Introduces TaskManager: Calling spawn using the task manager would force the future-to-be-spawned to accept a cancellation token so that all tasks using the task manager are cancellable.
• DaService:
  • Replaces tokio::spawn for each transaction queue call with a single worker.
  • Replaces an additional long running task for logging
• Sequencer
  • Spawns tasks using TaskManager.
  • Since most tasks which are running in the sequencer are running in a single tokio task, handling ctrl-c event would be sufficient since tokio::select is going to execute the ctrl-c signal to stop all current tasks. This means that no other branch of tokio::select is going to be executed concurrently.
• Full Node
  • Spawns tasks using TaskManager.
  • Use cancellation token to shutdown DA handler
• Prover
  • Spawns tasks using TaskManager.
  • Use cancellation token to shutdown DA handler

Task cancellation

For tasks which are spawned by the task manager, all tasks are expected to use the cancellation_token. Most of the uses of the token would come inside the tokio::select. This means that if all futures inside this select are running in a single task, notifying the task with the cancellation token means that we have to finish processing whatever the task is currently doing in-hand and return control back to tokio::select in order for the cancellation token to be seen as cancelled and then the component currently running is stopped.

Linked Issues

• Fixes MockDA spawns a task per DA transaction / commitment the sequencer sends #1200
• Related to Research replacing tokio::spawns with a task manager #923

Refs:

• Graceful shutdown

[kpp]

bitcoin-da should be updated too

[rakanalh]

bitcoin-da should be updated too

bitcoin-da does not use tokio::spawn anywhere. Are there specific parts you'd like to update?

[kpp]

 172   │     pub fn spawn_da_queue(
 173   │         self: Arc<Self>,
 174   │         mut rx: UnboundedReceiver<SenderWithNotifier<TxidWrapper>>,
 175   │     ) {
 176   │         // This is a queue of inscribe requests
 177   │         tokio::task::spawn_blocking(|| {

[rakanalh]

 172   │     pub fn spawn_da_queue(
 173   │         self: Arc<Self>,
 174   │         mut rx: UnboundedReceiver<SenderWithNotifier<TxidWrapper>>,
 175   │     ) {
 176   │         // This is a queue of inscribe requests
 177   │         tokio::task::spawn_blocking(|| {

I haven't look at the code here yet and maybe it does need some refactoring. However, based on Tokio's documentation for spawn_blocking:

Be aware that tasks spawned using spawn_blocking cannot be aborted because they are not async. If you call abort on a spawn_blocking task, then this will not have any effect, and the task will continue running normally. The exception is if the task has not started running yet; in that case, calling abort may prevent the task from starting.

We might be able to pull off cancellation of a blocking task using a cancellation token... i'll look into that.

[kpp]

So there are 2 things a tokio task is doing - to process get_fee_rate (can be a separate task) and to send_transaction_with_fee_rate - which is mining a tx prefix inside.

[rakanalh]

So there are 2 things a tokio task is doing - to process get_fee_rate (can be a separate task) and to send_transaction_with_fee_rate - which is mining a tx prefix inside.

get_fee_rate cannot be a separate task because we're fetching the value so that we can pass it to send_transaction_with_fee_rate

[codecov]

Codecov Report

Attention: Patch coverage is 80.98160% with 31 lines in your changes missing coverage. Please review.

Project coverage is 77.8%. Comparing base (68f32a2) to head (fa1deea).
Report is 1 commits behind head on nightly.

Files with missing lines	Patch %	Lines
crates/bitcoin-da/src/service.rs	28.5%	20 Missing ⚠️
crates/common/src/tasks/manager.rs	73.9%	6 Missing ⚠️
crates/fullnode/src/runner.rs	96.1%	2 Missing ⚠️
bin/citrea/src/rollup/bitcoin.rs	0.0%	1 Missing ⚠️
crates/fullnode/src/lib.rs	0.0%	1 Missing ⚠️
crates/prover/src/lib.rs	0.0%	1 Missing ⚠️

Additional details and impacted files

Files with missing lines	Coverage Δ
crates/fullnode/src/da_block_handler.rs	74.0% <100.0%> (+<0.1%)	⬆️
crates/prover/src/da_block_handler.rs	81.2% <100.0%> (-0.2%)	⬇️
crates/prover/src/runner.rs	87.8% <100.0%> (+3.4%)	⬆️
crates/sequencer/src/sequencer.rs	91.7% <100.0%> (+1.4%)	⬆️
...ates/sovereign-sdk/adapters/mock-da/src/service.rs	95.7% <100.0%> (+<0.1%)	⬆️
bin/citrea/src/rollup/bitcoin.rs	0.0% <0.0%> (ø)
crates/fullnode/src/lib.rs	50.0% <0.0%> (ø)
crates/prover/src/lib.rs	50.0% <0.0%> (ø)
crates/fullnode/src/runner.rs	89.2% <96.1%> (+3.4%)	⬆️
crates/common/src/tasks/manager.rs	73.9% <73.9%> (ø)
... and 1 more

... and 1 file with indirect coverage changes

[kpp]

LGMT but please note my comments

[ercecan]

LGTM overall but I believe l1 and l2 sync can also be managed with task manager? I would like to hear your thoughts on this

[rakanalh]

Shouldn't l1_sync_worker also be spawned with task_manager?

LGTM overall but I believe l1 and l2 sync can also be managed with task manager? I would like to hear your thoughts on this

if we can also pass the cancellation_token into l2 sync, we can make sure we committed db correctly before shutting down and that would be great

The reason we don't spawn sync_l1 and sync_l2 by the task manager is because it is already managed by it's own tokio::select loop. On our cancellation request, sync_l2 for example would finish it's work before ctrl_c signal is handled if a sync task was already running. For the tasks that we spawn as separate tasks, we would use the task manager to cancel them.

So, tasks which are running outside the select loop in sequencer / full node / prover would be cancelled by task manager's cancellation token. But the ones running inside would be cancelled by the ctrl_c signal.