lib.rs 73.9 KB
Newer Older
Fedor Sakharov's avatar
Fedor Sakharov committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 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/>.

//! # Overseer
//!
//! `overseer` implements the Overseer architecture described in the
20
//! [implementers-guide](https://github.com/paritytech/polkadot/blob/master/roadmap/implementers-guide/guide.md).
Fedor Sakharov's avatar
Fedor Sakharov committed
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
//! For the motivations behind implementing the overseer itself you should
//! check out that guide, documentation in this crate will be mostly discussing
//! technical stuff.
//!
//! An `Overseer` is something that allows spawning/stopping and overseing
//! asynchronous tasks as well as establishing a well-defined and easy to use
//! protocol that the tasks can use to communicate with each other. It is desired
//! that this protocol is the only way tasks communicate with each other, however
//! at this moment there are no foolproof guards against other ways of communication.
//!
//! The `Overseer` is instantiated with a pre-defined set of `Subsystems` that
//! share the same behavior from `Overseer`'s point of view.
//!
//! ```text
//!                              +-----------------------------+
//!                              |         Overseer            |
//!                              +-----------------------------+
//!
//!             ................|  Overseer "holds" these and uses |..............
//!             .                  them to (re)start things                      .
//!             .                                                                .
//!             .  +-------------------+                +---------------------+  .
//!             .  |   Subsystem1      |                |   Subsystem2        |  .
//!             .  +-------------------+                +---------------------+  .
//!             .           |                                       |            .
//!             ..................................................................
//!                         |                                       |
//!                       start()                                 start()
//!                         V                                       V
//!             ..................| Overseer "runs" these |.......................
//!             .  +--------------------+               +---------------------+  .
//!             .  | SubsystemInstance1 |               | SubsystemInstance2  |  .
//!             .  +--------------------+               +---------------------+  .
//!             ..................................................................
//! ```

57
58
59
60
61
// #![deny(unused_results)]
// unused dependencies can not work for test and examples at the same time
// yielding false positives
#![warn(missing_docs)]

Fedor Sakharov's avatar
Fedor Sakharov committed
62
63
use std::fmt::Debug;
use std::pin::Pin;
64
use std::sync::Arc;
Fedor Sakharov's avatar
Fedor Sakharov committed
65
66
use std::task::Poll;
use std::time::Duration;
67
use std::collections::{hash_map, HashMap};
Fedor Sakharov's avatar
Fedor Sakharov committed
68
69
70
71

use futures::channel::{mpsc, oneshot};
use futures::{
	pending, poll, select,
72
	future::BoxFuture,
73
	stream::{self, FuturesUnordered},
Fedor Sakharov's avatar
Fedor Sakharov committed
74
75
76
77
78
	Future, FutureExt, SinkExt, StreamExt,
};
use futures_timer::Delay;
use streamunordered::{StreamYield, StreamUnordered};

asynchronous rob's avatar
asynchronous rob committed
79
use polkadot_primitives::v1::{Block, BlockNumber, Hash};
80
use client::{BlockImportNotification, BlockchainEvents, FinalityNotification};
81

82
use polkadot_subsystem::messages::{
83
	CandidateValidationMessage, CandidateBackingMessage,
84
	CandidateSelectionMessage, ChainApiMessage, StatementDistributionMessage,
85
	AvailabilityDistributionMessage, BitfieldSigningMessage, BitfieldDistributionMessage,
86
87
	ProvisionerMessage, PoVDistributionMessage, RuntimeApiMessage,
	AvailabilityStoreMessage, NetworkBridgeMessage, AllMessages, CollationGenerationMessage, CollatorProtocolMessage,
88
89
90
};
pub use polkadot_subsystem::{
	Subsystem, SubsystemContext, OverseerSignal, FromOverseer, SubsystemError, SubsystemResult,
91
	SpawnedSubsystem, ActiveLeavesUpdate, DummySubsystem,
92
};
93
use polkadot_node_subsystem_util::metrics::{self, prometheus};
94
use polkadot_node_primitives::SpawnNamed;
95

Fedor Sakharov's avatar
Fedor Sakharov committed
96
97
98
99
100

// A capacity of bounded channels inside the overseer.
const CHANNEL_CAPACITY: usize = 1024;
// A graceful `Overseer` teardown time delay.
const STOP_DELAY: u64 = 1;
101
102
103
// Target for logs.
const LOG_TARGET: &'static str = "overseer";

Fedor Sakharov's avatar
Fedor Sakharov committed
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
/// A type of messages that are sent from [`Subsystem`] to [`Overseer`].
///
/// It wraps a system-wide [`AllMessages`] type that represents all possible
/// messages in the system.
///
/// [`AllMessages`]: enum.AllMessages.html
/// [`Subsystem`]: trait.Subsystem.html
/// [`Overseer`]: struct.Overseer.html
enum ToOverseer {
	/// This is a message sent by a `Subsystem`.
	SubsystemMessage(AllMessages),

	/// A message that wraps something the `Subsystem` is desiring to
	/// spawn on the overseer and a `oneshot::Sender` to signal the result
	/// of the spawn.
	SpawnJob {
120
		name: &'static str,
Fedor Sakharov's avatar
Fedor Sakharov committed
121
122
		s: BoxFuture<'static, ()>,
	},
123
124
125
126
127
128
129

	/// Same as `SpawnJob` but for blocking tasks to be executed on a
	/// dedicated thread pool.
	SpawnBlockingJob {
		name: &'static str,
		s: BoxFuture<'static, ()>,
	},
Fedor Sakharov's avatar
Fedor Sakharov committed
130
131
}

132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
/// An event telling the `Overseer` on the particular block
/// that has been imported or finalized.
///
/// This structure exists solely for the purposes of decoupling
/// `Overseer` code from the client code and the necessity to call
/// `HeaderBackend::block_number_from_id()`.
pub struct BlockInfo {
	/// hash of the block.
	pub hash: Hash,
	/// hash of the parent block.
	pub parent_hash: Hash,
	/// block's number.
	pub number: BlockNumber,
}

147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
impl From<BlockImportNotification<Block>> for BlockInfo {
	fn from(n: BlockImportNotification<Block>) -> Self {
		BlockInfo {
			hash: n.hash,
			parent_hash: n.header.parent_hash,
			number: n.header.number,
		}
	}
}

impl From<FinalityNotification<Block>> for BlockInfo {
	fn from(n: FinalityNotification<Block>) -> Self {
		BlockInfo {
			hash: n.hash,
			parent_hash: n.header.parent_hash,
			number: n.header.number,
		}
	}
}

167
/// Some event from the outer world.
Fedor Sakharov's avatar
Fedor Sakharov committed
168
enum Event {
169
170
	BlockImported(BlockInfo),
	BlockFinalized(BlockInfo),
Fedor Sakharov's avatar
Fedor Sakharov committed
171
	MsgToSubsystem(AllMessages),
172
	ExternalRequest(ExternalRequest),
Fedor Sakharov's avatar
Fedor Sakharov committed
173
174
175
	Stop,
}

176
177
178
179
/// Some request from outer world.
enum ExternalRequest {
	WaitForActivation {
		hash: Hash,
180
		response_channel: oneshot::Sender<SubsystemResult<()>>,
181
182
183
	},
}

Fedor Sakharov's avatar
Fedor Sakharov committed
184
185
186
/// A handler used to communicate with the [`Overseer`].
///
/// [`Overseer`]: struct.Overseer.html
187
#[derive(Clone)]
Fedor Sakharov's avatar
Fedor Sakharov committed
188
189
190
191
192
193
pub struct OverseerHandler {
	events_tx: mpsc::Sender<Event>,
}

impl OverseerHandler {
	/// Inform the `Overseer` that that some block was imported.
194
	pub async fn block_imported(&mut self, block: BlockInfo) -> SubsystemResult<()> {
195
		self.events_tx.send(Event::BlockImported(block)).await.map_err(Into::into)
Fedor Sakharov's avatar
Fedor Sakharov committed
196
197
198
	}

	/// Send some message to one of the `Subsystem`s.
199
200
	pub async fn send_msg(&mut self, msg: impl Into<AllMessages>) -> SubsystemResult<()> {
		self.events_tx.send(Event::MsgToSubsystem(msg.into())).await.map_err(Into::into)
Fedor Sakharov's avatar
Fedor Sakharov committed
201
202
203
	}

	/// Inform the `Overseer` that that some block was finalized.
204
	pub async fn block_finalized(&mut self, block: BlockInfo) -> SubsystemResult<()> {
205
206
		self.events_tx.send(Event::BlockFinalized(block)).await.map_err(Into::into)
	}
Fedor Sakharov's avatar
Fedor Sakharov committed
207

208
209
210
211
212
213
214
	/// Wait for a block with the given hash to be in the active-leaves set.
	/// This method is used for external code like `Proposer` that doesn't subscribe to Overseer's signals.
	///
	/// The response channel responds if the hash was activated and is closed if the hash was deactivated.
	/// Note that due the fact the overseer doesn't store the whole active-leaves set, only deltas,
	/// the response channel may never return if the hash was deactivated before this call.
	/// In this case, it's the caller's responsibility to ensure a timeout is set.
215
	pub async fn wait_for_activation(&mut self, hash: Hash, response_channel: oneshot::Sender<SubsystemResult<()>>) -> SubsystemResult<()> {
216
217
218
219
		self.events_tx.send(Event::ExternalRequest(ExternalRequest::WaitForActivation {
			hash,
			response_channel
		})).await.map_err(Into::into)
Fedor Sakharov's avatar
Fedor Sakharov committed
220
221
222
223
	}

	/// Tell `Overseer` to shutdown.
	pub async fn stop(&mut self) -> SubsystemResult<()> {
224
		self.events_tx.send(Event::Stop).await.map_err(Into::into)
Fedor Sakharov's avatar
Fedor Sakharov committed
225
226
227
	}
}

228
/// Glues together the [`Overseer`] and `BlockchainEvents` by forwarding
229
230
231
232
233
/// import and finality notifications into the [`OverseerHandler`].
///
/// [`Overseer`]: struct.Overseer.html
/// [`OverseerHandler`]: struct.OverseerHandler.html
pub async fn forward_events<P: BlockchainEvents<Block>>(
234
	client: Arc<P>,
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
	mut handler: OverseerHandler,
) -> SubsystemResult<()> {
	let mut finality = client.finality_notification_stream();
	let mut imports = client.import_notification_stream();

	loop {
		select! {
			f = finality.next() => {
				match f {
					Some(block) => {
						handler.block_finalized(block.into()).await?;
					}
					None => break,
				}
			},
			i = imports.next() => {
				match i {
					Some(block) => {
						handler.block_imported(block.into()).await?;
					}
					None => break,
				}
			},
			complete => break,
		}
	}

	Ok(())
}

Fedor Sakharov's avatar
Fedor Sakharov committed
265
266
267
268
269
270
impl Debug for ToOverseer {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		match self {
			ToOverseer::SubsystemMessage(msg) => {
				write!(f, "OverseerMessage::SubsystemMessage({:?})", msg)
			}
271
272
			ToOverseer::SpawnJob { .. } => write!(f, "OverseerMessage::Spawn(..)"),
			ToOverseer::SpawnBlockingJob { .. } => write!(f, "OverseerMessage::SpawnBlocking(..)")
Fedor Sakharov's avatar
Fedor Sakharov committed
273
274
275
276
277
278
279
		}
	}
}

/// A running instance of some [`Subsystem`].
///
/// [`Subsystem`]: trait.Subsystem.html
280
struct SubsystemInstance<M> {
Fedor Sakharov's avatar
Fedor Sakharov committed
281
282
283
284
285
286
287
288
289
290
	tx: mpsc::Sender<FromOverseer<M>>,
}

/// A context type that is given to the [`Subsystem`] upon spawning.
/// It can be used by [`Subsystem`] to communicate with other [`Subsystem`]s
/// or to spawn it's [`SubsystemJob`]s.
///
/// [`Overseer`]: struct.Overseer.html
/// [`Subsystem`]: trait.Subsystem.html
/// [`SubsystemJob`]: trait.SubsystemJob.html
291
292
#[derive(Debug)]
pub struct OverseerSubsystemContext<M>{
Fedor Sakharov's avatar
Fedor Sakharov committed
293
294
295
296
	rx: mpsc::Receiver<FromOverseer<M>>,
	tx: mpsc::Sender<ToOverseer>,
}

297
298
299
300
301
#[async_trait::async_trait]
impl<M: Send + 'static> SubsystemContext for OverseerSubsystemContext<M> {
	type Message = M;

	async fn try_recv(&mut self) -> Result<Option<FromOverseer<M>>, ()> {
Fedor Sakharov's avatar
Fedor Sakharov committed
302
303
304
305
306
307
308
		match poll!(self.rx.next()) {
			Poll::Ready(Some(msg)) => Ok(Some(msg)),
			Poll::Ready(None) => Err(()),
			Poll::Pending => Ok(None),
		}
	}

309
	async fn recv(&mut self) -> SubsystemResult<FromOverseer<M>> {
310
311
312
313
314
		self.rx.next().await
			.ok_or(SubsystemError::Context(
				"No more messages in rx queue to process"
				.to_owned()
			))
Fedor Sakharov's avatar
Fedor Sakharov committed
315
316
	}

317
318
319
	async fn spawn(&mut self, name: &'static str, s: Pin<Box<dyn Future<Output = ()> + Send>>)
		-> SubsystemResult<()>
	{
Fedor Sakharov's avatar
Fedor Sakharov committed
320
		self.tx.send(ToOverseer::SpawnJob {
321
			name,
Fedor Sakharov's avatar
Fedor Sakharov committed
322
			s,
323
		}).await.map_err(Into::into)
Fedor Sakharov's avatar
Fedor Sakharov committed
324
325
	}

326
327
328
329
330
331
	async fn spawn_blocking(&mut self, name: &'static str, s: Pin<Box<dyn Future<Output = ()> + Send>>)
		-> SubsystemResult<()>
	{
		self.tx.send(ToOverseer::SpawnBlockingJob {
			name,
			s,
332
		}).await.map_err(Into::into)
333
334
	}

335
	async fn send_message(&mut self, msg: AllMessages) -> SubsystemResult<()> {
336
		self.tx.send(ToOverseer::SubsystemMessage(msg)).await.map_err(Into::into)
Fedor Sakharov's avatar
Fedor Sakharov committed
337
338
	}

339
340
341
342
	async fn send_messages<T>(&mut self, msgs: T) -> SubsystemResult<()>
		where T: IntoIterator<Item = AllMessages> + Send, T::IntoIter: Send
	{
		let mut msgs = stream::iter(msgs.into_iter().map(ToOverseer::SubsystemMessage).map(Ok));
343
		self.tx.send_all(&mut msgs).await.map_err(Into::into)
Fedor Sakharov's avatar
Fedor Sakharov committed
344
345
346
347
348
349
350
351
352
353
	}
}

/// A subsystem that we oversee.
///
/// Ties together the [`Subsystem`] itself and it's running instance
/// (which may be missing if the [`Subsystem`] is not running at the moment
/// for whatever reason).
///
/// [`Subsystem`]: trait.Subsystem.html
354
struct OverseenSubsystem<M> {
Fedor Sakharov's avatar
Fedor Sakharov committed
355
356
357
358
	instance: Option<SubsystemInstance<M>>,
}

/// The `Overseer` itself.
359
pub struct Overseer<S> {
360
361
	/// A candidate validation subsystem.
	candidate_validation_subsystem: OverseenSubsystem<CandidateValidationMessage>,
Fedor Sakharov's avatar
Fedor Sakharov committed
362

363
	/// A candidate backing subsystem.
364
	candidate_backing_subsystem: OverseenSubsystem<CandidateBackingMessage>,
Fedor Sakharov's avatar
Fedor Sakharov committed
365

366
367
368
369
370
371
372
373
374
	/// A candidate selection subsystem.
	candidate_selection_subsystem: OverseenSubsystem<CandidateSelectionMessage>,

	/// A statement distribution subsystem.
	statement_distribution_subsystem: OverseenSubsystem<StatementDistributionMessage>,

	/// An availability distribution subsystem.
	availability_distribution_subsystem: OverseenSubsystem<AvailabilityDistributionMessage>,

375
376
377
	/// A bitfield signing subsystem.
	bitfield_signing_subsystem: OverseenSubsystem<BitfieldSigningMessage>,

378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
	/// A bitfield distribution subsystem.
	bitfield_distribution_subsystem: OverseenSubsystem<BitfieldDistributionMessage>,

	/// A provisioner subsystem.
	provisioner_subsystem: OverseenSubsystem<ProvisionerMessage>,

	/// A PoV distribution subsystem.
	pov_distribution_subsystem: OverseenSubsystem<PoVDistributionMessage>,

	/// A runtime API subsystem.
	runtime_api_subsystem: OverseenSubsystem<RuntimeApiMessage>,

	/// An availability store subsystem.
	availability_store_subsystem: OverseenSubsystem<AvailabilityStoreMessage>,

	/// A network bridge subsystem.
	network_bridge_subsystem: OverseenSubsystem<NetworkBridgeMessage>,

396
	/// A Chain API subsystem.
397
	chain_api_subsystem: OverseenSubsystem<ChainApiMessage>,
398

399
400
401
402
403
404
	/// A Collation Generation subsystem.
	collation_generation_subsystem: OverseenSubsystem<CollationGenerationMessage>,

	/// A Collator Protocol subsystem.
	collator_protocol_subsystem: OverseenSubsystem<CollatorProtocolMessage>,

Fedor Sakharov's avatar
Fedor Sakharov committed
405
406
407
408
	/// Spawner to spawn tasks to.
	s: S,

	/// Here we keep handles to spawned subsystems to be notified when they terminate.
409
	running_subsystems: FuturesUnordered<BoxFuture<'static, SubsystemResult<()>>>,
Fedor Sakharov's avatar
Fedor Sakharov committed
410
411
412
413
414
415

	/// Gather running subsystms' outbound streams into one.
	running_subsystems_rx: StreamUnordered<mpsc::Receiver<ToOverseer>>,

	/// Events that are sent to the overseer from the outside world
	events_rx: mpsc::Receiver<Event>,
416

417
	/// External listeners waiting for a hash to be in the active-leave set.
418
	activation_external_listeners: HashMap<Hash, Vec<oneshot::Sender<SubsystemResult<()>>>>,
419

420
421
422
423
424
425
	/// A set of leaves that `Overseer` starts working with.
	///
	/// Drained at the beginning of `run` and never used again.
	leaves: Vec<(Hash, BlockNumber)>,

	/// The set of the "active leaves".
426
	active_leaves: HashMap<Hash, BlockNumber>,
427
428
429

	/// Various Prometheus metrics.
	metrics: Metrics,
Fedor Sakharov's avatar
Fedor Sakharov committed
430
431
}

432
433
434
435
436
437
438
439
/// This struct is passed as an argument to create a new instance of an [`Overseer`].
///
/// As any entity that satisfies the interface may act as a [`Subsystem`] this allows
/// mocking in the test code:
///
/// Each [`Subsystem`] is supposed to implement some interface that is generic over
/// message type that is specific to this [`Subsystem`]. At the moment not all
/// subsystems are implemented and the rest can be mocked with the [`DummySubsystem`].
440
441
442
443
pub struct AllSubsystems<
	CV = (), CB = (), CS = (), SD = (), AD = (), BS = (), BD = (), P = (),
	PoVD = (), RA = (), AS = (), NB = (), CA = (), CG = (), CP = ()
> {
444
445
446
447
448
449
450
451
452
453
	/// A candidate validation subsystem.
	pub candidate_validation: CV,
	/// A candidate backing subsystem.
	pub candidate_backing: CB,
	/// A candidate selection subsystem.
	pub candidate_selection: CS,
	/// A statement distribution subsystem.
	pub statement_distribution: SD,
	/// An availability distribution subsystem.
	pub availability_distribution: AD,
454
455
	/// A bitfield signing subsystem.
	pub bitfield_signing: BS,
456
457
458
459
460
461
462
463
464
465
466
467
	/// A bitfield distribution subsystem.
	pub bitfield_distribution: BD,
	/// A provisioner subsystem.
	pub provisioner: P,
	/// A PoV distribution subsystem.
	pub pov_distribution: PoVD,
	/// A runtime API subsystem.
	pub runtime_api: RA,
	/// An availability store subsystem.
	pub availability_store: AS,
	/// A network bridge subsystem.
	pub network_bridge: NB,
468
469
	/// A Chain API subsystem.
	pub chain_api: CA,
470
471
472
473
	/// A Collation Generation subsystem.
	pub collation_generation: CG,
	/// A Collator Protocol subsystem.
	pub collator_protocol: CP,
474
475
}

476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
impl<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP>
	AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP>
{
	/// Create a new instance of [`AllSubsystems`].
	///
	/// Each subsystem is set to [`DummySystem`].
	///
	///# Note
	///
	/// Because of a bug in rustc it is required that when calling this function,
	/// you provide a "random" type for the first generic parameter:
	///
	/// ```
	/// polkadot_overseer::AllSubsystems::<()>::dummy();
	/// ```
	pub fn dummy() -> AllSubsystems<
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem,
		DummySubsystem
	> {
		AllSubsystems {
			candidate_validation: DummySubsystem,
			candidate_backing: DummySubsystem,
			candidate_selection: DummySubsystem,
			statement_distribution: DummySubsystem,
			availability_distribution: DummySubsystem,
			bitfield_signing: DummySubsystem,
			bitfield_distribution: DummySubsystem,
			provisioner: DummySubsystem,
			pov_distribution: DummySubsystem,
			runtime_api: DummySubsystem,
			availability_store: DummySubsystem,
			network_bridge: DummySubsystem,
			chain_api: DummySubsystem,
			collation_generation: DummySubsystem,
			collator_protocol: DummySubsystem,
		}
	}

	/// Replace the `candidate_validation` instance in `self`.
	pub fn replace_candidate_validation<NEW>(
		self,
		candidate_validation: NEW,
	) -> AllSubsystems<NEW, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `candidate_backing` instance in `self`.
	pub fn replace_candidate_backing<NEW>(
		self,
		candidate_backing: NEW,
	) -> AllSubsystems<CV, NEW, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `candidate_selection` instance in `self`.
	pub fn replace_candidate_selection<NEW>(
		self,
		candidate_selection: NEW,
	) -> AllSubsystems<CV, CB, NEW, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `statement_distribution` instance in `self`.
	pub fn replace_statement_distribution<NEW>(
		self,
		statement_distribution: NEW,
	) -> AllSubsystems<CV, CB, CS, NEW, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `availability_distribution` instance in `self`.
	pub fn replace_availability_distribution<NEW>(
		self,
		availability_distribution: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, NEW, BS, BD, P, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `bitfield_signing` instance in `self`.
	pub fn replace_bitfield_signing<NEW>(
		self,
		bitfield_signing: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, NEW, BD, P, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `bitfield_distribution` instance in `self`.
	pub fn replace_bitfield_distribution<NEW>(
		self,
		bitfield_distribution: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, NEW, P, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `provisioner` instance in `self`.
	pub fn replace_provisioner<NEW>(
		self,
		provisioner: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, NEW, PoVD, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `pov_distribution` instance in `self`.
	pub fn replace_pov_distribution<NEW>(
		self,
		pov_distribution: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, NEW, RA, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `runtime_api` instance in `self`.
	pub fn replace_runtime_api<NEW>(
		self,
		runtime_api: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, NEW, AS, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `availability_store` instance in `self`.
	pub fn replace_availability_store<NEW>(
		self,
		availability_store: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, NEW, NB, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `network_bridge` instance in `self`.
	pub fn replace_network_bridge<NEW>(
		self,
		network_bridge: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NEW, CA, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `chain_api` instance in `self`.
	pub fn replace_chain_api<NEW>(
		self,
		chain_api: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, NEW, CG, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api,
			collation_generation: self.collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `collation_generation` instance in `self`.
	pub fn replace_collation_generation<NEW>(
		self,
		collation_generation: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, NEW, CP> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation,
			collator_protocol: self.collator_protocol,
		}
	}

	/// Replace the `collator_protocol` instance in `self`.
	pub fn replace_collator_protocol<NEW>(
		self,
		collator_protocol: NEW,
	) -> AllSubsystems<CV, CB, CS, SD, AD, BS, BD, P, PoVD, RA, AS, NB, CA, CG, NEW> {
		AllSubsystems {
			candidate_validation: self.candidate_validation,
			candidate_backing: self.candidate_backing,
			candidate_selection: self.candidate_selection,
			statement_distribution: self.statement_distribution,
			availability_distribution: self.availability_distribution,
			bitfield_signing: self.bitfield_signing,
			bitfield_distribution: self.bitfield_distribution,
			provisioner: self.provisioner,
			pov_distribution: self.pov_distribution,
			runtime_api: self.runtime_api,
			availability_store: self.availability_store,
			network_bridge: self.network_bridge,
			chain_api: self.chain_api,
			collation_generation: self.collation_generation,
			collator_protocol,
		}
	}
}

888
889
890
891
892
/// Overseer Prometheus metrics.
#[derive(Clone)]
struct MetricsInner {
	activated_heads_total: prometheus::Counter<prometheus::U64>,
	deactivated_heads_total: prometheus::Counter<prometheus::U64>,
893
	messages_relayed_total: prometheus::Counter<prometheus::U64>,
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
}

#[derive(Default, Clone)]
struct Metrics(Option<MetricsInner>);

impl Metrics {
	fn on_head_activated(&self) {
		if let Some(metrics) = &self.0 {
			metrics.activated_heads_total.inc();
		}
	}

	fn on_head_deactivated(&self) {
		if let Some(metrics) = &self.0 {
			metrics.deactivated_heads_total.inc();
		}
	}
911
912
913
914
915
916

	fn on_message_relayed(&self) {
		if let Some(metrics) = &self.0 {
			metrics.messages_relayed_total.inc();
		}
	}
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
}

impl metrics::Metrics for Metrics {
	fn try_register(registry: &prometheus::Registry) -> Result<Self, prometheus::PrometheusError> {
		let metrics = MetricsInner {
			activated_heads_total: prometheus::register(
				prometheus::Counter::new(
					"parachain_activated_heads_total",
					"Number of activated heads."
				)?,
				registry,
			)?,
			deactivated_heads_total: prometheus::register(
				prometheus::Counter::new(
					"parachain_deactivated_heads_total",
					"Number of deactivated heads."
				)?,
				registry,
			)?,
936
937
938
939
940
941
942
			messages_relayed_total: prometheus::register(
				prometheus::Counter::new(
					"parachain_messages_relayed_total",
					"Number of messages relayed by Overseer."
				)?,
				registry,
			)?,
943
944
945
946
947
		};
		Ok(Metrics(Some(metrics)))
	}
}

Fedor Sakharov's avatar
Fedor Sakharov committed
948
949
impl<S> Overseer<S>
where
950
	S: SpawnNamed,
Fedor Sakharov's avatar
Fedor Sakharov committed
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
{
	/// Create a new intance of the `Overseer` with a fixed set of [`Subsystem`]s.
	///
	/// ```text
	///                  +------------------------------------+
	///                  |            Overseer                |
	///                  +------------------------------------+
	///                    /            |             |      \
	///      ................. subsystems...................................
	///      . +-----------+    +-----------+   +----------+   +---------+ .
	///      . |           |    |           |   |          |   |         | .
	///      . +-----------+    +-----------+   +----------+   +---------+ .
	///      ...............................................................
	///                              |
	///                        probably `spawn`
	///                            a `job`
	///                              |
	///                              V
	///                         +-----------+
	///                         |           |
	///                         +-----------+
	///
	/// ```
	///
	/// [`Subsystem`]: trait.Subsystem.html
	///
	/// # Example
	///
	/// The [`Subsystems`] may be any type as long as they implement an expected interface.
980
981
	/// Here, we create a mock validation subsystem and a few dummy ones and start the `Overseer` with them.
	/// For the sake of simplicity the termination of the example is done with a timeout.
Fedor Sakharov's avatar
Fedor Sakharov committed
982
983
984
985
	/// ```
	/// # use std::time::Duration;
	/// # use futures::{executor, pin_mut, select, FutureExt};
	/// # use futures_timer::Delay;
986
	/// # use polkadot_overseer::{Overseer, AllSubsystems};
987
	/// # use polkadot_subsystem::{
988
989
	/// #     Subsystem, DummySubsystem, SpawnedSubsystem, SubsystemContext,
	/// #     messages::CandidateValidationMessage,
Fedor Sakharov's avatar
Fedor Sakharov committed
990
991
992
	/// # };
	///
	/// struct ValidationSubsystem;
993
994
	///
	/// impl<C> Subsystem<C> for ValidationSubsystem
995
	///     where C: SubsystemContext<Message=CandidateValidationMessage>
996
	/// {
Fedor Sakharov's avatar
Fedor Sakharov committed
997
	///     fn start(
998
	///         self,
999
	///         mut ctx: C,
Fedor Sakharov's avatar
Fedor Sakharov committed
1000
	///     ) -> SpawnedSubsystem {
For faster browsing, not all history is shown. View entire blame