Git-Mirrors/veilid
1
0
Fork 0
mirror of https://gitlab.com/veilid/veilid.git synced 2024-10-01 01:26:08 -04:00
Code Issues Packages Projects Releases Wiki Activity

refactor for flows

Browse Source
This commit is contained in:
Christien Rioux 2023-11-04 14:08:39 -04:00
parent facb343160
commit 0640342556
32 changed files with 468 additions and 424 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
veilid-core/src/network_manager/connection_handle.rs
View File

@ -2,8 +2,8 @@ use super::*;
#[derive(Clone, Debug)]
pub struct ConnectionHandle {
_id: NetworkConnectionId,
descriptor: ConnectionDescriptor,
connection_id: NetworkConnectionId,
flow: Flow,
channel: flume::Sender<(Option<Id>, Vec<u8>)>,
}
@ -15,23 +15,32 @@ pub enum ConnectionHandleSendResult {
impl ConnectionHandle {
pub(super) fn new(
id: NetworkConnectionId,
descriptor: ConnectionDescriptor,
connection_id: NetworkConnectionId,
flow: Flow,
channel: flume::Sender<(Option<Id>, Vec<u8>)>,
) -> Self {
Self {
_id: id,
descriptor,
connection_id,
flow,
channel,
}
}
// pub fn connection_id(&self) -> NetworkConnectionId {
// self.id
// }
#[allow(dead_code)]
pub fn connection_id(&self) -> NetworkConnectionId {
self.connection_id
}
pub fn connection_descriptor(&self) -> ConnectionDescriptor {
self.descriptor
#[allow(dead_code)]
pub fn flow(&self) -> Flow {
self.flow
}
pub fn unique_flow(&self) -> UniqueFlow {
UniqueFlow {
flow: self.flow,
connection_id: Some(self.connection_id),
}
}
// #[cfg_attr(feature="verbose-tracing", instrument(level="trace", skip(self, message), fields(message.len = message.len())))]
@ -57,7 +66,7 @@ impl ConnectionHandle {
impl PartialEq for ConnectionHandle {
fn eq(&self, other: &Self) -> bool {
self.descriptor == other.descriptor
self.connection_id == other.connection_id && self.flow == other.flow
}
}

62
veilid-core/src/network_manager/connection_manager.rs
View File

@ -16,23 +16,25 @@ enum ConnectionManagerEvent {
#[derive(Debug)]
pub(crate) struct ConnectionRefScope {
connection_manager: ConnectionManager,
descriptor: ConnectionDescriptor,
id: NetworkConnectionId,
}
impl ConnectionRefScope {
pub fn new(connection_manager: ConnectionManager, descriptor: ConnectionDescriptor) -> Self {
connection_manager.connection_ref(descriptor, ConnectionRefKind::AddRef);
Self {
connection_manager,
descriptor,
pub fn try_new(connection_manager: ConnectionManager, id: NetworkConnectionId) -> Option<Self> {
if !connection_manager.connection_ref(id, ConnectionRefKind::AddRef) {
return None;
}
Some(Self {
connection_manager,
id,
})
}
}
impl Drop for ConnectionRefScope {
fn drop(&mut self) {
self.connection_manager
.connection_ref(self.descriptor, ConnectionRefKind::RemoveRef);
.connection_ref(self.id, ConnectionRefKind::RemoveRef);
}
}
@ -163,7 +165,7 @@ impl ConnectionManager {
fn should_protect_connection(&self, conn: &NetworkConnection) -> Option<NodeRef> {
let netman = self.network_manager();
let routing_table = netman.routing_table();
let remote_address = conn.connection_descriptor().remote_address().address();
let remote_address = conn.flow().remote_address().address();
let Some(routing_domain) = routing_table.routing_domain_for_address(remote_address) else {
return None;
};
@ -173,8 +175,8 @@ impl ConnectionManager {
let relay_nr = rn.filtered_clone(
NodeRefFilter::new()
.with_routing_domain(routing_domain)
.with_address_type(conn.connection_descriptor().address_type())
.with_protocol_type(conn.connection_descriptor().protocol_type()),
.with_address_type(conn.flow().address_type())
.with_protocol_type(conn.flow().protocol_type()),
);
let dids = relay_nr.all_filtered_dial_info_details();
for did in dids {
@ -231,7 +233,7 @@ impl ConnectionManager {
}
Err(ConnectionTableAddError::AddressFilter(conn, e)) => {
// Connection filtered
let desc = conn.connection_descriptor();
let desc = conn.flow();
let _ = inner.sender.send(ConnectionManagerEvent::Dead(conn));
return Ok(NetworkResult::no_connection_other(format!(
"connection filtered: {:?} ({})",
@ -240,7 +242,7 @@ impl ConnectionManager {
}
Err(ConnectionTableAddError::AlreadyExists(conn)) => {
// Connection already exists
let desc = conn.connection_descriptor();
let desc = conn.flow();
log_net!(debug "== Connection already exists: {:?}", conn.debug_print(get_aligned_timestamp()));
let _ = inner.sender.send(ConnectionManagerEvent::Dead(conn));
return Ok(NetworkResult::no_connection_other(format!(
@ -250,7 +252,7 @@ impl ConnectionManager {
}
Err(ConnectionTableAddError::TableFull(conn)) => {
// Connection table is full
let desc = conn.connection_descriptor();
let desc = conn.flow();
log_net!(debug "== Connection table full: {:?}", conn.debug_print(get_aligned_timestamp()));
let _ = inner.sender.send(ConnectionManagerEvent::Dead(conn));
return Ok(NetworkResult::no_connection_other(format!(
@ -263,10 +265,8 @@ impl ConnectionManager {
}
// Returns a network connection if one already is established
pub fn get_connection(&self, descriptor: ConnectionDescriptor) -> Option<ConnectionHandle> {
self.arc
.connection_table
.peek_connection_by_descriptor(descriptor)
pub fn get_connection(&self, flow: Flow) -> Option<ConnectionHandle> {
self.arc.connection_table.peek_connection_by_flow(flow)
}
// Returns a network connection if one already is established
@ -275,13 +275,11 @@ impl ConnectionManager {
}
// Protects a network connection if one already is established
fn connection_ref(&self, descriptor: ConnectionDescriptor, kind: ConnectionRefKind) {
self.arc
.connection_table
.ref_connection_by_descriptor(descriptor, kind);
fn connection_ref(&self, id: NetworkConnectionId, kind: ConnectionRefKind) -> bool {
self.arc.connection_table.ref_connection_by_id(id, kind)
}
pub fn connection_ref_scope(&self, descriptor: ConnectionDescriptor) -> ConnectionRefScope {
ConnectionRefScope::new(self.clone(), descriptor)
pub fn try_connection_ref_scope(&self, id: NetworkConnectionId) -> Option<ConnectionRefScope> {
ConnectionRefScope::try_new(self.clone(), id)
}
/// Called when we want to create a new connection or get the current one that already exists
@ -385,14 +383,22 @@ impl ConnectionManager {
// Process async commands
while let Ok(Ok(event)) = receiver.recv_async().timeout_at(stop_token.clone()).await {
match event {
ConnectionManagerEvent::Accepted(conn) => {
ConnectionManagerEvent::Accepted(prot_conn) => {
// Async lock on the remote address for atomicity per remote
let _lock_guard = self
.arc
.address_lock_table
.lock_tag(prot_conn.flow().remote_address().socket_addr())
.await;
let mut inner = self.arc.inner.lock();
match &mut *inner {
Some(inner) => {
// Register the connection
// We don't care if this fails, since nobody here asked for the inbound connection.
// If it does, we just drop the connection
let _ = self.on_new_protocol_network_connection(inner, conn);
let _ = self.on_new_protocol_network_connection(inner, prot_conn);
}
None => {
// If this somehow happens, we're shutting down
@ -400,6 +406,12 @@ impl ConnectionManager {
};
}
ConnectionManagerEvent::Dead(mut conn) => {
let _lock_guard = self
.arc
.address_lock_table
.lock_tag(conn.flow().remote_address().socket_addr())
.await;
conn.close();
conn.await;
}

82
veilid-core/src/network_manager/connection_table.rs
View File

@ -38,7 +38,7 @@ struct ConnectionTableInner {
max_connections: Vec<usize>,
conn_by_id: Vec<LruCache<NetworkConnectionId, NetworkConnection>>,
protocol_index_by_id: BTreeMap<NetworkConnectionId, usize>,
id_by_descriptor: BTreeMap<ConnectionDescriptor, NetworkConnectionId>,
id_by_flow: BTreeMap<Flow, NetworkConnectionId>,
ids_by_remote: BTreeMap<PeerAddress, Vec<NetworkConnectionId>>,
address_filter: AddressFilter,
}
@ -67,7 +67,7 @@ impl ConnectionTable {
LruCache::new_unbounded(),
],
protocol_index_by_id: BTreeMap::new(),
id_by_descriptor: BTreeMap::new(),
id_by_flow: BTreeMap::new(),
ids_by_remote: BTreeMap::new(),
address_filter,
})),
@ -105,7 +105,7 @@ impl ConnectionTable {
}
}
inner.protocol_index_by_id.clear();
inner.id_by_descriptor.clear();
inner.id_by_flow.clear();
inner.ids_by_remote.clear();
unord
};
@ -151,14 +151,14 @@ impl ConnectionTable {
) -> Result<Option<NetworkConnection>, ConnectionTableAddError> {
// Get indices for network connection table
let id = network_connection.connection_id();
let descriptor = network_connection.connection_descriptor();
let protocol_index = Self::protocol_to_index(descriptor.protocol_type());
let remote = descriptor.remote();
let flow = network_connection.flow();
let protocol_index = Self::protocol_to_index(flow.protocol_type());
let remote = flow.remote();
let mut inner = self.inner.lock();
// Two connections to the same descriptor should be rejected (soft rejection)
if inner.id_by_descriptor.contains_key(&descriptor) {
// Two connections to the same flow should be rejected (soft rejection)
if inner.id_by_flow.contains_key(&flow) {
return Err(ConnectionTableAddError::already_exists(network_connection));
}
@ -169,14 +169,14 @@ impl ConnectionTable {
if inner.protocol_index_by_id.get(&id).is_some() {
panic!("duplicate id to protocol index: {:#?}", network_connection);
}
if let Some(ids) = inner.ids_by_remote.get(&descriptor.remote()) {
if let Some(ids) = inner.ids_by_remote.get(&flow.remote()) {
if ids.contains(&id) {
panic!("duplicate id by remote: {:#?}", network_connection);
}
}
// Filter by ip for connection limits
let ip_addr = descriptor.remote_address().ip_addr();
let ip_addr = flow.remote_address().ip_addr();
match inner.address_filter.add_connection(ip_addr) {
Ok(()) => {}
Err(e) => {
@ -217,12 +217,26 @@ impl ConnectionTable {
// add connection records
inner.protocol_index_by_id.insert(id, protocol_index);
inner.id_by_descriptor.insert(descriptor, id);
inner.id_by_flow.insert(flow, id);
inner.ids_by_remote.entry(remote).or_default().push(id);
Ok(out_conn)
}
//#[instrument(level = "trace", skip(self), ret)]
pub fn peek_connection_by_flow(&self, flow: Flow) -> Option<ConnectionHandle> {
if flow.protocol_type() == ProtocolType::UDP {
return None;
}
let inner = self.inner.lock();
let id = *inner.id_by_flow.get(&flow)?;
let protocol_index = Self::protocol_to_index(flow.protocol_type());
let out = inner.conn_by_id[protocol_index].peek(&id).unwrap();
Some(out.get_handle())
}
//#[instrument(level = "trace", skip(self), ret)]
pub fn touch_connection_by_id(&self, id: NetworkConnectionId) {
let mut inner = self.inner.lock();
@ -233,45 +247,21 @@ impl ConnectionTable {
}
//#[instrument(level = "trace", skip(self), ret)]
pub fn peek_connection_by_descriptor(
pub fn ref_connection_by_id(
&self,
descriptor: ConnectionDescriptor,
) -> Option<ConnectionHandle> {
if descriptor.protocol_type() == ProtocolType::UDP {
return None;
}
let inner = self.inner.lock();
let id = *inner.id_by_descriptor.get(&descriptor)?;
let protocol_index = Self::protocol_to_index(descriptor.protocol_type());
let out = inner.conn_by_id[protocol_index].peek(&id).unwrap();
Some(out.get_handle())
}
//#[instrument(level = "trace", skip(self), ret)]
pub fn ref_connection_by_descriptor(
&self,
descriptor: ConnectionDescriptor,
id: NetworkConnectionId,
ref_type: ConnectionRefKind,
) -> bool {
if descriptor.protocol_type() == ProtocolType::UDP {
return false;
}
let mut inner = self.inner.lock();
let Some(id) = inner.id_by_descriptor.get(&descriptor).copied() else {
log_net!(error "failed to ref descriptor: {:?} ({:?})", descriptor, ref_type);
let Some(protocol_index) = inner.protocol_index_by_id.get(&id).copied() else {
// Sometimes network connections die before we can ref/unref them
return false;
};
let protocol_index = Self::protocol_to_index(descriptor.protocol_type());
let out = inner.conn_by_id[protocol_index].get_mut(&id).unwrap();
match ref_type {
ConnectionRefKind::AddRef => out.add_ref(),
ConnectionRefKind::RemoveRef => out.remove_ref(),
}
true
}
@ -299,7 +289,7 @@ impl ConnectionTable {
if let Some(best_port) = best_port {
for id in all_ids_by_remote {
let nc = inner.conn_by_id[protocol_index].peek(id).unwrap();
if let Some(local_addr) = nc.connection_descriptor().local() {
if let Some(local_addr) = nc.flow().local() {
if local_addr.port() == best_port {
let nc = inner.conn_by_id[protocol_index].get(id).unwrap();
return Some(nc.get_handle());
@ -326,13 +316,13 @@ impl ConnectionTable {
// pub fn drain_filter<F>(&self, mut filter: F) -> Vec<NetworkConnection>
// where
// F: FnMut(ConnectionDescriptor) -> bool,
// F: FnMut(Flow) -> bool,
// {
// let mut inner = self.inner.lock();
// let mut filtered_ids = Vec::new();
// for cbi in &mut inner.conn_by_id {
// for (id, conn) in cbi {
// if filter(conn.connection_descriptor()) {
// if filter(conn.flow()) {
// filtered_ids.push(*id);
// }
// }
@ -359,11 +349,11 @@ impl ConnectionTable {
let protocol_index = inner.protocol_index_by_id.remove(&id).unwrap();
// conn_by_id
let conn = inner.conn_by_id[protocol_index].remove(&id).unwrap();
// id_by_descriptor
let descriptor = conn.connection_descriptor();
inner.id_by_descriptor.remove(&descriptor).unwrap();
// id_by_flow
let flow = conn.flow();
inner.id_by_flow.remove(&flow).unwrap();
// ids_by_remote
let remote = descriptor.remote();
let remote = flow.remote();
let ids = inner.ids_by_remote.get_mut(&remote).unwrap();
for (n, elem) in ids.iter().enumerate() {
if *elem == id {

11
veilid-core/src/network_manager/direct_boot.rs
View File

@ -3,10 +3,7 @@ use super::*;
impl NetworkManager {
// Direct bootstrap request handler (separate fallback mechanism from cheaper TXT bootstrap mechanism)
#[instrument(level = "trace", skip(self), ret, err)]
pub(crate) async fn handle_boot_request(
&self,
descriptor: ConnectionDescriptor,
) -> EyreResult<NetworkResult<()>> {
pub(crate) async fn handle_boot_request(&self, flow: Flow) -> EyreResult<NetworkResult<()>> {
let routing_table = self.routing_table();
// Get a bunch of nodes with the various
@ -22,14 +19,14 @@ impl NetworkManager {
// Reply with a chunk of signed routing table
match self
.net()
.send_data_to_existing_connection(descriptor, json_bytes)
.send_data_to_existing_flow(flow, json_bytes)
.await?
{
None => {
SendDataToExistingFlowResult::Sent(_) => {
// Bootstrap reply was sent
Ok(NetworkResult::value(()))
}
Some(_) => Ok(NetworkResult::no_connection_other(
SendDataToExistingFlowResult::NotSent(_) => Ok(NetworkResult::no_connection_other(
"bootstrap reply could not be sent",
)),
}

55
veilid-core/src/network_manager/mod.rs
View File

@ -95,8 +95,8 @@ pub(crate) struct SendDataMethod {
pub contact_method: NodeContactMethod,
/// Pre-relayed contact method
pub opt_relayed_contact_method: Option<NodeContactMethod>,
/// The connection used to send the data
pub connection_descriptor: ConnectionDescriptor,
/// The specific flow used to send the data
pub unique_flow: UniqueFlow,
}
/// Mechanism required to contact another node
@ -128,6 +128,11 @@ struct NodeContactMethodCacheKey {
#[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd, Hash)]
struct PublicAddressCheckCacheKey(ProtocolType, AddressType);
enum SendDataToExistingFlowResult {
Sent(UniqueFlow),
NotSent(Vec<u8>),
}
// The mutable state of the network manager
struct NetworkManagerInner {
stats: NetworkManagerStats,
@ -661,7 +666,7 @@ impl NetworkManager {
#[instrument(level = "trace", skip(self), err)]
pub async fn handle_signal(
&self,
signal_connection_descriptor: ConnectionDescriptor,
signal_flow: Flow,
signal_info: SignalInfo,
) -> EyreResult<NetworkResult<()>> {
match signal_info {
@ -685,7 +690,7 @@ impl NetworkManager {
};
// Restrict reverse connection to same sequencing requirement as inbound signal
if signal_connection_descriptor.protocol_type().is_ordered() {
if signal_flow.protocol_type().is_ordered() {
peer_nr.set_sequencing(Sequencing::EnsureOrdered);
}
@ -730,7 +735,7 @@ impl NetworkManager {
// Do our half of the hole punch by sending an empty packet
// Both sides will do this and then the receipt will get sent over the punched hole
let connection_descriptor = network_result_try!(
let unique_flow = network_result_try!(
self.net()
.send_data_to_dial_info(
hole_punch_dial_info_detail.dial_info.clone(),
@ -742,7 +747,7 @@ impl NetworkManager {
// XXX: do we need a delay here? or another hole punch packet?
// Set the hole punch as our 'last connection' to ensure we return the receipt over the direct hole punch
peer_nr.set_last_connection(connection_descriptor, get_aligned_timestamp());
peer_nr.set_last_flow(unique_flow.flow, get_aligned_timestamp());
// Return the receipt using the same dial info send the receipt to it
rpc.rpc_call_return_receipt(Destination::direct(peer_nr), receipt)
@ -867,28 +872,20 @@ impl NetworkManager {
// network protocol handler. Processes the envelope, authenticates and decrypts the RPC message
// and passes it to the RPC handler
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", ret, err, skip(self, data), fields(data.len = data.len())))]
async fn on_recv_envelope(
&self,
data: &mut [u8],
connection_descriptor: ConnectionDescriptor,
) -> EyreResult<bool> {
async fn on_recv_envelope(&self, data: &mut [u8], flow: Flow) -> EyreResult<bool> {
#[cfg(feature = "verbose-tracing")]
let root = span!(
parent: None,
Level::TRACE,
"on_recv_envelope",
"data.len" = data.len(),
"descriptor" = ?connection_descriptor
"flow" = ?flow
);
#[cfg(feature = "verbose-tracing")]
let _root_enter = root.enter();
log_net!(
"envelope of {} bytes received from {:?}",
data.len(),
connection_descriptor
);
let remote_addr = connection_descriptor.remote_address().ip_addr();
log_net!("envelope of {} bytes received from {:?}", data.len(), flow);
let remote_addr = flow.remote_address().ip_addr();
// Network accounting
self.stats_packet_rcvd(remote_addr, ByteCount::new(data.len() as u64));
@ -910,18 +907,18 @@ impl NetworkManager {
// Get the routing domain for this data
let routing_domain = match self
.routing_table()
.routing_domain_for_address(connection_descriptor.remote_address().address())
.routing_domain_for_address(flow.remote_address().address())
{
Some(rd) => rd,
None => {
log_net!(debug "no routing domain for envelope received from {:?}", connection_descriptor);
log_net!(debug "no routing domain for envelope received from {:?}", flow);
return Ok(false);
}
};
// Is this a direct bootstrap request instead of an envelope?
if data[0..4] == *BOOT_MAGIC {
network_result_value_or_log!(self.handle_boot_request(connection_descriptor).await? => [ format!(": connection_descriptor={:?}", connection_descriptor) ] {});
network_result_value_or_log!(self.handle_boot_request(flow).await? => [ format!(": flow={:?}", flow) ] {});
return Ok(true);
}
@ -968,7 +965,7 @@ impl NetworkManager {
log_net!(debug
"Timestamp behind: {}ms ({})",
timestamp_to_secs(ts.saturating_sub(ets).as_u64()) * 1000f64,
connection_descriptor.remote()
flow.remote()
);
return Ok(false);
}
@ -978,7 +975,7 @@ impl NetworkManager {
log_net!(debug
"Timestamp ahead: {}ms ({})",
timestamp_to_secs(ets.saturating_sub(ts).as_u64()) * 1000f64,
connection_descriptor.remote()
flow.remote()
);
return Ok(false);
}
@ -1033,7 +1030,7 @@ impl NetworkManager {
if let Some(mut relay_nr) = some_relay_nr {
// Ensure the protocol used to forward is of the same sequencing requirement
// Address type is allowed to change if connectivity is better
if connection_descriptor.protocol_type().is_ordered() {
if flow.protocol_type().is_ordered() {
relay_nr.set_sequencing(Sequencing::EnsureOrdered);
};
@ -1080,7 +1077,7 @@ impl NetworkManager {
// Cache the envelope information in the routing table
let source_noderef = match routing_table.register_node_with_existing_connection(
envelope.get_sender_typed_id(),
connection_descriptor,
flow,
ts,
) {
Ok(v) => v,
@ -1093,13 +1090,7 @@ impl NetworkManager {
source_noderef.add_envelope_version(envelope.get_version());
// Pass message to RPC system
rpc.enqueue_direct_message(
envelope,
source_noderef,
connection_descriptor,
routing_domain,
body,
)?;
rpc.enqueue_direct_message(envelope, source_noderef, flow, routing_domain, body)?;
// Inform caller that we dealt with the envelope locally
Ok(true)

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

@ -578,75 +578,79 @@ impl Network {
}
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_to_existing_connection(
pub async fn send_data_to_existing_flow(
&self,
descriptor: ConnectionDescriptor,
flow: Flow,
data: Vec<u8>,
) -> EyreResult<Option<Vec<u8>>> {
) -> EyreResult<SendDataToExistingFlowResult> {
let data_len = data.len();
// Handle connectionless protocol
if descriptor.protocol_type() == ProtocolType::UDP {
if flow.protocol_type() == ProtocolType::UDP {
// send over the best udp socket we have bound since UDP is not connection oriented
let peer_socket_addr = descriptor.remote().socket_addr();
let peer_socket_addr = flow.remote().socket_addr();
if let Some(ph) = self.find_best_udp_protocol_handler(
&peer_socket_addr,
&descriptor.local().map(|sa| sa.socket_addr()),
&flow.local().map(|sa| sa.socket_addr()),
) {
network_result_value_or_log!(ph.clone()
.send_message(data.clone(), peer_socket_addr)
.await
.wrap_err("sending data to existing connection")? => [ format!(": data.len={}, descriptor={:?}", data.len(), descriptor) ]
{ return Ok(Some(data)); } );
.wrap_err("sending data to existing connection")? => [ format!(": data.len={}, flow={:?}", data.len(), flow) ]
{ return Ok(SendDataToExistingFlowResult::NotSent(data)); } );
// Network accounting
self.network_manager()
.stats_packet_sent(peer_socket_addr.ip(), ByteCount::new(data_len as u64));
// Data was consumed
return Ok(None);
let unique_flow = UniqueFlow {
flow,
connection_id: None,
};
return Ok(SendDataToExistingFlowResult::Sent(unique_flow));
}
}
// Handle connection-oriented protocols
// Try to send to the exact existing connection if one exists
if let Some(conn) = self.connection_manager().get_connection(descriptor) {
if let Some(conn) = self.connection_manager().get_connection(flow) {
// connection exists, send over it
match conn.send_async(data).await {
ConnectionHandleSendResult::Sent => {
// Network accounting
self.network_manager().stats_packet_sent(
descriptor.remote().socket_addr().ip(),
flow.remote().socket_addr().ip(),
ByteCount::new(data_len as u64),
);
// Data was consumed
return Ok(None);
return Ok(SendDataToExistingFlowResult::Sent(conn.unique_flow()));
}
ConnectionHandleSendResult::NotSent(data) => {
// Couldn't send
// Pass the data back out so we don't own it any more
return Ok(Some(data));
return Ok(SendDataToExistingFlowResult::NotSent(data));
}
}
}
// Connection didn't exist
// Pass the data back out so we don't own it any more
Ok(Some(data))
Ok(SendDataToExistingFlowResult::NotSent(data))
}
// Send data directly to a dial info, possibly without knowing which node it is going to
// Returns a descriptor for the connection used to send the data
// Returns a flow for the connection used to send the data
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_to_dial_info(
&self,
dial_info: DialInfo,
data: Vec<u8>,
) -> EyreResult<NetworkResult<ConnectionDescriptor>> {
) -> EyreResult<NetworkResult<UniqueFlow>> {
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
let connection_descriptor;
let unique_flow;
if dial_info.protocol_type() == ProtocolType::UDP {
// Handle connectionless protocol
let peer_socket_addr = dial_info.to_socket_addr();
@ -658,10 +662,14 @@ impl Network {
));
}
};
connection_descriptor = network_result_try!(ph
let flow = network_result_try!(ph
.send_message(data, peer_socket_addr)
.await
.wrap_err("failed to send data to dial info")?);
unique_flow = UniqueFlow {
flow,
connection_id: None,
};
} else {
// Handle connection-oriented protocols
let conn = network_result_try!(
@ -676,14 +684,14 @@ impl Network {
"failed to send",
)));
}
connection_descriptor = conn.connection_descriptor();
unique_flow = conn.unique_flow();
}
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.ip_addr(), ByteCount::new(data_len as u64));
Ok(NetworkResult::value(connection_descriptor))
Ok(NetworkResult::value(unique_flow))
})
.await
}

6
veilid-core/src/network_manager/native/network_udp.rs
View File

@ -65,16 +65,16 @@ impl Network {
.timeout_at(stop_token.clone())
.await
{
Ok(Ok((size, descriptor))) => {
Ok(Ok((size, flow))) => {
// Network accounting
network_manager.stats_packet_rcvd(
descriptor.remote_address().ip_addr(),
flow.remote_address().ip_addr(),
ByteCount::new(size as u64),
);
// Pass it up for processing
if let Err(e) = network_manager
.on_recv_envelope(&mut data[..size], descriptor)
.on_recv_envelope(&mut data[..size], flow)
.await
{
log_net!(debug "failed to process received udp envelope: {}", e);

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

@ -45,13 +45,13 @@ impl ProtocolNetworkConnection {
}
}
pub fn descriptor(&self) -> ConnectionDescriptor {
pub fn flow(&self) -> Flow {
match self {
// Self::Dummy(d) => d.descriptor(),
Self::RawTcp(t) => t.descriptor(),
Self::WsAccepted(w) => w.descriptor(),
Self::Ws(w) => w.descriptor(),
Self::Wss(w) => w.descriptor(),
// Self::Dummy(d) => d.flow(),
Self::RawTcp(t) => t.flow(),
Self::WsAccepted(w) => w.flow(),
Self::Ws(w) => w.flow(),
Self::Wss(w) => w.flow(),
}
}

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

@ -3,7 +3,7 @@ use futures_util::{AsyncReadExt, AsyncWriteExt};
use sockets::*;
pub struct RawTcpNetworkConnection {
descriptor: ConnectionDescriptor,
flow: Flow,
stream: AsyncPeekStream,
}
@ -14,12 +14,12 @@ impl fmt::Debug for RawTcpNetworkConnection {
}
impl RawTcpNetworkConnection {
pub fn new(descriptor: ConnectionDescriptor, stream: AsyncPeekStream) -> Self {
Self { descriptor, stream }
pub fn new(flow: Flow, stream: AsyncPeekStream) -> Self {
Self { flow, stream }
}
pub fn descriptor(&self) -> ConnectionDescriptor {
self.descriptor
pub fn flow(&self) -> Flow {
self.flow
}
#[cfg_attr(
@ -152,7 +152,7 @@ impl RawTcpProtocolHandler {
ProtocolType::TCP,
);
let conn = ProtocolNetworkConnection::RawTcp(RawTcpNetworkConnection::new(
ConnectionDescriptor::new(peer_addr, SocketAddress::from_socket_addr(local_addr)),
Flow::new(peer_addr, SocketAddress::from_socket_addr(local_addr)),
ps,
));
@ -186,7 +186,7 @@ impl RawTcpProtocolHandler {
// Wrap the stream in a network connection and return it
let conn = ProtocolNetworkConnection::RawTcp(RawTcpNetworkConnection::new(
ConnectionDescriptor::new(
Flow::new(
PeerAddress::new(
SocketAddress::from_socket_addr(socket_addr),
ProtocolType::TCP,

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

@ -17,9 +17,9 @@ impl RawUdpProtocolHandler {
}
}
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.len, ret.descriptor)))]
pub async fn recv_message(&self, data: &mut [u8]) -> io::Result<(usize, ConnectionDescriptor)> {
let (message_len, descriptor) = loop {
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.len, ret.flow)))]
pub async fn recv_message(&self, data: &mut [u8]) -> io::Result<(usize, Flow)> {
let (message_len, flow) = loop {
// Get a packet
let (size, remote_addr) = network_result_value_or_log!(self.socket.recv_from(data).await.into_network_result()? => continue);
@ -64,33 +64,33 @@ impl RawUdpProtocolHandler {
// Copy assemble message out if we got one
data[0..message.len()].copy_from_slice(&message);
// Return a connection descriptor and the amount of data in the message
// Return a flow and the amount of data in the message
let peer_addr = PeerAddress::new(
SocketAddress::from_socket_addr(remote_addr),
ProtocolType::UDP,
);
let local_socket_addr = self.socket.local_addr()?;
let descriptor = ConnectionDescriptor::new(
let flow = Flow::new(
peer_addr,
SocketAddress::from_socket_addr(local_socket_addr),
);
break (message.len(), descriptor);
break (message.len(), flow);
};
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.len", message_len);
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.descriptor", format!("{:?}", descriptor).as_str());
Ok((message_len, descriptor))
tracing::Span::current().record("ret.flow", format!("{:?}", flow).as_str());
Ok((message_len, flow))
}
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.descriptor)))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.flow)))]
pub async fn send_message(
&self,
data: Vec<u8>,
remote_addr: SocketAddr,
) -> io::Result<NetworkResult<ConnectionDescriptor>> {
) -> io::Result<NetworkResult<Flow>> {
if data.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large UDP message");
}
@ -121,21 +121,21 @@ impl RawUdpProtocolHandler {
.await?
);
// Return a connection descriptor for the sent message
// Return a flow for the sent message
let peer_addr = PeerAddress::new(
SocketAddress::from_socket_addr(remote_addr),
ProtocolType::UDP,
);
let local_socket_addr = self.socket.local_addr()?;
let descriptor = ConnectionDescriptor::new(
let flow = Flow::new(
peer_addr,
SocketAddress::from_socket_addr(local_socket_addr),
);
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.descriptor", format!("{:?}", descriptor).as_str());
Ok(NetworkResult::value(descriptor))
tracing::Span::current().record("ret.flow", format!("{:?}", flow).as_str());
Ok(NetworkResult::value(flow))
}
#[instrument(level = "trace", err)]

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

@ -54,7 +54,7 @@ pub struct WebsocketNetworkConnection<T>
where
T: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
descriptor: ConnectionDescriptor,
flow: Flow,
stream: CloneStream<WebSocketStream<T>>,
}
@ -71,15 +71,15 @@ impl<T> WebsocketNetworkConnection<T>
where
T: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
pub fn new(descriptor: ConnectionDescriptor, stream: WebSocketStream<T>) -> Self {
pub fn new(flow: Flow, stream: WebSocketStream<T>) -> Self {
Self {
descriptor,
flow,
stream: CloneStream::new(stream),
}
}
pub fn descriptor(&self) -> ConnectionDescriptor {
self.descriptor
pub fn flow(&self) -> Flow {
self.flow
}
#[cfg_attr(
@ -286,7 +286,7 @@ impl WebsocketProtocolHandler {
PeerAddress::new(SocketAddress::from_socket_addr(socket_addr), protocol_type);
let conn = ProtocolNetworkConnection::WsAccepted(WebsocketNetworkConnection::new(
ConnectionDescriptor::new(peer_addr, SocketAddress::from_socket_addr(local_addr)),
Flow::new(peer_addr, SocketAddress::from_socket_addr(local_addr)),
ws_stream,
));
@ -335,8 +335,8 @@ impl WebsocketProtocolHandler {
#[cfg(feature = "rt-tokio")]
let tcp_stream = tcp_stream.compat();
// Make our connection descriptor
let descriptor = ConnectionDescriptor::new(
// Make our flow
let flow = Flow::new(
dial_info.peer_address(),
SocketAddress::from_socket_addr(actual_local_addr),
);
@ -350,14 +350,14 @@ impl WebsocketProtocolHandler {
.map_err(to_io_error_other)?;
Ok(NetworkResult::Value(ProtocolNetworkConnection::Wss(
WebsocketNetworkConnection::new(descriptor, ws_stream),
WebsocketNetworkConnection::new(flow, ws_stream),
)))
} else {
let (ws_stream, _response) = client_async(request, tcp_stream)
.await
.map_err(to_io_error_other)?;
Ok(NetworkResult::Value(ProtocolNetworkConnection::Ws(
WebsocketNetworkConnection::new(descriptor, ws_stream),
WebsocketNetworkConnection::new(flow, ws_stream),
)))
}
}

54
veilid-core/src/network_manager/network_connection.rs
View File

@ -47,12 +47,12 @@ cfg_if::cfg_if! {
// #[derive(Debug)]
// pub struct DummyNetworkConnection {
// descriptor: ConnectionDescriptor,
// flow: Flow,
// }
// impl DummyNetworkConnection {
// pub fn descriptor(&self) -> ConnectionDescriptor {
// self.descriptor
// pub fn flow(&self) -> Flow {
// self.flow
// }
// pub fn close(&self) -> io::Result<NetworkResult<()>> {
// Ok(NetworkResult::Value(()))
@ -83,12 +83,10 @@ pub struct NetworkConnectionStats {
}
pub type NetworkConnectionId = AlignedU64;
#[derive(Debug)]
pub(in crate::network_manager) struct NetworkConnection {
connection_id: NetworkConnectionId,
descriptor: ConnectionDescriptor,
flow: Flow,
processor: Option<MustJoinHandle<()>>,
established_time: Timestamp,
stats: Arc<Mutex<NetworkConnectionStats>>,
@ -108,13 +106,13 @@ impl Drop for NetworkConnection {
impl NetworkConnection {
pub(super) fn dummy(id: NetworkConnectionId, descriptor: ConnectionDescriptor) -> Self {
pub(super) fn dummy(id: NetworkConnectionId, flow: Flow) -> Self {
// Create handle for sending (dummy is immediately disconnected)
let (sender, _receiver) = flume::bounded(get_concurrency() as usize);
Self {
connection_id: id,
descriptor,
flow,
processor: None,
established_time: get_aligned_timestamp(),
stats: Arc::new(Mutex::new(NetworkConnectionStats {
@ -134,8 +132,8 @@ impl NetworkConnection {
protocol_connection: ProtocolNetworkConnection,
connection_id: NetworkConnectionId,
) -> Self {
// Get descriptor
let descriptor = protocol_connection.descriptor();
// Get flow
let flow = protocol_connection.flow();
// Create handle for sending
let (sender, receiver) = flume::bounded(get_concurrency() as usize);
@ -155,7 +153,7 @@ impl NetworkConnection {
local_stop_token,
manager_stop_token,
connection_id,
descriptor,
flow,
receiver,
protocol_connection,
stats.clone(),
@ -164,7 +162,7 @@ impl NetworkConnection {
// Return the connection
Self {
connection_id,
descriptor,
flow,
processor: Some(processor),
established_time: get_aligned_timestamp(),
stats,
@ -179,12 +177,20 @@ impl NetworkConnection {
self.connection_id
}
pub fn connection_descriptor(&self) -> ConnectionDescriptor {
self.descriptor
pub fn flow(&self) -> Flow {
self.flow
}
#[allow(dead_code)]
pub fn unique_flow(&self) -> UniqueFlow {
UniqueFlow {
flow: self.flow,
connection_id: Some(self.connection_id),
}
}
pub fn get_handle(&self) -> ConnectionHandle {
ConnectionHandle::new(self.connection_id, self.descriptor, self.sender.clone())
ConnectionHandle::new(self.connection_id, self.flow, self.sender.clone())
}
pub fn is_in_use(&self) -> bool {
@ -264,7 +270,7 @@ impl NetworkConnection {
local_stop_token: StopToken,
manager_stop_token: StopToken,
connection_id: NetworkConnectionId,
descriptor: ConnectionDescriptor,
flow: Flow,
receiver: flume::Receiver<(Option<Id>, Vec<u8>)>,
protocol_connection: ProtocolNetworkConnection,
stats: Arc<Mutex<NetworkConnectionStats>>,
@ -272,7 +278,7 @@ impl NetworkConnection {
Box::pin(async move {
log_net!(
"== Starting process_connection loop for id={}, {:?}", connection_id,
descriptor
flow
);
let network_manager = connection_manager.network_manager();
@ -286,7 +292,7 @@ impl NetworkConnection {
let new_timer = || {
sleep(connection_manager.connection_inactivity_timeout_ms()).then(|_| async {
// timeout
log_net!("== Connection timeout on {:?}", descriptor);
log_net!("== Connection timeout on {:?}", flow);
RecvLoopAction::Timeout
})
};
@ -341,7 +347,7 @@ impl NetworkConnection {
.then(|res| async {
match res {
Ok(v) => {
let peer_address = protocol_connection.descriptor().remote();
let peer_address = protocol_connection.flow().remote();
// Check to see if it is punished
if address_filter.is_ip_addr_punished(peer_address.socket_addr().ip()) {
@ -367,7 +373,7 @@ impl NetworkConnection {
// Pass received messages up to the network manager for processing
if let Err(e) = network_manager
.on_recv_envelope(message.as_mut_slice(), descriptor)
.on_recv_envelope(message.as_mut_slice(), flow)
.await
{
log_net!(debug "failed to process received envelope: {}", e);
@ -424,8 +430,8 @@ impl NetworkConnection {
}
log_net!(
"== Connection loop finished descriptor={:?}",
descriptor
"== Connection loop finished flow={:?}",
flow
);
// Let the connection manager know the receive loop exited
@ -443,8 +449,8 @@ impl NetworkConnection {
pub fn debug_print(&self, cur_ts: Timestamp) -> String {
format!("{} <- {} | {} | est {} sent {} rcvd {} refcount {}{}",
self.descriptor.remote_address(),
self.descriptor.local().map(|x| x.to_string()).unwrap_or("---".to_owned()),
self.flow.remote_address(),
self.flow.local().map(|x| x.to_string()).unwrap_or("---".to_owned()),
self.connection_id.as_u64(),
debug_duration(cur_ts.as_u64().saturating_sub(self.established_time.as_u64())),
self.stats().last_message_sent_time.map(|ts| debug_duration(cur_ts.as_u64().saturating_sub(ts.as_u64())) ).unwrap_or("---".to_owned()),

149
veilid-core/src/network_manager/send_data.rs
View File

@ -3,10 +3,10 @@ use super::*;
impl NetworkManager {
/// Send raw data to a node
///
/// We may not have dial info for a node, but have an existing connection for it
/// because an inbound connection happened first, and no FindNodeQ has happened to that
/// node yet to discover its dial info. The existing connection should be tried first
/// in this case, if it matches the node ref's filters and no more permissive connection
/// We may not have dial info for a node, but have an existing flow for it
/// because an inbound flow happened first, and no FindNodeQ has happened to that
/// node yet to discover its dial info. The existing flow should be tried first
/// in this case, if it matches the node ref's filters and no more permissive flow
/// could be established.
///
/// Sending to a node requires determining a NetworkClass compatible contact method
@ -19,27 +19,26 @@ impl NetworkManager {
let this = self.clone();
Box::pin(
async move {
// First try to send data to the last socket we've seen this peer on
let data = if let Some(connection_descriptor) = destination_node_ref.last_connection() {
// First try to send data to the last flow we've seen this peer on
let data = if let Some(flow) = destination_node_ref.last_flow() {
match this
.net()
.send_data_to_existing_connection(connection_descriptor, data)
.send_data_to_existing_flow(flow, data)
.await?
{
None => {
// Update timestamp for this last connection since we just sent to it
SendDataToExistingFlowResult::Sent(unique_flow) => {
// Update timestamp for this last flow since we just sent to it
destination_node_ref
.set_last_connection(connection_descriptor, get_aligned_timestamp());
.set_last_flow(unique_flow.flow, get_aligned_timestamp());
return Ok(NetworkResult::value(SendDataMethod {
opt_relayed_contact_method: None,
contact_method: NodeContactMethod::Existing,
connection_descriptor,
unique_flow,
}));
}
Some(data) => {
// Couldn't send data to existing connection
SendDataToExistingFlowResult::NotSent(data) => {
// Couldn't send data to existing flow
// so pass the data back out
data
}
@ -135,30 +134,32 @@ impl NetworkManager {
data: Vec<u8>,
) -> EyreResult<NetworkResult<SendDataMethod>> {
// First try to send data to the last connection we've seen this peer on
let Some(connection_descriptor) = target_node_ref.last_connection() else {
let Some(flow) = target_node_ref.last_flow() else {
return Ok(NetworkResult::no_connection_other(
format!("should have found an existing connection: {}", target_node_ref)
));
};
if self
let unique_flow = match self
.net()
.send_data_to_existing_connection(connection_descriptor, data)
.send_data_to_existing_flow(flow, data)
.await?
.is_some()
{
return Ok(NetworkResult::no_connection_other(
"failed to send to existing connection",
));
}
SendDataToExistingFlowResult::Sent(unique_flow) => unique_flow,
SendDataToExistingFlowResult::NotSent(_) => {
return Ok(NetworkResult::no_connection_other(
"failed to send to existing flow",
));
}
};
// Update timestamp for this last connection since we just sent to it
target_node_ref.set_last_connection(connection_descriptor, get_aligned_timestamp());
target_node_ref.set_last_flow(flow, get_aligned_timestamp());
Ok(NetworkResult::value(SendDataMethod{
contact_method: NodeContactMethod::Existing,
opt_relayed_contact_method: None,
connection_descriptor
unique_flow
}))
}
@ -169,30 +170,32 @@ impl NetworkManager {
data: Vec<u8>,
) -> EyreResult<NetworkResult<SendDataMethod>> {
// Try to send data to the last socket we've seen this peer on
let Some(connection_descriptor) = target_node_ref.last_connection() else {
let Some(flow) = target_node_ref.last_flow() else {
return Ok(NetworkResult::no_connection_other(
format!("Node is not reachable and has no existing connection: {}", target_node_ref)
));
};
if self
let unique_flow = match self
.net()
.send_data_to_existing_connection(connection_descriptor, data)
.await?
.is_some()
.send_data_to_existing_flow(flow, data)
.await?
{
return Ok(NetworkResult::no_connection_other(
format!("failed to send to unreachable node over existing connection: {:?}", connection_descriptor)
));
}
SendDataToExistingFlowResult::Sent(unique_flow) => unique_flow,
SendDataToExistingFlowResult::NotSent(_) => {
return Ok(NetworkResult::no_connection_other(
format!("failed to send to unreachable node over existing connection: {:?}", flow)
));
}
};
// Update timestamp for this last connection since we just sent to it
target_node_ref.set_last_connection(connection_descriptor, get_aligned_timestamp());
target_node_ref.set_last_flow(flow, get_aligned_timestamp());
Ok(NetworkResult::value(SendDataMethod {
connection_descriptor,
contact_method: NodeContactMethod::Existing,
opt_relayed_contact_method: None,
unique_flow,
}))
}
@ -204,24 +207,24 @@ impl NetworkManager {
data: Vec<u8>,
) -> EyreResult<NetworkResult<SendDataMethod>> {
// First try to send data to the last socket we've seen this peer on
let data = if let Some(connection_descriptor) = target_node_ref.last_connection() {
let data = if let Some(flow) = target_node_ref.last_flow() {
match self
.net()
.send_data_to_existing_connection(connection_descriptor, data)
.send_data_to_existing_flow(flow, data)
.await?
{
None => {
SendDataToExistingFlowResult::Sent(unique_flow) => {
// Update timestamp for this last connection since we just sent to it
target_node_ref
.set_last_connection(connection_descriptor, get_aligned_timestamp());
.set_last_flow(flow, get_aligned_timestamp());
return Ok(NetworkResult::value(SendDataMethod{
contact_method: NodeContactMethod::Existing,
opt_relayed_contact_method: None,
connection_descriptor
unique_flow
}));
}
Some(data) => {
SendDataToExistingFlowResult::NotSent(data) => {
// Couldn't send data to existing connection
// so pass the data back out
data
@ -232,14 +235,14 @@ impl NetworkManager {
data
};
let connection_descriptor = network_result_try!(
let unique_flow = network_result_try!(
self.do_reverse_connect(relay_nr.clone(), target_node_ref.clone(), data)
.await?
);
Ok(NetworkResult::value(SendDataMethod {
connection_descriptor,
contact_method: NodeContactMethod::SignalReverse(relay_nr, target_node_ref),
opt_relayed_contact_method: None,
unique_flow,
}))
}
@ -251,24 +254,24 @@ impl NetworkManager {
data: Vec<u8>,
) -> EyreResult<NetworkResult<SendDataMethod>> {
// First try to send data to the last socket we've seen this peer on
let data = if let Some(connection_descriptor) = target_node_ref.last_connection() {
let data = if let Some(flow) = target_node_ref.last_flow() {
match self
.net()
.send_data_to_existing_connection(connection_descriptor, data)
.send_data_to_existing_flow(flow, data)
.await?
{
None => {
SendDataToExistingFlowResult::Sent(unique_flow) => {
// Update timestamp for this last connection since we just sent to it
target_node_ref
.set_last_connection(connection_descriptor, get_aligned_timestamp());
.set_last_flow(flow, get_aligned_timestamp());
return Ok(NetworkResult::value(SendDataMethod{
contact_method: NodeContactMethod::Existing,
opt_relayed_contact_method: None,
connection_descriptor
unique_flow
}));
}
Some(data) => {
SendDataToExistingFlowResult::NotSent(data) => {
// Couldn't send data to existing connection
// so pass the data back out
data
@ -279,12 +282,12 @@ impl NetworkManager {
data
};
let connection_descriptor =
let unique_flow =
network_result_try!(self.do_hole_punch(relay_nr.clone(), target_node_ref.clone(), data).await?);
Ok(NetworkResult::value(SendDataMethod {
connection_descriptor,
contact_method: NodeContactMethod::SignalHolePunch(relay_nr, target_node_ref),
opt_relayed_contact_method: None,
unique_flow,
}))
}
@ -299,31 +302,31 @@ impl NetworkManager {
let node_ref = node_ref.filtered_clone(NodeRefFilter::from(dial_info.make_filter()));
// First try to send data to the last socket we've seen this peer on
let data = if let Some(connection_descriptor) = node_ref.last_connection() {
let data = if let Some(flow) = node_ref.last_flow() {
#[cfg(feature = "verbose-tracing")]
debug!(
"ExistingConnection: {:?} for {:?}",
connection_descriptor, node_ref
flow, node_ref
);
match self
.net()
.send_data_to_existing_connection(connection_descriptor, data)
.send_data_to_existing_flow(flow, data)
.await?
{
None => {
SendDataToExistingFlowResult::Sent(unique_flow) => {
// Update timestamp for this last connection since we just sent to it
node_ref.set_last_connection(connection_descriptor, get_aligned_timestamp());
node_ref.set_last_flow(flow, get_aligned_timestamp());
return Ok(NetworkResult::value(SendDataMethod{
contact_method: NodeContactMethod::Existing,
opt_relayed_contact_method: None,
connection_descriptor
unique_flow
}));
}
Some(d) => {
SendDataToExistingFlowResult::NotSent(d) => {
// Connection couldn't send, kill it
node_ref.clear_last_connection(connection_descriptor);
node_ref.clear_last_connection(flow);
d
}
}
@ -332,16 +335,16 @@ impl NetworkManager {
};
// New direct connection was necessary for this dial info
let connection_descriptor =
let unique_flow =
network_result_try!(self.net().send_data_to_dial_info(dial_info.clone(), data).await?);
// If we connected to this node directly, save off the last connection so we can use it again
node_ref.set_last_connection(connection_descriptor, get_aligned_timestamp());
node_ref.set_last_flow(unique_flow.flow, get_aligned_timestamp());
Ok(NetworkResult::value(SendDataMethod {
connection_descriptor,
contact_method: NodeContactMethod::Direct(dial_info),
opt_relayed_contact_method: None,
unique_flow,
}))
}
@ -528,7 +531,7 @@ impl NetworkManager {
relay_nr: NodeRef,
target_nr: NodeRef,
data: Vec<u8>,
) -> EyreResult<NetworkResult<ConnectionDescriptor>> {
) -> EyreResult<NetworkResult<UniqueFlow>> {
// Build a return receipt for the signal
let receipt_timeout = ms_to_us(
self.unlocked_inner
@ -592,14 +595,14 @@ impl NetworkManager {
}
// And now use the existing connection to send over
if let Some(descriptor) = inbound_nr.last_connection() {
if let Some(flow) = inbound_nr.last_flow() {
match self
.net()
.send_data_to_existing_connection(descriptor, data)
.send_data_to_existing_flow(flow, data)
.await?
{
None => Ok(NetworkResult::value(descriptor)),
Some(_) => Ok(NetworkResult::no_connection_other(
SendDataToExistingFlowResult::Sent(unique_flow) => Ok(NetworkResult::value(unique_flow)),
SendDataToExistingFlowResult::NotSent(_) => Ok(NetworkResult::no_connection_other(
"unable to send over reverse connection",
)),
}
@ -620,7 +623,7 @@ impl NetworkManager {
relay_nr: NodeRef,
target_nr: NodeRef,
data: Vec<u8>,
) -> EyreResult<NetworkResult<ConnectionDescriptor>> {
) -> EyreResult<NetworkResult<UniqueFlow>> {
// Ensure we are filtered down to UDP (the only hole punch protocol supported today)
assert!(target_nr
.filter_ref()
@ -660,7 +663,7 @@ impl NetworkManager {
// Do our half of the hole punch by sending an empty packet
// Both sides will do this and then the receipt will get sent over the punched hole
// Don't bother storing the returned connection descriptor as the 'last connection' because the other side of the hole
// Don't bother storing the returned flow as the 'last flow' because the other side of the hole
// punch should come through and create a real 'last connection' for us if this succeeds
network_result_try!(
self.net()
@ -710,14 +713,14 @@ impl NetworkManager {
}
// And now use the existing connection to send over
if let Some(descriptor) = inbound_nr.last_connection() {
if let Some(flow) = inbound_nr.last_flow() {
match self
.net()
.send_data_to_existing_connection(descriptor, data)
.send_data_to_existing_flow(flow, data)
.await?
{
None => Ok(NetworkResult::value(descriptor)),
Some(_) => Ok(NetworkResult::no_connection_other(
SendDataToExistingFlowResult::Sent(unique_flow) => Ok(NetworkResult::value(unique_flow)),
SendDataToExistingFlowResult::NotSent(_) => Ok(NetworkResult::no_connection_other(
"unable to send over hole punch",
)),
}

21
veilid-core/src/network_manager/tasks/public_address_check.rs
View File

@ -31,7 +31,7 @@ impl NetworkManager {
pub fn report_local_network_socket_address(
&self,
_socket_address: SocketAddress,
_connection_descriptor: ConnectionDescriptor,
_flow: Flow,
_reporting_peer: NodeRef,
) {
// XXX: Nothing here yet.
@ -43,11 +43,11 @@ impl NetworkManager {
pub fn report_public_internet_socket_address(
&self,
socket_address: SocketAddress, // the socket address as seen by the remote peer
connection_descriptor: ConnectionDescriptor, // the connection descriptor used
flow: Flow, // the flow used
reporting_peer: NodeRef, // the peer's noderef reporting the socket address
) {
#[cfg(feature = "network-result-extra")]
debug!("report_global_socket_address\nsocket_address: {:#?}\nconnection_descriptor: {:#?}\nreporting_peer: {:#?}", socket_address, connection_descriptor, reporting_peer);
debug!("report_global_socket_address\nsocket_address: {:#?}\nflow: {:#?}\nreporting_peer: {:#?}", socket_address, flow, reporting_peer);
// Ignore these reports if we are currently detecting public dial info
let net = self.net();
@ -77,10 +77,7 @@ impl NetworkManager {
});
// Get the ip(block) this report is coming from
let reporting_ipblock = ip_to_ipblock(
ip6_prefix_size,
connection_descriptor.remote_address().ip_addr(),
);
let reporting_ipblock = ip_to_ipblock(ip6_prefix_size, flow.remote_address().ip_addr());
// Reject public address reports from nodes that we know are behind symmetric nat or
// nodes that must be using a relay for everything
@ -105,10 +102,8 @@ impl NetworkManager {
let mut inner = self.inner.lock();
let inner = &mut *inner;
let addr_proto_type_key = PublicAddressCheckCacheKey(
connection_descriptor.protocol_type(),
connection_descriptor.address_type(),
);
let addr_proto_type_key =
PublicAddressCheckCacheKey(flow.protocol_type(), flow.address_type());
if inner
.public_address_inconsistencies_table
.get(&addr_proto_type_key)
@ -136,7 +131,7 @@ impl NetworkManager {
NetworkClass::InboundCapable
) {
// Get the dial info filter for this connection so we can check if we have any public dialinfo that may have changed
let dial_info_filter = connection_descriptor.make_dial_info_filter();
let dial_info_filter = flow.make_dial_info_filter();
// Get current external ip/port from registered global dialinfo
let current_addresses: BTreeSet<SocketAddress> = routing_table
@ -267,7 +262,7 @@ impl NetworkManager {
net.set_needs_public_dial_info_check(bad_public_address_detection_punishment);
} else {
warn!("Public address may have changed. Restarting the server may be required.");
warn!("report_global_socket_address\nsocket_address: {:#?}\nconnection_descriptor: {:#?}\nreporting_peer: {:#?}", socket_address, connection_descriptor, reporting_peer);
warn!("report_global_socket_address\nsocket_address: {:#?}\nflow: {:#?}\nreporting_peer: {:#?}", socket_address, flow, reporting_peer);
warn!(
"public_address_check_cache: {:#?}",
inner.public_address_check_cache

36
veilid-core/src/network_manager/tests/test_connection_table.rs
View File

@ -9,12 +9,12 @@ pub async fn test_add_get_remove() {
let address_filter = AddressFilter::new(config.clone(), mock_routing_table());
let table = ConnectionTable::new(config, address_filter);
let a1 = ConnectionDescriptor::new_no_local(PeerAddress::new(
let a1 = Flow::new_no_local(PeerAddress::new(
SocketAddress::new(Address::IPV4(Ipv4Addr::new(192, 168, 0, 1)), 8080),
ProtocolType::TCP,
));
let a2 = a1;
let a3 = ConnectionDescriptor::new(
let a3 = Flow::new(
PeerAddress::new(
SocketAddress::new(Address::IPV6(Ipv6Addr::new(191, 0, 0, 0, 0, 0, 0, 1)), 8090),
ProtocolType::TCP,
@ -26,7 +26,7 @@ pub async fn test_add_get_remove() {
0,
))),
);
let a4 = ConnectionDescriptor::new(
let a4 = Flow::new(
PeerAddress::new(
SocketAddress::new(Address::IPV6(Ipv6Addr::new(192, 0, 0, 0, 0, 0, 0, 1)), 8090),
ProtocolType::TCP,
@ -38,7 +38,7 @@ pub async fn test_add_get_remove() {
0,
))),
);
let a5 = ConnectionDescriptor::new(
let a5 = Flow::new(
PeerAddress::new(
SocketAddress::new(Address::IPV6(Ipv6Addr::new(192, 0, 0, 0, 0, 0, 0, 1)), 8090),
ProtocolType::WSS,
@ -59,12 +59,12 @@ pub async fn test_add_get_remove() {
let c4 = NetworkConnection::dummy(4.into(), a4);
let c5 = NetworkConnection::dummy(5.into(), a5);
assert_eq!(a1, c2.connection_descriptor());
assert_ne!(a3, c4.connection_descriptor());
assert_ne!(a4, c5.connection_descriptor());
assert_eq!(a1, c2.flow());
assert_ne!(a3, c4.flow());
assert_ne!(a4, c5.flow());
assert_eq!(table.connection_count(), 0);
assert_eq!(table.peek_connection_by_descriptor(a1), None);
assert_eq!(table.peek_connection_by_flow(a1), None);
table.add_connection(c1).unwrap();
assert!(table.add_connection(c1b).is_err());
@ -72,26 +72,26 @@ pub async fn test_add_get_remove() {
assert!(table.remove_connection_by_id(4.into()).is_none());
assert!(table.remove_connection_by_id(5.into()).is_none());
assert_eq!(table.connection_count(), 1);
assert_eq!(table.peek_connection_by_descriptor(a1), Some(c1h.clone()));
assert_eq!(table.peek_connection_by_descriptor(a1), Some(c1h.clone()));
assert_eq!(table.peek_connection_by_flow(a1), Some(c1h.clone()));
assert_eq!(table.peek_connection_by_flow(a1), Some(c1h.clone()));
assert_eq!(table.connection_count(), 1);
assert_err!(table.add_connection(c2));
assert_eq!(table.connection_count(), 1);
assert_eq!(table.peek_connection_by_descriptor(a1), Some(c1h.clone()));
assert_eq!(table.peek_connection_by_descriptor(a1), Some(c1h.clone()));
assert_eq!(table.peek_connection_by_flow(a1), Some(c1h.clone()));
assert_eq!(table.peek_connection_by_flow(a1), Some(c1h.clone()));
assert_eq!(table.connection_count(), 1);
assert_eq!(
table
.remove_connection_by_id(1.into())
.map(|c| c.connection_descriptor())
.map(|c| c.flow())
.unwrap(),
a1
);
assert_eq!(table.connection_count(), 0);
assert!(table.remove_connection_by_id(2.into()).is_none());
assert_eq!(table.connection_count(), 0);
assert_eq!(table.peek_connection_by_descriptor(a2), None);
assert_eq!(table.peek_connection_by_descriptor(a1), None);
assert_eq!(table.peek_connection_by_flow(a2), None);
assert_eq!(table.peek_connection_by_flow(a1), None);
assert_eq!(table.connection_count(), 0);
let c1 = NetworkConnection::dummy(6.into(), a1);
table.add_connection(c1).unwrap();
@ -103,21 +103,21 @@ pub async fn test_add_get_remove() {
assert_eq!(
table
.remove_connection_by_id(6.into())
.map(|c| c.connection_descriptor())
.map(|c| c.flow())
.unwrap(),
a2
);
assert_eq!(
table
.remove_connection_by_id(3.into())
.map(|c| c.connection_descriptor())
.map(|c| c.flow())
.unwrap(),
a3
);
assert_eq!(
table
.remove_connection_by_id(4.into())
.map(|c| c.connection_descriptor())
.map(|c| c.flow())
.unwrap(),
a4
);

4
veilid-core/src/network_manager/types/dial_info_filter.rs
View File

@ -97,8 +97,8 @@ impl From<AddressType> for DialInfoFilter {
}
}
impl From<ConnectionDescriptor> for DialInfoFilter {
fn from(other: ConnectionDescriptor) -> Self {
impl From<Flow> for DialInfoFilter {
fn from(other: Flow) -> Self {
Self {
protocol_type_set: ProtocolTypeSet::from(other.protocol_type()),
address_type_set: AddressTypeSet::from(other.address_type()),

23
veilid-core/src/network_manager/types/connection_descriptor.rs → veilid-core/src/network_manager/types/flow.rs
View File

@ -3,17 +3,21 @@ use super::*;
/// Represents the 5-tuple of an established connection
/// Not used to specify connections to create, that is reserved for DialInfo
///
/// ConnectionDescriptors should never be from unspecified local addresses for connection oriented protocols
/// Abstracts both connections to 'connection oriented' protocols (TCP/WS/WSS), but also datagram protocols (UDP)
///
/// Flows should never be from UNSPECIFIED local addresses for connection oriented protocols
/// If the medium does not allow local addresses, None should have been used or 'new_no_local'
/// If we are specifying only a port, then the socket's 'local_address()' should have been used, since an
/// established connection is always from a real address to another real address.
///
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct ConnectionDescriptor {
pub struct Flow {
remote: PeerAddress,
local: Option<SocketAddress>,
}
impl ConnectionDescriptor {
impl Flow {
pub fn new(remote: PeerAddress, local: SocketAddress) -> Self {
assert!(!remote.protocol_type().is_ordered() || !local.address().is_unspecified());
@ -50,7 +54,7 @@ impl ConnectionDescriptor {
}
}
impl MatchesDialInfoFilter for ConnectionDescriptor {
impl MatchesDialInfoFilter for Flow {
fn matches_filter(&self, filter: &DialInfoFilter) -> bool {
if !filter.protocol_type_set.contains(self.protocol_type()) {
return false;
@ -61,3 +65,14 @@ impl MatchesDialInfoFilter for ConnectionDescriptor {
true
}
}
/// UniqueFlow is a record a specific flow that may or may not currently exist
/// The NetworkConnectionId associated with each flow may represent a low level network connection
/// and will be unique with high probability per low-level connection
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct UniqueFlow {
pub flow: Flow,
pub connection_id: Option<NetworkConnectionId>,
}
pub type NetworkConnectionId = AlignedU64;

4
veilid-core/src/network_manager/types/mod.rs
View File

@ -1,9 +1,9 @@
mod address;
mod address_type;
mod connection_descriptor;
mod dial_info;
mod dial_info_class;
mod dial_info_filter;
mod flow;
mod low_level_protocol_type;
mod network_class;
mod peer_address;
@ -15,10 +15,10 @@ use super::*;
pub use address::*;
pub use address_type::*;
pub use connection_descriptor::*;
pub use dial_info::*;
pub use dial_info_class::*;
pub use dial_info_filter::*;
pub use flow::*;
pub use low_level_protocol_type::*;
pub use network_class::*;
pub use peer_address::*;

14
veilid-core/src/network_manager/wasm/mod.rs
View File

@ -248,11 +248,11 @@ impl Network {
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_to_existing_connection(
&self,
descriptor: ConnectionDescriptor,
flow: Flow,
data: Vec<u8>,
) -> EyreResult<Option<Vec<u8>>> {
let data_len = data.len();
match descriptor.protocol_type() {
match flow.protocol_type() {
ProtocolType::UDP => {
bail!("no support for UDP protocol")
}
@ -265,13 +265,13 @@ impl Network {
// Handle connection-oriented protocols
// Try to send to the exact existing connection if one exists
if let Some(conn) = self.connection_manager().get_connection(descriptor) {
if let Some(conn) = self.connection_manager().get_connection(flow) {
// connection exists, send over it
match conn.send_async(data).await {
ConnectionHandleSendResult::Sent => {
// Network accounting
self.network_manager().stats_packet_sent(
descriptor.remote().socket_addr().ip(),
flow.remote().socket_addr().ip(),
ByteCount::new(data_len as u64),
);
@ -295,7 +295,7 @@ impl Network {
&self,
dial_info: DialInfo,
data: Vec<u8>,
) -> EyreResult<NetworkResult<ConnectionDescriptor>> {
) -> EyreResult<NetworkResult<Flow>> {
self.record_dial_info_failure(dial_info.clone(), async move {
let data_len = data.len();
if dial_info.protocol_type() == ProtocolType::UDP {
@ -318,13 +318,13 @@ impl Network {
"failed to send",
)));
}
let connection_descriptor = conn.connection_descriptor();
let flow = conn.flow();
// Network accounting
self.network_manager()
.stats_packet_sent(dial_info.ip_addr(), ByteCount::new(data_len as u64));
Ok(NetworkResult::value(connection_descriptor))
Ok(NetworkResult::value(flow))
})
.await
}

6
veilid-core/src/network_manager/wasm/protocol/mod.rs
View File

@ -35,10 +35,10 @@ impl ProtocolNetworkConnection {
}
}
pub fn descriptor(&self) -> ConnectionDescriptor {
pub fn flow(&self) -> Flow {
match self {
// Self::Dummy(d) => d.descriptor(),
Self::Ws(w) => w.descriptor(),
// Self::Dummy(d) => d.flow(),
Self::Ws(w) => w.flow(),
}
}
pub async fn close(&self) -> io::Result<NetworkResult<()>> {

14
veilid-core/src/network_manager/wasm/protocol/ws.rs
View File

@ -34,7 +34,7 @@ fn to_io(err: WsErr) -> io::Error {
#[derive(Clone)]
pub struct WebsocketNetworkConnection {
descriptor: ConnectionDescriptor,
flow: Flow,
inner: Arc<WebsocketNetworkConnectionInner>,
}
@ -45,9 +45,9 @@ impl fmt::Debug for WebsocketNetworkConnection {
}
impl WebsocketNetworkConnection {
pub fn new(descriptor: ConnectionDescriptor, ws_meta: WsMeta, ws_stream: WsStream) -> Self {
pub fn new(flow: Flow, ws_meta: WsMeta, ws_stream: WsStream) -> Self {
Self {
descriptor,
flow,
inner: Arc::new(WebsocketNetworkConnectionInner {
ws_meta,
ws_stream: CloneStream::new(ws_stream),
@ -55,8 +55,8 @@ impl WebsocketNetworkConnection {
}
}
pub fn descriptor(&self) -> ConnectionDescriptor {
self.descriptor
pub fn flow(&self) -> Flow {
self.flow
}
#[cfg_attr(
@ -147,9 +147,9 @@ impl WebsocketProtocolHandler {
.into_network_result())
.into_network_result()?);
// Make our connection descriptor
// Make our flow
let wnc = WebsocketNetworkConnection::new(
ConnectionDescriptor::new_no_local(dial_info.peer_address()),
Flow::new_no_local(dial_info.peer_address()),
wsmeta,
wsio,
);

86
veilid-core/src/routing_table/bucket_entry.rs
View File

@ -36,7 +36,7 @@ pub(crate) enum BucketEntryState {
}
#[derive(Debug, Clone, Eq, PartialEq, PartialOrd, Ord, Hash)]
pub(crate) struct LastConnectionKey(ProtocolType, AddressType);
pub(crate) struct LastFlowKey(ProtocolType, AddressType);
/// Bucket entry information specific to the LocalNetwork RoutingDomain
#[derive(Debug, Serialize, Deserialize)]
@ -75,9 +75,9 @@ pub(crate) struct BucketEntryInner {
/// has the same timestamp, because if we change our own IP address or network class it may be possible for nodes that were
/// unreachable may now be reachable with the same SignedNodeInfo/DialInfo
updated_since_last_network_change: bool,
/// The last connection descriptors used to contact this node, per protocol type
/// The last flows used to contact this node, per protocol type
#[serde(skip)]
last_connections: BTreeMap<LastConnectionKey, (ConnectionDescriptor, Timestamp)>,
last_flows: BTreeMap<LastFlowKey, (Flow, Timestamp)>,
/// The node info for this entry on the publicinternet routing domain
public_internet: BucketEntryPublicInternet,
/// The node info for this entry on the localnetwork routing domain
@ -304,7 +304,7 @@ impl BucketEntryInner {
// The latest connection would have been the one we got the new node info
// over so that connection is still valid.
if node_info_changed {
self.clear_last_connections_except_latest();
self.clear_last_flows_except_latest();
}
}
@ -333,7 +333,7 @@ impl BucketEntryInner {
}
// Check connections
let last_connections = self.last_connections(
let last_connections = self.last_flows(
rti,
true,
NodeRefFilter::from(routing_domain),
@ -387,7 +387,7 @@ impl BucketEntryInner {
}
// Check connections
let mut best_routing_domain: Option<RoutingDomain> = None;
let last_connections = self.last_connections(
let last_connections = self.last_flows(
rti,
true,
NodeRefFilter::from(routing_domain_set),
@ -408,77 +408,77 @@ impl BucketEntryInner {
best_routing_domain
}
fn descriptor_to_key(&self, last_connection: ConnectionDescriptor) -> LastConnectionKey {
LastConnectionKey(
last_connection.protocol_type(),
last_connection.address_type(),
fn flow_to_key(&self, last_flow: Flow) -> LastFlowKey {
LastFlowKey(
last_flow.protocol_type(),
last_flow.address_type(),
)
}
// Stores a connection descriptor in this entry's table of last connections
pub fn set_last_connection(&mut self, last_connection: ConnectionDescriptor, timestamp: Timestamp) {
// Stores a flow in this entry's table of last flows
pub fn set_last_flow(&mut self, last_flow: Flow, timestamp: Timestamp) {
if self.is_punished {
// Don't record connection if this entry is currently punished
return;
}
let key = self.descriptor_to_key(last_connection);
self.last_connections
.insert(key, (last_connection, timestamp));
let key = self.flow_to_key(last_flow);
self.last_flows
.insert(key, (last_flow, timestamp));
}
// Removes a connection descriptor in this entry's table of last connections
pub fn clear_last_connection(&mut self, last_connection: ConnectionDescriptor) {
let key = self.descriptor_to_key(last_connection);
self.last_connections
// Removes a flow in this entry's table of last flows
pub fn remove_last_flow(&mut self, last_flow: Flow) {
let key = self.flow_to_key(last_flow);
self.last_flows
.remove(&key);
}
// Clears the table of last connections to ensure we create new ones and drop any existing ones
pub fn clear_last_connections(&mut self) {
self.last_connections.clear();
// Clears the table of last flows to ensure we create new ones and drop any existing ones
pub fn clear_last_flows(&mut self) {
self.last_flows.clear();
}
// Clears the table of last connections except the most recent one
pub fn clear_last_connections_except_latest(&mut self) {
if self.last_connections.is_empty() {
// Clears the table of last flows except the most recent one
pub fn clear_last_flows_except_latest(&mut self) {
if self.last_flows.is_empty() {
// No last_connections
return;
}
let mut dead_keys = Vec::with_capacity(self.last_connections.len()-1);
let mut most_recent_connection = None;
let mut most_recent_connection_time = 0u64;
for (k, v) in &self.last_connections {
let mut dead_keys = Vec::with_capacity(self.last_flows.len()-1);
let mut most_recent_flow = None;
let mut most_recent_flow_time = 0u64;
for (k, v) in &self.last_flows {
let lct = v.1.as_u64();
if lct > most_recent_connection_time {
most_recent_connection = Some(k);
most_recent_connection_time = lct;
if lct > most_recent_flow_time {
most_recent_flow = Some(k);
most_recent_flow_time = lct;
}
}
let Some(most_recent_connection) = most_recent_connection else {
let Some(most_recent_flow) = most_recent_flow else {
return;
};
for k in self.last_connections.keys() {
if k != most_recent_connection {
for k in self.last_flows.keys() {
if k != most_recent_flow {
dead_keys.push(k.clone());
}
}
for dk in dead_keys {
self.last_connections.remove(&dk);
self.last_flows.remove(&dk);
}
}
// Gets all the 'last connections' that match a particular filter, and their accompanying timestamps of last use
pub(super) fn last_connections(
// Gets all the 'last flows' that match a particular filter, and their accompanying timestamps of last use
pub(super) fn last_flows(
&self,
rti: &RoutingTableInner,
only_live: bool,
filter: NodeRefFilter,
) -> Vec<(ConnectionDescriptor, Timestamp)> {
) -> Vec<(Flow, Timestamp)> {
let connection_manager =
rti.unlocked_inner.network_manager.connection_manager();
let mut out: Vec<(ConnectionDescriptor, Timestamp)> = self
.last_connections
let mut out: Vec<(Flow, Timestamp)> = self
.last_flows
.iter()
.filter_map(|(k, v)| {
let include = {
@ -564,7 +564,7 @@ impl BucketEntryInner {
pub fn set_punished(&mut self, punished: bool) {
self.is_punished = punished;
if punished {
self.clear_last_connections();
self.clear_last_flows();
}
}
@ -845,7 +845,7 @@ impl BucketEntry {
unsupported_node_ids: TypedKeyGroup::new(),
envelope_support: Vec::new(),
updated_since_last_network_change: false,
last_connections: BTreeMap::new(),
last_flows: BTreeMap::new(),
local_network: BucketEntryLocalNetwork {
last_seen_our_node_info_ts: Timestamp::new(0u64),
signed_node_info: None,

8
veilid-core/src/routing_table/mod.rs
View File

@ -90,7 +90,7 @@ pub type BucketIndex = (CryptoKind, usize);
#[derive(Debug, Clone, Copy)]
pub(crate) struct RecentPeersEntry {
pub last_connection: ConnectionDescriptor,
pub last_connection: Flow,
}
pub(crate) struct RoutingTableUnlockedInner {
@ -662,13 +662,13 @@ impl RoutingTable {
pub fn register_node_with_existing_connection(
&self,
node_id: TypedKey,
descriptor: ConnectionDescriptor,
flow: Flow,
timestamp: Timestamp,
) -> EyreResult<NodeRef> {
self.inner.write().register_node_with_existing_connection(
self.clone(),
node_id,
descriptor,
flow,
timestamp,
)
}
@ -698,7 +698,7 @@ impl RoutingTable {
for e in &recent_peers {
let mut dead = true;
if let Ok(Some(nr)) = self.lookup_node_ref(*e) {
if let Some(last_connection) = nr.last_connection() {
if let Some(last_connection) = nr.last_flow() {
out.push((*e, RecentPeersEntry { last_connection }));
dead = false;
}

20
veilid-core/src/routing_table/node_ref.rs
View File

@ -273,13 +273,13 @@ pub(crate) trait NodeRefBase: Sized {
/// Get the most recent 'last connection' to this node
/// Filtered first and then sorted by ordering preference and then by most recent
fn last_connection(&self) -> Option<ConnectionDescriptor> {
fn last_flow(&self) -> Option<Flow> {
self.operate(|rti, e| {
// apply sequencing to filter and get sort
let sequencing = self.common().sequencing;
let filter = self.common().filter.unwrap_or_default();
let (ordered, filter) = filter.with_sequencing(sequencing);
let mut last_connections = e.last_connections(rti, true, filter);
let mut last_connections = e.last_flows(rti, true, filter);
if ordered {
last_connections.sort_by(|a, b| {
@ -292,19 +292,19 @@ pub(crate) trait NodeRefBase: Sized {
}
fn clear_last_connections(&self) {
self.operate_mut(|_rti, e| e.clear_last_connections())
self.operate_mut(|_rti, e| e.clear_last_flows())
}
fn set_last_connection(&self, connection_descriptor: ConnectionDescriptor, ts: Timestamp) {
fn set_last_flow(&self, flow: Flow, ts: Timestamp) {
self.operate_mut(|rti, e| {
e.set_last_connection(connection_descriptor, ts);
rti.touch_recent_peer(e.best_node_id(), connection_descriptor);
e.set_last_flow(flow, ts);
rti.touch_recent_peer(e.best_node_id(), flow);
})
}
fn clear_last_connection(&self, connection_descriptor: ConnectionDescriptor) {
fn clear_last_connection(&self, flow: Flow) {
self.operate_mut(|_rti, e| {
e.clear_last_connection(connection_descriptor);
e.remove_last_flow(flow);
})
}
@ -325,6 +325,10 @@ pub(crate) trait NodeRefBase: Sized {
self.stats_failed_to_send(get_aligned_timestamp(), false);
}
fn report_failed_route_test(&self) {
self.stats_failed_to_send(get_aligned_timestamp(), false);
}
fn stats_question_sent(&self, ts: Timestamp, bytes: Timestamp, expects_answer: bool) {
self.operate_mut(|rti, e| {
rti.transfer_stats_accounting().add_up(bytes);

4
veilid-core/src/routing_table/node_ref_filter.rs
View File

@ -112,8 +112,8 @@ impl From<AddressType> for NodeRefFilter {
}
}
impl From<ConnectionDescriptor> for NodeRefFilter {
fn from(other: ConnectionDescriptor) -> Self {
impl From<Flow> for NodeRefFilter {
fn from(other: Flow) -> Self {
Self {
routing_domain_set: RoutingDomainSet::all(),
dial_info_filter: DialInfoFilter::from(other),

33
veilid-core/src/routing_table/route_spec_store/mod.rs
View File

@ -661,9 +661,9 @@ impl RouteSpecStore {
)]
async fn test_allocated_route(&self, private_route_id: RouteId) -> VeilidAPIResult<bool> {
// Make loopback route to test with
let dest = {
// Get the best private route for this id
let (key, hop_count) = {
let (dest, hops) = {
// Get the best allocated route for this id
let (key, hops) = {
let inner = &mut *self.inner.lock();
let Some(rssd) = inner.content.get_detail(&private_route_id) else {
apibail_invalid_argument!(
@ -675,9 +675,10 @@ impl RouteSpecStore {
let Some(key) = rssd.get_best_route_set_key() else {
apibail_internal!("no best key to test allocated route");
};
// Match the private route's hop length for safety route length
let hop_count = rssd.hop_count();
(key, hop_count)
// Get the hops so we can match the route's hop length for safety
// route length as well as marking nodes as unreliable if this fails
let hops = rssd.hops_node_refs();
(key, hops)
};
// Get the private route to send to
@ -686,6 +687,8 @@ impl RouteSpecStore {
let stability = Stability::Reliable;
// Routes should test with the most likely to succeed sequencing they are capable of
let sequencing = Sequencing::PreferOrdered;
// Hop count for safety spec should match the private route spec
let hop_count = hops.len();
let safety_spec = SafetySpec {
preferred_route: Some(private_route_id),
@ -695,10 +698,13 @@ impl RouteSpecStore {
};
let safety_selection = SafetySelection::Safe(safety_spec);
Destination::PrivateRoute {
private_route,
safety_selection,
}
(
Destination::PrivateRoute {
private_route,
safety_selection,
},
hops,
)
};
// Test with double-round trip ping to self
@ -706,7 +712,12 @@ impl RouteSpecStore {
let _res = match rpc_processor.rpc_call_status(dest).await? {
NetworkResult::Value(v) => v,
_ => {
// Did not error, but did not come back, just return false
// Did not error, but did not come back, mark the nodes as failed to send, and then return false
// This will prevent those node from immediately being included in the next allocated route,
// avoiding the same route being constructed to replace this one when it is removed.
for hop in hops {
hop.report_failed_route_test();
}
return Ok(false);
}
};

3
veilid-core/src/routing_table/route_spec_store/route_set_spec_detail.rs
View File

@ -86,6 +86,9 @@ impl RouteSetSpecDetail {
pub fn hop_count(&self) -> usize {
self.hop_node_refs.len()
}
pub fn hops_node_refs(&self) -> Vec<NodeRef> {
self.hop_node_refs.clone()
}
pub fn hop_node_ref(&self, idx: usize) -> Option<NodeRef> {
self.hop_node_refs.get(idx).cloned()
}

9
veilid-core/src/routing_table/routing_table_inner.rs
View File

@ -378,7 +378,7 @@ impl RoutingTableInner {
for bucket in &self.buckets[&ck] {
for entry in bucket.entries() {
entry.1.with_mut_inner(|e| {
e.clear_last_connections();
e.clear_last_flows();
});
}
}
@ -853,7 +853,7 @@ impl RoutingTableInner {
&mut self,
outer_self: RoutingTable,
node_id: TypedKey,
descriptor: ConnectionDescriptor,
flow: Flow,
timestamp: Timestamp,
) -> EyreResult<NodeRef> {
let nr = self.create_node_ref(outer_self, &TypedKeyGroup::from(node_id), |_rti, e| {
@ -861,8 +861,7 @@ impl RoutingTableInner {
e.touch_last_seen(timestamp);
})?;
// set the most recent node address for connection finding and udp replies
nr.locked_mut(self)
.set_last_connection(descriptor, timestamp);
nr.locked_mut(self).set_last_flow(flow, timestamp);
Ok(nr)
}
@ -912,7 +911,7 @@ impl RoutingTableInner {
}
}
pub fn touch_recent_peer(&mut self, node_id: TypedKey, last_connection: ConnectionDescriptor) {
pub fn touch_recent_peer(&mut self, node_id: TypedKey, last_connection: Flow) {
self.recent_peers
.insert(node_id, RecentPeersEntry { last_connection });
}

18
veilid-core/src/rpc_processor/mod.rs
View File

@ -54,8 +54,8 @@ struct RPCMessageHeaderDetailDirect {
envelope: Envelope,
/// The noderef of the peer that sent the message (not the original sender). Ensures node doesn't get evicted from routing table until we're done with it
peer_noderef: NodeRef,
/// The connection from the peer sent the message (not the original sender)
connection_descriptor: ConnectionDescriptor,
/// The flow from the peer sent the message (not the original sender)
flow: Flow,
/// The routing domain the message was sent through
routing_domain: RoutingDomain,
}
@ -189,7 +189,7 @@ struct WaitableReply {
safety_route: Option<PublicKey>,
remote_private_route: Option<PublicKey>,
reply_private_route: Option<PublicKey>,
_connection_ref_scope: ConnectionRefScope,
_opt_connection_ref_scope: Option<ConnectionRefScope>,
}
/////////////////////////////////////////////////////////////////////
@ -1158,7 +1158,7 @@ impl RPCProcessor {
// Log rpc send
#[cfg(feature = "verbose-tracing")]
debug!(target: "rpc_message", dir = "send", kind = "question", op_id = op_id.as_u64(), desc = operation.kind().desc(), ?dest, protect);
debug!(target: "rpc_message", dir = "send", kind = "question", op_id = op_id.as_u64(), desc = operation.kind().desc(), ?dest);
// Produce rendered operation
let RenderedOperation {
@ -1224,10 +1224,10 @@ impl RPCProcessor {
// Ref the connection so it doesn't go away until we're done with the waitable reply
let connection_ref_scope = self
let opt_connection_ref_scope = send_data_method.unique_flow.connection_id.and_then(|id| self
.network_manager()
.connection_manager()
.connection_ref_scope(send_data_method.connection_descriptor);
.try_connection_ref_scope(id));
// Pass back waitable reply completion
Ok(NetworkResult::value(WaitableReply {
@ -1239,7 +1239,7 @@ impl RPCProcessor {
safety_route,
remote_private_route,
reply_private_route,
_connection_ref_scope: connection_ref_scope,
_opt_connection_ref_scope: opt_connection_ref_scope,
}))
}
@ -1660,7 +1660,7 @@ impl RPCProcessor {
&self,
envelope: Envelope,
peer_noderef: NodeRef,
connection_descriptor: ConnectionDescriptor,
flow: Flow,
routing_domain: RoutingDomain,
body: Vec<u8>,
) -> EyreResult<()> {
@ -1668,7 +1668,7 @@ impl RPCProcessor {
detail: RPCMessageHeaderDetail::Direct(RPCMessageHeaderDetailDirect {
envelope,
peer_noderef,
connection_descriptor,
flow,
routing_domain,
}),
timestamp: get_aligned_timestamp(),

6
veilid-core/src/rpc_processor/rpc_signal.rs
View File

@ -49,8 +49,8 @@ impl RPCProcessor {
// Can't allow anything other than direct packets here, as handling reverse connections
// or anything like via signals over private routes would deanonymize the route
let connection_descriptor = match &msg.header.detail {
RPCMessageHeaderDetail::Direct(d) => d.connection_descriptor,
let flow = match &msg.header.detail {
RPCMessageHeaderDetail::Direct(d) => d.flow,
RPCMessageHeaderDetail::SafetyRouted(_) | RPCMessageHeaderDetail::PrivateRouted(_) => {
return Ok(NetworkResult::invalid_message("signal must be direct"));
}
@ -70,7 +70,7 @@ impl RPCProcessor {
let network_manager = self.network_manager();
let signal_info = signal.destructure();
network_manager
.handle_signal(connection_descriptor, signal_info)
.handle_signal(flow, signal_info)
.await
.map_err(RPCError::network)
}

8
veilid-core/src/rpc_processor/rpc_status.rs
View File

@ -162,13 +162,13 @@ impl RPCProcessor {
.network_manager()
.report_public_internet_socket_address(
sender_info.socket_address,
send_data_method.connection_descriptor,
send_data_method.unique_flow.flow,
target,
),
RoutingDomain::LocalNetwork => {
self.network_manager().report_local_network_socket_address(
sender_info.socket_address,
send_data_method.connection_descriptor,
send_data_method.unique_flow.flow,
target,
)
}
@ -208,7 +208,7 @@ impl RPCProcessor {
let (node_status, sender_info) = match &msg.header.detail {
RPCMessageHeaderDetail::Direct(detail) => {
let connection_descriptor = detail.connection_descriptor;
let flow = detail.flow;
let routing_domain = detail.routing_domain;
// Ensure the node status from the question is the kind for the routing domain we received the request in
@ -222,7 +222,7 @@ impl RPCProcessor {
// Get the peer address in the returned sender info
let sender_info = SenderInfo {
socket_address: *connection_descriptor.remote_address(),
socket_address: *flow.remote_address(),
};
// Make status answer

1
veilid-server/Cargo.toml
View File

@ -34,6 +34,7 @@ rt-tokio = [
]
tracking = ["veilid-core/tracking"]
network-result-extra = ["veilid-core/network-result-extra"]
verbose-tracing = ["veilid-core/verbose-tracing"]
[dependencies]
veilid-core = { path = "../veilid-core", default-features = false }