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

WIP cleanup before machine switch

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2022-10-05 09:49:23 +01:00
parent 932e137e27
commit 31056a90cb
15 changed files with 332 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -651,7 +651,6 @@ dependencies = [
"config",
"crypto",
"dirs",
"fluvio-wasm-timer",
"futures",
"gateway-client",
"gateway-requests",
@@ -673,6 +672,7 @@ dependencies = [
"validator-client",
"wasm-bindgen",
"wasm-bindgen-futures",
"wasm-timer",
]
[[package]]
clients/client-core/Cargo.toml
+3 -2
View File
@@ -37,8 +37,9 @@ version = "0.4"
[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen]
version = "=0.2.78"
[target."cfg(target_arch = \"wasm32\")".dependencies.fluvio-wasm-timer]
version = "0.2.5"
[target."cfg(target_arch = \"wasm32\")".dependencies.wasm-timer]
git = "https://github.com/jstuczyn/wasm-timer"
rev = "2859241251dd00c54a9c99b74727b99bd7feffb4"
[target."cfg(not(target_arch = \"wasm32\"))".dependencies.task]
path = "../../common/task"
clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs
+32 -35
View File
@@ -20,8 +20,6 @@ pub(super) struct AcknowledgementListener {
ack_key: Arc<AckKey>,
ack_receiver: AcknowledgementReceiver,
action_sender: ActionSender,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
impl AcknowledgementListener {
@@ -29,14 +27,11 @@ impl AcknowledgementListener {
ack_key: Arc<AckKey>,
ack_receiver: AcknowledgementReceiver,
action_sender: ActionSender,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
AcknowledgementListener {
ack_key,
ack_receiver,
action_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -71,39 +66,41 @@ impl AcknowledgementListener {
.unwrap();
}
pub(super) async fn run(&mut self) {
debug!("Started AcknowledgementListener");
async fn handle_ack_receiver_item(&mut self, item: Vec<Vec<u8>>) {
// realistically we would only be getting one ack at the time
for ack in item {
self.on_ack(ack).await;
}
}
// todo: make it non-wasm compatible
debug!("Started AcknowledgementListener");
while let Some(acks) = self.ack_receiver.next().await {
// realistically we would only be getting one ack at the time
for ack in acks {
self.on_ack(ack).await;
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
debug!("Started AcknowledgementListener with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
acks = self.ack_receiver.next() => match acks {
Some(acks) => self.handle_ack_receiver_item(acks).await,
None => {
log::trace!("AcknowledgementListener: Stopping since channel closed");
break;
}
},
_ = shutdown.recv() => {
log::trace!("AcknowledgementListener: Received shutdown");
}
}
}
error!("TODO: error msg. Or maybe panic?")
assert!(shutdown.is_shutdown_poll());
log::debug!("AcknowledgementListener: Exiting");
}
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// acks = self.ack_receiver.next() => match acks {
// Some(acks) => {
// // realistically we would only be getting one ack at the time
// for ack in acks {
// self.on_ack(ack).await;
// }
// },
// None => {
// log::trace!("AcknowledgementListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("AcknowledgementListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("AcknowledgementListener: Exiting");
#[cfg(target_arch = "wasm32")]
pub(super) async fn run(&mut self) {
debug!("Started AcknowledgementListener without graceful shutdown support");
while let Some(acks) = self.ack_receiver.next().await {
self.handle_ack_receiver_item(acks).await
}
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs
+36 -39
View File
@@ -102,17 +102,12 @@ pub(super) struct ActionController {
/// Channel for notifying `RetransmissionRequestListener` about expired acknowledgements.
retransmission_sender: RetransmissionRequestSender,
/// Listen for shutdown notifications
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
impl ActionController {
pub(super) fn new(
config: Config,
retransmission_sender: RetransmissionRequestSender,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> (Self, ActionSender) {
let (sender, receiver) = mpsc::unbounded();
(
@@ -122,8 +117,6 @@ impl ActionController {
pending_acks_timers: NonExhaustiveDelayQueue::new(),
incoming_actions: receiver,
retransmission_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
},
sender,
)
@@ -255,42 +248,46 @@ impl ActionController {
}
}
pub(super) async fn run(&mut self) {
// TODO: shutdown without wasm etc etc
loop {
// at some point there will be a global shutdown signal here as the third option
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
debug!("Started ActionController with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
action = self.incoming_actions.next() => match action {
Some(action) => self.process_action(action),
None => {
log::trace!(
"ActionController: Stopping since incoming actions channel closed"
);
break;
}
},
expired_ack = self.pending_acks_timers.next() => match expired_ack {
Some(expired_ack) => self.handle_expired_ack_timer(expired_ack),
None => {
log::trace!("ActionController: Stopping since ack channel closed");
break;
}
},
_ = shutdown.recv() => {
log::trace!("ActionController: Received shutdown");
}
}
}
assert!(shutdown.is_shutdown_poll());
log::debug!("ActionController: Exiting");
}
#[cfg(target_arch = "wasm32")]
pub(super) async fn run(&mut self) {
debug!("Started ActionController without graceful shutdown support");
loop {
tokio::select! {
// we NEVER expect for ANY sender to get dropped so unwrap here is fine
action = self.incoming_actions.next() => self.process_action(action.unwrap()),
// pending ack queue Stream CANNOT return a `None` so unwrap here is fine
expired_ack = self.pending_acks_timers.next() => self.handle_expired_ack_timer(expired_ack.unwrap())
}
}
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// action = self.incoming_actions.next() => match action {
// Some(action) => self.process_action(action),
// None => {
// log::trace!(
// "ActionController: Stopping since incoming actions channel closed"
// );
// break;
// }
// },
// expired_ack = self.pending_acks_timers.next() => match expired_ack {
// Some(expired_ack) => self.handle_expired_ack_timer(expired_ack),
// None => {
// log::trace!("ActionController: Stopping since ack channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("ActionController: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("ActionController: Exiting");
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs
+26 -27
View File
@@ -38,8 +38,6 @@ where
topology_access: TopologyAccessor,
#[cfg(feature = "reply-surb")]
reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
impl<R> InputMessageListener<R>
@@ -58,7 +56,6 @@ where
real_message_sender: BatchRealMessageSender,
topology_access: TopologyAccessor,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
InputMessageListener {
ack_key,
@@ -70,8 +67,6 @@ where
topology_access,
#[cfg(feature = "reply-surb")]
reply_key_storage,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -193,31 +188,35 @@ where
}
}
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
debug!("Started InputMessageListener with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
input_msg = self.input_receiver.next() => match input_msg {
Some(input_msg) => {
self.on_input_message(input_msg).await;
},
None => {
log::trace!("InputMessageListener: Stopping since channel closed");
break;
}
},
_ = shutdown.recv() => {
log::trace!("InputMessageListener: Received shutdown");
}
}
}
assert!(shutdown.is_shutdown_poll());
log::debug!("InputMessageListener: Exiting");
}
#[cfg(target_arch = "wasm32")]
pub(super) async fn run(&mut self) {
// TODO: shutdown without wasm etc etc
debug!("Started InputMessageListener");
debug!("Started InputMessageListener without graceful shutdown support");
while let Some(input_msg) = self.input_receiver.next().await {
self.on_input_message(input_msg).await;
}
// debug!("Started InputMessageListener");
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// input_msg = self.input_receiver.next() => match input_msg {
// Some(input_msg) => {
// self.on_input_message(input_msg).await;
// },
// None => {
// log::trace!("InputMessageListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("InputMessageListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("InputMessageListener: Exiting");
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs
+75 -61
View File
@@ -146,11 +146,11 @@ pub(super) struct AcknowledgementController<R>
where
R: CryptoRng + Rng,
{
acknowledgement_listener: Option<AcknowledgementListener>,
input_message_listener: Option<InputMessageListener<R>>,
retransmission_request_listener: Option<RetransmissionRequestListener<R>>,
sent_notification_listener: Option<SentNotificationListener>,
action_controller: Option<ActionController>,
acknowledgement_listener: AcknowledgementListener,
input_message_listener: InputMessageListener<R>,
retransmission_request_listener: RetransmissionRequestListener<R>,
sent_notification_listener: SentNotificationListener,
action_controller: ActionController,
}
impl<R> AcknowledgementController<R>
@@ -166,18 +166,13 @@ where
ack_recipient: Recipient,
connectors: AcknowledgementControllerConnectors,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
let action_config =
action_controller::Config::new(config.ack_wait_addition, config.ack_wait_multiplier);
let (action_controller, action_sender) = ActionController::new(
action_config,
retransmission_tx,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
let (action_controller, action_sender) =
ActionController::new(action_config, retransmission_tx);
let message_preparer = MessagePreparer::new(
rng,
@@ -191,8 +186,6 @@ where
Arc::clone(&ack_key),
connectors.ack_receiver,
action_sender.clone(),
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
// will listen for any new messages from the client
@@ -206,8 +199,6 @@ where
topology_access.clone(),
#[cfg(feature = "reply-surb")]
reply_key_storage,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
// will listen for any ack timeouts and trigger retransmission
@@ -219,72 +210,95 @@ where
connectors.real_message_sender,
retransmission_rx,
topology_access,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
// will listen for events indicating the packet was sent through the network so that
// the retransmission timer should be started.
let sent_notification_listener = SentNotificationListener::new(
connectors.sent_notifier,
action_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
);
let sent_notification_listener =
SentNotificationListener::new(connectors.sent_notifier, action_sender);
AcknowledgementController {
acknowledgement_listener: Some(acknowledgement_listener),
input_message_listener: Some(input_message_listener),
retransmission_request_listener: Some(retransmission_request_listener),
sent_notification_listener: Some(sent_notification_listener),
action_controller: Some(action_controller),
acknowledgement_listener,
input_message_listener,
retransmission_request_listener,
sent_notification_listener,
action_controller,
}
}
pub(super) async fn run(&mut self) {
let mut acknowledgement_listener = self.acknowledgement_listener.take().unwrap();
let mut input_message_listener = self.input_message_listener.take().unwrap();
let mut retransmission_request_listener =
self.retransmission_request_listener.take().unwrap();
let mut sent_notification_listener = self.sent_notification_listener.take().unwrap();
let mut action_controller = self.action_controller.take().unwrap();
#[cfg(not(target_arch = "wasm32"))]
pub(super) fn start_with_shutdown(self, shutdown: task::ShutdownListener) {
let mut acknowledgement_listener = self.acknowledgement_listener;
let mut input_message_listener = self.input_message_listener;
let mut retransmission_request_listener = self.retransmission_request_listener;
let mut sent_notification_listener = self.sent_notification_listener;
let mut action_controller = self.action_controller;
// the below are log messages are errors as at the current stage we do not expect any of
// the task to ever finish. This will of course change once we introduce
// graceful shutdowns.
let ack_listener_fut = spawn_future(async move {
acknowledgement_listener.run().await;
let shutdown_handle = shutdown.clone();
spawn_future(async move {
acknowledgement_listener
.run_with_shutdown(shutdown_handle)
.await;
debug!("The acknowledgement listener has finished execution!");
});
let input_listener_fut = spawn_future(async move {
input_message_listener.run().await;
let shutdown_handle = shutdown.clone();
spawn_future(async move {
input_message_listener
.run_with_shutdown(shutdown_handle)
.await;
debug!("The input listener has finished execution!");
});
let retransmission_req_fut = spawn_future(async move {
retransmission_request_listener.run().await;
let shutdown_handle = shutdown.clone();
spawn_future(async move {
retransmission_request_listener
.run_with_shutdown(shutdown_handle)
.await;
debug!("The retransmission request listener has finished execution!");
});
let sent_notification_fut = spawn_future(async move {
sent_notification_listener.run().await;
let shutdown_handle = shutdown.clone();
spawn_future(async move {
sent_notification_listener
.run_with_shutdown(shutdown_handle)
.await;
debug!("The sent notification listener has finished execution!");
});
let action_controller_fut = spawn_future(async move {
action_controller.run().await;
spawn_future(async move {
action_controller.run_with_shutdown(shutdown).await;
debug!("The controller has finished execution!");
});
// // technically we don't have to bring `AcknowledgementController` back to a valid state
// // but we can do it, so why not? Perhaps it might be useful if we wanted to allow
// // for restarts of certain modules without killing the entire process.
// self.acknowledgement_listener = Some(ack_listener_fut.await.unwrap());
// self.input_message_listener = Some(input_listener_fut.await.unwrap());
// self.retransmission_request_listener = Some(retransmission_req_fut.await.unwrap());
// self.sent_notification_listener = Some(sent_notification_fut.await.unwrap());
// self.action_controller = Some(action_controller_fut.await.unwrap());
}
#[allow(dead_code)]
pub(super) fn start(mut self) {
spawn_future(async move { self.run().await })
#[cfg(target_arch = "wasm32")]
pub(super) fn start(self) {
let mut acknowledgement_listener = self.acknowledgement_listener;
let mut input_message_listener = self.input_message_listener;
let mut retransmission_request_listener = self.retransmission_request_listener;
let mut sent_notification_listener = self.sent_notification_listener;
let mut action_controller = self.action_controller;
spawn_future(async move {
acknowledgement_listener.run().await;
error!("The acknowledgement listener has finished execution!");
});
spawn_future(async move {
input_message_listener.run().await;
error!("The input listener has finished execution!");
});
spawn_future(async move {
retransmission_request_listener.run().await;
error!("The retransmission request listener has finished execution!");
});
spawn_future(async move {
sent_notification_listener.run().await;
error!("The sent notification listener has finished execution!");
});
spawn_future(async move {
action_controller.run().await;
error!("The controller has finished execution!");
});
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs
+25 -25
View File
@@ -30,8 +30,6 @@ where
real_message_sender: BatchRealMessageSender,
request_receiver: RetransmissionRequestReceiver,
topology_access: TopologyAccessor,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
impl<R> RetransmissionRequestListener<R>
@@ -47,7 +45,6 @@ where
real_message_sender: BatchRealMessageSender,
request_receiver: RetransmissionRequestReceiver,
topology_access: TopologyAccessor,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
RetransmissionRequestListener {
ack_key,
@@ -57,8 +54,6 @@ where
real_message_sender,
request_receiver,
topology_access,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -128,29 +123,34 @@ where
.unwrap();
}
pub(super) async fn run(&mut self) {
debug!("Started RetransmissionRequestListener");
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
debug!("Started RetransmissionRequestListener with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
timed_out_ack = self.request_receiver.next() => match timed_out_ack {
Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await,
None => {
log::trace!("RetransmissionRequestListener: Stopping since channel closed");
break;
}
},
_ = shutdown.recv() => {
log::trace!("RetransmissionRequestListener: Received shutdown");
}
}
}
assert!(shutdown.is_shutdown_poll());
log::debug!("RetransmissionRequestListener: Exiting");
}
#[cfg(target_arch = "wasm32")]
pub(super) async fn run(&mut self) {
debug!("Started RetransmissionRequestListener without graceful shutdown support");
// TODO: shutdown without wasm etc etc
while let Some(timed_out_ack) = self.request_receiver.next().await {
self.on_retransmission_request(timed_out_ack).await;
}
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// timed_out_ack = self.request_receiver.next() => match timed_out_ack {
// Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await,
// None => {
// log::trace!("RetransmissionRequestListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("RetransmissionRequestListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("RetransmissionRequestListener: Exiting");
}
}
clients/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs
+27 -27
View File
@@ -17,21 +17,16 @@ use task::ShutdownListener;
pub(super) struct SentNotificationListener {
sent_notifier: SentPacketNotificationReceiver,
action_sender: ActionSender,
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
impl SentNotificationListener {
pub(super) fn new(
sent_notifier: SentPacketNotificationReceiver,
action_sender: ActionSender,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
SentNotificationListener {
sent_notifier,
action_sender,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -50,31 +45,36 @@ impl SentNotificationListener {
.unwrap();
}
pub(super) async fn run(&mut self) {
debug!("Started SentNotificationListener");
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
debug!("Started SentNotificationListener with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
frag_id = self.sent_notifier.next() => match frag_id {
Some(frag_id) => {
self.on_sent_message(frag_id).await;
}
None => {
log::trace!("SentNotificationListener: Stopping since channel closed");
break;
}
},
_ = shutdown.recv() => {
log::trace!("SentNotificationListener: Received shutdown");
}
}
}
assert!(shutdown.is_shutdown_poll());
log::debug!("SentNotificationListener: Exiting");
}
#[cfg(target_arch = "wasm32")]
pub(super) async fn run(&mut self) {
debug!("Started SentNotificationListener without graceful shutdown support");
// TODO: shutdown without wasm etc etc
while let Some(frag_id) = self.sent_notifier.next().await {
self.on_sent_message(frag_id).await;
}
// while !self.shutdown.is_shutdown() {
// tokio::select! {
// frag_id = self.sent_notifier.next() => match frag_id {
// Some(frag_id) => {
// self.on_sent_message(frag_id).await;
// }
// None => {
// log::trace!("SentNotificationListener: Stopping since channel closed");
// break;
// }
// },
// _ = self.shutdown.recv() => {
// log::trace!("SentNotificationListener: Received shutdown");
// }
// }
// }
// assert!(self.shutdown.is_shutdown_poll());
// log::debug!("SentNotificationListener: Exiting");
}
}
clients/client-core/src/client/real_messages_control/mod.rs
+21 -28
View File
@@ -82,8 +82,8 @@ pub struct RealMessagesController<R>
where
R: CryptoRng + Rng,
{
out_queue_control: Option<OutQueueControl<R>>,
ack_control: Option<AcknowledgementController<R>>,
out_queue_control: OutQueueControl<R>,
ack_control: AcknowledgementController<R>,
}
// obviously when we finally make shared rng that is on 'higher' level, this should become
@@ -96,7 +96,6 @@ impl RealMessagesController<OsRng> {
mix_sender: BatchMixMessageSender,
topology_access: TopologyAccessor,
#[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
let rng = OsRng;
@@ -126,8 +125,6 @@ impl RealMessagesController<OsRng> {
ack_controller_connectors,
#[cfg(feature = "reply-surb")]
reply_key_storage,
#[cfg(not(target_arch = "wasm32"))]
shutdown.clone(),
);
let out_queue_config = real_traffic_stream::Config::new(
@@ -145,40 +142,36 @@ impl RealMessagesController<OsRng> {
rng,
config.self_recipient,
topology_access,
#[cfg(not(target_arch = "wasm32"))]
shutdown,
);
RealMessagesController {
out_queue_control: Some(out_queue_control),
ack_control: Some(ack_control),
out_queue_control,
ack_control,
}
}
pub(super) async fn run(&mut self) {
let mut out_queue_control = self.out_queue_control.take().unwrap();
let mut ack_control = self.ack_control.take().unwrap();
#[cfg(not(target_arch = "wasm32"))]
pub fn start_with_shutdown(self, shutdown: task::ShutdownListener) {
let mut out_queue_control = self.out_queue_control;
let mut ack_control = self.ack_control;
// the below are log messages are errors as at the current stage we do not expect any of
// the task to ever finish. This will of course change once we introduce
// graceful shutdowns.
let out_queue_control_fut = spawn_future(async move {
out_queue_control.run_out_queue_control().await;
let shutdown_handle = shutdown.clone();
spawn_future(async move {
out_queue_control.run_with_shutdown(shutdown_handle).await;
debug!("The out queue controller has finished execution!");
});
let ack_control_fut = spawn_future(async move {
ack_control.run().await;
debug!("The acknowledgement controller has finished execution!");
});
// // technically we don't have to bring `RealMessagesController` back to a valid state
// // but we can do it, so why not? Perhaps it might be useful if we wanted to allow
// // for restarts of certain modules without killing the entire process.
// self.out_queue_control = Some(out_queue_control_fut.await.unwrap());
// self.ack_control = Some(ack_control_fut.await.unwrap());
ack_control.start_with_shutdown(shutdown);
}
#[cfg(target_arch = "wasm32")]
pub fn start(mut self) {
spawn_future(async move { self.run().await })
let mut out_queue_control = self.out_queue_control;
let mut ack_control = self.ack_control;
spawn_future(async move {
out_queue_control.run().await;
debug!("The out queue controller has finished execution!");
});
ack_control.start();
}
}
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+55 -42
View File
@@ -24,7 +24,7 @@ use std::time::Duration;
use tokio::time;
#[cfg(target_arch = "wasm32")]
use fluvio_wasm_timer as wasm_timer;
use wasm_timer;
#[cfg(not(target_arch = "wasm32"))]
use task::ShutdownListener;
@@ -95,10 +95,6 @@ where
/// Buffer containing all real messages received. It is first exhausted before more are pulled.
received_buffer: VecDeque<RealMessage>,
/// Listens for shutdown signals
#[cfg(not(target_arch = "wasm32"))]
shutdown: ShutdownListener,
}
pub(crate) struct RealMessage {
@@ -199,7 +195,6 @@ where
rng: R,
our_full_destination: Recipient,
topology_access: TopologyAccessor,
#[cfg(not(target_arch = "wasm32"))] shutdown: ShutdownListener,
) -> Self {
#[cfg(not(target_arch = "wasm32"))]
let next_delay = Box::pin(time::sleep(Default::default()));
@@ -218,8 +213,6 @@ where
rng,
topology_access,
received_buffer: VecDeque::with_capacity(0), // we won't be putting any data into this guy directly
#[cfg(not(target_arch = "wasm32"))]
shutdown,
}
}
@@ -295,49 +288,69 @@ where
tokio::task::yield_now().await;
}
// Send messages at certain rate and if no real traffic is available, send cover message.
async fn run_normal_out_queue(&mut self) {
// // Send messages at certain rate and if no real traffic is available, send cover message.
// async fn run_normal_out_queue(&mut self) {
// // we should set initial delay only when we actually start the stream
// let sampled =
// sample_poisson_duration(&mut self.rng, self.config.average_message_sending_delay);
//
// #[cfg(not(target_arch = "wasm32"))]
// let next_delay = Box::pin(time::sleep(sampled));
//
// #[cfg(target_arch = "wasm32")]
// let next_delay = Box::pin(wasm_timer::Delay::new(sampled));
//
// self.next_delay = next_delay;
//
// // TODO: fix it for non-wasm
// }
// pub(crate) async fn run_out_queue_control(&mut self) {
// debug!("Starting out queue controller...");
// self.run_normal_out_queue().await
// }
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
debug!("Started OutQueueControl with graceful shutdown support");
// we should set initial delay only when we actually start the stream
let sampled =
sample_poisson_duration(&mut self.rng, self.config.average_message_sending_delay);
self.next_delay = Box::pin(time::sleep(sampled));
#[cfg(not(target_arch = "wasm32"))]
let next_delay = Box::pin(time::sleep(sampled));
while !shutdown.is_shutdown() {
tokio::select! {
biased;
_ = shutdown.recv() => {
log::trace!("OutQueueControl: Received shutdown");
}
next_message = self.next() => match next_message {
Some(next_message) => {
self.on_message(next_message).await;
},
None => {
log::trace!("OutQueueControl: Stopping since channel closed");
break;
}
}
}
}
assert!(shutdown.is_shutdown_poll());
log::debug!("OutQueueControl: Exiting");
}
#[cfg(target_arch = "wasm32")]
let next_delay = Box::pin(wasm_timer::Delay::new(sampled));
#[cfg(target_arch = "wasm32")]
pub(super) async fn run(&mut self) {
debug!("Started OutQueueControl without graceful shutdown support");
self.next_delay = next_delay;
// we should set initial delay only when we actually start the stream
let sampled =
sample_poisson_duration(&mut self.rng, self.config.average_message_sending_delay);
self.next_delay = Box::pin(wasm_timer::Delay::new(sampled));
// TODO: fix it for non-wasm
while let Some(next_message) = self.next().await {
self.on_message(next_message).await;
}
// let mut shutdown = self.shutdown.clone();
// while !shutdown.is_shutdown() {
// tokio::select! {
// biased;
// _ = shutdown.recv() => {
// log::trace!("OutQueueControl: Received shutdown");
// }
// next_message = self.next() => match next_message {
// Some(next_message) => {
// self.on_message(next_message).await;
// },
// None => {
// log::trace!("OutQueueControl: Stopping since channel closed");
// break;
// }
// }
// }
// }
// assert!(shutdown.is_shutdown_poll());
// log::debug!("OutQueueControl: Exiting");
}
pub(crate) async fn run_out_queue_control(&mut self) {
debug!("Starting out queue controller...");
self.run_normal_out_queue().await
}
}
clients/client-core/src/client/topology_control.rs
+22 -17
View File
@@ -306,28 +306,33 @@ impl TopologyRefresher {
self.topology_accessor.is_routable().await
}
pub fn start(mut self, #[cfg(not(target_arch = "wasm32"))] mut shutdown: ShutdownListener) {
// TODO: usual non-wasm, etc, etc
#[cfg(not(target_arch = "wasm32"))]
pub async fn start_with_shutdown(mut self, mut shutdown: task::ShutdownListener) {
debug!("Started TopologyRefresher with graceful shutdown support");
while !shutdown.is_shutdown() {
tokio::select! {
_ = tokio::time::sleep(self.refresh_rate) => {
self.refresh().await;
},
_ = shutdown.recv() => {
log::trace!("TopologyRefresher: Received shutdown");
},
}
}
assert!(shutdown.is_shutdown_poll());
log::debug!("TopologyRefresher: Exiting");
}
#[cfg(target_arch = "wasm32")]
pub fn start(mut self) {
debug!("Started TopologyRefresher without graceful shutdown support");
spawn_future(async move {
loop {
// TODO: this won't work in wasm
tokio::time::sleep(self.refresh_rate).await;
wasm_timer::Delay::new(self.refresh_rate).await;
self.refresh().await;
}
// while !shutdown.is_shutdown() {
// tokio::select! {
// _ = tokio::time::sleep(self.refresh_rate) => {
// self.refresh().await;
// },
// _ = shutdown.recv() => {
// log::trace!("TopologyRefresher: Received shutdown");
// },
// }
// }
// assert!(shutdown.is_shutdown_poll());
// log::debug!("TopologyRefresher: Exiting");
})
}
}
clients/client-core/src/lib.rs
-2
View File
@@ -5,8 +5,6 @@ pub mod config;
pub mod error;
pub mod init;
// TODO: move those to separate lower modules and conditionally re-export them accordingly to make intellij happier about name clash
#[cfg(target_arch = "wasm32")]
pub(crate) fn spawn_future<F>(future: F)
where
clients/webassembly/js-example/index.js
+5
View File
@@ -61,6 +61,11 @@ async function main() {
// only really useful if you want to adjust some settings like traffic rate
// (if not needed you can just pass a null)
const debug = default_debug();
debug.loop_cover_traffic_average_delay_ms = BigInt(60_000);
// note: we still have poisson distribution so, on average, we will be sending SOME packet every 20ms
debug.message_sending_average_delay_ms = BigInt(20);
debug.average_packet_delay_ms = BigInt(10);
debug.average_ack_delay_ms = BigInt(10);
const config = new Config("my-awesome-wasm-client", validator, gatewayEndpoint, debug)
clients/webassembly/src/client/mod.rs
+3 -2
View File
@@ -242,6 +242,7 @@ impl NymClient {
}
console_log!("Starting topology refresher...");
// TODO: re-enable
// topology_refresher.start();
}
@@ -354,14 +355,14 @@ impl NymClient {
// Right now it's impossible to have async exported functions to take `&mut self` rather than mut self
// TODO: try Rc<RefCell<Self>> approach?
pub async fn send_message(mut self, message: String, recipient: String) -> Self {
pub async fn send_message(self, message: String, recipient: String) -> Self {
console_log!("Sending {} to {}", message, recipient);
let message_bytes = message.into_bytes();
self.send_binary_message(message_bytes, recipient).await
}
pub async fn send_binary_message(mut self, message: Vec<u8>, recipient: String) -> Self {
pub async fn send_binary_message(self, message: Vec<u8>, recipient: String) -> Self {
console_log!("Sending {} bytes to {}", message.len(), recipient);
let recipient = Recipient::try_from_base58_string(recipient).unwrap();
clients/webassembly/src/lib.rs
+1
View File
@@ -5,6 +5,7 @@ use wasm_bindgen::prelude::*;
#[cfg(target_arch = "wasm32")]
mod client;
#[cfg(target_arch = "wasm32")]
pub mod gateway_selector;
#[wasm_bindgen]