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

Update to latest libp2p (#673)

Browse Source
This commit is contained in:
Pierre Krieger
2018-09-06 14:56:41 +02:00
committed by Gav Wood
parent b538733a24
commit af3ad2109f
5 changed files with 345 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/substrate/network-libp2p/Cargo.toml
+2 -2
View File
@@ -11,7 +11,7 @@ bytes = "0.4"
error-chain = { version = "0.12", default-features = false }
fnv = "1.0"
futures = "0.1"
libp2p = { git = "https://github.com/libp2p/rust-libp2p", rev = "e2960b4317b22d64c4fca7fa77c6124a44a92f88", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
libp2p = { git = "https://github.com/tomaka/libp2p-rs", rev = "8ac9b65e669ebbd9deeffa9d4566c95f631cfed5", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
ethcore-io = { git = "https://github.com/paritytech/parity.git" }
ethkey = { git = "https://github.com/paritytech/parity.git" }
ethereum-types = "0.3"
@@ -25,7 +25,7 @@ serde_json = "1.0.24"
tokio = "0.1"
tokio-io = "0.1"
tokio-timer = "0.2"
unsigned-varint = { version = "0.2", features = ["codec"] }
unsigned-varint = { version = "0.2.1", features = ["codec"] }
[dev-dependencies]
assert_matches = "1.2"
substrate/substrate/network-libp2p/src/custom_proto.rs
+5 -5
View File
@@ -64,7 +64,7 @@ pub struct RegisteredProtocolOutput<T> {
/// Stream where incoming messages are received. The stream ends whenever
/// either side is closed.
pub incoming: Box<Stream<Item = (PacketId, Bytes), Error = IoError>>,
pub incoming: Box<Stream<Item = (PacketId, Bytes), Error = IoError> + Send>,
}
impl<T> RegisteredProtocol<T> {
@@ -101,8 +101,8 @@ impl<T> RegisteredProtocol<T> {
// `Maf` is short for `MultiaddressFuture`
impl<T, C, Maf> ConnectionUpgrade<C, Maf> for RegisteredProtocol<T>
where C: AsyncRead + AsyncWrite + 'static, // TODO: 'static :-/
Maf: Future<Item = Multiaddr, Error = IoError> + 'static, // TODO: 'static :(
where C: AsyncRead + AsyncWrite + Send + 'static, // TODO: 'static :-/
Maf: Future<Item = Multiaddr, Error = IoError> + Send + 'static, // TODO: 'static :(
{
type NamesIter = VecIntoIter<(Bytes, Self::UpgradeIdentifier)>;
type UpgradeIdentifier = u8; // Protocol version
@@ -252,8 +252,8 @@ impl<T> Default for RegisteredProtocols<T> {
}
impl<T, C, Maf> ConnectionUpgrade<C, Maf> for RegisteredProtocols<T>
where C: AsyncRead + AsyncWrite + 'static, // TODO: 'static :-/
Maf: Future<Item = Multiaddr, Error = IoError> + 'static, // TODO: 'static :(
where C: AsyncRead + AsyncWrite + Send + 'static, // TODO: 'static :-/
Maf: Future<Item = Multiaddr, Error = IoError> + Send + 'static, // TODO: 'static :(
{
type NamesIter = VecIntoIter<(Bytes, Self::UpgradeIdentifier)>;
type UpgradeIdentifier = (usize,
substrate/substrate/network-libp2p/src/service.rs
+150 -66
View File
@@ -395,7 +395,7 @@ fn init_thread(
});
let shared = shared.clone();
base.and_then(move |(peer_id, stream), endpoint, remote_addr| {
Transport::and_then(base, move |(peer_id, stream), endpoint, remote_addr| {
remote_addr.and_then(move |remote_addr| {
if &peer_id == shared.kad_system.local_peer_id() {
// TODO: this happens very frequently for now and floods the logs
@@ -575,13 +575,13 @@ struct TransportOutput<S> {
/// Enum of all the possible protocols our service handles.
enum FinalUpgrade<C> {
Kad(NodeIndex, KadConnecController, Box<Stream<Item = KadIncomingRequest, Error = IoError>>),
Kad(NodeIndex, KadConnecController, Box<Stream<Item = KadIncomingRequest, Error = IoError> + Send>),
/// The remote identification system, and the multiaddress we see the remote as.
IdentifyListener(NodeIndex, IdentifySender<C>, Multiaddr),
/// The remote information about the address they see us as.
IdentifyDialer(NodeIndex, IdentifyInfo, Multiaddr),
PingDialer(NodeIndex, ping::Pinger, Box<Future<Item = (), Error = IoError>>),
PingListener(NodeIndex, Box<Future<Item = (), Error = IoError>>),
PingDialer(NodeIndex, ping::Pinger, Box<Future<Item = (), Error = IoError> + Send>),
PingListener(NodeIndex, Box<Future<Item = (), Error = IoError> + Send>),
/// `Custom` means anything not in the core libp2p and is handled
/// by `CustomProtoConnectionUpgrade`.
Custom(NodeIndex, RegisteredProtocolOutput<Arc<NetworkProtocolHandler + Send + Sync>>),
@@ -613,8 +613,8 @@ impl<C> From<(NodeIndex, IdentifyOutput<C>, Multiaddr)> for FinalUpgrade<C> {
fn listener_handle<'a, C>(
shared: Arc<Shared>,
upgrade: FinalUpgrade<C>,
) -> Box<Future<Item = (), Error = IoError> + 'a>
where C: AsyncRead + AsyncWrite + 'a {
) -> Box<Future<Item = (), Error = IoError> + Send + 'a>
where C: AsyncRead + AsyncWrite + Send + 'a {
match upgrade {
FinalUpgrade::Kad(node_index, controller, kademlia_stream) => {
trace!(target: "sub-libp2p", "Opened kademlia substream with #{:?}", node_index);
@@ -674,7 +674,7 @@ fn handle_kademlia_connection(
shared: Arc<Shared>,
node_index: NodeIndex,
controller: KadConnecController,
kademlia_stream: Box<Stream<Item = KadIncomingRequest, Error = IoError>>
kademlia_stream: Box<Stream<Item = KadIncomingRequest, Error = IoError> + Send>
) -> Result<impl Future<Item = (), Error = IoError>, IoError> {
let kad_connec = match shared.network_state.kad_connection(node_index) {
Some(kad) => kad,
@@ -767,7 +767,7 @@ fn handle_custom_connection(
shared: Arc<Shared>,
node_index: NodeIndex,
custom_proto_out: RegisteredProtocolOutput<Arc<NetworkProtocolHandler + Send + Sync>>
) -> impl Future<Item = (), Error = IoError> {
) -> Box<Future<Item = (), Error = IoError> + Send> {
let handler = custom_proto_out.custom_data;
let protocol_id = custom_proto_out.protocol_id;
@@ -781,7 +781,7 @@ fn handle_custom_connection(
protocol_id,
) {
Some(c) => c,
None => return future::Either::A(future::err(IoErrorKind::Other.into())),
None => return Box::new(future::err(IoErrorKind::Other.into())) as Box<_>,
};
if let UniqueConnecState::Full = unique_connec.state() {
@@ -790,7 +790,7 @@ fn handle_custom_connection(
node_index,
custom_proto_out.protocol_id
);
return future::Either::A(future::ok(()))
return Box::new(future::ok(())) as Box<_>
}
struct ProtoDisconnectGuard {
@@ -873,7 +873,7 @@ fn handle_custom_connection(
current_peer: Some(node_index),
}, &node_index);
future::Either::B(final_fut)
Box::new(final_fut) as Box<_>
}
/// Randomly discovers peers to connect to.
@@ -883,13 +883,24 @@ fn handle_custom_connection(
fn start_kademlia_discovery<T, To, St, C>(
shared: Arc<Shared>,
transport: T,
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError>>>
) -> impl Future<Item = (), Error = IoError>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
T::MultiaddrFuture: 'static,
To: AsyncRead + AsyncWrite + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
C: 'static {
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError> + Send>>
) -> Box<Future<Item = (), Error = IoError> + Send>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + Send + 'static,
T::Dial: Send,
T::MultiaddrFuture: Send + 'static,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
To: AsyncRead + AsyncWrite + Send + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
St::Dial: Send,
St::MultiaddrFuture: Send,
St::Listener: Send,
St::ListenerUpgrade: Send,
St::Incoming: Send,
St::IncomingUpgrade: Send,
C: Send + 'static {
let kad_init = shared.kad_system.perform_initialization({
let shared = shared.clone();
let transport = transport.clone();
@@ -923,7 +934,7 @@ fn start_kademlia_discovery<T, To, St, C>(
.and_then(|(_, rest)| rest);
// Note that we use a Box in order to speed compilation time.
Box::new(final_future) as Box<Future<Item = _, Error = _>>
Box::new(final_future) as Box<Future<Item = _, Error = _> + Send>
}
/// Performs a kademlia request to a random node.
@@ -932,13 +943,24 @@ fn start_kademlia_discovery<T, To, St, C>(
fn perform_kademlia_query<T, To, St, C>(
shared: Arc<Shared>,
transport: T,
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError>>>
) -> impl Future<Item = (), Error = IoError>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
T::MultiaddrFuture: 'static,
To: AsyncRead + AsyncWrite + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
C: 'static {
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError> + Send>>
) -> Box<Future<Item = (), Error = IoError> + Send>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + Send + 'static,
T::MultiaddrFuture: Send + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
To: AsyncRead + AsyncWrite + Send + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Send + Clone + 'static,
St::Dial: Send,
St::MultiaddrFuture: Send,
St::Listener: Send,
St::ListenerUpgrade: Send,
St::Incoming: Send,
St::IncomingUpgrade: Send,
C: Send + 'static {
// Query the node IDs that are closest to a random ID.
// Note that the randomness doesn't have to be secure, as this only
// influences which nodes we end up being connected to.
@@ -974,20 +996,31 @@ fn perform_kademlia_query<T, To, St, C>(
.map(|_| ());
// Note that we use a `Box` in order to speed up compilation.
Box::new(future) as Box<Future<Item = _, Error = _>>
Box::new(future) as Box<Future<Item = _, Error = _> + Send>
}
/// Connects to additional nodes, if necessary.
fn connect_to_nodes<T, To, St, C>(
shared: Arc<Shared>,
base_transport: T,
swarm_controller: &SwarmController<St, Box<Future<Item = (), Error = IoError>>>
swarm_controller: &SwarmController<St, Box<Future<Item = (), Error = IoError> + Send>>
)
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
T::MultiaddrFuture: 'static,
To: AsyncRead + AsyncWrite + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
C: 'static {
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + Send + 'static,
T::MultiaddrFuture: Send + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
To: AsyncRead + AsyncWrite + Send + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
St::Dial: Send,
St::MultiaddrFuture: Send,
St::Listener: Send,
St::ListenerUpgrade: Send,
St::Incoming: Send,
St::IncomingUpgrade: Send,
C: Send + 'static {
let (addrs, _will_change) = shared.network_state.outgoing_connections_to_attempt();
for (peer, addr) in addrs.into_iter() {
@@ -1021,13 +1054,24 @@ fn open_peer_custom_proto<T, To, St, C>(
addr: Multiaddr,
expected_peer_id: Option<PeerstorePeerId>,
proto: RegisteredProtocol<Arc<NetworkProtocolHandler + Send + Sync>>,
swarm_controller: &SwarmController<St, Box<Future<Item = (), Error = IoError>>>
swarm_controller: &SwarmController<St, Box<Future<Item = (), Error = IoError> + Send>>
)
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
T::MultiaddrFuture: 'static,
To: AsyncRead + AsyncWrite + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
C: 'static,
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + Send + 'static,
T::MultiaddrFuture: Send + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
To: AsyncRead + AsyncWrite + Send + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
St::Dial: Send,
St::MultiaddrFuture: Send,
St::Listener: Send,
St::ListenerUpgrade: Send,
St::Incoming: Send,
St::IncomingUpgrade: Send,
C: Send + 'static,
{
let proto_id = proto.id();
@@ -1083,13 +1127,24 @@ fn obtain_kad_connection<T, To, St, C>(
shared: Arc<Shared>,
who: PeerstorePeerId,
transport: T,
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError>>>
) -> impl Future<Item = KadConnecController, Error = IoError>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
T::MultiaddrFuture: 'static,
To: AsyncRead + AsyncWrite + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
C: 'static {
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError> + Send>>
) -> Box<Future<Item = KadConnecController, Error = IoError> + Send>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + Send + 'static,
T::MultiaddrFuture: Send + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
To: AsyncRead + AsyncWrite + Send + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
St::Dial: Send,
St::MultiaddrFuture: Send,
St::Listener: Send,
St::ListenerUpgrade: Send,
St::Incoming: Send,
St::IncomingUpgrade: Send,
C: Send + 'static {
let kad_upgrade = shared.kad_upgrade.clone();
let transport = transport
.and_then(move |out, endpoint, client_addr| {
@@ -1121,7 +1176,7 @@ fn obtain_kad_connection<T, To, St, C>(
.and_then(|(kad, _)| kad.ok_or_else(|| IoErrorKind::ConnectionRefused.into()));
// Note that we use a Box in order to speed up compilation.
Box::new(future) as Box<Future<Item = _, Error = _>>
Box::new(future) as Box<Future<Item = _, Error = _> + Send>
}
/// Processes the identification information that we received about a node.
@@ -1170,13 +1225,24 @@ fn process_identify_info(
fn start_periodic_updates<T, To, St, C>(
shared: Arc<Shared>,
transport: T,
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError>>>
) -> impl Future<Item = (), Error = IoError>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
T::MultiaddrFuture: 'static,
To: AsyncRead + AsyncWrite + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
C: 'static {
swarm_controller: SwarmController<St, Box<Future<Item = (), Error = IoError> + Send>>
) -> Box<Future<Item = (), Error = IoError> + Send>
where T: MuxedTransport<Output = TransportOutput<To>> + Clone + Send + 'static,
T::MultiaddrFuture: Send + 'static,
T::Dial: Send,
T::Listener: Send,
T::ListenerUpgrade: Send,
T::Incoming: Send,
T::IncomingUpgrade: Send,
To: AsyncRead + AsyncWrite + Send + 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
St::Dial: Send,
St::MultiaddrFuture: Send,
St::Listener: Send,
St::ListenerUpgrade: Send,
St::Incoming: Send,
St::IncomingUpgrade: Send,
C: Send + 'static {
let ping_transport = transport.clone()
.and_then(move |out, endpoint, client_addr| {
let node_index = out.node_index;
@@ -1212,7 +1278,7 @@ fn start_periodic_updates<T, To, St, C>(
});
// Note that we use a Box in order to speed compilation time.
Box::new(fut) as Box<Future<Item = _, Error = _>>
Box::new(fut) as Box<Future<Item = _, Error = _> + Send>
}
/// Pings all the nodes we're connected to and disconnects any node that
@@ -1222,14 +1288,32 @@ fn periodic_updates<Tp, Tid, St, C>(
shared: Arc<Shared>,
ping_transport: Tp,
identify_transport: Tid,
swarm_controller: &SwarmController<St, Box<Future<Item = (), Error = IoError>>>
) -> impl Future<Item = (), Error = IoError>
where Tp: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
Tp::MultiaddrFuture: 'static,
Tid: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
Tid::MultiaddrFuture: 'static,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
C: 'static {
swarm_controller: &SwarmController<St, Box<Future<Item = (), Error = IoError> + Send>>
) -> Box<Future<Item = (), Error = IoError> + Send>
where Tp: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
Tp::MultiaddrFuture: Send + 'static,
Tp::Dial: Send,
Tp::MultiaddrFuture: Send,
Tp::Listener: Send,
Tp::ListenerUpgrade: Send,
Tp::Incoming: Send,
Tp::IncomingUpgrade: Send,
Tid: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
Tid::MultiaddrFuture: Send + 'static,
Tid::Dial: Send,
Tid::MultiaddrFuture: Send,
Tid::Listener: Send,
Tid::ListenerUpgrade: Send,
Tid::Incoming: Send,
Tid::IncomingUpgrade: Send,
St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + Send + 'static,
St::Dial: Send,
St::MultiaddrFuture: Send,
St::Listener: Send,
St::ListenerUpgrade: Send,
St::Incoming: Send,
St::IncomingUpgrade: Send,
C: Send + 'static {
trace!(target: "sub-libp2p", "Periodic update cycle");
let mut ping_futures = Vec::new();
@@ -1287,7 +1371,7 @@ fn periodic_updates<Tp, Tid, St, C>(
});
// Note that we use a Box in order to speed up compilation.
Box::new(future) as Box<Future<Item = _, Error = _>>
Box::new(future) as Box<Future<Item = _, Error = _> + Send>
}
/// Since new protocols are added after the networking starts, we have to load the protocols list
@@ -1296,8 +1380,8 @@ fn periodic_updates<Tp, Tid, St, C>(
struct DelayedProtosList(Arc<Shared>);
// `Maf` is short for `MultiaddressFuture`
impl<C, Maf> ConnectionUpgrade<C, Maf> for DelayedProtosList
where C: AsyncRead + AsyncWrite + 'static, // TODO: 'static :-/
Maf: Future<Item = Multiaddr, Error = IoError> + 'static, // TODO: 'static :(
where C: AsyncRead + AsyncWrite + Send + 'static, // TODO: 'static :-/
Maf: Future<Item = Multiaddr, Error = IoError> + Send + 'static, // TODO: 'static :(
{
type NamesIter = <RegisteredProtocols<Arc<NetworkProtocolHandler + Send + Sync>> as ConnectionUpgrade<C, Maf>>::NamesIter;
type UpgradeIdentifier = <RegisteredProtocols<Arc<NetworkProtocolHandler + Send + Sync>> as ConnectionUpgrade<C, Maf>>::UpgradeIdentifier;
substrate/substrate/network-libp2p/src/transport.rs
+3 -2
View File
@@ -15,7 +15,7 @@
// along with Substrate. If not, see <http://www.gnu.org/licenses/>.
use libp2p::{self, PeerId, Transport, mplex, secio, yamux};
use libp2p::core::{MuxedTransport, either, upgrade};
use libp2p::core::{either, upgrade, transport::BoxedMuxed};
use libp2p::transport_timeout::TransportTimeout;
use std::time::Duration;
use std::usize;
@@ -24,7 +24,7 @@ use tokio_io::{AsyncRead, AsyncWrite};
/// Builds the transport that serves as a common ground for all connections.
pub fn build_transport(
local_private_key: secio::SecioKeyPair
) -> impl MuxedTransport<Output = (PeerId, impl AsyncRead + AsyncWrite)> + Clone {
) -> BoxedMuxed<(PeerId, impl AsyncRead + AsyncWrite)> {
let mut mplex_config = mplex::MplexConfig::new();
mplex_config.max_buffer_len_behaviour(mplex::MaxBufferBehaviour::Block);
mplex_config.max_buffer_len(usize::MAX);
@@ -46,4 +46,5 @@ pub fn build_transport(
.map(|(key, substream), _| (key.into_peer_id(), substream));
TransportTimeout::new(base, Duration::from_secs(20))
.boxed_muxed()
}