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

Remove NetworkSpecialization (#4665)

Browse Source 
* remove networkspecialization

* Fix most of the fallout

* get all tests compiling

Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>
This commit is contained in:
Robert Habermeier
2020-02-21 05:02:12 -08:00
committed by GitHub
parent e8000e7429
commit 0090fe979b
20 changed files with 136 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/client/network/src/service.rs
+32 -52
View File
@@ -45,9 +45,8 @@ use crate::{transport, config::NonReservedPeerMode, ReputationChange};
use crate::config::{Params, TransportConfig};
use crate::error::Error;
use crate::network_state::{NetworkState, NotConnectedPeer as NetworkStateNotConnectedPeer, Peer as NetworkStatePeer};
use crate::protocol::{self, Protocol, Context, PeerInfo};
use crate::protocol::{self, Protocol, PeerInfo};
use crate::protocol::{event::Event, light_dispatch::{AlwaysBadChecker, RequestData}};
use crate::protocol::specialization::NetworkSpecialization;
use crate::protocol::sync::SyncState;
/// Minimum Requirements for a Hash within Networking
@@ -101,7 +100,7 @@ impl ReportHandle {
}
/// Substrate network service. Handles network IO and manages connectivity.
pub struct NetworkService<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> {
pub struct NetworkService<B: BlockT + 'static, H: ExHashT> {
/// Number of peers we're connected to.
num_connected: Arc<AtomicUsize>,
/// The local external addresses.
@@ -116,19 +115,19 @@ pub struct NetworkService<B: BlockT + 'static, S: NetworkSpecialization<B>, H: E
/// nodes it should be connected to or not.
peerset: PeersetHandle,
/// Channel that sends messages to the actual worker.
to_worker: mpsc::UnboundedSender<ServiceToWorkerMsg<B, H, S>>,
to_worker: mpsc::UnboundedSender<ServiceToWorkerMsg<B, H>>,
/// Marker to pin the `H` generic. Serves no purpose except to not break backwards
/// compatibility.
_marker: PhantomData<H>,
}
impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker<B, S, H> {
impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
/// Creates the network service.
///
/// Returns a `NetworkWorker` that implements `Future` and must be regularly polled in order
/// for the network processing to advance. From it, you can extract a `NetworkService` using
/// `worker.service()`. The `NetworkService` can be shared through the codebase.
pub fn new(params: Params<B, S, H>) -> Result<NetworkWorker<B, S, H>, Error> {
pub fn new(params: Params<B, H>) -> Result<NetworkWorker<B, H>, Error> {
let (to_worker, from_worker) = mpsc::unbounded();
if let Some(ref path) = params.network_config.net_config_path {
@@ -205,7 +204,6 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
},
params.chain.clone(),
checker.clone(),
params.specialization,
params.transaction_pool,
params.finality_proof_provider.clone(),
params.finality_proof_request_builder,
@@ -215,7 +213,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
)?;
// Build the swarm.
let (mut swarm, bandwidth): (Swarm::<B, S, H>, _) = {
let (mut swarm, bandwidth): (Swarm::<B, H>, _) = {
let user_agent = format!(
"{} ({})",
params.network_config.client_version,
@@ -263,14 +261,14 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
// Listen on multiaddresses.
for addr in &params.network_config.listen_addresses {
if let Err(err) = Swarm::<B, S, H>::listen_on(&mut swarm, addr.clone()) {
if let Err(err) = Swarm::<B, H>::listen_on(&mut swarm, addr.clone()) {
warn!(target: "sub-libp2p", "Can't listen on {} because: {:?}", addr, err)
}
}
// Add external addresses.
for addr in &params.network_config.public_addresses {
Swarm::<B, S, H>::add_external_address(&mut swarm, addr.clone());
Swarm::<B, H>::add_external_address(&mut swarm, addr.clone());
}
let external_addresses = Arc::new(Mutex::new(Vec::new()));
@@ -351,13 +349,13 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
/// Return a `NetworkService` that can be shared through the code base and can be used to
/// manipulate the worker.
pub fn service(&self) -> &Arc<NetworkService<B, S, H>> {
pub fn service(&self) -> &Arc<NetworkService<B, H>> {
&self.service
}
/// You must call this when a new block is imported by the client.
pub fn on_block_imported(&mut self, hash: B::Hash, header: B::Header, data: Vec<u8>, is_best: bool) {
self.network_service.user_protocol_mut().on_block_imported(hash, &header, data, is_best);
pub fn on_block_imported(&mut self, header: B::Header, data: Vec<u8>, is_best: bool) {
self.network_service.user_protocol_mut().on_block_imported(&header, data, is_best);
}
/// You must call this when a new block is finalized by the client.
@@ -415,9 +413,9 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
};
NetworkState {
peer_id: Swarm::<B, S, H>::local_peer_id(&swarm).to_base58(),
listened_addresses: Swarm::<B, S, H>::listeners(&swarm).cloned().collect(),
external_addresses: Swarm::<B, S, H>::external_addresses(&swarm).cloned().collect(),
peer_id: Swarm::<B, H>::local_peer_id(&swarm).to_base58(),
listened_addresses: Swarm::<B, H>::listeners(&swarm).cloned().collect(),
external_addresses: Swarm::<B, H>::external_addresses(&swarm).cloned().collect(),
average_download_per_sec: self.service.bandwidth.average_download_per_sec(),
average_upload_per_sec: self.service.bandwidth.average_upload_per_sec(),
connected_peers,
@@ -446,7 +444,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkWorker
}
}
impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkService<B, S, H> {
impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
/// Writes a message on an open notifications channel. Has no effect if the notifications
/// channel with this protocol name is closed.
///
@@ -545,15 +543,6 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkServic
.unbounded_send(ServiceToWorkerMsg::RequestJustification(hash.clone(), number));
}
/// Execute a closure with the chain-specific network specialization.
pub fn with_spec<F>(&self, f: F)
where F: FnOnce(&mut S, &mut dyn Context<B>) + Send + 'static
{
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::ExecuteWithSpec(Box::new(f)));
}
/// Are we in the process of downloading the chain?
pub fn is_major_syncing(&self) -> bool {
self.is_major_syncing.load(Ordering::Relaxed)
@@ -641,8 +630,8 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> NetworkServic
}
}
impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> sp_consensus::SyncOracle
for NetworkService<B, S, H>
impl<B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle
for NetworkService<B, H>
{
fn is_major_syncing(&mut self) -> bool {
NetworkService::is_major_syncing(self)
@@ -653,8 +642,8 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> sp_consensus:
}
}
impl<'a, B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> sp_consensus::SyncOracle
for &'a NetworkService<B, S, H>
impl<'a, B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle
for &'a NetworkService<B, H>
{
fn is_major_syncing(&mut self) -> bool {
NetworkService::is_major_syncing(self)
@@ -674,10 +663,9 @@ pub trait NetworkStateInfo {
fn local_peer_id(&self) -> PeerId;
}
impl<B, S, H> NetworkStateInfo for NetworkService<B, S, H>
impl<B, H> NetworkStateInfo for NetworkService<B, H>
where
B: sp_runtime::traits::Block,
S: NetworkSpecialization<B>,
H: ExHashT,
{
/// Returns the local external addresses.
@@ -694,12 +682,11 @@ impl<B, S, H> NetworkStateInfo for NetworkService<B, S, H>
/// Messages sent from the `NetworkService` to the `NetworkWorker`.
///
/// Each entry corresponds to a method of `NetworkService`.
enum ServiceToWorkerMsg<B: BlockT, H: ExHashT, S: NetworkSpecialization<B>> {
enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
PropagateExtrinsic(H),
PropagateExtrinsics,
RequestJustification(B::Hash, NumberFor<B>),
AnnounceBlock(B::Hash, Vec<u8>),
ExecuteWithSpec(Box<dyn FnOnce(&mut S, &mut dyn Context<B>) + Send>),
GetValue(record::Key),
PutValue(record::Key, Vec<u8>),
AddKnownAddress(PeerId, Multiaddr),
@@ -721,7 +708,7 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT, S: NetworkSpecialization<B>> {
///
/// You are encouraged to poll this in a separate background thread or task.
#[must_use = "The NetworkWorker must be polled in order for the network to work"]
pub struct NetworkWorker<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> {
pub struct NetworkWorker<B: BlockT + 'static, H: ExHashT> {
/// Updated by the `NetworkWorker` and loaded by the `NetworkService`.
external_addresses: Arc<Mutex<Vec<Multiaddr>>>,
/// Updated by the `NetworkWorker` and loaded by the `NetworkService`.
@@ -729,20 +716,20 @@ pub struct NetworkWorker<B: BlockT + 'static, S: NetworkSpecialization<B>, H: Ex
/// Updated by the `NetworkWorker` and loaded by the `NetworkService`.
is_major_syncing: Arc<AtomicBool>,
/// The network service that can be extracted and shared through the codebase.
service: Arc<NetworkService<B, S, H>>,
service: Arc<NetworkService<B, H>>,
/// The *actual* network.
network_service: Swarm<B, S, H>,
network_service: Swarm<B, H>,
/// The import queue that was passed as initialization.
import_queue: Box<dyn ImportQueue<B>>,
/// Messages from the `NetworkService` and that must be processed.
from_worker: mpsc::UnboundedReceiver<ServiceToWorkerMsg<B, H, S>>,
from_worker: mpsc::UnboundedReceiver<ServiceToWorkerMsg<B, H>>,
/// Receiver for queries from the light client that must be processed.
light_client_rqs: Option<mpsc::UnboundedReceiver<RequestData<B>>>,
/// Senders for events that happen on the network.
event_streams: Vec<mpsc::UnboundedSender<Event>>,
}
impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for NetworkWorker<B, S, H> {
impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
type Output = Result<(), io::Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context) -> Poll<Self::Output> {
@@ -769,11 +756,6 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for Ne
};
match msg {
ServiceToWorkerMsg::ExecuteWithSpec(task) => {
let protocol = this.network_service.user_protocol_mut();
let (mut context, spec) = protocol.specialization_lock();
task(spec, &mut context);
},
ServiceToWorkerMsg::AnnounceBlock(hash, data) =>
this.network_service.user_protocol_mut().announce_block(hash, data),
ServiceToWorkerMsg::RequestJustification(hash, number) =>
@@ -839,7 +821,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for Ne
// Update the variables shared with the `NetworkService`.
this.num_connected.store(this.network_service.user_protocol_mut().num_connected_peers(), Ordering::Relaxed);
{
let external_addresses = Swarm::<B, S, H>::external_addresses(&this.network_service).cloned().collect();
let external_addresses = Swarm::<B, H>::external_addresses(&this.network_service).cloned().collect();
*this.external_addresses.lock() = external_addresses;
}
this.is_major_syncing.store(match this.network_service.user_protocol_mut().sync_state() {
@@ -851,20 +833,18 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for Ne
}
}
impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Unpin for NetworkWorker<B, S, H> {
impl<B: BlockT + 'static, H: ExHashT> Unpin for NetworkWorker<B, H> {
}
/// The libp2p swarm, customized for our needs.
type Swarm<B, S, H> = libp2p::swarm::Swarm<
Behaviour<B, S, H>
>;
type Swarm<B, H> = libp2p::swarm::Swarm<Behaviour<B, H>>;
// Implementation of `import_queue::Link` trait using the available local variables.
struct NetworkLink<'a, B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> {
protocol: &'a mut Swarm<B, S, H>,
struct NetworkLink<'a, B: BlockT, H: ExHashT> {
protocol: &'a mut Swarm<B, H>,
}
impl<'a, B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Link<B> for NetworkLink<'a, B, S, H> {
impl<'a, B: BlockT, H: ExHashT> Link<B> for NetworkLink<'a, B, H> {
fn blocks_processed(
&mut self,
imported: usize,