liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

add direct bootstrap fallback

Browse Source
This commit is contained in:
John Smith
2022-06-25 15:28:27 -04:00
parent 0adcc70bc9
commit a475028c75
12 changed files with 405 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
veilid-core/src/network_manager/mod.rs
+20
View File
@@ -1091,6 +1091,26 @@ impl NetworkManager {
}
}
// Direct bootstrap request
pub async fn boot_request(&self, dial_info: DialInfo) -> Result<Vec<PeerInfo>, String> {
let timeout_ms = {
let c = self.config.get();
c.network.rpc.timeout_ms
};
// Send boot magic to requested peer address
let data = BOOT_MAGIC.to_vec();
let out_data: Vec<u8> = self
.net()
.send_recv_data_unbound_to_dial_info(dial_info, data, timeout_ms)
.await?;
let bootstrap_peerinfo: Vec<PeerInfo> =
deserialize_json(std::str::from_utf8(&out_data).map_err(map_to_string)?)
.map_err(map_to_string)?;
Ok(bootstrap_peerinfo)
}
// Called when a packet potentially containing an RPC envelope is received by a low-level
// network protocol handler. Processes the envelope, authenticates and decrypts the RPC message
// and passes it to the RPC handler
veilid-core/src/network_manager/native/mod.rs
+44
View File
@@ -297,6 +297,50 @@ impl Network {
res
}
// Send data to a dial info, unbound, using a new connection from a random port
// Waits for a specified amount of time to receive a single response
// This creates a short-lived connection in the case of connection-oriented protocols
// for the purpose of sending this one message.
// This bypasses the connection table as it is not a 'node to node' connection.
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len(), ret.len))]
pub async fn send_recv_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
let data_len = data.len();
let out = match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawUdpProtocolHandler::send_recv_unbound_message(peer_socket_addr, data, timeout_ms)
.await?
}
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawTcpProtocolHandler::send_recv_unbound_message(peer_socket_addr, data, timeout_ms)
.await?
}
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::send_recv_unbound_message(
dial_info.clone(),
data,
timeout_ms,
)
.await?
}
};
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), data_len as u64);
self.network_manager()
.stats_packet_rcvd(dial_info.to_ip_addr(), out.len() as u64);
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
pub async fn send_data_to_existing_connection(
&self,
veilid-core/src/network_manager/native/protocol/mod.rs
+31
View File
@@ -51,6 +51,37 @@ impl ProtocolNetworkConnection {
}
}
pub async fn send_recv_unbound_message(
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
udp::RawUdpProtocolHandler::send_recv_unbound_message(
peer_socket_addr,
data,
timeout_ms,
)
.await
}
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
tcp::RawTcpProtocolHandler::send_recv_unbound_message(
peer_socket_addr,
data,
timeout_ms,
)
.await
}
ProtocolType::WS | ProtocolType::WSS => {
ws::WebsocketProtocolHandler::send_recv_unbound_message(dial_info, data, timeout_ms)
.await
}
}
}
pub fn descriptor(&self) -> ConnectionDescriptor {
match self {
Self::Dummy(d) => d.descriptor(),
veilid-core/src/network_manager/native/protocol/tcp.rs
+57 -10
View File
@@ -41,7 +41,7 @@ impl RawTcpNetworkConnection {
.map_err(map_to_string)
}
async fn send_internal(mut stream: AsyncPeekStream, message: Vec<u8>) -> Result<(), String> {
async fn send_internal(stream: &mut AsyncPeekStream, message: Vec<u8>) -> Result<(), String> {
log_net!("sending TCP message of size {}", message.len());
if message.len() > MAX_MESSAGE_SIZE {
return Err("sending too large TCP message".to_owned());
@@ -55,16 +55,13 @@ impl RawTcpNetworkConnection {
#[instrument(level="trace", err, skip(self, message), fields(message.len = message.len()))]
pub async fn send(&self, message: Vec<u8>) -> Result<(), String> {
let stream = self.stream.clone();
Self::send_internal(stream, message).await
let mut stream = self.stream.clone();
Self::send_internal(&mut stream, message).await
}
#[instrument(level="trace", err, skip(self), fields(message.len))]
pub async fn recv(&self) -> Result<Vec<u8>, String> {
pub async fn recv_internal(stream: &mut AsyncPeekStream) -> Result<Vec<u8>, String> {
let mut header = [0u8; 4];
let mut stream = self.stream.clone();
stream
.read_exact(&mut header)
.await
@@ -80,7 +77,14 @@ impl RawTcpNetworkConnection {
let mut out: Vec<u8> = vec![0u8; len];
stream.read_exact(&mut out).await.map_err(map_to_string)?;
tracing::Span::current().record("message.len", &out.len());
Ok(out)
}
#[instrument(level="trace", err, skip(self), fields(ret.len))]
pub async fn recv(&self) -> Result<Vec<u8>, String> {
let mut stream = self.stream.clone();
let out = Self::recv_internal(&mut stream).await?;
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
}
@@ -212,11 +216,54 @@ impl RawTcpProtocolHandler {
// .local_addr()
// .map_err(map_to_string)
// .map_err(logthru_net!("could not get local address from TCP stream"))?;
let ps = AsyncPeekStream::new(ts.clone());
let mut ps = AsyncPeekStream::new(ts.clone());
// Send directly from the raw network connection
// this builds the connection and tears it down immediately after the send
RawTcpNetworkConnection::send_internal(ps, data).await
RawTcpNetworkConnection::send_internal(&mut ps, data).await
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len(), ret.len))]
pub async fn send_recv_unbound_message(
socket_addr: SocketAddr,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound TCP message".to_owned());
}
trace!(
"sending unbound message of length {} to {}",
data.len(),
socket_addr
);
// Make a shared socket
let socket = new_unbound_shared_tcp_socket(socket2::Domain::for_address(socket_addr))?;
// Non-blocking connect to remote address
let ts = nonblocking_connect(socket, socket_addr)
.await
.map_err(map_to_string)
.map_err(logthru_net!(error "remote_addr={}", socket_addr))?;
// See what local address we ended up with and turn this into a stream
// let actual_local_address = ts
// .local_addr()
// .map_err(map_to_string)
// .map_err(logthru_net!("could not get local address from TCP stream"))?;
let mut ps = AsyncPeekStream::new(ts.clone());
// Send directly from the raw network connection
// this builds the connection and tears it down immediately after the send
RawTcpNetworkConnection::send_internal(&mut ps, data).await?;
let out = timeout(timeout_ms, RawTcpNetworkConnection::recv_internal(&mut ps))
.await
.map_err(map_to_string)??;
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
}
veilid-core/src/network_manager/native/protocol/udp.rs
+58
View File
@@ -68,6 +68,7 @@ impl RawUdpProtocolHandler {
}
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len()))]
pub async fn send_unbound_message(
socket_addr: SocketAddr,
data: Vec<u8>,
@@ -104,4 +105,61 @@ impl RawUdpProtocolHandler {
Ok(())
}
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len(), ret.len))]
pub async fn send_recv_unbound_message(
socket_addr: SocketAddr,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound UDP message".to_owned())
.map_err(logthru_net!(error));
}
log_net!(
"sending unbound message of length {} to {}",
data.len(),
socket_addr
);
// get local wildcard address for bind
let local_socket_addr = match socket_addr {
SocketAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0),
SocketAddr::V6(_) => {
SocketAddr::new(IpAddr::V6(Ipv6Addr::new(0, 0, 0, 0, 0, 0, 0, 0)), 0)
}
};
// get unspecified bound socket
let socket = UdpSocket::bind(local_socket_addr)
.await
.map_err(map_to_string)
.map_err(logthru_net!(error "failed to bind unbound udp socket"))?;
let len = socket
.send_to(&data, socket_addr)
.await
.map_err(map_to_string)
.map_err(logthru_net!(error "failed unbound udp send: addr={}", socket_addr))?;
if len != data.len() {
return Err("UDP partial unbound send".to_owned()).map_err(logthru_net!(error));
}
// receive single response
let mut out = vec![0u8; MAX_MESSAGE_SIZE];
let (len, from_addr) = timeout(timeout_ms, socket.recv_from(&mut out))
.await
.map_err(map_to_string)?
.map_err(map_to_string)?;
// if the from address is not the same as the one we sent to, then drop this
if from_addr != socket_addr {
return Err(format!(
"Unbound response received from wrong address: addr={}",
from_addr,
));
}
out.resize(len, 0u8);
tracing::Span::current().record("ret.len", &len);
Ok(out)
}
}
veilid-core/src/network_manager/native/protocol/ws.rs
+26 -8
View File
@@ -59,7 +59,7 @@ where
.map_err(map_to_string)
}
#[instrument(level="trace", err, skip(self, message), fields(message.len = message.len()))]
#[instrument(level = "trace", err, skip(self, message), fields(message.len = message.len()))]
pub async fn send(&self, message: Vec<u8>) -> Result<(), String> {
if message.len() > MAX_MESSAGE_SIZE {
return Err("received too large WS message".to_owned());
@@ -72,7 +72,7 @@ where
.map_err(logthru_net!(error "failed to send websocket message"))
}
#[instrument(level="trace", err, skip(self), fields(message.len))]
#[instrument(level = "trace", err, skip(self), fields(ret.len))]
pub async fn recv(&self) -> Result<Vec<u8>, String> {
let out = match self.stream.clone().next().await {
Some(Ok(Message::Binary(v))) => v,
@@ -89,7 +89,7 @@ where
if out.len() > MAX_MESSAGE_SIZE {
Err("sending too large WS message".to_owned()).map_err(logthru_net!(error))
} else {
tracing::Span::current().record("message.len", &out.len());
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
}
@@ -283,11 +283,6 @@ impl WebsocketProtocolHandler {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound WS message".to_owned());
}
trace!(
"sending unbound websocket message of length {} to {}",
data.len(),
dial_info,
);
let protconn = Self::connect_internal(None, dial_info.clone())
.await
@@ -295,6 +290,29 @@ impl WebsocketProtocolHandler {
protconn.send(data).await
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len(), ret.len))]
pub async fn send_recv_unbound_message(
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound WS message".to_owned());
}
let protconn = Self::connect_internal(None, dial_info.clone())
.await
.map_err(|e| format!("failed to connect websocket for unbound message: {}", e))?;
protconn.send(data).await?;
let out = timeout(timeout_ms, protconn.recv())
.await
.map_err(map_to_string)??;
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
}
impl ProtocolAcceptHandler for WebsocketProtocolHandler {
veilid-core/src/network_manager/wasm/mod.rs
+40
View File
@@ -79,6 +79,46 @@ impl Network {
res
}
// Send data to a dial info, unbound, using a new connection from a random port
// Waits for a specified amount of time to receive a single response
// This creates a short-lived connection in the case of connection-oriented protocols
// for the purpose of sending this one message.
// This bypasses the connection table as it is not a 'node to node' connection.
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len(), ret.len))]
pub async fn send_recv_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
let data_len = data.len();
let out = match dial_info.protocol_type() {
ProtocolType::UDP => {
return Err("no support for UDP protocol".to_owned()).map_err(logthru_net!(error))
}
ProtocolType::TCP => {
return Err("no support for TCP protocol".to_owned()).map_err(logthru_net!(error))
}
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::send_recv_unbound_message(
dial_info.clone(),
data,
timeout_ms,
)
.await?
}
};
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), data_len as u64);
self.network_manager()
.stats_packet_rcvd(dial_info.to_ip_addr(), out.len() as u64);
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
pub async fn send_data_to_existing_connection(
&self,
veilid-core/src/network_manager/wasm/protocol/mod.rs
+23 -4
View File
@@ -18,10 +18,10 @@ impl ProtocolNetworkConnection {
) -> Result<ProtocolNetworkConnection, String> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
panic!("UDP dial info is not support on WASM targets");
panic!("UDP dial info is not supported on WASM targets");
}
ProtocolType::TCP => {
panic!("TCP dial info is not support on WASM targets");
panic!("TCP dial info is not supported on WASM targets");
}
ProtocolType::WS | ProtocolType::WSS => {
ws::WebsocketProtocolHandler::connect(local_address, dial_info).await
@@ -35,10 +35,10 @@ impl ProtocolNetworkConnection {
) -> Result<(), String> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
panic!("UDP dial info is not support on WASM targets");
panic!("UDP dial info is not supported on WASM targets");
}
ProtocolType::TCP => {
panic!("TCP dial info is not support on WASM targets");
panic!("TCP dial info is not supported on WASM targets");
}
ProtocolType::WS | ProtocolType::WSS => {
ws::WebsocketProtocolHandler::send_unbound_message(dial_info, data).await
@@ -46,6 +46,25 @@ impl ProtocolNetworkConnection {
}
}
pub async fn send_recv_unbound_message(
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
panic!("UDP dial info is not supported on WASM targets");
}
ProtocolType::TCP => {
panic!("TCP dial info is not supported on WASM targets");
}
ProtocolType::WS | ProtocolType::WSS => {
ws::WebsocketProtocolHandler::send_recv_unbound_message(dial_info, data, timeout_ms)
.await
}
}
}
pub fn descriptor(&self) -> ConnectionDescriptor {
match self {
Self::Dummy(d) => d.descriptor(),
veilid-core/src/network_manager/wasm/protocol/ws.rs
+32 -8
View File
@@ -36,10 +36,12 @@ impl WebsocketNetworkConnection {
self.descriptor.clone()
}
#[instrument(level = "trace", err, skip(self))]
pub async fn close(&self) -> Result<(), String> {
self.inner.ws_meta.close().await.map_err(map_to_string).map(drop)
}
#[instrument(level = "trace", err, skip(self, message), fields(message.len = message.len()))]
pub async fn send(&self, message: Vec<u8>) -> Result<(), String> {
if message.len() > MAX_MESSAGE_SIZE {
return Err("sending too large WS message".to_owned()).map_err(logthru_net!(error));
@@ -49,6 +51,8 @@ impl WebsocketNetworkConnection {
.map_err(|_| "failed to send to websocket".to_owned())
.map_err(logthru_net!(error))
}
#[instrument(level = "trace", err, skip(self), fields(ret.len))]
pub async fn recv(&self) -> Result<Vec<u8>, String> {
let out = match self.inner.ws_stream.clone().next().await {
Some(WsMessage::Binary(v)) => v,
@@ -74,6 +78,7 @@ impl WebsocketNetworkConnection {
pub struct WebsocketProtocolHandler {}
impl WebsocketProtocolHandler {
#[instrument(level = "trace", err)]
pub async fn connect(
local_address: Option<SocketAddr>,
dial_info: DialInfo,
@@ -99,28 +104,47 @@ impl WebsocketProtocolHandler {
.map_err(logthru_net!(error))?;
// Make our connection descriptor
Ok(ProtocolNetworkConnection::Ws(WebsocketNetworkConnection::new(ConnectionDescriptor::new_no_local(
dial_info.to_peer_address(),
), wsmeta, wsio)))
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len()))]
pub async fn send_unbound_message(dial_info: DialInfo, data: Vec<u8>) -> Result<(), String> {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound WS message".to_owned());
}
trace!(
"sending unbound websocket message of length {} to {}",
data.len(),
dial_info,
);
// Make the real connection
let conn = Self::connect(None, dial_info)
.await
.map_err(|e| format!("failed to connect websocket for unbound message: {}", e))?;
conn.send(data).await
conn.send(data).await
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len(), ret.len))]
pub async fn send_recv_unbound_message(
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> Result<Vec<u8>, String> {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound WS message".to_owned());
}
let conn = Self::connect(None, dial_info.clone())
.await
.map_err(|e| format!("failed to connect websocket for unbound message: {}", e))?;
conn.send(data).await?;
let out = timeout(timeout_ms, conn.recv())
.await
.map_err(map_to_string)??;
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
}