pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-13 23:21:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

remove provisioner checks (#4254)

Browse Source 
* chore/provisioner: move metrics to a separate module

* avoid the duplicate names

* reduce all checks

* fixup tests

* Update node/core/provisioner/src/lib.rs

Co-authored-by: Zeke Mostov <z.mostov@gmail.com>

* chore: fmt

* chore: spellcheck

* doc

* remove the enum anti-pattern

* guide update - remove all the responsibilities

* add another trivial check

* Update node/core/provisioner/src/metrics.rs

Co-authored-by: Andronik Ordian <write@reusable.software>

* Update roadmap/implementers-guide/src/node/utility/provisioner.md

Co-authored-by: Andronik Ordian <write@reusable.software>

* Update node/core/provisioner/src/metrics.rs

Co-authored-by: Andronik Ordian <write@reusable.software>

Co-authored-by: Zeke Mostov <z.mostov@gmail.com>
Co-authored-by: Andronik Ordian <write@reusable.software>
This commit is contained in:
Bernhard Schuster
2021-11-19 19:15:59 +01:00
committed by GitHub
parent eee4bb2577
commit d5d916a915
8 changed files with 321 additions and 886 deletions
Download Patch File Download Diff File Expand all files Collapse all files
polkadot/node/core/provisioner/src/lib.rs
+60 -346
View File
@@ -14,12 +14,11 @@
// You should have received a copy of the GNU General Public License
// along with Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! The provisioner is responsible for assembling a relay chain block
//! from a set of available parachain candidates of its choice.
//! The provisioner is responsible for assembling a set of items, from which the
//! runtime will pick a subset and create a relay chain block.
#![deny(missing_docs, unused_crate_dependencies)]
use bitvec::vec::BitVec;
use futures::{
channel::{mpsc, oneshot},
prelude::*,
@@ -29,25 +28,24 @@ use polkadot_node_subsystem::{
errors::{ChainApiError, RuntimeApiError},
jaeger,
messages::{
CandidateBackingMessage, ChainApiMessage, DisputeCoordinatorMessage, ProvisionableData,
CandidateBackingMessage, DisputeCoordinatorMessage, ProvisionableData,
ProvisionerInherentData, ProvisionerMessage,
},
PerLeafSpan, SubsystemSender,
};
use polkadot_node_subsystem_util::{
self as util,
metrics::{self, prometheus},
request_availability_cores, request_persisted_validation_data, JobSender, JobSubsystem,
JobTrait,
};
use polkadot_node_subsystem_util::{self as util, JobSender, JobSubsystem, JobTrait};
use polkadot_primitives::v1::{
BackedCandidate, BlockNumber, CandidateReceipt, CoreState, DisputeStatement,
DisputeStatementSet, Hash, MultiDisputeStatementSet, OccupiedCoreAssumption,
SignedAvailabilityBitfield, ValidatorIndex,
BackedCandidate, CandidateHash, CandidateReceipt, DisputeStatement, DisputeStatementSet, Hash,
Id as ParaId, MultiDisputeStatementSet, SignedAvailabilityBitfield,
SignedAvailabilityBitfields,
};
use std::{collections::BTreeMap, pin::Pin, sync::Arc};
use std::{collections::HashSet, pin::Pin, sync::Arc};
use thiserror::Error;
mod metrics;
pub use self::metrics::*;
#[cfg(test)]
mod tests;
@@ -106,40 +104,17 @@ pub enum Error {
#[error(transparent)]
Util(#[from] util::Error),
#[error("failed to get availability cores")]
CanceledAvailabilityCores(#[source] oneshot::Canceled),
#[error("failed to get persisted validation data")]
CanceledPersistedValidationData(#[source] oneshot::Canceled),
#[error("failed to get block number")]
CanceledBlockNumber(#[source] oneshot::Canceled),
#[error("failed to get backed candidates")]
CanceledBackedCandidates(#[source] oneshot::Canceled),
#[error("failed to get votes on dispute")]
CanceledCandidateVotes(#[source] oneshot::Canceled),
#[error(transparent)]
ChainApi(#[from] ChainApiError),
#[error(transparent)]
Runtime(#[from] RuntimeApiError),
#[error("failed to send message to ChainAPI")]
ChainApiMessageSend(#[source] mpsc::SendError),
#[error("failed to send message to CandidateBacking to get backed candidates")]
GetBackedCandidatesSend(#[source] mpsc::SendError),
#[error("failed to send return message with Inherents")]
InherentDataReturnChannel,
#[error(
"backed candidate does not correspond to selected candidate; check logic in provisioner"
)]
BackedCandidateOrderingProblem,
}
impl JobTrait for ProvisioningJob {
@@ -193,11 +168,10 @@ impl ProvisioningJob {
sender: &mut impl SubsystemSender,
span: PerLeafSpan,
) -> Result<(), Error> {
use ProvisionerMessage::{ProvisionableData, RequestInherentData};
loop {
futures::select! {
msg = self.receiver.next() => match msg {
Some(RequestInherentData(_, return_sender)) => {
Some(ProvisionerMessage::RequestInherentData(_, return_sender)) => {
let _span = span.child("req-inherent-data");
let _timer = self.metrics.time_request_inherent_data();
@@ -207,7 +181,7 @@ impl ProvisioningJob {
self.awaiting_inherent.push(return_sender);
}
}
Some(ProvisionableData(_, data)) => {
Some(ProvisionerMessage::ProvisionableData(_, data)) => {
let span = span.child("provisionable-data");
let _timer = self.metrics.time_provisionable_data();
@@ -235,8 +209,8 @@ impl ProvisioningJob {
) {
if let Err(err) = send_inherent_data(
self.relay_parent,
&self.signed_bitfields,
&self.backed_candidates,
self.signed_bitfields.clone(),
self.backed_candidates.clone(),
return_senders,
sender,
)
@@ -268,46 +242,25 @@ impl ProvisioningJob {
}
}
type CoreAvailability = BitVec<bitvec::order::Lsb0, u8>;
/// The provisioner is the subsystem best suited to choosing which specific
/// backed candidates and availability bitfields should be assembled into the
/// block. To engage this functionality, a
/// `ProvisionerMessage::RequestInherentData` is sent; the response is a set of
/// non-conflicting candidates and the appropriate bitfields. Non-conflicting
/// means that there are never two distinct parachain candidates included for
/// the same parachain and that new parachain candidates cannot be included
/// until the previous one either gets declared available or expired.
///
/// The main complication here is going to be around handling
/// occupied-core-assumptions. We might have candidates that are only
/// includable when some bitfields are included. And we might have candidates
/// that are not includable when certain bitfields are included.
///
/// When we're choosing bitfields to include, the rule should be simple:
/// maximize availability. So basically, include all bitfields. And then
/// choose a coherent set of candidates along with that.
/// The provisioner is the subsystem best suited on the node side,
/// yet it lacks sufficient information to do weight based inherents limiting.
/// This does the minimalistic checks and forwards a most likely
/// too large set of bitfields, candidates, and dispute votes to
/// the runtime. The `fn create_inherent` in the runtime is responsible
/// to use a subset of these.
async fn send_inherent_data(
relay_parent: Hash,
bitfields: &[SignedAvailabilityBitfield],
candidates: &[CandidateReceipt],
bitfields: SignedAvailabilityBitfields,
candidate_receipts: Vec<CandidateReceipt>,
return_senders: Vec<oneshot::Sender<ProvisionerInherentData>>,
from_job: &mut impl SubsystemSender,
) -> Result<(), Error> {
let availability_cores = request_availability_cores(relay_parent, from_job)
.await
.await
.map_err(|err| Error::CanceledAvailabilityCores(err))??;
let backed_candidates =
collect_backed_candidates(candidate_receipts, relay_parent, from_job).await?;
let bitfields = select_availability_bitfields(&availability_cores, bitfields);
let candidates =
select_candidates(&availability_cores, &bitfields, candidates, relay_parent, from_job)
.await?;
let disputes = collect_disputes(from_job).await?;
let disputes = select_disputes(from_job).await?;
let inherent_data =
ProvisionerInherentData { bitfields, backed_candidates: candidates, disputes };
let inherent_data = ProvisionerInherentData { bitfields, backed_candidates, disputes };
for return_sender in return_senders {
return_sender
@@ -318,120 +271,33 @@ async fn send_inherent_data(
Ok(())
}
/// In general, we want to pick all the bitfields. However, we have the following constraints:
///
/// - not more than one per validator
/// - each 1 bit must correspond to an occupied core
///
/// If we have too many, an arbitrary selection policy is fine. For purposes of maximizing availability,
/// we pick the one with the greatest number of 1 bits.
///
/// Note: This does not enforce any sorting precondition on the output; the ordering there will be unrelated
/// to the sorting of the input.
fn select_availability_bitfields(
cores: &[CoreState],
bitfields: &[SignedAvailabilityBitfield],
) -> Vec<SignedAvailabilityBitfield> {
let mut selected: BTreeMap<ValidatorIndex, SignedAvailabilityBitfield> = BTreeMap::new();
'a: for bitfield in bitfields.iter().cloned() {
if bitfield.payload().0.len() != cores.len() {
continue
}
let is_better = selected
.get(&bitfield.validator_index())
.map_or(true, |b| b.payload().0.count_ones() < bitfield.payload().0.count_ones());
if !is_better {
continue
}
for (idx, _) in cores.iter().enumerate().filter(|v| !v.1.is_occupied()) {
// Bit is set for an unoccupied core - invalid
if *bitfield.payload().0.get(idx).as_deref().unwrap_or(&false) {
continue 'a
}
}
let _ = selected.insert(bitfield.validator_index(), bitfield);
}
selected.into_iter().map(|(_, b)| b).collect()
}
/// Determine which cores are free, and then to the degree possible, pick a candidate appropriate to each free core.
async fn select_candidates(
availability_cores: &[CoreState],
bitfields: &[SignedAvailabilityBitfield],
candidates: &[CandidateReceipt],
/// Collect backed candidates with a matching `relay_parent`.
async fn collect_backed_candidates(
candidate_receipts: Vec<CandidateReceipt>,
relay_parent: Hash,
sender: &mut impl SubsystemSender,
) -> Result<Vec<BackedCandidate>, Error> {
let block_number = get_block_number_under_construction(relay_parent, sender).await?;
let mut selected_candidates =
Vec::with_capacity(candidates.len().min(availability_cores.len()));
for (core_idx, core) in availability_cores.iter().enumerate() {
let (scheduled_core, assumption) = match core {
CoreState::Scheduled(scheduled_core) => (scheduled_core, OccupiedCoreAssumption::Free),
CoreState::Occupied(occupied_core) => {
if bitfields_indicate_availability(core_idx, bitfields, &occupied_core.availability)
{
if let Some(ref scheduled_core) = occupied_core.next_up_on_available {
(scheduled_core, OccupiedCoreAssumption::Included)
} else {
continue
}
} else {
if occupied_core.time_out_at != block_number {
continue
}
if let Some(ref scheduled_core) = occupied_core.next_up_on_time_out {
(scheduled_core, OccupiedCoreAssumption::TimedOut)
} else {
continue
}
}
},
CoreState::Free => continue,
};
let validation_data = match request_persisted_validation_data(
relay_parent,
scheduled_core.para_id,
assumption,
sender,
)
.await
.await
.map_err(|err| Error::CanceledPersistedValidationData(err))??
{
Some(v) => v,
None => continue,
};
let computed_validation_data_hash = validation_data.hash();
// we arbitrarily pick the first of the backed candidates which match the appropriate selection criteria
if let Some(candidate) = candidates.iter().find(|backed_candidate| {
let descriptor = &backed_candidate.descriptor;
descriptor.para_id == scheduled_core.para_id &&
descriptor.persisted_validation_data_hash == computed_validation_data_hash
}) {
let candidate_hash = candidate.hash();
tracing::trace!(
target: LOG_TARGET,
"Selecting candidate {}. para_id={} core={}",
candidate_hash,
candidate.descriptor.para_id,
core_idx,
);
selected_candidates.push(candidate_hash);
}
}
let max_one_candidate_per_para = HashSet::<ParaId>::with_capacity(candidate_receipts.len());
let selected_candidates = candidate_receipts
.into_iter()
.filter(|candidate_receipt| {
// assure the follow up query `GetBackedCandidate` succeeds
candidate_receipt.descriptor().relay_parent == relay_parent
})
.scan(max_one_candidate_per_para, |unique, candidate_receipt| {
let para_id = candidate_receipt.descriptor().para_id;
if unique.insert(para_id) {
Some(candidate_receipt.hash())
} else {
tracing::debug!(
target: LOG_TARGET,
?para_id,
"Duplicate candidate detected for para, only submitting one",
);
None
}
})
.collect::<Vec<CandidateHash>>();
// now get the backed candidates corresponding to these candidate receipts
let (tx, rx) = oneshot::channel();
@@ -445,106 +311,18 @@ async fn select_candidates(
.into(),
)
.await;
let mut candidates = rx.await.map_err(|err| Error::CanceledBackedCandidates(err))?;
// `selected_candidates` is generated in ascending order by core index, and `GetBackedCandidates`
// _should_ preserve that property, but let's just make sure.
//
// We can't easily map from `BackedCandidate` to `core_idx`, but we know that every selected candidate
// maps to either 0 or 1 backed candidate, and the hashes correspond. Therefore, by checking them
// in order, we can ensure that the backed candidates are also in order.
let mut backed_idx = 0;
for selected in selected_candidates {
if selected ==
candidates.get(backed_idx).ok_or(Error::BackedCandidateOrderingProblem)?.hash()
{
backed_idx += 1;
}
}
if candidates.len() != backed_idx {
Err(Error::BackedCandidateOrderingProblem)?;
}
// keep only one candidate with validation code.
let mut with_validation_code = false;
candidates.retain(|c| {
if c.candidate.commitments.new_validation_code.is_some() {
if with_validation_code {
return false
}
with_validation_code = true;
}
true
});
let backed_candidates = rx.await.map_err(|err| Error::CanceledBackedCandidates(err))?;
tracing::debug!(
target: LOG_TARGET,
"Selected {} candidates for {} cores",
candidates.len(),
availability_cores.len(),
"Selected {} backed candidates ready to be sanitized by the runtime",
backed_candidates.len(),
);
Ok(candidates)
Ok(backed_candidates)
}
/// Produces a block number 1 higher than that of the relay parent
/// in the event of an invalid `relay_parent`, returns `Ok(0)`
async fn get_block_number_under_construction(
relay_parent: Hash,
sender: &mut impl SubsystemSender,
) -> Result<BlockNumber, Error> {
let (tx, rx) = oneshot::channel();
sender.send_message(ChainApiMessage::BlockNumber(relay_parent, tx).into()).await;
match rx.await.map_err(|err| Error::CanceledBlockNumber(err))? {
Ok(Some(n)) => Ok(n + 1),
Ok(None) => Ok(0),
Err(err) => Err(err.into()),
}
}
/// The availability bitfield for a given core is the transpose
/// of a set of signed availability bitfields. It goes like this:
///
/// - construct a transverse slice along `core_idx`
/// - bitwise-or it with the availability slice
/// - count the 1 bits, compare to the total length; true on 2/3+
fn bitfields_indicate_availability(
core_idx: usize,
bitfields: &[SignedAvailabilityBitfield],
availability: &CoreAvailability,
) -> bool {
let mut availability = availability.clone();
let availability_len = availability.len();
for bitfield in bitfields {
let validator_idx = bitfield.validator_index().0 as usize;
match availability.get_mut(validator_idx) {
None => {
// in principle, this function might return a `Result<bool, Error>` so that we can more clearly express this error condition
// however, in practice, that would just push off an error-handling routine which would look a whole lot like this one.
// simpler to just handle the error internally here.
tracing::warn!(
target: LOG_TARGET,
validator_idx = %validator_idx,
availability_len = %availability_len,
"attempted to set a transverse bit at idx {} which is greater than bitfield size {}",
validator_idx,
availability_len,
);
return false
},
Some(mut bit_mut) => *bit_mut |= bitfield.payload().0[core_idx],
}
}
3 * availability.count_ones() >= 2 * availability.len()
}
async fn select_disputes(
async fn collect_disputes(
sender: &mut impl SubsystemSender,
) -> Result<MultiDisputeStatementSet, Error> {
let (tx, rx) = oneshot::channel();
@@ -560,6 +338,8 @@ async fn select_disputes(
// 2. Disputes are expected to be rare because they come with heavy slashing.
sender.send_message(DisputeCoordinatorMessage::RecentDisputes(tx).into()).await;
// TODO: scrape concluded disputes from on-chain to limit the number of disputes
// TODO: <https://github.com/paritytech/polkadot/issues/4329>
let recent_disputes = match rx.await {
Ok(r) => r,
Err(oneshot::Canceled) => {
@@ -614,71 +394,5 @@ async fn select_disputes(
.collect())
}
#[derive(Clone)]
struct MetricsInner {
inherent_data_requests: prometheus::CounterVec<prometheus::U64>,
request_inherent_data: prometheus::Histogram,
provisionable_data: prometheus::Histogram,
}
/// Provisioner metrics.
#[derive(Default, Clone)]
pub struct Metrics(Option<MetricsInner>);
impl Metrics {
fn on_inherent_data_request(&self, response: Result<(), ()>) {
if let Some(metrics) = &self.0 {
match response {
Ok(()) => metrics.inherent_data_requests.with_label_values(&["succeeded"]).inc(),
Err(()) => metrics.inherent_data_requests.with_label_values(&["failed"]).inc(),
}
}
}
/// Provide a timer for `request_inherent_data` which observes on drop.
fn time_request_inherent_data(
&self,
) -> Option<metrics::prometheus::prometheus::HistogramTimer> {
self.0.as_ref().map(|metrics| metrics.request_inherent_data.start_timer())
}
/// Provide a timer for `provisionable_data` which observes on drop.
fn time_provisionable_data(&self) -> Option<metrics::prometheus::prometheus::HistogramTimer> {
self.0.as_ref().map(|metrics| metrics.provisionable_data.start_timer())
}
}
impl metrics::Metrics for Metrics {
fn try_register(registry: &prometheus::Registry) -> Result<Self, prometheus::PrometheusError> {
let metrics = MetricsInner {
inherent_data_requests: prometheus::register(
prometheus::CounterVec::new(
prometheus::Opts::new(
"parachain_inherent_data_requests_total",
"Number of InherentData requests served by provisioner.",
),
&["success"],
)?,
registry,
)?,
request_inherent_data: prometheus::register(
prometheus::Histogram::with_opts(prometheus::HistogramOpts::new(
"parachain_provisioner_request_inherent_data",
"Time spent within `provisioner::request_inherent_data`",
))?,
registry,
)?,
provisionable_data: prometheus::register(
prometheus::Histogram::with_opts(prometheus::HistogramOpts::new(
"parachain_provisioner_provisionable_data",
"Time spent within `provisioner::provisionable_data`",
))?,
registry,
)?,
};
Ok(Metrics(Some(metrics)))
}
}
/// The provisioning subsystem.
pub type ProvisioningSubsystem<Spawner> = JobSubsystem<ProvisioningJob, Spawner>;
pub type ProvisionerSubsystem<Spawner> = JobSubsystem<ProvisioningJob, Spawner>;
polkadot/node/core/provisioner/src/metrics.rs
+85
View File
@@ -0,0 +1,85 @@
// 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 polkadot_node_subsystem_util::metrics::{self, prometheus};
#[derive(Clone)]
struct MetricsInner {
inherent_data_requests: prometheus::CounterVec<prometheus::U64>,
request_inherent_data: prometheus::Histogram,
provisionable_data: prometheus::Histogram,
}
/// Provisioner metrics.
#[derive(Default, Clone)]
pub struct Metrics(Option<MetricsInner>);
impl Metrics {
pub(crate) fn on_inherent_data_request(&self, response: Result<(), ()>) {
if let Some(metrics) = &self.0 {
match response {
Ok(()) => metrics.inherent_data_requests.with_label_values(&["succeeded"]).inc(),
Err(()) => metrics.inherent_data_requests.with_label_values(&["failed"]).inc(),
}
}
}
/// Provide a timer for `request_inherent_data` which observes on drop.
pub(crate) fn time_request_inherent_data(
&self,
) -> Option<metrics::prometheus::prometheus::HistogramTimer> {
self.0.as_ref().map(|metrics| metrics.request_inherent_data.start_timer())
}
/// Provide a timer for `provisionable_data` which observes on drop.
pub(crate) fn time_provisionable_data(
&self,
) -> Option<metrics::prometheus::prometheus::HistogramTimer> {
self.0.as_ref().map(|metrics| metrics.provisionable_data.start_timer())
}
}
impl metrics::Metrics for Metrics {
fn try_register(registry: &prometheus::Registry) -> Result<Self, prometheus::PrometheusError> {
let metrics = MetricsInner {
inherent_data_requests: prometheus::register(
prometheus::CounterVec::new(
prometheus::Opts::new(
"parachain_inherent_data_requests_total",
"Number of InherentData requests served by provisioner.",
),
&["success"],
)?,
registry,
)?,
request_inherent_data: prometheus::register(
prometheus::Histogram::with_opts(prometheus::HistogramOpts::new(
"parachain_provisioner_request_inherent_data_time",
"Time spent within `provisioner::request_inherent_data`",
))?,
registry,
)?,
provisionable_data: prometheus::register(
prometheus::Histogram::with_opts(prometheus::HistogramOpts::new(
"parachain_provisioner_provisionable_data_time",
"Time spent within `provisioner::provisionable_data`",
))?,
registry,
)?,
};
Ok(Metrics(Some(metrics)))
}
}
polkadot/node/core/provisioner/src/tests.rs
+146 -469
View File
@@ -1,481 +1,158 @@
// 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 super::*;
use bitvec::bitvec;
use polkadot_primitives::v1::{OccupiedCore, ScheduledCore};
use futures::{channel::mpsc, future};
pub fn occupied_core(para_id: u32) -> CoreState {
CoreState::Occupied(OccupiedCore {
group_responsible: para_id.into(),
next_up_on_available: None,
occupied_since: 100_u32,
time_out_at: 200_u32,
next_up_on_time_out: None,
availability: bitvec![bitvec::order::Lsb0, u8; 0; 32],
candidate_descriptor: Default::default(),
candidate_hash: Default::default(),
})
fn default_bitvec(n_cores: usize) -> bitvec::vec::BitVec<bitvec::order::Lsb0, u8> {
bitvec::vec::BitVec::repeat(false, n_cores)
}
pub fn build_occupied_core<Builder>(para_id: u32, builder: Builder) -> CoreState
where
Builder: FnOnce(&mut OccupiedCore),
use crate::collect_backed_candidates;
use polkadot_node_subsystem::messages::AllMessages;
use polkadot_node_subsystem_test_helpers::TestSubsystemSender;
use polkadot_primitives::v1::{
BlockNumber, CandidateCommitments, CandidateDescriptor, CommittedCandidateReceipt, Hash,
PersistedValidationData,
};
fn test_harness<OverseerFactory, Overseer, TestFactory, Test>(
overseer_factory: OverseerFactory,
test_factory: TestFactory,
) where
OverseerFactory: FnOnce(mpsc::UnboundedReceiver<AllMessages>) -> Overseer,
Overseer: Future<Output = ()>,
TestFactory: FnOnce(TestSubsystemSender) -> Test,
Test: Future<Output = ()>,
{
let mut core = match occupied_core(para_id) {
CoreState::Occupied(core) => core,
_ => unreachable!(),
};
let (tx, rx) = polkadot_node_subsystem_test_helpers::sender_receiver();
let overseer = overseer_factory(rx);
let test = test_factory(tx);
builder(&mut core);
futures::pin_mut!(overseer, test);
CoreState::Occupied(core)
let _ = futures::executor::block_on(future::join(overseer, test));
}
pub fn default_bitvec(n_cores: usize) -> CoreAvailability {
bitvec![bitvec::order::Lsb0, u8; 0; n_cores]
}
pub fn scheduled_core(id: u32) -> ScheduledCore {
ScheduledCore { para_id: id.into(), ..Default::default() }
}
mod select_availability_bitfields {
use super::{super::*, default_bitvec, occupied_core};
use futures::executor::block_on;
use polkadot_primitives::v1::{SigningContext, ValidatorId, ValidatorIndex};
use sp_application_crypto::AppKey;
use sp_keystore::{testing::KeyStore, CryptoStore, SyncCryptoStorePtr};
use std::sync::Arc;
async fn signed_bitfield(
keystore: &SyncCryptoStorePtr,
field: CoreAvailability,
validator_idx: ValidatorIndex,
) -> SignedAvailabilityBitfield {
let public = CryptoStore::sr25519_generate_new(&**keystore, ValidatorId::ID, None)
.await
.expect("generated sr25519 key");
SignedAvailabilityBitfield::sign(
&keystore,
field.into(),
&<SigningContext<Hash>>::default(),
validator_idx,
&public.into(),
)
.await
.ok()
.flatten()
.expect("Should be signed")
}
#[test]
fn not_more_than_one_per_validator() {
let keystore: SyncCryptoStorePtr = Arc::new(KeyStore::new());
let mut bitvec = default_bitvec(2);
bitvec.set(0, true);
bitvec.set(1, true);
let cores = vec![occupied_core(0), occupied_core(1)];
// we pass in three bitfields with two validators
// this helps us check the postcondition that we get two bitfields back, for which the validators differ
let bitfields = vec![
block_on(signed_bitfield(&keystore, bitvec.clone(), ValidatorIndex(0))),
block_on(signed_bitfield(&keystore, bitvec.clone(), ValidatorIndex(1))),
block_on(signed_bitfield(&keystore, bitvec, ValidatorIndex(1))),
];
let mut selected_bitfields = select_availability_bitfields(&cores, &bitfields);
selected_bitfields.sort_by_key(|bitfield| bitfield.validator_index());
assert_eq!(selected_bitfields.len(), 2);
assert_eq!(selected_bitfields[0], bitfields[0]);
// we don't know which of the (otherwise equal) bitfields will be selected
assert!(selected_bitfields[1] == bitfields[1] || selected_bitfields[1] == bitfields[2]);
}
#[test]
fn each_corresponds_to_an_occupied_core() {
let keystore: SyncCryptoStorePtr = Arc::new(KeyStore::new());
let bitvec = default_bitvec(3);
// invalid: bit on free core
let mut bitvec0 = bitvec.clone();
bitvec0.set(0, true);
// invalid: bit on scheduled core
let mut bitvec1 = bitvec.clone();
bitvec1.set(1, true);
// valid: bit on occupied core.
let mut bitvec2 = bitvec.clone();
bitvec2.set(2, true);
let cores =
vec![CoreState::Free, CoreState::Scheduled(Default::default()), occupied_core(2)];
let bitfields = vec![
block_on(signed_bitfield(&keystore, bitvec0, ValidatorIndex(0))),
block_on(signed_bitfield(&keystore, bitvec1, ValidatorIndex(1))),
block_on(signed_bitfield(&keystore, bitvec2.clone(), ValidatorIndex(2))),
];
let selected_bitfields = select_availability_bitfields(&cores, &bitfields);
// selects only the valid bitfield
assert_eq!(selected_bitfields.len(), 1);
assert_eq!(selected_bitfields[0].payload().0, bitvec2);
}
#[test]
fn more_set_bits_win_conflicts() {
let keystore: SyncCryptoStorePtr = Arc::new(KeyStore::new());
let mut bitvec = default_bitvec(2);
bitvec.set(0, true);
let mut bitvec1 = bitvec.clone();
bitvec1.set(1, true);
let cores = vec![occupied_core(0), occupied_core(1)];
let bitfields = vec![
block_on(signed_bitfield(&keystore, bitvec, ValidatorIndex(1))),
block_on(signed_bitfield(&keystore, bitvec1.clone(), ValidatorIndex(1))),
];
let selected_bitfields = select_availability_bitfields(&cores, &bitfields);
assert_eq!(selected_bitfields.len(), 1);
assert_eq!(selected_bitfields[0].payload().0, bitvec1.clone());
}
#[test]
fn more_complex_bitfields() {
let keystore: SyncCryptoStorePtr = Arc::new(KeyStore::new());
let cores = vec![occupied_core(0), occupied_core(1), occupied_core(2), occupied_core(3)];
let mut bitvec0 = default_bitvec(4);
bitvec0.set(0, true);
bitvec0.set(2, true);
let mut bitvec1 = default_bitvec(4);
bitvec1.set(1, true);
let mut bitvec2 = default_bitvec(4);
bitvec2.set(2, true);
let mut bitvec3 = default_bitvec(4);
bitvec3.set(0, true);
bitvec3.set(1, true);
bitvec3.set(2, true);
bitvec3.set(3, true);
// these are out of order but will be selected in order. The better
// bitfield for 3 will be selected.
let bitfields = vec![
block_on(signed_bitfield(&keystore, bitvec2.clone(), ValidatorIndex(3))),
block_on(signed_bitfield(&keystore, bitvec3.clone(), ValidatorIndex(3))),
block_on(signed_bitfield(&keystore, bitvec0.clone(), ValidatorIndex(0))),
block_on(signed_bitfield(&keystore, bitvec2.clone(), ValidatorIndex(2))),
block_on(signed_bitfield(&keystore, bitvec1.clone(), ValidatorIndex(1))),
];
let selected_bitfields = select_availability_bitfields(&cores, &bitfields);
assert_eq!(selected_bitfields.len(), 4);
assert_eq!(selected_bitfields[0].payload().0, bitvec0);
assert_eq!(selected_bitfields[1].payload().0, bitvec1);
assert_eq!(selected_bitfields[2].payload().0, bitvec2);
assert_eq!(selected_bitfields[3].payload().0, bitvec3);
}
}
mod select_candidates {
use super::{super::*, build_occupied_core, default_bitvec, occupied_core, scheduled_core};
use polkadot_node_subsystem::messages::{
AllMessages, RuntimeApiMessage,
RuntimeApiRequest::{
AvailabilityCores, PersistedValidationData as PersistedValidationDataReq,
},
};
use polkadot_node_subsystem_test_helpers::TestSubsystemSender;
use polkadot_primitives::v1::{
BlockNumber, CandidateCommitments, CandidateDescriptor, CommittedCandidateReceipt,
PersistedValidationData,
};
const BLOCK_UNDER_PRODUCTION: BlockNumber = 128;
fn test_harness<OverseerFactory, Overseer, TestFactory, Test>(
overseer_factory: OverseerFactory,
test_factory: TestFactory,
) where
OverseerFactory: FnOnce(mpsc::UnboundedReceiver<AllMessages>) -> Overseer,
Overseer: Future<Output = ()>,
TestFactory: FnOnce(TestSubsystemSender) -> Test,
Test: Future<Output = ()>,
{
let (tx, rx) = polkadot_node_subsystem_test_helpers::sender_receiver();
let overseer = overseer_factory(rx);
let test = test_factory(tx);
futures::pin_mut!(overseer, test);
let _ = futures::executor::block_on(future::join(overseer, test));
}
// For test purposes, we always return this set of availability cores:
//
// [
// 0: Free,
// 1: Scheduled(default),
// 2: Occupied(no next_up set),
// 3: Occupied(next_up_on_available set but not available),
// 4: Occupied(next_up_on_available set and available),
// 5: Occupied(next_up_on_time_out set but not timeout),
// 6: Occupied(next_up_on_time_out set and timeout but available),
// 7: Occupied(next_up_on_time_out set and timeout and not available),
// 8: Occupied(both next_up set, available),
// 9: Occupied(both next_up set, not available, no timeout),
// 10: Occupied(both next_up set, not available, timeout),
// 11: Occupied(next_up_on_available and available, but different successor para_id)
// ]
fn mock_availability_cores() -> Vec<CoreState> {
use std::ops::Not;
use CoreState::{Free, Scheduled};
vec![
// 0: Free,
Free,
// 1: Scheduled(default),
Scheduled(scheduled_core(1)),
// 2: Occupied(no next_up set),
occupied_core(2),
// 3: Occupied(next_up_on_available set but not available),
build_occupied_core(3, |core| {
core.next_up_on_available = Some(scheduled_core(3));
}),
// 4: Occupied(next_up_on_available set and available),
build_occupied_core(4, |core| {
core.next_up_on_available = Some(scheduled_core(4));
core.availability = core.availability.clone().not();
}),
// 5: Occupied(next_up_on_time_out set but not timeout),
build_occupied_core(5, |core| {
core.next_up_on_time_out = Some(scheduled_core(5));
}),
// 6: Occupied(next_up_on_time_out set and timeout but available),
build_occupied_core(6, |core| {
core.next_up_on_time_out = Some(scheduled_core(6));
core.time_out_at = BLOCK_UNDER_PRODUCTION;
core.availability = core.availability.clone().not();
}),
// 7: Occupied(next_up_on_time_out set and timeout and not available),
build_occupied_core(7, |core| {
core.next_up_on_time_out = Some(scheduled_core(7));
core.time_out_at = BLOCK_UNDER_PRODUCTION;
}),
// 8: Occupied(both next_up set, available),
build_occupied_core(8, |core| {
core.next_up_on_available = Some(scheduled_core(8));
core.next_up_on_time_out = Some(scheduled_core(8));
core.availability = core.availability.clone().not();
}),
// 9: Occupied(both next_up set, not available, no timeout),
build_occupied_core(9, |core| {
core.next_up_on_available = Some(scheduled_core(9));
core.next_up_on_time_out = Some(scheduled_core(9));
}),
// 10: Occupied(both next_up set, not available, timeout),
build_occupied_core(10, |core| {
core.next_up_on_available = Some(scheduled_core(10));
core.next_up_on_time_out = Some(scheduled_core(10));
core.time_out_at = BLOCK_UNDER_PRODUCTION;
}),
// 11: Occupied(next_up_on_available and available, but different successor para_id)
build_occupied_core(11, |core| {
core.next_up_on_available = Some(scheduled_core(12));
core.availability = core.availability.clone().not();
}),
]
}
async fn mock_overseer(
mut receiver: mpsc::UnboundedReceiver<AllMessages>,
expected: Vec<BackedCandidate>,
) {
use ChainApiMessage::BlockNumber;
use RuntimeApiMessage::Request;
while let Some(from_job) = receiver.next().await {
match from_job {
AllMessages::ChainApi(BlockNumber(_relay_parent, tx)) =>
tx.send(Ok(Some(BLOCK_UNDER_PRODUCTION - 1))).unwrap(),
AllMessages::RuntimeApi(Request(
_parent_hash,
PersistedValidationDataReq(_para_id, _assumption, tx),
)) => tx.send(Ok(Some(Default::default()))).unwrap(),
AllMessages::RuntimeApi(Request(_parent_hash, AvailabilityCores(tx))) =>
tx.send(Ok(mock_availability_cores())).unwrap(),
AllMessages::CandidateBacking(CandidateBackingMessage::GetBackedCandidates(
_,
_,
sender,
)) => {
let _ = sender.send(expected.clone());
},
_ => panic!("Unexpected message: {:?}", from_job),
}
async fn mock_overseer(
mut receiver: mpsc::UnboundedReceiver<AllMessages>,
expected: Vec<BackedCandidate>,
) {
while let Some(from_job) = receiver.next().await {
match from_job {
AllMessages::CandidateBacking(CandidateBackingMessage::GetBackedCandidates(
_,
_,
sender,
)) => {
let _ = sender.send(expected.clone());
},
_ => panic!("Unexpected message: {:?}", from_job),
}
}
#[test]
fn can_succeed() {
test_harness(
|r| mock_overseer(r, Vec::new()),
|mut tx: TestSubsystemSender| async move {
select_candidates(&[], &[], &[], Default::default(), &mut tx).await.unwrap();
},
)
}
// this tests that only the appropriate candidates get selected.
// To accomplish this, we supply a candidate list containing one candidate per possible core;
// the candidate selection algorithm must filter them to the appropriate set
#[test]
fn selects_correct_candidates() {
let mock_cores = mock_availability_cores();
let n_cores = mock_cores.len();
let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
let candidate_template = CandidateReceipt {
descriptor: CandidateDescriptor {
persisted_validation_data_hash: empty_hash,
..Default::default()
},
commitments_hash: CandidateCommitments::default().hash(),
};
let candidates: Vec<_> = std::iter::repeat(candidate_template)
.take(mock_cores.len())
.enumerate()
.map(|(idx, mut candidate)| {
candidate.descriptor.para_id = idx.into();
candidate
})
.cycle()
.take(mock_cores.len() * 3)
.enumerate()
.map(|(idx, mut candidate)| {
if idx < mock_cores.len() {
// first go-around: use candidates which should work
candidate
} else if idx < mock_cores.len() * 2 {
// for the second repetition of the candidates, give them the wrong hash
candidate.descriptor.persisted_validation_data_hash = Default::default();
candidate
} else {
// third go-around: right hash, wrong para_id
candidate.descriptor.para_id = idx.into();
candidate
}
})
.collect();
// why those particular indices? see the comments on mock_availability_cores()
let expected_candidates: Vec<_> =
[1, 4, 7, 8, 10].iter().map(|&idx| candidates[idx].clone()).collect();
let expected_backed = expected_candidates
.iter()
.map(|c| BackedCandidate {
candidate: CommittedCandidateReceipt {
descriptor: c.descriptor.clone(),
..Default::default()
},
validity_votes: Vec::new(),
validator_indices: default_bitvec(n_cores),
})
.collect();
test_harness(
|r| mock_overseer(r, expected_backed),
|mut tx: TestSubsystemSender| async move {
let result =
select_candidates(&mock_cores, &[], &candidates, Default::default(), &mut tx)
.await
.unwrap();
result.into_iter().for_each(|c| {
assert!(
expected_candidates.iter().any(|c2| c.candidate.corresponds_to(c2)),
"Failed to find candidate: {:?}",
c,
)
});
},
)
}
#[test]
fn selects_max_one_code_upgrade() {
let mock_cores = mock_availability_cores();
let n_cores = mock_cores.len();
let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
// why those particular indices? see the comments on mock_availability_cores()
// the first candidate with code is included out of [1, 4, 7, 8, 10].
let cores = [1, 7, 10];
let cores_with_code = [1, 4, 8];
let committed_receipts: Vec<_> = (0..mock_cores.len())
.map(|i| CommittedCandidateReceipt {
descriptor: CandidateDescriptor {
para_id: i.into(),
persisted_validation_data_hash: empty_hash,
..Default::default()
},
commitments: CandidateCommitments {
new_validation_code: if cores_with_code.contains(&i) {
Some(vec![].into())
} else {
None
},
..Default::default()
},
..Default::default()
})
.collect();
let candidates: Vec<_> = committed_receipts.iter().map(|r| r.to_plain()).collect();
let expected_candidates: Vec<_> =
cores.iter().map(|&idx| candidates[idx].clone()).collect();
let expected_backed: Vec<_> = cores
.iter()
.map(|&idx| BackedCandidate {
candidate: committed_receipts[idx].clone(),
validity_votes: Vec::new(),
validator_indices: default_bitvec(n_cores),
})
.collect();
test_harness(
|r| mock_overseer(r, expected_backed),
|mut tx: TestSubsystemSender| async move {
let result =
select_candidates(&mock_cores, &[], &candidates, Default::default(), &mut tx)
.await
.unwrap();
result.into_iter().for_each(|c| {
assert!(
expected_candidates.iter().any(|c2| c.candidate.corresponds_to(c2)),
"Failed to find candidate: {:?}",
c,
)
});
},
)
}
}
#[test]
fn can_succeed() {
test_harness(
|r| mock_overseer(r, Vec::new()),
|mut tx: TestSubsystemSender| async move {
collect_backed_candidates(Vec::new(), Default::default(), &mut tx)
.await
.unwrap();
},
)
}
// this tests that only the appropriate candidates get selected.
// To accomplish this, we supply a candidate list containing one candidate per possible core;
// the candidate selection algorithm must filter them to the appropriate set
#[test]
fn selects_correct_candidates() {
let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
let candidate_template = CandidateReceipt {
descriptor: CandidateDescriptor {
persisted_validation_data_hash: empty_hash,
..Default::default()
},
commitments_hash: CandidateCommitments::default().hash(),
};
let n_cores = 5;
let candidate_receipts: Vec<_> = std::iter::repeat(candidate_template)
.take(n_cores)
.enumerate()
.map(|(idx, mut candidate)| {
candidate.descriptor.para_id = idx.into();
candidate
})
.cycle()
.take(n_cores * 4)
.enumerate()
.map(|(idx, mut candidate_receipt)| {
if idx < n_cores {
// first go-around: use candidates which should work
candidate_receipt
} else if idx < n_cores * 2 {
// for the second repetition of the candidates, give them the wrong hash
candidate_receipt.descriptor.persisted_validation_data_hash = Default::default();
candidate_receipt
} else if idx < n_cores * 3 {
// third go-around: right hash, wrong para_id
candidate_receipt.descriptor.para_id = idx.into();
candidate_receipt
} else {
// fourth go-around: wrong relay parent, this is the only thing that is checked
candidate_receipt.descriptor.relay_parent = Hash::repeat_byte(0xFF);
candidate_receipt
}
})
.collect();
// candidates now contains 1/3 valid canidates, and 2/3 invalid
// but we don't check them in them here, so they should be passed alright
let expected_candidate_receipts =
candidate_receipts.iter().take(n_cores * 3).cloned().collect::<Vec<_>>();
let expected_backed = expected_candidate_receipts
.iter()
.map(|candidate_receipt| BackedCandidate {
candidate: CommittedCandidateReceipt {
descriptor: candidate_receipt.descriptor.clone(),
..Default::default()
},
validity_votes: Vec::new(),
validator_indices: default_bitvec(n_cores),
})
.collect();
test_harness(
|r| mock_overseer(r, expected_backed),
|mut tx: TestSubsystemSender| async move {
let result = collect_backed_candidates(candidate_receipts, Default::default(), &mut tx)
.await
.unwrap();
result.into_iter().for_each(|c| {
assert!(
expected_candidate_receipts.iter().any(|c2| c.candidate.corresponds_to(c2)),
"Failed to find candidate: {:?}",
c,
)
});
},
)
}