spawn availability store and approval voting subsystems as blocking tasks (#2521)

* spawn availability store and approval voting subsystems as blocking tasks

* refactor
This commit is contained in:
Robert Habermeier
2021-02-24 14:22:48 -06:00
committed by GitHub
parent c3ef81ab30
commit 3cdaa88509
+28 -2
View File
@@ -1457,6 +1457,7 @@ where
all_subsystems.candidate_validation, all_subsystems.candidate_validation,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let candidate_backing_subsystem = spawn( let candidate_backing_subsystem = spawn(
@@ -1466,6 +1467,7 @@ where
all_subsystems.candidate_backing, all_subsystems.candidate_backing,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let candidate_selection_subsystem = spawn( let candidate_selection_subsystem = spawn(
@@ -1475,6 +1477,7 @@ where
all_subsystems.candidate_selection, all_subsystems.candidate_selection,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let statement_distribution_subsystem = spawn( let statement_distribution_subsystem = spawn(
@@ -1484,6 +1487,7 @@ where
all_subsystems.statement_distribution, all_subsystems.statement_distribution,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let availability_distribution_subsystem = spawn( let availability_distribution_subsystem = spawn(
@@ -1493,6 +1497,7 @@ where
all_subsystems.availability_distribution, all_subsystems.availability_distribution,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let availability_recovery_subsystem = spawn( let availability_recovery_subsystem = spawn(
@@ -1502,6 +1507,7 @@ where
all_subsystems.availability_recovery, all_subsystems.availability_recovery,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let bitfield_signing_subsystem = spawn( let bitfield_signing_subsystem = spawn(
@@ -1511,6 +1517,7 @@ where
all_subsystems.bitfield_signing, all_subsystems.bitfield_signing,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let bitfield_distribution_subsystem = spawn( let bitfield_distribution_subsystem = spawn(
@@ -1520,6 +1527,7 @@ where
all_subsystems.bitfield_distribution, all_subsystems.bitfield_distribution,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let provisioner_subsystem = spawn( let provisioner_subsystem = spawn(
@@ -1529,6 +1537,7 @@ where
all_subsystems.provisioner, all_subsystems.provisioner,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let pov_distribution_subsystem = spawn( let pov_distribution_subsystem = spawn(
@@ -1538,6 +1547,7 @@ where
all_subsystems.pov_distribution, all_subsystems.pov_distribution,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let runtime_api_subsystem = spawn( let runtime_api_subsystem = spawn(
@@ -1547,6 +1557,7 @@ where
all_subsystems.runtime_api, all_subsystems.runtime_api,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let availability_store_subsystem = spawn( let availability_store_subsystem = spawn(
@@ -1556,6 +1567,7 @@ where
all_subsystems.availability_store, all_subsystems.availability_store,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Blocking,
)?; )?;
let network_bridge_subsystem = spawn( let network_bridge_subsystem = spawn(
@@ -1565,6 +1577,7 @@ where
all_subsystems.network_bridge, all_subsystems.network_bridge,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let chain_api_subsystem = spawn( let chain_api_subsystem = spawn(
@@ -1574,6 +1587,7 @@ where
all_subsystems.chain_api, all_subsystems.chain_api,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let collation_generation_subsystem = spawn( let collation_generation_subsystem = spawn(
@@ -1583,6 +1597,7 @@ where
all_subsystems.collation_generation, all_subsystems.collation_generation,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
@@ -1593,6 +1608,7 @@ where
all_subsystems.collator_protocol, all_subsystems.collator_protocol,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let approval_distribution_subsystem = spawn( let approval_distribution_subsystem = spawn(
@@ -1602,6 +1618,7 @@ where
all_subsystems.approval_distribution, all_subsystems.approval_distribution,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Regular,
)?; )?;
let approval_voting_subsystem = spawn( let approval_voting_subsystem = spawn(
@@ -1611,6 +1628,7 @@ where
all_subsystems.approval_voting, all_subsystems.approval_voting,
&metrics, &metrics,
&mut seed, &mut seed,
TaskKind::Blocking,
)?; )?;
let leaves = leaves let leaves = leaves
@@ -1972,6 +1990,11 @@ where
} }
} }
enum TaskKind {
Regular,
Blocking,
}
fn spawn<S: SpawnNamed, M: Send + 'static>( fn spawn<S: SpawnNamed, M: Send + 'static>(
spawner: &mut S, spawner: &mut S,
futures: &mut FuturesUnordered<BoxFuture<'static, SubsystemResult<()>>>, futures: &mut FuturesUnordered<BoxFuture<'static, SubsystemResult<()>>>,
@@ -1979,6 +2002,7 @@ fn spawn<S: SpawnNamed, M: Send + 'static>(
s: impl Subsystem<OverseerSubsystemContext<M>>, s: impl Subsystem<OverseerSubsystemContext<M>>,
metrics: &Metrics, metrics: &Metrics,
seed: &mut u64, seed: &mut u64,
task_kind: TaskKind,
) -> SubsystemResult<OverseenSubsystem<M>> { ) -> SubsystemResult<OverseenSubsystem<M>> {
let (to_tx, to_rx) = metered::channel(CHANNEL_CAPACITY, "subsystem_spawn"); let (to_tx, to_rx) = metered::channel(CHANNEL_CAPACITY, "subsystem_spawn");
let ctx = OverseerSubsystemContext::new( let ctx = OverseerSubsystemContext::new(
@@ -2004,7 +2028,10 @@ fn spawn<S: SpawnNamed, M: Send + 'static>(
let _ = tx.send(()); let _ = tx.send(());
}); });
spawner.spawn(name, fut); match task_kind {
TaskKind::Regular => spawner.spawn(name, fut),
TaskKind::Blocking => spawner.spawn_blocking(name, fut),
}
futures.push(Box::pin(rx.map(|e| { tracing::warn!(err = ?e, "dropping error"); Ok(()) }))); futures.push(Box::pin(rx.map(|e| { tracing::warn!(err = ?e, "dropping error"); Ok(()) })));
@@ -2018,7 +2045,6 @@ fn spawn<S: SpawnNamed, M: Send + 'static>(
}) })
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use std::sync::atomic; use std::sync::atomic;