Skip to content
This repository has been archived by the owner on Nov 15, 2023. It is now read-only.

Commit

Permalink
Merge branch 'master' into ao-update-deps
Browse files Browse the repository at this point in the history
* master:
  babe: expose next epoch data (#7829)
  Fix master build (#7837)
  Improve spans of pallet macro (#7830)
  fix : remove `_{ }` syntax from benchmark macro (#7822)
  client/network: Use request response for block requests (#7478)
  • Loading branch information
ordian committed Jan 6, 2021
2 parents d245f58 + 8927543 commit 2451372
Show file tree
Hide file tree
Showing 81 changed files with 1,243 additions and 1,632 deletions.
14 changes: 7 additions & 7 deletions .maintain/sentry-node/docker-compose.yml
Expand Up @@ -47,9 +47,9 @@ services:
- "--validator"
- "--alice"
- "--sentry-nodes"
- "/dns/sentry-a/tcp/30333/p2p/QmV7EhW6J6KgmNdr558RH1mPx2xGGznW7At4BhXzntRFsi"
- "/dns/sentry-a/tcp/30333/p2p/12D3KooWSCufgHzV4fCwRijfH2k3abrpAJxTKxEvN1FDuRXA2U9x"
- "--reserved-nodes"
- "/dns/sentry-a/tcp/30333/p2p/QmV7EhW6J6KgmNdr558RH1mPx2xGGznW7At4BhXzntRFsi"
- "/dns/sentry-a/tcp/30333/p2p/12D3KooWSCufgHzV4fCwRijfH2k3abrpAJxTKxEvN1FDuRXA2U9x"
# Not only bind to localhost.
- "--unsafe-ws-external"
- "--unsafe-rpc-external"
Expand Down Expand Up @@ -83,11 +83,11 @@ services:
- "--port"
- "30333"
- "--sentry"
- "/dns/validator-a/tcp/30333/p2p/QmRpheLN4JWdAnY7HGJfWFNbfkQCb6tFf4vvA6hgjMZKrR"
- "/dns/validator-a/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp"
- "--reserved-nodes"
- "/dns/validator-a/tcp/30333/p2p/QmRpheLN4JWdAnY7HGJfWFNbfkQCb6tFf4vvA6hgjMZKrR"
- "/dns/validator-a/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp"
- "--bootnodes"
- "/dns/validator-b/tcp/30333/p2p/QmSVnNf9HwVMT1Y4cK1P6aoJcEZjmoTXpjKBmAABLMnZEk"
- "/dns/validator-b/tcp/30333/p2p/12D3KooWHdiAxVd8uMQR1hGWXccidmfCwLqcMpGwR6QcTP6QRMuD"
- "--no-telemetry"
- "--rpc-cors"
- "all"
Expand Down Expand Up @@ -118,9 +118,9 @@ services:
- "--validator"
- "--bob"
- "--bootnodes"
- "/dns/validator-a/tcp/30333/p2p/QmRpheLN4JWdAnY7HGJfWFNbfkQCb6tFf4vvA6hgjMZKrR"
- "/dns/validator-a/tcp/30333/p2p/12D3KooWEyoppNCUx8Yx66oV9fJnriXwCcXwDDUA2kj6vnc6iDEp"
- "--bootnodes"
- "/dns/sentry-a/tcp/30333/p2p/QmV7EhW6J6KgmNdr558RH1mPx2xGGznW7At4BhXzntRFsi"
- "/dns/sentry-a/tcp/30333/p2p/12D3KooWSCufgHzV4fCwRijfH2k3abrpAJxTKxEvN1FDuRXA2U9x"
- "--no-telemetry"
- "--rpc-cors"
- "all"
Expand Down
2 changes: 2 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 4 additions & 0 deletions bin/node/runtime/src/lib.rs
Expand Up @@ -1180,6 +1180,10 @@ impl_runtime_apis! {
Babe::current_epoch()
}

fn next_epoch() -> sp_consensus_babe::Epoch {
Babe::next_epoch()
}

fn generate_key_ownership_proof(
_slot_number: sp_consensus_babe::SlotNumber,
authority_id: sp_consensus_babe::AuthorityId,
Expand Down
153 changes: 53 additions & 100 deletions client/network/src/behaviour.rs
Expand Up @@ -17,20 +17,22 @@
// along with this program. If not, see <https://www.gnu.org/licenses/>.

use crate::{
config::{ProtocolId, Role}, block_requests, light_client_handler,
peer_info, request_responses, discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
config::{ProtocolId, Role}, light_client_handler, peer_info, request_responses,
discovery::{DiscoveryBehaviour, DiscoveryConfig, DiscoveryOut},
protocol::{message::Roles, CustomMessageOutcome, NotificationsSink, Protocol},
ObservedRole, DhtEvent, ExHashT,
};

use bytes::Bytes;
use codec::Encode as _;
use futures::channel::oneshot;
use libp2p::NetworkBehaviour;
use libp2p::core::{Multiaddr, PeerId, PublicKey};
use libp2p::identify::IdentifyInfo;
use libp2p::kad::record;
use libp2p::swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters};
use log::debug;
use prost::Message;
use sp_consensus::{BlockOrigin, import_queue::{IncomingBlock, Origin}};
use sp_runtime::{traits::{Block as BlockT, NumberFor}, Justification};
use std::{
Expand All @@ -42,7 +44,7 @@ use std::{
};

pub use crate::request_responses::{
ResponseFailure, InboundFailure, RequestFailure, OutboundFailure, RequestId, SendRequestError
ResponseFailure, InboundFailure, RequestFailure, OutboundFailure, RequestId,
};

/// General behaviour of the network. Combines all protocols together.
Expand All @@ -58,8 +60,6 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
discovery: DiscoveryBehaviour,
/// Generic request-reponse protocols.
request_responses: request_responses::RequestResponsesBehaviour,
/// Block request handling.
block_requests: block_requests::BlockRequests<B>,
/// Light client request handling.
light_client_handler: light_client_handler::LightClientHandler<B>,

Expand All @@ -70,6 +70,11 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
/// Role of our local node, as originally passed from the configuration.
#[behaviour(ignore)]
role: Role,

/// Protocol name used to send out block requests via
/// [`request_responses::RequestResponsesBehaviour`].
#[behaviour(ignore)]
block_request_protocol_name: String,
}

/// Event generated by `Behaviour`.
Expand All @@ -93,34 +98,18 @@ pub enum BehaviourOut<B: BlockT> {
result: Result<Duration, ResponseFailure>,
},

/// A request initiated using [`Behaviour::send_request`] has succeeded or failed.
RequestFinished {
/// Request that has succeeded.
request_id: RequestId,
/// Response sent by the remote or reason for failure.
result: Result<Vec<u8>, RequestFailure>,
},

/// Started a new request with the given node.
///
/// This event is for statistics purposes only. The request and response handling are entirely
/// internal to the behaviour.
OpaqueRequestStarted {
peer: PeerId,
/// Protocol name of the request.
protocol: String,
},
/// Finished, successfully or not, a previously-started request.
/// A request has succeeded or failed.
///
/// This event is for statistics purposes only. The request and response handling are entirely
/// internal to the behaviour.
OpaqueRequestFinished {
/// Who we were requesting.
/// This event is generated for statistics purposes.
RequestFinished {
/// Peer that we send a request to.
peer: PeerId,
/// Protocol name of the request.
protocol: String,
/// How long before the response came or the request got cancelled.
request_duration: Duration,
/// Name of the protocol in question.
protocol: Cow<'static, str>,
/// Duration the request took.
duration: Duration,
/// Result of the request.
result: Result<(), RequestFailure>,
},

/// Opened a substream with the given node with the given notifications protocol.
Expand Down Expand Up @@ -180,21 +169,28 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
role: Role,
user_agent: String,
local_public_key: PublicKey,
block_requests: block_requests::BlockRequests<B>,
light_client_handler: light_client_handler::LightClientHandler<B>,
disco_config: DiscoveryConfig,
request_response_protocols: Vec<request_responses::ProtocolConfig>,
// Block request protocol config.
block_request_protocol_config: request_responses::ProtocolConfig,
// All remaining request protocol configs.
mut request_response_protocols: Vec<request_responses::ProtocolConfig>,
) -> Result<Self, request_responses::RegisterError> {
// Extract protocol name and add to `request_response_protocols`.
let block_request_protocol_name = block_request_protocol_config.name.to_string();
request_response_protocols.push(block_request_protocol_config);

Ok(Behaviour {
substrate,
peer_info: peer_info::PeerInfoBehaviour::new(user_agent, local_public_key),
discovery: disco_config.finish(),
request_responses:
request_responses::RequestResponsesBehaviour::new(request_response_protocols.into_iter())?,
block_requests,
light_client_handler,
events: VecDeque::new(),
role,

block_request_protocol_name,
})
}

Expand Down Expand Up @@ -236,13 +232,14 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
}

/// Initiates sending a request.
///
/// An error is returned if we are not connected to the target peer of if the protocol doesn't
/// match one that has been registered.
pub fn send_request(&mut self, target: &PeerId, protocol: &str, request: Vec<u8>)
-> Result<RequestId, SendRequestError>
{
self.request_responses.send_request(target, protocol, request)
pub fn send_request(
&mut self,
target: &PeerId,
protocol: &str,
request: Vec<u8>,
pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
) {
self.request_responses.send_request(target, protocol, request, pending_response)
}

/// Registers a new notifications protocol.
Expand Down Expand Up @@ -331,28 +328,20 @@ Behaviour<B, H> {
self.events.push_back(BehaviourOut::BlockImport(origin, blocks)),
CustomMessageOutcome::JustificationImport(origin, hash, nb, justification) =>
self.events.push_back(BehaviourOut::JustificationImport(origin, hash, nb, justification)),
CustomMessageOutcome::BlockRequest { target, request } => {
match self.block_requests.send_request(&target, request) {
block_requests::SendRequestOutcome::Ok => {
self.events.push_back(BehaviourOut::OpaqueRequestStarted {
peer: target,
protocol: self.block_requests.protocol_name().to_owned(),
});
},
block_requests::SendRequestOutcome::Replaced { request_duration, .. } => {
self.events.push_back(BehaviourOut::OpaqueRequestFinished {
peer: target.clone(),
protocol: self.block_requests.protocol_name().to_owned(),
request_duration,
});
self.events.push_back(BehaviourOut::OpaqueRequestStarted {
peer: target,
protocol: self.block_requests.protocol_name().to_owned(),
});
}
block_requests::SendRequestOutcome::NotConnected |
block_requests::SendRequestOutcome::EncodeError(_) => {},
CustomMessageOutcome::BlockRequest { target, request, pending_response } => {
let mut buf = Vec::with_capacity(request.encoded_len());
if let Err(err) = request.encode(&mut buf) {
log::warn!(
target: "sync",
"Failed to encode block request {:?}: {:?}",
request, err
);
return
}

self.request_responses.send_request(
&target, &self.block_request_protocol_name, buf, pending_response,
);
},
CustomMessageOutcome::NotificationStreamOpened { remote, protocols, roles, notifications_sink } => {
let role = reported_roles_to_observed_role(&self.role, &remote, roles);
Expand Down Expand Up @@ -401,51 +390,15 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<request_responses::Even
result,
});
}

request_responses::Event::RequestFinished { request_id, result } => {
request_responses::Event::RequestFinished { peer, protocol, duration, result } => {
self.events.push_back(BehaviourOut::RequestFinished {
request_id,
result,
peer, protocol, duration, result,
});
},
}
}
}

impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<block_requests::Event<B>> for Behaviour<B, H> {
fn inject_event(&mut self, event: block_requests::Event<B>) {
match event {
block_requests::Event::AnsweredRequest { peer, total_handling_time } => {
self.events.push_back(BehaviourOut::InboundRequest {
peer,
protocol: self.block_requests.protocol_name().to_owned().into(),
result: Ok(total_handling_time),
});
},
block_requests::Event::Response { peer, response, request_duration } => {
self.events.push_back(BehaviourOut::OpaqueRequestFinished {
peer: peer.clone(),
protocol: self.block_requests.protocol_name().to_owned(),
request_duration,
});
let ev = self.substrate.on_block_response(peer, response);
self.inject_event(ev);
}
block_requests::Event::RequestCancelled { peer, request_duration, .. } |
block_requests::Event::RequestTimeout { peer, request_duration, .. } => {
// There doesn't exist any mechanism to report cancellations or timeouts yet, so
// we process them by disconnecting the node.
self.events.push_back(BehaviourOut::OpaqueRequestFinished {
peer: peer.clone(),
protocol: self.block_requests.protocol_name().to_owned(),
request_duration,
});
self.substrate.on_block_request_failed(&peer);
}
}
}
}

impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<peer_info::PeerInfoEvent>
for Behaviour<B, H> {
fn inject_event(&mut self, event: peer_info::PeerInfoEvent) {
Expand Down

0 comments on commit 2451372

Please sign in to comment.