Skip to content

Commit

Permalink
Modular block request handler (#1524)
Browse files Browse the repository at this point in the history
Submit the outstanding PRs from the old repos(these were already
reviewed and approved before the repo rorg, but not yet submitted):
Main PR: paritytech/substrate#14014
Companion PRs: paritytech/polkadot#7134,
paritytech/cumulus#2489

The changes in the PR:
1. ChainSync currently calls into the block request handler directly.
Instead, move the block request handler behind a trait. This allows new
protocols to be plugged into ChainSync.
2. BuildNetworkParams is changed so that custom relay protocol
implementations can be (optionally) passed in during network creation
time. If custom protocol is not specified, it defaults to the existing
block handler
3. BlockServer and BlockDownloader traits are introduced for the
protocol implementation. The existing block handler has been changed to
implement these traits
4. Other changes:
[X] Make TxHash serializable. This is needed for exchanging the
serialized hash in the relay protocol messages
[X] Clean up types no longer used(OpaqueBlockRequest,
OpaqueBlockResponse)

---------

Co-authored-by: Dmitry Markin <dmitry@markin.tech>
Co-authored-by: command-bot <>
  • Loading branch information
rahulksnv and dmitry-markin authored Sep 15, 2023
1 parent c6df364 commit b35b28c
Show file tree
Hide file tree
Showing 14 changed files with 370 additions and 256 deletions.
1 change: 1 addition & 0 deletions cumulus/client/service/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,7 @@ where
import_queue,
block_announce_validator_builder: Some(Box::new(move |_| block_announce_validator)),
warp_sync_params,
block_relay: None,
})
}

Expand Down
1 change: 1 addition & 0 deletions polkadot/node/service/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -894,6 +894,7 @@ pub fn new_full<OverseerGenerator: OverseerGen>(
import_queue,
block_announce_validator_builder: None,
warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)),
block_relay: None,
})?;

if config.offchain_worker.enabled {
Expand Down
1 change: 1 addition & 0 deletions substrate/bin/node-template/node/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,7 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
import_queue,
block_announce_validator_builder: None,
warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)),
block_relay: None,
})?;

if config.offchain_worker.enabled {
Expand Down
1 change: 1 addition & 0 deletions substrate/bin/node/cli/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -388,6 +388,7 @@ pub fn new_full_base(
import_queue,
block_announce_validator_builder: None,
warp_sync_params: Some(WarpSyncParams::WithProvider(warp_sync)),
block_relay: None,
})?;

let role = config.role.clone();
Expand Down
33 changes: 2 additions & 31 deletions substrate/client/network/common/src/sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,12 @@ pub mod message;
pub mod metrics;
pub mod warp;

use crate::{role::Roles, sync::message::BlockAnnounce, types::ReputationChange};
use crate::{role::Roles, types::ReputationChange};
use futures::Stream;

use libp2p_identity::PeerId;

use message::{BlockData, BlockRequest, BlockResponse};
use message::{BlockAnnounce, BlockRequest, BlockResponse};
use sc_consensus::{import_queue::RuntimeOrigin, IncomingBlock};
use sp_consensus::BlockOrigin;
use sp_runtime::{
Expand Down Expand Up @@ -226,28 +226,6 @@ impl fmt::Debug for OpaqueStateResponse {
}
}

/// Wrapper for implementation-specific block request.
///
/// NOTE: Implementation must be able to encode and decode it for network purposes.
pub struct OpaqueBlockRequest(pub Box<dyn Any + Send>);

impl fmt::Debug for OpaqueBlockRequest {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
f.debug_struct("OpaqueBlockRequest").finish()
}
}

/// Wrapper for implementation-specific block response.
///
/// NOTE: Implementation must be able to encode and decode it for network purposes.
pub struct OpaqueBlockResponse(pub Box<dyn Any + Send>);

impl fmt::Debug for OpaqueBlockResponse {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
f.debug_struct("OpaqueBlockResponse").finish()
}
}

/// Provides high-level status of syncing.
#[async_trait::async_trait]
pub trait SyncStatusProvider<Block: BlockT>: Send + Sync {
Expand Down Expand Up @@ -392,13 +370,6 @@ pub trait ChainSync<Block: BlockT>: Send {
/// Return some key metrics.
fn metrics(&self) -> Metrics;

/// Access blocks from implementation-specific block response.
fn block_response_into_blocks(
&self,
request: &BlockRequest<Block>,
response: OpaqueBlockResponse,
) -> Result<Vec<BlockData<Block>>, String>;

/// Advance the state of `ChainSync`
fn poll(&mut self, cx: &mut std::task::Context) -> Poll<()>;

Expand Down
72 changes: 72 additions & 0 deletions substrate/client/network/sync/src/block_relay_protocol.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
// Copyright Parity Technologies (UK) Ltd.
// This file is part of Substrate.

// Substrate 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.

// Substrate 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 Substrate. If not, see <http://www.gnu.org/licenses/>.

//! Block relay protocol related definitions.
use futures::channel::oneshot;
use libp2p::PeerId;
use sc_network::request_responses::{ProtocolConfig, RequestFailure};
use sc_network_common::sync::message::{BlockData, BlockRequest};
use sp_runtime::traits::Block as BlockT;
use std::sync::Arc;

/// The serving side of the block relay protocol. It runs a single instance
/// of the server task that processes the incoming protocol messages.
#[async_trait::async_trait]
pub trait BlockServer<Block: BlockT>: Send {
/// Starts the protocol processing.
async fn run(&mut self);
}

/// The client side stub to download blocks from peers. This is a handle
/// that can be used to initiate concurrent downloads.
#[async_trait::async_trait]
pub trait BlockDownloader<Block: BlockT>: Send + Sync {
/// Performs the protocol specific sequence to fetch the blocks from the peer.
/// Output: if the download succeeds, the response is a `Vec<u8>` which is
/// in a format specific to the protocol implementation. The block data
/// can be extracted from this response using [`BlockDownloader::block_response_into_blocks`].
async fn download_blocks(
&self,
who: PeerId,
request: BlockRequest<Block>,
) -> Result<Result<Vec<u8>, RequestFailure>, oneshot::Canceled>;

/// Parses the protocol specific response to retrieve the block data.
fn block_response_into_blocks(
&self,
request: &BlockRequest<Block>,
response: Vec<u8>,
) -> Result<Vec<BlockData<Block>>, BlockResponseError>;
}

/// Errors returned by [`BlockDownloader::block_response_into_blocks`].
#[derive(Debug)]
pub enum BlockResponseError {
/// Failed to decode the response bytes.
DecodeFailed(String),

/// Failed to extract the blocks from the decoded bytes.
ExtractionFailed(String),
}

/// Block relay specific params for network creation, specified in
/// ['sc_service::BuildNetworkParams'].
pub struct BlockRelayParams<Block: BlockT> {
pub server: Box<dyn BlockServer<Block>>,
pub downloader: Arc<dyn BlockDownloader<Block>>,
pub request_response_config: ProtocolConfig,
}
Loading

0 comments on commit b35b28c

Please sign in to comment.