GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Switch to PollSender

Browse Source
This commit is contained in:
Drazen
2024-06-12 17:58:07 +02:00
committed by durch
parent 575056ac67
commit eed87ff4c9
14 changed files with 48 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3
View File
@@ -4180,6 +4180,7 @@ dependencies = [
"tokio",
"tokio-stream",
"tokio-tungstenite",
"tokio-util",
"tungstenite",
"url",
"wasm-bindgen",
@@ -4956,6 +4957,7 @@ dependencies = [
"time",
"tokio",
"tokio-tungstenite",
"tokio-util",
"url",
"zeroize",
]
@@ -5300,6 +5302,7 @@ dependencies = [
"tap",
"thiserror",
"tokio",
"tokio-util",
"url",
]
common/client-core/Cargo.toml
+4 -1
View File
@@ -26,6 +26,7 @@ tap = { workspace = true }
thiserror = { workspace = true }
url = { workspace = true, features = ["serde"] }
tokio = { workspace = true, features = ["macros"] }
tokio-util = { workspace = true }
time = { workspace = true }
zeroize = { workspace = true }
@@ -47,7 +48,9 @@ nym-validator-client = { path = "../client-libs/validator-client", default-featu
nym-task = { path = "../task" }
nym-credential-storage = { path = "../credential-storage" }
nym-network-defaults = { path = "../network-defaults" }
nym-client-core-config-types = { path = "./config-types", features = ["disk-persistence"] }
nym-client-core-config-types = { path = "./config-types", features = [
"disk-persistence",
] }
nym-client-core-surb-storage = { path = "./surb-storage" }
nym-client-core-gateways-storage = { path = "./gateways-storage" }
common/client-core/src/client/base_client/mod.rs
+7 -2
View File
@@ -59,6 +59,7 @@ use std::fmt::Debug;
use std::os::raw::c_int as RawFd;
use std::path::Path;
use std::sync::Arc;
use tokio_util::sync::PollSender;
use url::Url;
#[cfg(all(
@@ -82,7 +83,11 @@ impl ClientInput {
&self,
message: InputMessage,
) -> Result<(), tokio::sync::mpsc::error::SendError<InputMessage>> {
self.input_sender.send(message).await
if let Some(channel) = self.input_sender.get_ref() {
channel.send(message).await
} else {
Err(tokio::sync::mpsc::error::SendError(message))
}
}
}
@@ -804,7 +809,7 @@ where
client_input: ClientInputStatus::AwaitingProducer {
client_input: ClientInput {
connection_command_sender: client_connection_tx,
input_sender,
input_sender: PollSender::new(input_sender),
},
},
client_output: ClientOutputStatus::AwaitingConsumer {
common/client-core/src/client/inbound_messages.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketType;
use nym_task::connections::TransmissionLane;
pub type InputMessageSender = tokio::sync::mpsc::Sender<InputMessage>;
pub type InputMessageSender = tokio_util::sync::PollSender<InputMessage>;
pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
#[derive(Debug)]
common/socks5-client-core/Cargo.toml
+1
View File
@@ -19,6 +19,7 @@ serde = { workspace = true, features = ["derive"] } # for config serialization/d
tap = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "net", "signal"] }
tokio-util = { workspace = true }
url = { workspace = true }
nym-bandwidth-controller = { path = "../../common/bandwidth-controller" }
common/socks5-client-core/src/socks/client.rs
+1
View File
@@ -7,6 +7,7 @@ use super::{SocksVersion, RESERVED, SOCKS4_VERSION, SOCKS5_VERSION};
use crate::config;
use futures::channel::mpsc;
use futures::task::{Context, Poll};
use futures::SinkExt;
use log::*;
use nym_client_core::client::inbound_messages::{InputMessage, InputMessageSender};
use nym_service_providers_common::interface::{ProviderInterfaceVersion, RequestVersion};
common/socks5/proxy-helpers/src/ordered_sender.rs
+6 -4
View File
@@ -7,7 +7,7 @@ use log::{debug, error};
use nym_socks5_requests::{ConnectionId, SocketData};
use std::io;
pub(crate) struct OrderedMessageSender<F, S> {
pub(crate) struct OrderedMessageSender<F, S: Send> {
connection_id: ConnectionId,
// addresses are provided for better logging
local_destination_address: String,
@@ -18,7 +18,7 @@ pub(crate) struct OrderedMessageSender<F, S> {
mix_message_adapter: F,
}
impl<F, S> OrderedMessageSender<F, S>
impl<F, S: Send> OrderedMessageSender<F, S>
where
F: Fn(SocketData) -> S,
{
@@ -56,8 +56,10 @@ where
}
async fn send_message(&self, message: S) {
if self.mixnet_sender.send(message).await.is_err() {
panic!("BatchRealMessageReceiver has stopped receiving!")
if let Some(sender) = self.mixnet_sender.get_ref() {
if sender.send(message).await.is_err() {
panic!("BatchRealMessageReceiver has stopped receiving!")
}
}
}
common/socks5/proxy-helpers/src/proxy_runner/inbound.rs
+1 -1
View File
@@ -74,7 +74,7 @@ async fn wait_for_lane(
}
}
pub(super) async fn run_inbound<F, S>(
pub(super) async fn run_inbound<F, S: Send>(
mut reader: OwnedReadHalf,
mut message_sender: OrderedMessageSender<F, S>,
connection_id: ConnectionId,
common/socks5/proxy-helpers/src/proxy_runner/mod.rs
+2 -1
View File
@@ -6,6 +6,7 @@ use crate::ordered_sender::OrderedMessageSender;
use nym_socks5_requests::{ConnectionId, SocketData};
use nym_task::connections::LaneQueueLengths;
use nym_task::TaskClient;
use tokio_util::sync::PollSender;
use std::fmt::Debug;
use std::{sync::Arc, time::Duration};
use tokio::{net::TcpStream, sync::Notify};
@@ -35,7 +36,7 @@ impl From<(Vec<u8>, bool)> for ProxyMessage {
}
}
pub type MixProxySender<S> = tokio::sync::mpsc::Sender<S>;
pub type MixProxySender<S> = PollSender<S>;
pub type MixProxyReader<S> = tokio::sync::mpsc::Receiver<S>;
// TODO: when we finally get to implementing graceful shutdown,
common/statistics/src/collector.rs
+1 -1
View File
@@ -19,7 +19,7 @@ pub trait StatisticsCollector {
interval: Duration,
timestamp: DateTime<Utc>,
) -> StatsMessage;
async fn send_stats_message(&self, stats_message: StatsMessage) -> Result<(), StatsError>;
async fn send_stats_message(&mut self, stats_message: StatsMessage) -> Result<(), StatsError>;
async fn reset_stats(&mut self);
}
gateway/src/node/statistics/collector.rs
+1 -1
View File
@@ -52,7 +52,7 @@ impl StatisticsCollector for GatewayStatisticsCollector {
}
}
async fn send_stats_message(&self, stats_message: StatsMessage) -> Result<(), StatsError> {
async fn send_stats_message(&mut self, stats_message: StatsMessage) -> Result<(), StatsError> {
build_and_send_statistics_request(stats_message, self.statistics_service_url.to_string())
.await
}
service-providers/network-requester/Cargo.toml
+12 -5
View File
@@ -20,7 +20,7 @@ anyhow = { workspace = true }
addr = { workspace = true }
async-trait = { workspace = true }
bs58 = { workspace = true }
clap = { workspace = true, features = ["cargo", "derive"]}
clap = { workspace = true, features = ["cargo", "derive"] }
dirs = "4.0"
futures = { workspace = true }
humantime-serde = { workspace = true }
@@ -33,19 +33,26 @@ regex = { workspace = true }
reqwest = { workspace = true, features = ["json"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"]}
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"] }
tap = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = [ "net", "rt-multi-thread", "macros" ] }
tokio = { workspace = true, features = ["net", "rt-multi-thread", "macros"] }
tokio-tungstenite = { workspace = true }
tokio-util = { workspace = true }
url = { workspace = true }
time = { workspace = true }
zeroize = { workspace = true }
# internal
nym-async-file-watcher = { path = "../../common/async-file-watcher" }
nym-bin-common = { path = "../../common/bin-common", features = ["output_format"] }
nym-client-core = { path = "../../common/client-core", features = ["cli", "fs-gateways-storage", "fs-surb-storage"] }
nym-bin-common = { path = "../../common/bin-common", features = [
"output_format",
] }
nym-client-core = { path = "../../common/client-core", features = [
"cli",
"fs-gateways-storage",
"fs-surb-storage",
] }
nym-client-websocket-requests = { path = "../../clients/native/websocket-requests" }
nym-config = { path = "../../common/config" }
nym-credentials = { path = "../../common/credentials" }
service-providers/network-requester/src/core.rs
+6 -2
View File
@@ -10,6 +10,7 @@ use crate::{reply, socks5};
use async_trait::async_trait;
use futures::channel::{mpsc, oneshot};
use futures::stream::StreamExt;
use futures::SinkExt;
use log::{debug, warn};
use nym_bin_common::bin_info_owned;
use nym_client_core::client::mix_traffic::transceiver::GatewayTransceiver;
@@ -38,6 +39,7 @@ use nym_statistics_common::collector::StatisticsSender;
use nym_task::connections::LaneQueueLengths;
use nym_task::manager::TaskHandle;
use nym_task::TaskClient;
use tokio_util::sync::PollSender;
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -284,6 +286,8 @@ impl NRServiceProviderBuilder {
// going to be used by `mixnet_response_listener`
let (mix_input_sender, mix_input_receiver) = tokio::sync::mpsc::channel::<MixnetMessage>(1);
let mix_input_sender = PollSender::new(mix_input_sender);
// Controller for managing all active connections.
let (mut active_connections_controller, controller_sender) = Controller::new(
mixnet_client.connection_command_sender(),
@@ -443,7 +447,7 @@ impl NRServiceProvider {
return_address: reply::MixnetAddress,
biggest_packet_size: PacketSize,
controller_sender: ControllerSender,
mix_input_sender: MixProxySender<MixnetMessage>,
mut mix_input_sender: MixProxySender<MixnetMessage>,
lane_queue_lengths: LaneQueueLengths,
mut shutdown: TaskClient,
) {
@@ -537,7 +541,7 @@ impl NRServiceProvider {
.unwrap_or(traffic_config.primary_packet_size);
let controller_sender_clone = self.controller_sender.clone();
let mix_input_sender_clone = self.mix_input_sender.clone();
let mut mix_input_sender_clone = self.mix_input_sender.clone();
let lane_queue_lengths_clone = self.mixnet_client.shared_lane_queue_lengths();
let mut shutdown = self.shutdown.get_handle();
service-providers/network-requester/src/statistics/collector.rs
+2 -1
View File
@@ -5,6 +5,7 @@ use super::error::StatsError;
use crate::core::new_legacy_request_version;
use crate::reply::MixnetMessage;
use async_trait::async_trait;
use futures::SinkExt;
use log::*;
use nym_service_providers_common::interface::RequestVersion;
use nym_socks5_proxy_helpers::proxy_runner::MixProxySender;
@@ -165,7 +166,7 @@ impl StatisticsCollector for ServiceStatisticsCollector {
}
async fn send_stats_message(
&self,
&mut self,
stats_message: StatsMessage,
) -> Result<(), CommonStatsError> {
let msg = build_statistics_request_bytes(stats_message)?;