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

refactor

Browse Source
This commit is contained in:
John Smith
2022-07-13 09:51:56 -04:00
parent 007150c818
commit b90d453cef
13 changed files with 340 additions and 374 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
veilid-core/src/network_manager/native/mod.rs
View File

@@ -288,26 +288,35 @@ impl Network {
data: Vec<u8>,
) -> EyreResult<()> {
let data_len = data.len();
let res = match dial_info.protocol_type() {
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawUdpProtocolHandler::send_unbound_message(peer_socket_addr, data).await
let h = RawUdpProtocolHandler::new_unspecified_bound_handler(&peer_socket_addr)
.await
.wrap_err("create socket failure")?;
h.send_message(data, peer_socket_addr)
.await
.wrap_err("send message failure")?;
}
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawTcpProtocolHandler::send_unbound_message(peer_socket_addr, data).await
let pnc = RawTcpProtocolHandler::connect(None, peer_socket_addr)
.await
.wrap_err("connect failure")?;
pnc.send(data).await.wrap_err("send failure")?;
}
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::send_unbound_message(dial_info.clone(), data).await
let pnc = WebsocketProtocolHandler::connect(None, &dial_info)
.await
.wrap_err("connect failure")?;
pnc.send(data).await.wrap_err("send failure")?;
}
}
.wrap_err("low level network error");
if res.is_ok() {
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), data_len as u64);
}
res
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), data_len as u64);
Ok(())
}
// Send data to a dial info, unbound, using a new connection from a random port
@@ -315,43 +324,94 @@ impl Network {
// 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))]
#[instrument(level="trace", err, skip(self, data), fields(ret.timeout_or, data.len = data.len()))]
pub async fn send_recv_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> EyreResult<Vec<u8>> {
) -> EyreResult<TimeoutOr<Vec<u8>>> {
let data_len = data.len();
let out = match dial_info.protocol_type() {
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?
}
};
let h = RawUdpProtocolHandler::new_unspecified_bound_handler(&peer_socket_addr)
.await
.wrap_err("create socket failure")?;
h.send_message(data, peer_socket_addr)
.await
.wrap_err("send message failure")?;
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), data_len as u64);
// 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);
// receive single response
let mut out = vec![0u8; MAX_MESSAGE_SIZE];
let timeout_or_ret = timeout(timeout_ms, h.recv_message(&mut out))
.await
.into_timeout_or()
.into_result()
.wrap_err("recv_message failure")?;
let (recv_len, recv_addr) = match timeout_or_ret {
TimeoutOr::Value(v) => v,
TimeoutOr::Timeout => {
tracing::Span::current().record("ret.timeout_or", &"Timeout".to_owned());
return Ok(TimeoutOr::Timeout);
}
};
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
let recv_socket_addr = recv_addr.remote_address().to_socket_addr();
self.network_manager()
.stats_packet_rcvd(recv_socket_addr.ip(), recv_len as u64);
// if the from address is not the same as the one we sent to, then drop this
if recv_socket_addr != peer_socket_addr {
bail!("wrong address");
}
out.resize(recv_len, 0u8);
Ok(TimeoutOr::Value(out))
}
ProtocolType::TCP | ProtocolType::WS | ProtocolType::WSS => {
let pnc = match dial_info.protocol_type() {
ProtocolType::UDP => unreachable!(),
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawTcpProtocolHandler::connect(None, peer_socket_addr)
.await
.wrap_err("connect failure")?
}
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::connect(None, &dial_info)
.await
.wrap_err("connect failure")?
}
};
pnc.send(data).await.wrap_err("send failure")?;
self.network_manager()
.stats_packet_sent(dial_info.to_ip_addr(), data_len as u64);
let out = timeout(timeout_ms, pnc.recv())
.await
.into_timeout_or()
.into_result()
.wrap_err("recv failure")?;
tracing::Span::current().record(
"ret.timeout_or",
&match out {
TimeoutOr::<Vec<u8>>::Value(ref v) => format!("Value(len={})", v.len()),
TimeoutOr::<Vec<u8>>::Timeout => "Timeout".to_owned(),
},
);
if let TimeoutOr::Value(out) = &out {
self.network_manager()
.stats_packet_rcvd(dial_info.to_ip_addr(), out.len() as u64);
}
Ok(out)
}
}
}
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]

51
veilid-core/src/network_manager/native/protocol/mod.rs
View File

@@ -21,14 +21,14 @@ pub enum ProtocolNetworkConnection {
impl ProtocolNetworkConnection {
pub async fn connect(
local_address: Option<SocketAddr>,
dial_info: DialInfo,
dial_info: &DialInfo,
) -> io::Result<ProtocolNetworkConnection> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
panic!("Should not connect to UDP dialinfo");
}
ProtocolType::TCP => {
tcp::RawTcpProtocolHandler::connect(local_address, dial_info).await
tcp::RawTcpProtocolHandler::connect(local_address, dial_info.to_socket_addr()).await
}
ProtocolType::WS | ProtocolType::WSS => {
ws::WebsocketProtocolHandler::connect(local_address, dial_info).await
@@ -36,53 +36,6 @@ impl ProtocolNetworkConnection {
}
}
pub async fn send_unbound_message(dial_info: DialInfo, data: Vec<u8>) -> io::Result<()> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
udp::RawUdpProtocolHandler::send_unbound_message(peer_socket_addr, data).await
}
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
tcp::RawTcpProtocolHandler::send_unbound_message(peer_socket_addr, data).await
}
ProtocolType::WS | ProtocolType::WSS => {
ws::WebsocketProtocolHandler::send_unbound_message(dial_info, data).await
}
}
}
pub async fn send_recv_unbound_message(
dial_info: DialInfo,
data: Vec<u8>,
timeout_ms: u32,
) -> io::Result<Vec<u8>> {
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(),

139
veilid-core/src/network_manager/native/protocol/tcp.rs
View File

@@ -58,7 +58,7 @@ impl RawTcpNetworkConnection {
Self::send_internal(&mut stream, message).await
}
pub async fn recv_internal(stream: &mut AsyncPeekStream) -> io::Result<Vec<u8>> {
async fn recv_internal(stream: &mut AsyncPeekStream) -> io::Result<Vec<u8>> {
let mut header = [0u8; 4];
stream.read_exact(&mut header).await?;
@@ -141,21 +141,16 @@ impl RawTcpProtocolHandler {
#[instrument(level = "trace", err)]
pub async fn connect(
local_address: Option<SocketAddr>,
dial_info: DialInfo,
socket_addr: SocketAddr,
) -> io::Result<ProtocolNetworkConnection> {
// Get remote socket address to connect to
let remote_socket_addr = dial_info.to_socket_addr();
// Make a shared socket
let socket = match local_address {
Some(a) => new_bound_shared_tcp_socket(a)?,
None => {
new_unbound_shared_tcp_socket(socket2::Domain::for_address(remote_socket_addr))?
}
None => new_unbound_shared_tcp_socket(socket2::Domain::for_address(socket_addr))?,
};
// Non-blocking connect to remote address
let ts = nonblocking_connect(socket, remote_socket_addr).await?;
let ts = nonblocking_connect(socket, socket_addr).await?;
// See what local address we ended up with and turn this into a stream
let actual_local_address = ts.local_addr()?;
@@ -166,7 +161,10 @@ impl RawTcpProtocolHandler {
// Wrap the stream in a network connection and return it
let conn = ProtocolNetworkConnection::RawTcp(RawTcpNetworkConnection::new(
ConnectionDescriptor::new(
dial_info.to_peer_address(),
PeerAddress::new(
SocketAddress::from_socket_addr(socket_addr),
ProtocolType::TCP,
),
SocketAddress::from_socket_addr(actual_local_address),
),
ps,
@@ -175,79 +173,74 @@ impl RawTcpProtocolHandler {
Ok(conn)
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len()))]
pub async fn send_unbound_message(socket_addr: SocketAddr, data: Vec<u8>) -> io::Result<()> {
if data.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large unbound TCP message");
}
trace!(
"sending unbound message of length {} to {}",
data.len(),
socket_addr
);
// #[instrument(level = "trace", err, skip(data), fields(data.len = data.len()))]
// pub async fn send_unbound_message(socket_addr: SocketAddr, data: Vec<u8>) -> io::Result<()> {
// if data.len() > MAX_MESSAGE_SIZE {
// bail_io_error_other!("sending too large unbound TCP message");
// }
// // Make a shared socket
// let socket = new_unbound_shared_tcp_socket(socket2::Domain::for_address(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?;
// Non-blocking connect to remote address
let ts = nonblocking_connect(socket, socket_addr).await?;
// // 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"))?;
// 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"))?;
// #[cfg(feature = "rt-tokio")]
// let ts = ts.compat();
// let mut ps = AsyncPeekStream::new(ts);
#[cfg(feature = "rt-tokio")]
let ts = ts.compat();
let mut ps = AsyncPeekStream::new(ts);
// // 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
// }
// 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
}
// #[instrument(level = "trace", err, skip(data), fields(data.len = data.len(), ret.timeout_or))]
// pub async fn send_recv_unbound_message(
// socket_addr: SocketAddr,
// data: Vec<u8>,
// timeout_ms: u32,
// ) -> io::Result<TimeoutOr<Vec<u8>>> {
// if data.len() > MAX_MESSAGE_SIZE {
// bail_io_error_other!("sending too large unbound TCP message");
// }
#[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,
) -> io::Result<Vec<u8>> {
if data.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large unbound TCP message");
}
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))?;
// 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?;
// Non-blocking connect to remote address
let ts = nonblocking_connect(socket, socket_addr).await?;
// // 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"))?;
// #[cfg(feature = "rt-tokio")]
// let ts = ts.compat();
// let mut ps = AsyncPeekStream::new(ts);
// 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"))?;
#[cfg(feature = "rt-tokio")]
let ts = ts.compat();
let mut ps = AsyncPeekStream::new(ts);
// // 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
// .into_timeout_or()
// .into_result()?;
// 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(|e| e.to_io())??;
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
// tracing::Span::current().record(
// "ret.timeout_or",
// &match out {
// TimeoutOr::<Vec<u8>>::Value(ref v) => format!("Value(len={})", v.len()),
// TimeoutOr::<Vec<u8>>::Timeout => "Timeout".to_owned(),
// },
// );
// Ok(out)
// }
}
impl ProtocolAcceptHandler for RawTcpProtocolHandler {

109
veilid-core/src/network_manager/native/protocol/udp.rs
View File

@@ -60,68 +60,65 @@ impl RawUdpProtocolHandler {
Ok(())
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len()))]
pub async fn send_unbound_message(socket_addr: SocketAddr, data: Vec<u8>) -> io::Result<()> {
if data.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large unbound UDP message");
}
#[instrument(level = "trace", err)]
pub async fn new_unspecified_bound_handler(
socket_addr: &SocketAddr,
) -> io::Result<RawUdpProtocolHandler> {
// 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)
}
};
let local_socket_addr = compatible_unspecified_socket_addr(&socket_addr);
let socket = UdpSocket::bind(local_socket_addr).await?;
let len = socket.send_to(&data, socket_addr).await?;
if len != data.len() {
bail_io_error_other!("UDP partial unbound send")
}
Ok(())
Ok(RawUdpProtocolHandler::new(Arc::new(socket)))
}
#[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,
) -> io::Result<Vec<u8>> {
if data.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large unbound UDP message");
}
// #[instrument(level = "trace", err, skip(data), fields(data.len = data.len(), ret.timeout_or))]
// pub async fn send_recv_unbound_message(
// socket_addr: SocketAddr,
// data: Vec<u8>,
// timeout_ms: u32,
// ) -> io::Result<TimeoutOr<Vec<u8>>> {
// if data.len() > MAX_MESSAGE_SIZE {
// bail_io_error_other!("sending too large unbound UDP message");
// }
// 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 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?;
let len = socket.send_to(&data, socket_addr).await?;
if len != data.len() {
bail_io_error_other!("UDP partial unbound send");
}
// // get unspecified bound socket
// let socket = UdpSocket::bind(local_socket_addr).await?;
// let len = socket.send_to(&data, socket_addr).await?;
// if len != data.len() {
// bail_io_error_other!("UDP partial unbound send");
// }
// 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(|e| e.to_io())??;
// // receive single response
// let mut out = vec![0u8; MAX_MESSAGE_SIZE];
// let timeout_or_ret = timeout(timeout_ms, socket.recv_from(&mut out))
// .await
// .into_timeout_or()
// .into_result()?;
// let (len, from_addr) = match timeout_or_ret {
// TimeoutOr::Value(v) => v,
// TimeoutOr::Timeout => {
// tracing::Span::current().record("ret.timeout_or", &"Timeout".to_owned());
// return Ok(TimeoutOr::Timeout);
// }
// };
// if the from address is not the same as the one we sent to, then drop this
if from_addr != socket_addr {
bail_io_error_other!(format!(
"Unbound response received from wrong address: addr={}",
from_addr,
));
}
out.resize(len, 0u8);
tracing::Span::current().record("ret.len", &len);
Ok(out)
}
// // if the from address is not the same as the one we sent to, then drop this
// if from_addr != socket_addr {
// bail_io_error_other!(format!(
// "Unbound response received from wrong address: addr={}",
// from_addr,
// ));
// }
// out.resize(len, 0u8);
// tracing::Span::current().record("ret.timeout_or", &format!("Value(len={})", out.len()));
// Ok(TimeoutOr::Value(out))
// }
}

47
veilid-core/src/network_manager/native/protocol/ws.rs
View File

@@ -223,12 +223,13 @@ impl WebsocketProtocolHandler {
Ok(Some(conn))
}
async fn connect_internal(
#[instrument(level = "trace", err)]
pub async fn connect(
local_address: Option<SocketAddr>,
dial_info: DialInfo,
dial_info: &DialInfo,
) -> io::Result<ProtocolNetworkConnection> {
// Split dial info up
let (tls, scheme) = match &dial_info {
let (tls, scheme) = match dial_info {
DialInfo::WS(_) => (false, "ws"),
DialInfo::WSS(_) => (true, "wss"),
_ => panic!("invalid dialinfo for WS/WSS protocol"),
@@ -285,46 +286,6 @@ impl WebsocketProtocolHandler {
))
}
}
#[instrument(level = "trace", err)]
pub async fn connect(
local_address: Option<SocketAddr>,
dial_info: DialInfo,
) -> io::Result<ProtocolNetworkConnection> {
Self::connect_internal(local_address, dial_info).await
}
#[instrument(level = "trace", err, skip(data), fields(data.len = data.len()))]
pub async fn send_unbound_message(dial_info: DialInfo, data: Vec<u8>) -> io::Result<()> {
if data.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large unbound WS message");
}
let protconn = Self::connect_internal(None, dial_info.clone()).await?;
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,
) -> io::Result<Vec<u8>> {
if data.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large unbound WS message");
}
let protconn = Self::connect_internal(None, dial_info.clone()).await?;
protconn.send(data).await?;
let out = timeout(timeout_ms, protconn.recv())
.await
.map_err(|e| e.to_io())??;
tracing::Span::current().record("ret.len", &out.len());
Ok(out)
}
}
impl ProtocolAcceptHandler for WebsocketProtocolHandler {