2019-03-04 07:31:01 +00:00
|
|
|
use crate::error;
|
|
|
|
use crate::message_handler::{HandlerMessage, MessageHandler};
|
|
|
|
use crate::NetworkConfig;
|
2019-05-30 08:27:19 +00:00
|
|
|
use beacon_chain::{BeaconChain, BeaconChainTypes};
|
2019-03-20 04:09:24 +00:00
|
|
|
use eth2_libp2p::Service as LibP2PService;
|
2019-04-03 05:00:09 +00:00
|
|
|
use eth2_libp2p::Topic;
|
2019-03-20 04:09:24 +00:00
|
|
|
use eth2_libp2p::{Libp2pEvent, PeerId};
|
2019-03-25 12:02:51 +00:00
|
|
|
use eth2_libp2p::{PubsubMessage, RPCEvent};
|
2019-03-12 06:28:11 +00:00
|
|
|
use futures::prelude::*;
|
|
|
|
use futures::Stream;
|
2019-07-24 07:45:31 +00:00
|
|
|
use parking_lot::Mutex;
|
2019-03-18 12:34:44 +00:00
|
|
|
use slog::{debug, info, o, trace};
|
2019-05-08 09:38:18 +00:00
|
|
|
use std::marker::PhantomData;
|
2019-03-18 06:38:23 +00:00
|
|
|
use std::sync::Arc;
|
2019-03-12 06:28:11 +00:00
|
|
|
use tokio::runtime::TaskExecutor;
|
2019-07-16 07:28:15 +00:00
|
|
|
use tokio::sync::{mpsc, oneshot};
|
2019-03-04 07:31:01 +00:00
|
|
|
|
2019-03-20 04:09:24 +00:00
|
|
|
/// Service that handles communication between internal services and the eth2_libp2p network service.
|
2019-05-25 10:51:15 +00:00
|
|
|
pub struct Service<T: BeaconChainTypes> {
|
2019-07-24 07:45:31 +00:00
|
|
|
libp2p_service: Arc<Mutex<LibP2PService>>,
|
2019-04-03 05:23:09 +00:00
|
|
|
_libp2p_exit: oneshot::Sender<()>,
|
2019-07-24 07:45:31 +00:00
|
|
|
_network_send: mpsc::UnboundedSender<NetworkMessage>,
|
2019-05-25 10:51:15 +00:00
|
|
|
_phantom: PhantomData<T>, //message_handler: MessageHandler,
|
2019-05-08 09:38:18 +00:00
|
|
|
//message_handler_send: Sender<HandlerMessage>
|
2019-03-04 07:31:01 +00:00
|
|
|
}
|
|
|
|
|
2019-05-25 10:51:15 +00:00
|
|
|
impl<T: BeaconChainTypes + 'static> Service<T> {
|
2019-03-04 07:31:01 +00:00
|
|
|
pub fn new(
|
2019-05-25 10:51:15 +00:00
|
|
|
beacon_chain: Arc<BeaconChain<T>>,
|
2019-03-18 05:16:54 +00:00
|
|
|
config: &NetworkConfig,
|
2019-03-17 10:49:56 +00:00
|
|
|
executor: &TaskExecutor,
|
2019-03-04 07:31:01 +00:00
|
|
|
log: slog::Logger,
|
2019-07-16 07:28:15 +00:00
|
|
|
) -> error::Result<(Arc<Self>, mpsc::UnboundedSender<NetworkMessage>)> {
|
2019-03-18 12:18:25 +00:00
|
|
|
// build the network channel
|
2019-07-16 07:28:15 +00:00
|
|
|
let (network_send, network_recv) = mpsc::unbounded_channel::<NetworkMessage>();
|
2019-03-04 07:31:01 +00:00
|
|
|
// launch message handler thread
|
|
|
|
let message_handler_log = log.new(o!("Service" => "MessageHandler"));
|
2019-03-19 12:20:39 +00:00
|
|
|
let message_handler_send = MessageHandler::spawn(
|
2019-03-18 12:18:25 +00:00
|
|
|
beacon_chain,
|
|
|
|
network_send.clone(),
|
|
|
|
executor,
|
|
|
|
message_handler_log,
|
|
|
|
)?;
|
2019-03-04 07:31:01 +00:00
|
|
|
|
2019-03-21 02:15:14 +00:00
|
|
|
// launch libp2p service
|
|
|
|
let libp2p_log = log.new(o!("Service" => "Libp2p"));
|
2019-07-24 07:45:31 +00:00
|
|
|
let libp2p_service = Arc::new(Mutex::new(LibP2PService::new(config.clone(), libp2p_log)?));
|
2019-03-04 07:31:01 +00:00
|
|
|
|
2019-03-21 02:15:14 +00:00
|
|
|
// TODO: Spawn thread to handle libp2p messages and pass to message handler thread.
|
|
|
|
let libp2p_exit = spawn_service(
|
2019-07-24 07:45:31 +00:00
|
|
|
libp2p_service.clone(),
|
2019-03-18 12:18:25 +00:00
|
|
|
network_recv,
|
|
|
|
message_handler_send,
|
|
|
|
executor,
|
|
|
|
log,
|
|
|
|
)?;
|
2019-03-19 11:53:51 +00:00
|
|
|
let network_service = Service {
|
2019-07-24 07:45:31 +00:00
|
|
|
libp2p_service,
|
2019-04-03 05:23:09 +00:00
|
|
|
_libp2p_exit: libp2p_exit,
|
2019-07-24 07:45:31 +00:00
|
|
|
_network_send: network_send.clone(),
|
2019-05-08 09:38:18 +00:00
|
|
|
_phantom: PhantomData,
|
2019-03-12 06:28:11 +00:00
|
|
|
};
|
2019-03-04 07:31:01 +00:00
|
|
|
|
2019-03-19 11:53:51 +00:00
|
|
|
Ok((Arc::new(network_service), network_send))
|
2019-03-04 07:31:01 +00:00
|
|
|
}
|
2019-03-12 06:28:11 +00:00
|
|
|
|
2019-07-24 07:45:31 +00:00
|
|
|
pub fn libp2p_service(&self) -> Arc<Mutex<LibP2PService>> {
|
|
|
|
self.libp2p_service.clone()
|
2019-03-12 06:28:11 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn spawn_service(
|
2019-07-24 07:45:31 +00:00
|
|
|
libp2p_service: Arc<Mutex<LibP2PService>>,
|
2019-07-16 07:28:15 +00:00
|
|
|
network_recv: mpsc::UnboundedReceiver<NetworkMessage>,
|
|
|
|
message_handler_send: mpsc::UnboundedSender<HandlerMessage>,
|
2019-03-17 10:49:56 +00:00
|
|
|
executor: &TaskExecutor,
|
2019-03-12 06:28:11 +00:00
|
|
|
log: slog::Logger,
|
2019-07-16 07:28:15 +00:00
|
|
|
) -> error::Result<tokio::sync::oneshot::Sender<()>> {
|
|
|
|
let (network_exit, exit_rx) = tokio::sync::oneshot::channel();
|
2019-03-12 06:28:11 +00:00
|
|
|
|
|
|
|
// spawn on the current executor
|
|
|
|
executor.spawn(
|
|
|
|
network_service(
|
2019-03-21 02:15:14 +00:00
|
|
|
libp2p_service,
|
2019-03-12 06:28:11 +00:00
|
|
|
network_recv,
|
|
|
|
message_handler_send,
|
|
|
|
log.clone(),
|
|
|
|
)
|
|
|
|
// allow for manual termination
|
|
|
|
.select(exit_rx.then(|_| Ok(())))
|
|
|
|
.then(move |_| {
|
2019-03-18 07:22:01 +00:00
|
|
|
info!(log.clone(), "Network service shutdown");
|
2019-03-12 06:28:11 +00:00
|
|
|
Ok(())
|
|
|
|
}),
|
|
|
|
);
|
|
|
|
|
2019-03-18 12:18:25 +00:00
|
|
|
Ok(network_exit)
|
2019-03-12 06:28:11 +00:00
|
|
|
}
|
|
|
|
|
2019-03-25 12:02:51 +00:00
|
|
|
//TODO: Potentially handle channel errors
|
2019-03-12 06:28:11 +00:00
|
|
|
fn network_service(
|
2019-07-24 07:45:31 +00:00
|
|
|
libp2p_service: Arc<Mutex<LibP2PService>>,
|
2019-07-16 07:28:15 +00:00
|
|
|
mut network_recv: mpsc::UnboundedReceiver<NetworkMessage>,
|
|
|
|
mut message_handler_send: mpsc::UnboundedSender<HandlerMessage>,
|
2019-03-12 06:28:11 +00:00
|
|
|
log: slog::Logger,
|
2019-03-20 04:09:24 +00:00
|
|
|
) -> impl futures::Future<Item = (), Error = eth2_libp2p::error::Error> {
|
|
|
|
futures::future::poll_fn(move || -> Result<_, eth2_libp2p::error::Error> {
|
2019-07-16 07:28:15 +00:00
|
|
|
// only end the loop once both major polls are not ready.
|
|
|
|
let mut not_ready_count = 0;
|
|
|
|
while not_ready_count < 2 {
|
|
|
|
not_ready_count = 0;
|
|
|
|
// poll the network channel
|
|
|
|
match network_recv.poll() {
|
2019-07-24 07:45:31 +00:00
|
|
|
Ok(Async::Ready(Some(message))) => match message {
|
|
|
|
NetworkMessage::Send(peer_id, outgoing_message) => match outgoing_message {
|
|
|
|
OutgoingMessage::RPC(rpc_event) => {
|
|
|
|
trace!(log, "Sending RPC Event: {:?}", rpc_event);
|
|
|
|
libp2p_service.lock().swarm.send_rpc(peer_id, rpc_event);
|
2019-07-16 07:28:15 +00:00
|
|
|
}
|
2019-07-24 07:45:31 +00:00
|
|
|
},
|
|
|
|
NetworkMessage::Publish { topics, message } => {
|
|
|
|
debug!(log, "Sending pubsub message"; "topics" => format!("{:?}",topics));
|
|
|
|
libp2p_service.lock().swarm.publish(topics, *message);
|
2019-07-16 07:28:15 +00:00
|
|
|
}
|
2019-07-24 07:45:31 +00:00
|
|
|
},
|
2019-07-16 07:28:15 +00:00
|
|
|
Ok(Async::NotReady) => not_ready_count += 1,
|
|
|
|
Ok(Async::Ready(None)) => {
|
|
|
|
return Err(eth2_libp2p::error::Error::from("Network channel closed"));
|
|
|
|
}
|
|
|
|
Err(_) => {
|
|
|
|
return Err(eth2_libp2p::error::Error::from("Network channel error"));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// poll the swarm
|
2019-07-24 07:45:31 +00:00
|
|
|
match libp2p_service.lock().poll() {
|
2019-03-21 02:15:14 +00:00
|
|
|
Ok(Async::Ready(Some(event))) => match event {
|
|
|
|
Libp2pEvent::RPC(peer_id, rpc_event) => {
|
|
|
|
trace!(log, "RPC Event: RPC message received: {:?}", rpc_event);
|
|
|
|
message_handler_send
|
2019-07-16 07:28:15 +00:00
|
|
|
.try_send(HandlerMessage::RPC(peer_id, rpc_event))
|
2019-07-16 13:26:31 +00:00
|
|
|
.map_err(|_| "Failed to send RPC to handler")?;
|
2019-03-21 02:15:14 +00:00
|
|
|
}
|
|
|
|
Libp2pEvent::PeerDialed(peer_id) => {
|
|
|
|
debug!(log, "Peer Dialed: {:?}", peer_id);
|
|
|
|
message_handler_send
|
2019-07-16 07:28:15 +00:00
|
|
|
.try_send(HandlerMessage::PeerDialed(peer_id))
|
2019-07-16 12:32:37 +00:00
|
|
|
.map_err(|_| "Failed to send PeerDialed to handler")?;
|
|
|
|
}
|
|
|
|
Libp2pEvent::PeerDisconnected(peer_id) => {
|
|
|
|
debug!(log, "Peer Disconnected: {:?}", peer_id);
|
|
|
|
message_handler_send
|
2019-07-16 13:26:31 +00:00
|
|
|
.try_send(HandlerMessage::PeerDisconnected(peer_id))
|
2019-07-16 12:32:37 +00:00
|
|
|
.map_err(|_| "Failed to send PeerDisconnected to handler")?;
|
2019-03-21 02:15:14 +00:00
|
|
|
}
|
2019-03-25 12:02:51 +00:00
|
|
|
Libp2pEvent::PubsubMessage {
|
2019-04-03 05:23:09 +00:00
|
|
|
source, message, ..
|
2019-03-25 12:02:51 +00:00
|
|
|
} => {
|
|
|
|
//TODO: Decide if we need to propagate the topic upwards. (Potentially for
|
|
|
|
//attestations)
|
|
|
|
message_handler_send
|
2019-07-16 07:28:15 +00:00
|
|
|
.try_send(HandlerMessage::PubsubMessage(source, message))
|
2019-03-25 12:02:51 +00:00
|
|
|
.map_err(|_| " failed to send pubsub message to handler")?;
|
|
|
|
}
|
2019-03-21 02:15:14 +00:00
|
|
|
},
|
|
|
|
Ok(Async::Ready(None)) => unreachable!("Stream never ends"),
|
2019-07-16 07:28:15 +00:00
|
|
|
Ok(Async::NotReady) => not_ready_count += 1,
|
|
|
|
Err(_) => not_ready_count += 1,
|
2019-03-12 06:28:11 +00:00
|
|
|
}
|
|
|
|
}
|
2019-07-24 07:45:31 +00:00
|
|
|
|
2019-03-12 06:28:11 +00:00
|
|
|
Ok(Async::NotReady)
|
|
|
|
})
|
2019-03-04 07:31:01 +00:00
|
|
|
}
|
2019-03-18 12:18:25 +00:00
|
|
|
|
|
|
|
/// Types of messages that the network service can receive.
|
2019-07-16 12:32:37 +00:00
|
|
|
#[derive(Debug)]
|
2019-03-18 12:18:25 +00:00
|
|
|
pub enum NetworkMessage {
|
2019-03-21 02:15:14 +00:00
|
|
|
/// Send a message to libp2p service.
|
2019-03-18 12:18:25 +00:00
|
|
|
//TODO: Define typing for messages across the wire
|
2019-03-18 12:34:44 +00:00
|
|
|
Send(PeerId, OutgoingMessage),
|
2019-03-25 12:02:51 +00:00
|
|
|
/// Publish a message to pubsub mechanism.
|
2019-03-25 12:39:39 +00:00
|
|
|
Publish {
|
|
|
|
topics: Vec<Topic>,
|
2019-04-03 05:23:09 +00:00
|
|
|
message: Box<PubsubMessage>,
|
2019-03-25 12:39:39 +00:00
|
|
|
},
|
2019-03-18 12:34:44 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
/// Type of outgoing messages that can be sent through the network service.
|
2019-07-16 12:32:37 +00:00
|
|
|
#[derive(Debug)]
|
2019-03-18 12:34:44 +00:00
|
|
|
pub enum OutgoingMessage {
|
|
|
|
/// Send an RPC request/response.
|
|
|
|
RPC(RPCEvent),
|
2019-03-18 12:18:25 +00:00
|
|
|
}
|