fix binding issues
This commit is contained in:
@@ -196,6 +196,7 @@ impl AttachmentManager {
|
||||
if let Err(err) = netman.startup().await {
|
||||
error!("network startup failed: {}", err);
|
||||
netman.shutdown().await;
|
||||
restart = true;
|
||||
break;
|
||||
}
|
||||
|
||||
|
@@ -906,7 +906,7 @@ impl NetworkManager {
|
||||
return Ok(NetworkResult::timeout());
|
||||
}
|
||||
ReceiptEvent::Cancelled => {
|
||||
bail!("reverse connect receipt cancelled from {:?}", target_nr);
|
||||
return Ok(NetworkResult::no_connection_other(format!("reverse connect receipt cancelled from {}", target_nr)))
|
||||
}
|
||||
};
|
||||
|
||||
|
@@ -115,6 +115,15 @@ impl Network {
|
||||
// tcp_stream.peer_addr().unwrap(),
|
||||
// );
|
||||
|
||||
if let Err(e) = tcp_stream.set_linger(Some(core::time::Duration::from_secs(0))) {
|
||||
log_net!(debug "Couldn't set TCP linger: {}", e);
|
||||
return;
|
||||
}
|
||||
if let Err(e) = tcp_stream.set_nodelay(true) {
|
||||
log_net!(debug "Couldn't set TCP nodelay: {}", e);
|
||||
return;
|
||||
}
|
||||
|
||||
let listener_state = listener_state.clone();
|
||||
let connection_manager = connection_manager.clone();
|
||||
|
||||
|
@@ -33,7 +33,7 @@ cfg_if! {
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(level = "trace", ret, err)]
|
||||
#[instrument(level = "trace", ret)]
|
||||
pub fn new_unbound_shared_udp_socket(domain: Domain) -> io::Result<Socket> {
|
||||
let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
|
||||
if domain == Domain::IPV6 {
|
||||
@@ -49,7 +49,7 @@ pub fn new_unbound_shared_udp_socket(domain: Domain) -> io::Result<Socket> {
|
||||
Ok(socket)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace", ret, err)]
|
||||
#[instrument(level = "trace", ret)]
|
||||
pub fn new_bound_shared_udp_socket(local_address: SocketAddr) -> io::Result<Socket> {
|
||||
let domain = Domain::for_address(local_address);
|
||||
let socket = new_unbound_shared_udp_socket(domain)?;
|
||||
@@ -61,7 +61,7 @@ pub fn new_bound_shared_udp_socket(local_address: SocketAddr) -> io::Result<Sock
|
||||
Ok(socket)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace", ret, err)]
|
||||
#[instrument(level = "trace", ret)]
|
||||
pub fn new_bound_first_udp_socket(local_address: SocketAddr) -> io::Result<Socket> {
|
||||
let domain = Domain::for_address(local_address);
|
||||
let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
|
||||
@@ -95,7 +95,7 @@ pub fn new_bound_first_udp_socket(local_address: SocketAddr) -> io::Result<Socke
|
||||
Ok(socket)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace", ret, err)]
|
||||
#[instrument(level = "trace", ret)]
|
||||
pub fn new_unbound_shared_tcp_socket(domain: Domain) -> io::Result<Socket> {
|
||||
let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))?;
|
||||
if let Err(e) = socket.set_linger(Some(core::time::Duration::from_secs(0))) {
|
||||
@@ -117,7 +117,7 @@ pub fn new_unbound_shared_tcp_socket(domain: Domain) -> io::Result<Socket> {
|
||||
Ok(socket)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace", ret, err)]
|
||||
#[instrument(level = "trace", ret)]
|
||||
pub fn new_bound_shared_tcp_socket(local_address: SocketAddr) -> io::Result<Socket> {
|
||||
let domain = Domain::for_address(local_address);
|
||||
let socket = new_unbound_shared_tcp_socket(domain)?;
|
||||
@@ -129,7 +129,7 @@ pub fn new_bound_shared_tcp_socket(local_address: SocketAddr) -> io::Result<Sock
|
||||
Ok(socket)
|
||||
}
|
||||
|
||||
#[instrument(level = "trace", ret, err)]
|
||||
#[instrument(level = "trace", ret)]
|
||||
pub fn new_bound_first_tcp_socket(local_address: SocketAddr) -> io::Result<Socket> {
|
||||
let domain = Domain::for_address(local_address);
|
||||
|
||||
|
@@ -243,7 +243,24 @@ impl Network {
|
||||
bail!("No valid listen address: {}", listen_address);
|
||||
}
|
||||
let port = sockaddrs[0].port();
|
||||
if !self.bind_first_tcp_port(port) {
|
||||
|
||||
let mut attempts = 10;
|
||||
let mut success = false;
|
||||
while attempts >= 0 {
|
||||
if self.bind_first_tcp_port(port) {
|
||||
success = true;
|
||||
break;
|
||||
}
|
||||
attempts -= 1;
|
||||
|
||||
// Wait 5 seconds before trying again
|
||||
log_net!(debug
|
||||
"Binding TCP port at {} failed, waiting. Attempts remaining = {}",
|
||||
port, attempts
|
||||
);
|
||||
sleep(5000).await
|
||||
}
|
||||
if !success {
|
||||
bail!("Could not find free tcp port to listen on");
|
||||
}
|
||||
Ok((port, sockaddrs.iter().map(|s| s.ip()).collect()))
|
||||
|
@@ -5,7 +5,7 @@ impl RoutingTable {
|
||||
pub fn find_all_closest_peers(&self, key: TypedKey) -> NetworkResult<Vec<PeerInfo>> {
|
||||
let Some(own_peer_info) = self.get_own_peer_info(RoutingDomain::PublicInternet) else {
|
||||
// Our own node info is not yet available, drop this request.
|
||||
return NetworkResult::service_unavailable();
|
||||
return NetworkResult::service_unavailable("Not finding closest peers because our peer info is not yet available");
|
||||
};
|
||||
|
||||
// find N nodes closest to the target node in our routing table
|
||||
|
@@ -591,20 +591,8 @@ impl VeilidAPI {
|
||||
.map_err(VeilidAPIError::internal)?
|
||||
{
|
||||
NetworkResult::Value(v) => v,
|
||||
NetworkResult::Timeout => {
|
||||
return Ok("Timeout".to_owned());
|
||||
}
|
||||
NetworkResult::ServiceUnavailable => {
|
||||
return Ok("ServiceUnavailable".to_owned());
|
||||
}
|
||||
NetworkResult::NoConnection(e) => {
|
||||
return Ok(format!("NoConnection({})", e));
|
||||
}
|
||||
NetworkResult::AlreadyExists(e) => {
|
||||
return Ok(format!("AlreadyExists({})", e));
|
||||
}
|
||||
NetworkResult::InvalidMessage(e) => {
|
||||
return Ok(format!("InvalidMessage({})", e));
|
||||
r => {
|
||||
return Ok(r.to_string());
|
||||
}
|
||||
};
|
||||
|
||||
|
@@ -148,7 +148,10 @@ impl RoutingContext {
|
||||
let answer = match rpc_processor.rpc_call_app_call(dest, request).await {
|
||||
Ok(NetworkResult::Value(v)) => v,
|
||||
Ok(NetworkResult::Timeout) => apibail_timeout!(),
|
||||
Ok(NetworkResult::ServiceUnavailable) => apibail_try_again!(),
|
||||
Ok(NetworkResult::ServiceUnavailable(e)) => {
|
||||
log_network_result!(format!("app_call: ServiceUnavailable({})", e));
|
||||
apibail_try_again!()
|
||||
}
|
||||
Ok(NetworkResult::NoConnection(e)) | Ok(NetworkResult::AlreadyExists(e)) => {
|
||||
apibail_no_connection!(e);
|
||||
}
|
||||
@@ -173,7 +176,10 @@ impl RoutingContext {
|
||||
match rpc_processor.rpc_call_app_message(dest, message).await {
|
||||
Ok(NetworkResult::Value(())) => {}
|
||||
Ok(NetworkResult::Timeout) => apibail_timeout!(),
|
||||
Ok(NetworkResult::ServiceUnavailable) => apibail_try_again!(),
|
||||
Ok(NetworkResult::ServiceUnavailable(e)) => {
|
||||
log_network_result!(format!("app_message: ServiceUnavailable({})", e));
|
||||
apibail_try_again!()
|
||||
}
|
||||
Ok(NetworkResult::NoConnection(e)) | Ok(NetworkResult::AlreadyExists(e)) => {
|
||||
apibail_no_connection!(e);
|
||||
}
|
||||
|
Reference in New Issue
Block a user