Unverified Commit eeb6cc61 authored by Bernhard Schuster's avatar Bernhard Schuster Committed by GitHub
Browse files

Revert "remove connected disconnected state only (#3868)" (#3896)

This reverts commit 7bc35262.
parent 3e92a342
Pipeline #158128 canceled with stages
in 18 minutes and 53 seconds
......@@ -6842,7 +6842,6 @@ dependencies = [
"kvdb",
"kvdb-rocksdb",
"log",
"lru",
"pallet-babe",
"pallet-im-online",
"pallet-mmr-primitives",
......
......@@ -185,7 +185,6 @@ struct BehaveMaleficient;
impl OverseerGen for BehaveMaleficient {
fn generate<'a, Spawner, RuntimeClient>(
&self,
connector: OverseerConnector,
args: OverseerGenArgs<'a, Spawner, RuntimeClient>,
) -> Result<(Overseer<Spawner, Arc<RuntimeClient>>, OverseerHandler), Error>
where
......@@ -214,7 +213,7 @@ impl OverseerGen for BehaveMaleficient {
),
);
Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner, connector)
Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
.map_err(|e| e.into())
// A builder pattern will simplify this further
......
......@@ -37,7 +37,7 @@ use polkadot_cli::{
use polkadot_node_core_candidate_validation::CandidateValidationSubsystem;
use polkadot_node_subsystem::{
messages::{AllMessages, CandidateValidationMessage},
overseer::{self, OverseerConnector, OverseerHandle},
overseer::{self, OverseerHandle},
FromOverseer,
};
......@@ -86,7 +86,6 @@ struct BehaveMaleficient;
impl OverseerGen for BehaveMaleficient {
fn generate<'a, Spawner, RuntimeClient>(
&self,
connector: OverseerConnector,
args: OverseerGenArgs<'a, Spawner, RuntimeClient>,
) -> Result<(Overseer<Spawner, Arc<RuntimeClient>>, OverseerHandle), Error>
where
......@@ -114,7 +113,7 @@ impl OverseerGen for BehaveMaleficient {
},
);
Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner, connector)
Overseer::new(leaves, all_subsystems, registry, runtime_client, spawner)
.map_err(|e| e.into())
}
}
......
......@@ -29,8 +29,7 @@ use polkadot_node_subsystem_types::messages::{
use polkadot_overseer::{
self as overseer,
gen::{FromOverseer, SpawnedSubsystem},
AllMessages, AllSubsystems, HeadSupportsParachains, Overseer, OverseerConnector,
OverseerSignal, SubsystemError,
AllMessages, AllSubsystems, HeadSupportsParachains, Overseer, OverseerSignal, SubsystemError,
};
use polkadot_primitives::v1::Hash;
......@@ -174,15 +173,8 @@ fn main() {
.replace_candidate_validation(|_| Subsystem2)
.replace_candidate_backing(|orig| orig);
let (overseer, _handle) = Overseer::new(
vec![],
all_subsystems,
None,
AlwaysSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let (overseer, _handle) =
Overseer::new(vec![], all_subsystems, None, AlwaysSupportsParachains, spawner).unwrap();
let overseer_fut = overseer.run().fuse();
let timer_stream = timer_stream;
......
......@@ -130,7 +130,7 @@ pub(crate) fn impl_builder(info: &OverseerInfo) -> proc_macro2::TokenStream {
&mut self.handle
}
/// Obtain access to the overseer handle.
pub fn as_handle(&self) -> &#handle {
pub fn as_handle(&mut self) -> &#handle {
&self.handle
}
}
......
// Copyright 2020 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Polkadot is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
use crate::{AllMessages, OverseerSignal};
use polkadot_node_subsystem_types::errors::SubsystemError;
use polkadot_overseer_gen::{FromOverseer, SpawnedSubsystem, Subsystem, SubsystemContext};
/// A dummy subsystem that implements [`Subsystem`] for all
/// types of messages. Used for tests or as a placeholder.
#[derive(Clone, Copy, Debug)]
pub struct DummySubsystem;
impl<Context> Subsystem<Context, SubsystemError> for DummySubsystem
where
Context: SubsystemContext<
Signal = OverseerSignal,
Error = SubsystemError,
AllMessages = AllMessages,
>,
{
fn start(self, mut ctx: Context) -> SpawnedSubsystem<SubsystemError> {
let future = Box::pin(async move {
loop {
match ctx.recv().await {
Err(_) => return Ok(()),
Ok(FromOverseer::Signal(OverseerSignal::Conclude)) => return Ok(()),
Ok(overseer_msg) => {
tracing::debug!(
target: "dummy-subsystem",
"Discarding a message sent from overseer {:?}",
overseer_msg
);
continue
},
}
}
});
SpawnedSubsystem { name: "dummy-subsystem", future }
}
}
......@@ -70,6 +70,7 @@ use std::{
use futures::{channel::oneshot, future::BoxFuture, select, Future, FutureExt, StreamExt};
use lru::LruCache;
use parking_lot::RwLock;
use client::{BlockImportNotification, BlockchainEvents, FinalityNotification};
use polkadot_primitives::v1::{Block, BlockId, BlockNumber, Hash, ParachainHost};
......@@ -90,17 +91,12 @@ pub use polkadot_node_subsystem_types::{
jaeger, ActivatedLeaf, ActiveLeavesUpdate, LeafStatus, OverseerSignal,
};
/// Test helper supplements.
pub mod dummy;
pub use self::dummy::DummySubsystem;
// TODO legacy, to be deleted, left for easier integration
// TODO https://github.com/paritytech/polkadot/issues/3427
mod subsystems;
pub use self::subsystems::AllSubsystems;
pub use self::subsystems::{AllSubsystems, DummySubsystem};
/// Metrics re-exports of `polkadot-metrics`.
pub mod metrics;
mod metrics;
use self::metrics::Metrics;
use polkadot_node_metrics::{
......@@ -119,7 +115,7 @@ pub use polkadot_overseer_gen::{
/// Store 2 days worth of blocks, not accounting for forks,
/// in the LRU cache. Assumes a 6-second block time.
pub const KNOWN_LEAVES_CACHE_SIZE: usize = 2 * 24 * 3600 / 6;
const KNOWN_LEAVES_CACHE_SIZE: usize = 2 * 24 * 3600 / 6;
#[cfg(test)]
mod tests;
......@@ -145,12 +141,18 @@ where
///
/// [`Overseer`]: struct.Overseer.html
#[derive(Clone)]
pub struct Handle(pub OverseerHandle);
pub enum Handle {
/// Used only at initialization to break the cyclic dependency.
// TODO: refactor in https://github.com/paritytech/polkadot/issues/3427
Disconnected(Arc<RwLock<Option<OverseerHandle>>>),
/// A handle to the overseer.
Connected(OverseerHandle),
}
impl Handle {
/// Create a new [`Handle`].
pub fn new(raw: OverseerHandle) -> Self {
Self(raw)
/// Create a new disconnected [`Handle`].
pub fn new_disconnected() -> Self {
Self::Disconnected(Arc::new(RwLock::new(None)))
}
/// Inform the `Overseer` that that some block was imported.
......@@ -199,8 +201,58 @@ impl Handle {
/// Most basic operation, to stop a server.
async fn send_and_log_error(&mut self, event: Event) {
if self.0.send(event).await.is_err() {
tracing::info!(target: LOG_TARGET, "Failed to send an event to Overseer");
self.try_connect();
if let Self::Connected(ref mut handle) = self {
if handle.send(event).await.is_err() {
tracing::info!(target: LOG_TARGET, "Failed to send an event to Overseer");
}
} else {
tracing::warn!(target: LOG_TARGET, "Using a disconnected Handle to send to Overseer");
}
}
/// Whether the handle is disconnected.
pub fn is_disconnected(&self) -> bool {
match self {
Self::Disconnected(ref x) => x.read().is_none(),
_ => false,
}
}
/// Connect this handle and all disconnected clones of it to the overseer.
pub fn connect_to_overseer(&mut self, handle: OverseerHandle) {
match self {
Self::Disconnected(ref mut x) => {
let mut maybe_handle = x.write();
if maybe_handle.is_none() {
tracing::info!(target: LOG_TARGET, "🖇️ Connecting all Handles to Overseer");
*maybe_handle = Some(handle);
} else {
tracing::warn!(
target: LOG_TARGET,
"Attempting to connect a clone of a connected Handle",
);
}
},
_ => {
tracing::warn!(
target: LOG_TARGET,
"Attempting to connect an already connected Handle",
);
},
}
}
/// Try upgrading from `Self::Disconnected` to `Self::Connected` state
/// after calling `connect_to_overseer` on `self` or a clone of `self`.
fn try_connect(&mut self) {
if let Self::Disconnected(ref mut x) = self {
let guard = x.write();
if let Some(ref h) = *guard {
let handle = h.clone();
drop(guard);
*self = Self::Connected(handle);
}
}
}
}
......@@ -438,13 +490,12 @@ where
/// # use polkadot_primitives::v1::Hash;
/// # use polkadot_overseer::{
/// # self as overseer,
/// # Overseer,
/// # OverseerSignal,
/// # OverseerConnector,
/// # SubsystemSender as _,
/// # AllMessages,
/// # AllSubsystems,
/// # HeadSupportsParachains,
/// # Overseer,
/// # SubsystemError,
/// # gen::{
/// # SubsystemContext,
......@@ -498,7 +549,6 @@ where
/// None,
/// AlwaysSupportsParachains,
/// spawner,
/// OverseerConnector::default(),
/// ).unwrap();
///
/// let timer = Delay::new(Duration::from_millis(50)).fuse();
......@@ -565,7 +615,6 @@ where
prometheus_registry: Option<&prometheus::Registry>,
supports_parachains: SupportsParachains,
s: S,
connector: OverseerConnector,
) -> SubsystemResult<(Self, OverseerHandle)>
where
CV: Subsystem<OverseerSubsystemContext<CandidateValidationMessage>, SubsystemError> + Send,
......@@ -630,7 +679,7 @@ where
.supports_parachains(supports_parachains)
.metrics(metrics.clone())
.spawner(s)
.build_with_connector(connector)?;
.build()?;
// spawn the metrics metronome task
{
......
......@@ -17,7 +17,7 @@
//! Prometheus metrics related to the overseer and its channels.
use super::*;
pub use polkadot_node_metrics::metrics::{self, prometheus, Metrics as MetricsTrait};
use polkadot_node_metrics::metrics::{self, prometheus};
use parity_util_mem::MemoryAllocationSnapshot;
......@@ -110,7 +110,7 @@ impl Metrics {
}
}
impl MetricsTrait for Metrics {
impl metrics::Metrics for Metrics {
fn try_register(registry: &prometheus::Registry) -> Result<Self, prometheus::PrometheusError> {
let metrics = MetricsInner {
activated_heads_total: prometheus::register(
......
......@@ -19,9 +19,47 @@
//! In the future, everything should be set up using the generated
//! overseer builder pattern instead.
use crate::dummy::DummySubsystem;
use crate::{AllMessages, OverseerSignal};
use polkadot_node_subsystem_types::errors::SubsystemError;
use polkadot_overseer_all_subsystems_gen::AllSubsystemsGen;
use polkadot_overseer_gen::MapSubsystem;
use polkadot_overseer_gen::{
FromOverseer, MapSubsystem, SpawnedSubsystem, Subsystem, SubsystemContext,
};
/// A dummy subsystem that implements [`Subsystem`] for all
/// types of messages. Used for tests or as a placeholder.
#[derive(Clone, Copy, Debug)]
pub struct DummySubsystem;
impl<Context> Subsystem<Context, SubsystemError> for DummySubsystem
where
Context: SubsystemContext<
Signal = OverseerSignal,
Error = SubsystemError,
AllMessages = AllMessages,
>,
{
fn start(self, mut ctx: Context) -> SpawnedSubsystem<SubsystemError> {
let future = Box::pin(async move {
loop {
match ctx.recv().await {
Err(_) => return Ok(()),
Ok(FromOverseer::Signal(OverseerSignal::Conclude)) => return Ok(()),
Ok(overseer_msg) => {
tracing::debug!(
target: "dummy-subsystem",
"Discarding a message sent from overseer {:?}",
overseer_msg
);
continue
},
}
}
});
SpawnedSubsystem { name: "dummy-subsystem", future }
}
}
/// This struct is passed as an argument to create a new instance of an [`Overseer`].
///
......
......@@ -32,7 +32,7 @@ use polkadot_primitives::v1::{
ValidatorIndex,
};
use crate::{self as overseer, gen::Delay, HeadSupportsParachains, Overseer, OverseerConnector};
use crate::{self as overseer, gen::Delay, HeadSupportsParachains, Overseer};
use metered_channel as metered;
use assert_matches::assert_matches;
......@@ -164,16 +164,9 @@ fn overseer_works() {
.replace_candidate_validation(move |_| TestSubsystem1(s1_tx))
.replace_candidate_backing(move |_| TestSubsystem2(s2_tx));
let (overseer, handle) = Overseer::new(
vec![],
all_subsystems,
None,
MockSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let mut handle = Handle(handle);
let (overseer, handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
......@@ -234,10 +227,9 @@ fn overseer_metrics_work() {
Some(&registry),
MockSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let mut handle = Handle(handle);
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
......@@ -288,15 +280,8 @@ fn overseer_ends_on_subsystem_exit() {
executor::block_on(async move {
let all_subsystems =
AllSubsystems::<()>::dummy().replace_candidate_backing(|_| ReturnOnStart);
let (overseer, _handle) = Overseer::new(
vec![],
all_subsystems,
None,
MockSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let (overseer, _handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
overseer.run().await.unwrap();
})
......@@ -397,16 +382,10 @@ fn overseer_start_stop_works() {
let all_subsystems = AllSubsystems::<()>::dummy()
.replace_candidate_validation(move |_| TestSubsystem5(tx_5))
.replace_candidate_backing(move |_| TestSubsystem6(tx_6));
let (overseer, handle) = Overseer::new(
vec![first_block],
all_subsystems,
None,
MockSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let mut handle = Handle(handle);
let (overseer, handle) =
Overseer::new(vec![first_block], all_subsystems, None, MockSupportsParachains, spawner)
.unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
......@@ -507,10 +486,9 @@ fn overseer_finalize_works() {
None,
MockSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let mut handle = Handle(handle);
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
......@@ -595,16 +573,10 @@ fn do_not_send_empty_leaves_update_on_block_finalization() {
let all_subsystems =
AllSubsystems::<()>::dummy().replace_candidate_backing(move |_| TestSubsystem6(tx_5));
let (overseer, handle) = Overseer::new(
Vec::new(),
all_subsystems,
None,
MockSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let mut handle = Handle(handle);
let (overseer, handle) =
Overseer::new(Vec::new(), all_subsystems, None, MockSupportsParachains, spawner)
.unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
......@@ -877,17 +849,9 @@ fn overseer_all_subsystems_receive_signals_and_messages() {
dispute_distribution: subsystem.clone(),
chain_selection: subsystem.clone(),
};
let (overseer, handle) = Overseer::new(
vec![],
all_subsystems,
None,
MockSupportsParachains,
spawner,
OverseerConnector::default(),
)
.unwrap();
let mut handle = Handle(handle);
let (overseer, handle) =
Overseer::new(vec![], all_subsystems, None, MockSupportsParachains, spawner).unwrap();
let mut handle = Handle::Connected(handle);
let overseer_fut = overseer.run().fuse();
pin_mut!(overseer_fut);
......
......@@ -26,7 +26,6 @@ sc-keystore = { git = "https://github.com/paritytech/substrate", branch = "maste
sc-basic-authorship = { git = "https://github.com/paritytech/substrate", branch = "master" }
service = { package = "sc-service", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
telemetry = { package = "sc-telemetry", git = "https://github.com/paritytech/substrate", branch = "master" }
lru = "0.6"
# Substrate Primitives
sp-authority-discovery = { git = "https://github.com/paritytech/substrate", branch = "master" }
......
......@@ -54,7 +54,7 @@ use {
pub use sp_core::traits::SpawnNamed;
#[cfg(feature = "full-node")]
pub use {
polkadot_overseer::{Handle, Overseer, OverseerConnector, OverseerHandle},
polkadot_overseer::{Handle, Overseer, OverseerHandle},
polkadot_primitives::v1::ParachainHost,
sc_client_api::AuxStore,
sp_authority_discovery::AuthorityDiscoveryApi,
......@@ -68,8 +68,6 @@ use polkadot_subsystem::jaeger;
use std::{sync::Arc, time::Duration};
use prometheus_endpoint::Registry;
#[cfg(feature = "full-node")]
use service::KeystoreContainer;
use service::RpcHandlers;
use telemetry::TelemetryWorker;
#[cfg(feature = "full-node")]
......@@ -304,15 +302,14 @@ fn jaeger_launch_collector_with_agent(
}
#[cfg(feature = "full-node")]
type FullSelectChain = relay_chain_selection::SelectRelayChain<FullBackend>;
type FullSelectChain = relay_chain_selection::SelectRelayChainWithFallback<FullBackend>;
#[cfg(feature = "full-node")]
type FullGrandpaBlockImport<RuntimeApi, ExecutorDispatch, ChainSelection = FullSelectChain> =
grandpa::GrandpaBlockImport<
FullBackend,
Block,
FullClient<RuntimeApi, ExecutorDispatch>,
ChainSelection,
>;
type FullGrandpaBlockImport<RuntimeApi, ExecutorDispatch> = grandpa::GrandpaBlockImport<
FullBackend,
Block,
FullClient<RuntimeApi, ExecutorDispatch>,
FullSelectChain,
>;
#[cfg(feature = "light-node")]
type LightBackend = service::TLightBackendWithHash<Block, sp_runtime::traits::BlakeTwo256>;
......@@ -322,29 +319,36 @@ type LightClient<RuntimeApi, ExecutorDispatch> =
service::TLightClientWithBackend<Block, RuntimeApi, ExecutorDispatch, LightBackend>;
#[cfg(feature = "full-node")]
struct Basics<RuntimeApi, ExecutorDispatch>
where
RuntimeApi: ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>
+ Send
+ Sync
+ 'static,
RuntimeApi::RuntimeApi:
RuntimeApiCollection<StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,
ExecutorDispatch: NativeExecutionDispatch + 'static,
{
task_manager: TaskManager,
client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,
backend: Arc<FullBackend>,
keystore_container: KeystoreContainer,
telemetry: Option<Telemetry>,
}
#[cfg(feature = "full-node")]
fn new_partial_basics<RuntimeApi, ExecutorDispatch>(
fn new_partial<RuntimeApi, ExecutorDispatch>(
config: &mut Configuration,
jaeger_agent: Option<std::net::SocketAddr>,
telemetry_worker_handle: Option<TelemetryWorkerHandle>,
) -> Result<Basics<RuntimeApi, ExecutorDispatch>, Error>
) -> Result<
service::PartialComponents<
FullClient<RuntimeApi, ExecutorDispatch>,
FullBackend,
FullSelectChain,
sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,
sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>,
(
impl service::RpcExtensionBuilder,
(
babe::BabeBlockImport<
Block,
FullClient<RuntimeApi, ExecutorDispatch>,
FullGrandpaBlockImport<RuntimeApi, ExecutorDispatch>,
>,
grandpa::LinkHalf<Block, FullClient<RuntimeApi, ExecutorDispatch>, FullSelectChain>,
babe::BabeLink<Block>,
beefy_gadget::notification::BeefySignedCommitmentSender<Block>,
),
grandpa::SharedVoterState,
std::time::Duration, // slot-duration
Option<Telemetry>,
),
>,
Error,
>
where
RuntimeApi: ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>
+ Send
......@@ -387,62 +391,21 @@ where
)?;
let client = Arc::new(client);
jaeger_launch_collector_with_agent(task_manager.spawn_handle(), &*config, jaeger_agent)?;