veloren_network/api.rs
1use crate::{
2 channel::ProtocolsError,
3 message::{Message, partial_eq_bincode},
4 participant::{A2bStreamOpen, S2bShutdownBparticipant},
5 scheduler::{A2sConnect, Scheduler},
6};
7use bytes::Bytes;
8use hashbrown::HashMap;
9#[cfg(feature = "compression")]
10use lz_fear::raw::DecodeError;
11use network_protocol::{Bandwidth, InitProtocolError, Pid, Prio, Promises, Sid};
12#[cfg(feature = "metrics")]
13use prometheus::Registry;
14use serde::{Serialize, de::DeserializeOwned};
15use std::{
16 net::SocketAddr,
17 sync::{
18 Arc,
19 atomic::{AtomicBool, Ordering},
20 },
21 time::Duration,
22};
23use tokio::{
24 io,
25 runtime::Runtime,
26 sync::{Mutex, mpsc, oneshot, watch},
27};
28use tracing::*;
29
30type A2sDisconnect = Arc<Mutex<Option<mpsc::UnboundedSender<(Pid, S2bShutdownBparticipant)>>>>;
31
32/// Represents a Tcp, Quic, Udp or Mpsc connection address
33#[derive(Clone, Debug)]
34pub enum ConnectAddr {
35 Tcp(SocketAddr),
36 Udp(SocketAddr),
37 #[cfg(feature = "quic")]
38 Quic(SocketAddr, quinn::ClientConfig, String),
39 Mpsc(u64),
40}
41
42impl ConnectAddr {
43 /// Returns the `Some` if the protocol is TCP or QUIC and `None` if the
44 /// protocol is a local channel (mpsc).
45 pub fn socket_addr(&self) -> Option<SocketAddr> {
46 match self {
47 Self::Tcp(addr) => Some(*addr),
48 Self::Udp(addr) => Some(*addr),
49 Self::Mpsc(_) => None,
50 #[cfg(feature = "quic")]
51 Self::Quic(addr, _, _) => Some(*addr),
52 }
53 }
54}
55
56/// Represents a Tcp, Quic, Udp or Mpsc listen address
57#[derive(Clone, Debug)]
58pub enum ListenAddr {
59 Tcp(SocketAddr),
60 Udp(SocketAddr),
61 #[cfg(feature = "quic")]
62 Quic(SocketAddr, quinn::ServerConfig),
63 Mpsc(u64),
64}
65
66/// A Participant can throw different events, you are obligated to carefully
67/// empty the queue from time to time.
68#[derive(Clone, Debug)]
69pub enum ParticipantEvent {
70 ChannelCreated(ConnectAddr),
71 ChannelDeleted(ConnectAddr),
72}
73
74/// `Participants` are generated by the [`Network`] and represent a connection
75/// to a remote Participant. Look at the [`connect`] and [`connected`] method of
76/// [`Networks`] on how to generate `Participants`
77///
78/// [`Networks`]: crate::api::Network
79/// [`connect`]: Network::connect
80/// [`connected`]: Network::connected
81pub struct Participant {
82 local_pid: Pid,
83 remote_pid: Pid,
84 a2b_open_stream_s: mpsc::UnboundedSender<A2bStreamOpen>,
85 b2a_stream_opened_r: mpsc::UnboundedReceiver<Stream>,
86 b2a_event_r: mpsc::UnboundedReceiver<ParticipantEvent>,
87 b2a_bandwidth_stats_r: watch::Receiver<f32>,
88 a2s_disconnect_s: A2sDisconnect,
89}
90
91/// `Streams` represents a channel to send `n` messages with a certain priority
92/// and [`Promises`]. messages need always to be send between 2 `Streams`.
93///
94/// `Streams` are generated by the [`Participant`].
95/// Look at the [`open`] and [`opened`] method of [`Participant`] on how to
96/// generate `Streams`
97///
98/// Unlike [`Network`] and [`Participant`], `Streams` don't implement interior
99/// mutability, as multiple threads don't need access to the same `Stream`.
100///
101/// [`Networks`]: crate::api::Network
102/// [`open`]: Participant::open
103/// [`opened`]: Participant::opened
104#[derive(Debug)]
105pub struct Stream {
106 local_pid: Pid,
107 remote_pid: Pid,
108 sid: Sid,
109 #[expect(dead_code)]
110 prio: Prio,
111 promises: Promises,
112 #[expect(dead_code)]
113 guaranteed_bandwidth: Bandwidth,
114 send_closed: Arc<AtomicBool>,
115 a2b_msg_s: crossbeam_channel::Sender<(Sid, Bytes)>,
116 b2a_msg_recv_r: Option<async_channel::Receiver<Bytes>>,
117 a2b_close_stream_s: Option<mpsc::UnboundedSender<Sid>>,
118 output_limit: usize,
119}
120
121/// Error type thrown by [`Networks`](Network) methods
122#[derive(Debug)]
123pub enum NetworkError {
124 NetworkClosed,
125 ListenFailed(io::Error),
126 ConnectFailed(NetworkConnectError),
127}
128
129/// Error type thrown by [`Networks`](Network) connect
130#[derive(Debug)]
131pub enum NetworkConnectError {
132 /// Either a Pid UUID clash or you are trying to hijack a connection
133 InvalidSecret,
134 Handshake(InitProtocolError<ProtocolsError>),
135 Io(io::Error),
136}
137
138/// Error type thrown by [`Participants`](Participant) methods
139#[derive(Debug, PartialEq, Eq, Clone)]
140pub enum ParticipantError {
141 ///Participant was closed by remote side
142 ParticipantDisconnected,
143 ///Underlying Protocol failed and wasn't able to recover, expect some Data
144 /// loss unfortunately, there is no method to get the exact messages
145 /// that failed. This is also returned when local side tries to do
146 /// something while remote site gracefully disconnects
147 ProtocolFailedUnrecoverable,
148}
149
150/// Error type thrown by [`Streams`](Stream) methods
151/// A Compression Error should only happen if a client sends malicious code.
152/// A Deserialize Error probably means you are expecting Type X while you
153/// actually got send type Y.
154#[derive(Debug)]
155pub enum StreamError {
156 StreamClosed,
157 #[cfg(feature = "compression")]
158 Compression(DecodeError),
159 Deserialize(Box<bincode::error::DecodeError>),
160}
161
162/// All Parameters of a Stream, can be used to generate RawMessages
163#[derive(Debug, Clone)]
164pub struct StreamParams {
165 pub(crate) promises: Promises,
166}
167
168/// Use the `Network` to create connections to other [`Participants`]
169///
170/// The `Network` is the single source that handles all connections in your
171/// Application. You can pass it around multiple threads in an
172/// [`Arc`](std::sync::Arc) as all commands have internal mutability.
173///
174/// The `Network` has methods to [`connect`] to other [`Participants`] actively
175/// via their [`ConnectAddr`], or [`listen`] passively for [`connected`]
176/// [`Participants`] via [`ListenAddr`].
177///
178/// Too guarantee a clean shutdown, the [`Runtime`] MUST NOT be dropped before
179/// the Network.
180///
181/// # Examples
182/// ```rust
183/// use tokio::runtime::Runtime;
184/// use veloren_network::{Network, ConnectAddr, ListenAddr, Pid};
185///
186/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
187/// // Create a Network, listen on port `2999` to accept connections and connect to port `8080` to connect to a (pseudo) database Application
188/// let runtime = Runtime::new().unwrap();
189/// let mut network = Network::new(Pid::new(), &runtime);
190/// runtime.block_on(async{
191/// # //setup pseudo database!
192/// # let database = Network::new(Pid::new(), &runtime);
193/// # database.listen(ListenAddr::Tcp("127.0.0.1:8080".parse().unwrap())).await?;
194/// network.listen(ListenAddr::Tcp("127.0.0.1:2999".parse().unwrap())).await?;
195/// let database = network.connect(ConnectAddr::Tcp("127.0.0.1:8080".parse().unwrap())).await?;
196/// drop(network);
197/// # drop(database);
198/// # Ok(())
199/// })
200/// # }
201/// ```
202///
203/// [`Participants`]: crate::api::Participant
204/// [`Runtime`]: tokio::runtime::Runtime
205/// [`connect`]: Network::connect
206/// [`listen`]: Network::listen
207/// [`connected`]: Network::connected
208/// [`ConnectAddr`]: crate::api::ConnectAddr
209/// [`ListenAddr`]: crate::api::ListenAddr
210pub struct Network {
211 local_pid: Pid,
212 participant_disconnect_sender: Arc<Mutex<HashMap<Pid, A2sDisconnect>>>,
213 listen_sender: mpsc::UnboundedSender<(ListenAddr, oneshot::Sender<io::Result<()>>)>,
214 connect_sender: mpsc::UnboundedSender<A2sConnect>,
215 connected_receiver: mpsc::UnboundedReceiver<Participant>,
216 shutdown_network_s: Option<oneshot::Sender<oneshot::Sender<()>>>,
217}
218
219impl Network {
220 /// Generates a new `Network` to handle all connections in an Application
221 ///
222 /// # Arguments
223 /// * `participant_id` - provide it by calling [`Pid::new()`], usually you
224 /// don't want to reuse a Pid for 2 `Networks`
225 /// * `runtime` - provide a [`Runtime`], it's used to internally spawn
226 /// tasks. It is necessary to clean up in the non-async `Drop`. **All**
227 /// network related components **must** be dropped before the runtime is
228 /// stopped. dropping the runtime while a shutdown is still in progress
229 /// leaves the network in a bad state which might cause a panic!
230 ///
231 /// # Result
232 /// * `Self` - returns a `Network` which can be `Send` to multiple areas of
233 /// your code, including multiple threads. This is the base strct of this
234 /// crate.
235 ///
236 /// # Examples
237 /// ```rust
238 /// use tokio::runtime::Runtime;
239 /// use veloren_network::{Network, Pid};
240 ///
241 /// let runtime = Runtime::new().unwrap();
242 /// let network = Network::new(Pid::new(), &runtime);
243 /// ```
244 ///
245 /// Usually you only create a single `Network` for an application,
246 /// except when client and server are in the same application, then you
247 /// will want 2. However there are no technical limitations from
248 /// creating more.
249 ///
250 /// [`Pid::new()`]: network_protocol::Pid::new
251 /// [`Runtime`]: tokio::runtime::Runtime
252 pub fn new(participant_id: Pid, runtime: &Runtime) -> Self {
253 Self::internal_new(
254 participant_id,
255 runtime,
256 #[cfg(feature = "metrics")]
257 None,
258 usize::MAX,
259 )
260 }
261
262 /// See [`new`]
263 ///
264 /// # additional Arguments
265 /// * `registry` - Provide a Registry in order to collect Prometheus metrics
266 /// by this `Network`, `None` will deactivate Tracing. Tracing is done via
267 /// [`prometheus`]
268 /// * `output_limit` - Specify an approximate upper bound on the size of
269 /// uncompressed network messages, to mitigate the potential for DoS
270 /// attacks.
271 ///
272 /// # Examples
273 /// ```rust
274 /// use prometheus::Registry;
275 /// use tokio::runtime::Runtime;
276 /// use veloren_network::{Network, Pid};
277 ///
278 /// let runtime = Runtime::new().unwrap();
279 /// let registry = Registry::new();
280 /// let network = Network::new_with_registry(Pid::new(), &runtime, ®istry, 1 << 20);
281 /// ```
282 /// [`new`]: crate::api::Network::new
283 #[cfg(feature = "metrics")]
284 pub fn new_with_registry(
285 participant_id: Pid,
286 runtime: &Runtime,
287 registry: &Registry,
288 output_limit: usize,
289 ) -> Self {
290 Self::internal_new(participant_id, runtime, Some(registry), output_limit)
291 }
292
293 fn internal_new(
294 participant_id: Pid,
295 runtime: &Runtime,
296 #[cfg(feature = "metrics")] registry: Option<&Registry>,
297 output_limit: usize,
298 ) -> Self {
299 let p = participant_id;
300 let span = info_span!("network", ?p);
301 span.in_scope(|| trace!("Starting Network"));
302 let (scheduler, listen_sender, connect_sender, connected_receiver, shutdown_sender) =
303 Scheduler::new(
304 participant_id,
305 #[cfg(feature = "metrics")]
306 registry,
307 output_limit,
308 );
309 let participant_disconnect_sender = Arc::new(Mutex::new(HashMap::new()));
310 let (shutdown_network_s, shutdown_network_r) = oneshot::channel();
311 let f = Self::shutdown_mgr(
312 p,
313 shutdown_network_r,
314 Arc::clone(&participant_disconnect_sender),
315 shutdown_sender,
316 );
317 runtime.spawn(f);
318 runtime.spawn(
319 async move {
320 trace!("Starting scheduler in own thread");
321 scheduler.run().await;
322 trace!("Stopping scheduler and his own thread");
323 }
324 .instrument(info_span!("network", ?p)),
325 );
326 Self {
327 local_pid: participant_id,
328 participant_disconnect_sender,
329 listen_sender,
330 connect_sender,
331 connected_receiver,
332 shutdown_network_s: Some(shutdown_network_s),
333 }
334 }
335
336 /// starts listening on an [`ListenAddr`].
337 /// When the method returns the `Network` is ready to listen for incoming
338 /// connections OR has returned a [`NetworkError`] (e.g. port already used).
339 /// You can call [`connected`] to asynchrony wait for a [`Participant`] to
340 /// connect. You can call `listen` on multiple addresses, e.g. to
341 /// support multiple Protocols or NICs.
342 ///
343 /// # Examples
344 /// ```ignore
345 /// use tokio::runtime::Runtime;
346 /// use veloren_network::{Network, Pid, ListenAddr};
347 ///
348 /// # fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
349 /// // Create a Network, listen on port `2000` TCP on all NICs and `2001` UDP locally
350 /// let runtime = Runtime::new().unwrap();
351 /// let mut network = Network::new(Pid::new(), &runtime);
352 /// runtime.block_on(async {
353 /// network
354 /// .listen(ListenAddr::Tcp("127.0.0.1:2000".parse().unwrap()))
355 /// .await?;
356 /// network
357 /// .listen(ListenAddr::Udp("127.0.0.1:2001".parse().unwrap()))
358 /// .await?;
359 /// drop(network);
360 /// # Ok(())
361 /// })
362 /// # }
363 /// ```
364 ///
365 /// [`connected`]: Network::connected
366 /// [`ListenAddr`]: crate::api::ListenAddr
367 #[instrument(name="network", skip(self, address), fields(p = %self.local_pid))]
368 pub async fn listen(&self, address: ListenAddr) -> Result<(), NetworkError> {
369 let (s2a_result_s, s2a_result_r) = oneshot::channel::<io::Result<()>>();
370 debug!(?address, "listening on address");
371 self.listen_sender.send((address, s2a_result_s))?;
372 match s2a_result_r.await? {
373 //waiting guarantees that we either listened successfully or get an error like port in
374 // use
375 Ok(()) => Ok(()),
376 Err(e) => Err(NetworkError::ListenFailed(e)),
377 }
378 }
379
380 /// starts connection to an [`ConnectAddr`].
381 /// When the method returns the Network either returns a [`Participant`]
382 /// ready to open [`Streams`] on OR has returned a [`NetworkError`] (e.g.
383 /// can't connect, or invalid Handshake) # Examples
384 /// ```ignore
385 /// use tokio::runtime::Runtime;
386 /// use veloren_network::{Network, Pid, ListenAddr, ConnectAddr};
387 ///
388 /// # fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
389 /// // Create a Network, connect on port `2010` TCP and `2011` UDP like listening above
390 /// let runtime = Runtime::new().unwrap();
391 /// let network = Network::new(Pid::new(), &runtime);
392 /// # let remote = Network::new(Pid::new(), &runtime);
393 /// runtime.block_on(async {
394 /// # remote.listen(ListenAddr::Tcp("127.0.0.1:2010".parse().unwrap())).await?;
395 /// # remote.listen(ListenAddr::Udp("127.0.0.1:2011".parse().unwrap())).await?;
396 /// let p1 = network
397 /// .connect(ConnectAddr::Tcp("127.0.0.1:2010".parse().unwrap()))
398 /// .await?;
399 /// # //this doesn't work yet, so skip the test
400 /// # //TODO fixme!
401 /// # return Ok(());
402 /// let p2 = network
403 /// .connect(ConnectAddr::Udp("127.0.0.1:2011".parse().unwrap()))
404 /// .await?;
405 /// assert_eq!(&p1, &p2);
406 /// # Ok(())
407 /// })?;
408 /// drop(network);
409 /// # drop(remote);
410 /// # Ok(())
411 /// # }
412 /// ```
413 /// Usually the `Network` guarantees that a operation on a [`Participant`]
414 /// succeeds, e.g. by automatic retrying unless it fails completely e.g. by
415 /// disconnecting from the remote. If 2 [`ConnectAddr] you
416 /// `connect` to belongs to the same [`Participant`], you get the same
417 /// [`Participant`] as a result. This is useful e.g. by connecting to
418 /// the same [`Participant`] via multiple Protocols.
419 ///
420 /// [`Streams`]: crate::api::Stream
421 /// [`ConnectAddr`]: crate::api::ConnectAddr
422 #[instrument(name="network", skip(self, address), fields(p = %self.local_pid))]
423 pub async fn connect(&self, address: ConnectAddr) -> Result<Participant, NetworkError> {
424 let (pid_sender, pid_receiver) =
425 oneshot::channel::<Result<Participant, NetworkConnectError>>();
426 debug!(?address, "Connect to address");
427 self.connect_sender.send((address, pid_sender))?;
428 let participant = match pid_receiver.await? {
429 Ok(p) => p,
430 Err(e) => return Err(NetworkError::ConnectFailed(e)),
431 };
432 let remote_pid = participant.remote_pid;
433 trace!(?remote_pid, "connected");
434 self.participant_disconnect_sender
435 .lock()
436 .await
437 .insert(remote_pid, Arc::clone(&participant.a2s_disconnect_s));
438 Ok(participant)
439 }
440
441 /// Returns a [`Participant`] created from a [`ListenAddr`] you
442 /// called [`listen`] on before. This function will either return a
443 /// working [`Participant`] ready to open [`Streams`] on OR has returned
444 /// a [`NetworkError`] (e.g. Network got closed)
445 ///
446 /// # Examples
447 /// ```rust
448 /// use tokio::runtime::Runtime;
449 /// use veloren_network::{ConnectAddr, ListenAddr, Network, Pid};
450 ///
451 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
452 /// // Create a Network, listen on port `2020` TCP and opens returns their Pid
453 /// let runtime = Runtime::new().unwrap();
454 /// let mut network = Network::new(Pid::new(), &runtime);
455 /// # let remote = Network::new(Pid::new(), &runtime);
456 /// runtime.block_on(async {
457 /// network
458 /// .listen(ListenAddr::Tcp("127.0.0.1:2020".parse().unwrap()))
459 /// .await?;
460 /// # remote.connect(ConnectAddr::Tcp("127.0.0.1:2020".parse().unwrap())).await?;
461 /// while let Ok(participant) = network.connected().await {
462 /// println!("Participant connected: {}", participant.remote_pid());
463 /// # //skip test here as it would be a endless loop
464 /// # break;
465 /// }
466 /// drop(network);
467 /// # drop(remote);
468 /// # Ok(())
469 /// })
470 /// # }
471 /// ```
472 ///
473 /// [`Streams`]: crate::api::Stream
474 /// [`listen`]: crate::api::Network::listen
475 /// [`ListenAddr`]: crate::api::ListenAddr
476 #[instrument(name="network", skip(self), fields(p = %self.local_pid))]
477 pub async fn connected(&mut self) -> Result<Participant, NetworkError> {
478 let participant = self
479 .connected_receiver
480 .recv()
481 .await
482 .ok_or(NetworkError::NetworkClosed)?;
483 self.participant_disconnect_sender.lock().await.insert(
484 participant.remote_pid,
485 Arc::clone(&participant.a2s_disconnect_s),
486 );
487 Ok(participant)
488 }
489
490 /// Use a mgr to handle shutdown smoothly and not in `Drop`
491 #[instrument(name="network", skip(participant_disconnect_sender, shutdown_scheduler_s), fields(p = %local_pid))]
492 async fn shutdown_mgr(
493 local_pid: Pid,
494 shutdown_network_r: oneshot::Receiver<oneshot::Sender<()>>,
495 participant_disconnect_sender: Arc<Mutex<HashMap<Pid, A2sDisconnect>>>,
496 shutdown_scheduler_s: oneshot::Sender<()>,
497 ) {
498 trace!("waiting for shutdown triggerNetwork");
499 let return_s = shutdown_network_r.await;
500 trace!("Shutting down Participants of Network");
501 let mut finished_receiver_list = vec![];
502
503 for (remote_pid, a2s_disconnect_s) in participant_disconnect_sender.lock().await.drain() {
504 match a2s_disconnect_s.lock().await.take() {
505 Some(a2s_disconnect_s) => {
506 trace!(?remote_pid, "Participants will be closed");
507 let (finished_sender, finished_receiver) = oneshot::channel();
508 finished_receiver_list.push((remote_pid, finished_receiver));
509 // If the channel was already dropped, we can assume that the other side
510 // already released its resources.
511 let _ = a2s_disconnect_s
512 .send((remote_pid, (Duration::from_secs(10), finished_sender)));
513 },
514 None => trace!(?remote_pid, "Participant already disconnected gracefully"),
515 }
516 }
517 //wait after close is requested for all
518 for (remote_pid, finished_receiver) in finished_receiver_list.drain(..) {
519 match finished_receiver.await {
520 Ok(Ok(())) => trace!(?remote_pid, "disconnect successful"),
521 Ok(Err(e)) => info!(?remote_pid, ?e, "unclean disconnect"),
522 Err(e) => warn!(
523 ?remote_pid,
524 ?e,
525 "Failed to get a message back from the scheduler, seems like the network is \
526 already closed"
527 ),
528 }
529 }
530
531 trace!("Participants have shut down - next: Scheduler");
532 if let Err(()) = shutdown_scheduler_s.send(()) {
533 error!("Scheduler is closed, but nobody other should be able to close it")
534 };
535 if let Ok(return_s) = return_s
536 && return_s.send(()).is_err()
537 {
538 warn!("Network::drop stopped after a timeout and didn't wait for our shutdown");
539 };
540 debug!("Network has shut down");
541 }
542}
543
544impl Participant {
545 pub(crate) fn new(
546 local_pid: Pid,
547 remote_pid: Pid,
548 a2b_open_stream_s: mpsc::UnboundedSender<A2bStreamOpen>,
549 b2a_stream_opened_r: mpsc::UnboundedReceiver<Stream>,
550 b2a_event_r: mpsc::UnboundedReceiver<ParticipantEvent>,
551 b2a_bandwidth_stats_r: watch::Receiver<f32>,
552 a2s_disconnect_s: mpsc::UnboundedSender<(Pid, S2bShutdownBparticipant)>,
553 ) -> Self {
554 Self {
555 local_pid,
556 remote_pid,
557 a2b_open_stream_s,
558 b2a_stream_opened_r,
559 b2a_event_r,
560 b2a_bandwidth_stats_r,
561 a2s_disconnect_s: Arc::new(Mutex::new(Some(a2s_disconnect_s))),
562 }
563 }
564
565 /// Opens a [`Stream`] on this `Participant` with a certain Priority and
566 /// [`Promises`]
567 ///
568 /// # Arguments
569 /// * `prio` - defines which stream is processed first when limited on
570 /// bandwidth. See [`Prio`] for documentation.
571 /// * `promises` - use a combination of you preferred [`Promises`], see the
572 /// link for further documentation. You can combine them, e.g.
573 /// `Promises::ORDERED | Promises::CONSISTENCY` The Stream will then
574 /// guarantee that those promises are met.
575 /// * `bandwidth` - sets a guaranteed bandwidth which is reserved for this
576 /// stream. When excess bandwidth is available it will be used. See
577 /// [`Bandwidth`] for details.
578 ///
579 /// A [`ParticipantError`] might be thrown if the `Participant` is already
580 /// closed. [`Streams`] can be created without a answer from the remote
581 /// side, resulting in very fast creation and closing latency.
582 ///
583 /// # Examples
584 /// ```rust
585 /// use tokio::runtime::Runtime;
586 /// use veloren_network::{ConnectAddr, ListenAddr, Network, Pid, Promises};
587 ///
588 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
589 /// // Create a Network, connect on port 2100 and open a stream
590 /// let runtime = Runtime::new().unwrap();
591 /// let network = Network::new(Pid::new(), &runtime);
592 /// # let remote = Network::new(Pid::new(), &runtime);
593 /// runtime.block_on(async {
594 /// # remote.listen(ListenAddr::Tcp("127.0.0.1:2100".parse().unwrap())).await?;
595 /// let p1 = network
596 /// .connect(ConnectAddr::Tcp("127.0.0.1:2100".parse().unwrap()))
597 /// .await?;
598 /// let _s1 = p1
599 /// .open(4, Promises::ORDERED | Promises::CONSISTENCY, 1000)
600 /// .await?;
601 /// drop(network);
602 /// # drop(remote);
603 /// # Ok(())
604 /// })
605 /// # }
606 /// ```
607 ///
608 /// [`Prio`]: network_protocol::Prio
609 /// [`Bandwidth`]: network_protocol::Bandwidth
610 /// [`Promises`]: network_protocol::Promises
611 /// [`Streams`]: crate::api::Stream
612 #[instrument(name="network", skip(self, prio, promises, bandwidth), fields(p = %self.local_pid))]
613 pub async fn open(
614 &self,
615 prio: u8,
616 promises: Promises,
617 bandwidth: Bandwidth,
618 ) -> Result<Stream, ParticipantError> {
619 debug_assert!(prio <= network_protocol::HIGHEST_PRIO, "invalid prio");
620 let (p2a_return_stream_s, p2a_return_stream_r) = oneshot::channel::<Stream>();
621 if let Err(e) =
622 self.a2b_open_stream_s
623 .send((prio, promises, bandwidth, p2a_return_stream_s))
624 {
625 debug!(?e, "bParticipant is already closed, notifying");
626 return Err(ParticipantError::ParticipantDisconnected);
627 }
628 match p2a_return_stream_r.await {
629 Ok(stream) => {
630 let sid = stream.sid;
631 trace!(?sid, "opened stream");
632 Ok(stream)
633 },
634 Err(_) => {
635 debug!("p2a_return_stream_r failed, closing participant");
636 Err(ParticipantError::ParticipantDisconnected)
637 },
638 }
639 }
640
641 /// Use this method to handle [`Streams`] opened from remote site, like the
642 /// [`connected`] method of [`Network`]. This is the associated method
643 /// to [`open`]. It's guaranteed that the order of [`open`] and `opened`
644 /// is equal. The `nth` [`Streams`] on one side will represent the `nth` on
645 /// the other side. A [`ParticipantError`] might be thrown if the
646 /// `Participant` is already closed.
647 ///
648 /// # Examples
649 /// ```rust
650 /// use tokio::runtime::Runtime;
651 /// use veloren_network::{Network, Pid, ListenAddr, ConnectAddr, Promises};
652 ///
653 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
654 /// // Create a Network, connect on port 2110 and wait for the other side to open a stream
655 /// // Note: It's quite unusual to actively connect, but then wait on a stream to be connected, usually the Application taking initiative want's to also create the first Stream.
656 /// let runtime = Runtime::new().unwrap();
657 /// let mut network = Network::new(Pid::new(), &runtime);
658 /// # let mut remote = Network::new(Pid::new(), &runtime);
659 /// runtime.block_on(async {
660 /// # remote.listen(ListenAddr::Tcp("127.0.0.1:2110".parse().unwrap())).await?;
661 /// let mut p1 = network.connect(ConnectAddr::Tcp("127.0.0.1:2110".parse().unwrap())).await?;
662 /// # let p2 = remote.connected().await?;
663 /// # p2.open(4, Promises::ORDERED | Promises::CONSISTENCY, 0).await?;
664 /// let _s1 = p1.opened().await?;
665 /// drop(network);
666 /// # drop(remote);
667 /// # Ok(())
668 /// })
669 /// # }
670 /// ```
671 ///
672 /// [`Streams`]: crate::api::Stream
673 /// [`connected`]: Network::connected
674 /// [`open`]: Participant::open
675 #[instrument(name="network", skip(self), fields(p = %self.local_pid))]
676 pub async fn opened(&mut self) -> Result<Stream, ParticipantError> {
677 match self.b2a_stream_opened_r.recv().await {
678 Some(stream) => {
679 let sid = stream.sid;
680 debug!(?sid, "Receive opened stream");
681 Ok(stream)
682 },
683 None => {
684 debug!("stream_opened_receiver failed, closing participant");
685 Err(ParticipantError::ParticipantDisconnected)
686 },
687 }
688 }
689
690 /// disconnecting a `Participant` in a async way.
691 /// Use this rather than `Participant::Drop` if you want to close multiple
692 /// `Participants`.
693 ///
694 /// This function will wait for all [`Streams`] to properly close, including
695 /// all messages to be send before closing. If an error occurs with one
696 /// of the messages.
697 /// Except if the remote side already dropped the `Participant`
698 /// simultaneously, then messages won't be send
699 ///
700 /// There is NO `disconnected` function in `Participant`, if a `Participant`
701 /// is no longer reachable (e.g. as the network cable was unplugged) the
702 /// `Participant` will fail all action, but needs to be manually
703 /// disconnected, using this function.
704 ///
705 /// # Examples
706 /// ```rust
707 /// use tokio::runtime::Runtime;
708 /// use veloren_network::{Network, Pid, ListenAddr, ConnectAddr};
709 ///
710 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
711 /// // Create a Network, listen on port `2030` TCP and opens returns their Pid and close connection.
712 /// let runtime = Runtime::new().unwrap();
713 /// let mut network = Network::new(Pid::new(), &runtime);
714 /// # let mut remote = Network::new(Pid::new(), &runtime);
715 /// let err = runtime.block_on(async {
716 /// network
717 /// .listen(ListenAddr::Tcp("127.0.0.1:2030".parse().unwrap()))
718 /// .await?;
719 /// # let keep_alive = remote.connect(ConnectAddr::Tcp("127.0.0.1:2030".parse().unwrap())).await?;
720 /// while let Ok(participant) = network.connected().await {
721 /// println!("Participant connected: {}", participant.remote_pid());
722 /// participant.disconnect().await?;
723 /// # //skip test here as it would be a endless loop
724 /// # break;
725 /// }
726 /// # Ok(())
727 /// });
728 /// drop(network);
729 /// # drop(remote);
730 /// # err
731 /// # }
732 /// ```
733 ///
734 /// [`Streams`]: crate::api::Stream
735 #[instrument(name="network", skip(self), fields(p = %self.local_pid))]
736 pub async fn disconnect(self) -> Result<(), ParticipantError> {
737 // Remove, Close and try_unwrap error when unwrap fails!
738 debug!("Closing participant from network");
739
740 //Streams will be closed by BParticipant
741 match self.a2s_disconnect_s.lock().await.take() {
742 Some(a2s_disconnect_s) => {
743 let (finished_sender, finished_receiver) = oneshot::channel();
744 // Participant is connecting to Scheduler here, not as usual
745 // Participant<->BParticipant
746
747 // If this is already dropped, we can assume the other side already freed its
748 // resources.
749 let _ = a2s_disconnect_s
750 .send((self.remote_pid, (Duration::from_secs(120), finished_sender)));
751 match finished_receiver.await {
752 Ok(res) => {
753 match res {
754 Ok(()) => trace!("Participant is now closed"),
755 Err(ref e) => {
756 trace!(?e, "Error occurred during shutdown of participant")
757 },
758 };
759 res
760 },
761 Err(e) => {
762 //this is a bug. but as i am Participant i can't destroy the network
763 error!(
764 ?e,
765 "Failed to get a message back from the scheduler, seems like the \
766 network is already closed"
767 );
768 Err(ParticipantError::ProtocolFailedUnrecoverable)
769 },
770 }
771 },
772 None => {
773 warn!(
774 "seems like you are trying to disconnecting a participant after the network \
775 was already dropped. It was already dropped with the network!"
776 );
777 Err(ParticipantError::ParticipantDisconnected)
778 },
779 }
780 }
781
782 /// Use this method to query [`ParticipantEvent`]. Those are internal events
783 /// from the network crate that will get reported to the frontend.
784 /// E.g. Creation and Deletion of Channels.
785 ///
786 /// Make sure to call this function from time to time to not let events
787 /// stack up endlessly and create a memory leak.
788 ///
789 /// # Examples
790 /// ```rust
791 /// use tokio::runtime::Runtime;
792 /// use veloren_network::{Network, Pid, ListenAddr, ConnectAddr, Promises, ParticipantEvent};
793 ///
794 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
795 /// // Create a Network, connect on port 2040 and wait for the other side to open a stream
796 /// // Note: It's quite unusual to actively connect, but then wait on a stream to be connected, usually the Application taking initiative want's to also create the first Stream.
797 /// let runtime = Runtime::new().unwrap();
798 /// let mut network = Network::new(Pid::new(), &runtime);
799 /// # let mut remote = Network::new(Pid::new(), &runtime);
800 /// runtime.block_on(async {
801 /// # remote.listen(ListenAddr::Tcp("127.0.0.1:2040".parse().unwrap())).await?;
802 /// let mut p1 = network.connect(ConnectAddr::Tcp("127.0.0.1:2040".parse().unwrap())).await?;
803 /// # let p2 = remote.connected().await?;
804 /// let event = p1.fetch_event().await?;
805 /// drop(network);
806 /// # drop(remote);
807 /// # Ok(())
808 /// })
809 /// # }
810 /// ```
811 ///
812 /// [`ParticipantEvent`]: crate::api::ParticipantEvent
813 pub async fn fetch_event(&mut self) -> Result<ParticipantEvent, ParticipantError> {
814 match self.b2a_event_r.recv().await {
815 Some(event) => Ok(event),
816 None => {
817 debug!("event_receiver failed, closing participant");
818 Err(ParticipantError::ParticipantDisconnected)
819 },
820 }
821 }
822
823 /// use `try_fetch_event` to check for a [`ParticipantEvent`] . This
824 /// function does not block and returns immediately. It's intended for
825 /// use in non-async context only. Other then that, the same rules apply
826 /// than for [`fetch_event`].
827 ///
828 /// [`ParticipantEvent`]: crate::api::ParticipantEvent
829 /// [`fetch_event`]: Participant::fetch_event
830 pub fn try_fetch_event(&mut self) -> Result<Option<ParticipantEvent>, ParticipantError> {
831 match self.b2a_event_r.try_recv() {
832 Ok(event) => Ok(Some(event)),
833 Err(mpsc::error::TryRecvError::Empty) => Ok(None),
834 Err(mpsc::error::TryRecvError::Disconnected) => {
835 Err(ParticipantError::ParticipantDisconnected)
836 },
837 }
838 }
839
840 /// Returns the current approximation on the maximum bandwidth available.
841 /// This WILL fluctuate based on the amount/size of send messages.
842 pub fn bandwidth(&self) -> f32 { *self.b2a_bandwidth_stats_r.borrow() }
843
844 /// Returns the remote [`Pid`](network_protocol::Pid)
845 pub fn remote_pid(&self) -> Pid { self.remote_pid }
846}
847
848impl Stream {
849 pub(crate) fn new(
850 local_pid: Pid,
851 remote_pid: Pid,
852 sid: Sid,
853 prio: Prio,
854 promises: Promises,
855 guaranteed_bandwidth: Bandwidth,
856 send_closed: Arc<AtomicBool>,
857 a2b_msg_s: crossbeam_channel::Sender<(Sid, Bytes)>,
858 b2a_msg_recv_r: async_channel::Receiver<Bytes>,
859 a2b_close_stream_s: mpsc::UnboundedSender<Sid>,
860 output_limit: usize,
861 ) -> Self {
862 Self {
863 local_pid,
864 remote_pid,
865 sid,
866 prio,
867 promises,
868 guaranteed_bandwidth,
869 send_closed,
870 a2b_msg_s,
871 b2a_msg_recv_r: Some(b2a_msg_recv_r),
872 a2b_close_stream_s: Some(a2b_close_stream_s),
873 output_limit,
874 }
875 }
876
877 /// use to send a arbitrary message to the remote side, by having the remote
878 /// side also opened a `Stream` linked to this. the message will be
879 /// [`Serialized`], which actually is quite slow compared to most other
880 /// calculations done. A faster method [`send_raw`] exists, when extra
881 /// speed is needed. The other side needs to use the respective [`recv`]
882 /// function and know the type send.
883 ///
884 /// `send` is an exception to the `async` messages, as it's probably called
885 /// quite often so it doesn't wait for execution. Which also means, that
886 /// no feedback is provided. It's to assume that the Message got `send`
887 /// correctly. If a error occurred, the next call will return an Error.
888 /// If the [`Participant`] disconnected it will also be unable to be used
889 /// any more. A [`StreamError`] will be returned in the error case, e.g.
890 /// when the `Stream` got closed already.
891 ///
892 /// Note when a `Stream` is dropped locally, it will still send all
893 /// messages, though the `drop` will return immediately, however, when a
894 /// [`Participant`] gets gracefully shut down, all remaining messages
895 /// will be send. If the `Stream` is dropped from remote side no further
896 /// messages are send, because the remote side has no way of listening
897 /// to them either way. If the last channel is destroyed (e.g. by losing
898 /// the internet connection or non-graceful shutdown, pending messages
899 /// are also dropped.
900 ///
901 /// # Example
902 /// ```
903 /// # use veloren_network::Promises;
904 /// use tokio::runtime::Runtime;
905 /// use veloren_network::{Network, ListenAddr, ConnectAddr, Pid};
906 ///
907 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
908 /// // Create a Network, listen on Port `2200` and wait for a Stream to be opened, then answer `Hello World`
909 /// let runtime = Runtime::new().unwrap();
910 /// let mut network = Network::new(Pid::new(), &runtime);
911 /// # let remote = Network::new(Pid::new(), &runtime);
912 /// runtime.block_on(async {
913 /// network.listen(ListenAddr::Tcp("127.0.0.1:2200".parse().unwrap())).await?;
914 /// # let remote_p = remote.connect(ConnectAddr::Tcp("127.0.0.1:2200".parse().unwrap())).await?;
915 /// # // keep it alive
916 /// # let _stream_p = remote_p.open(4, Promises::ORDERED | Promises::CONSISTENCY, 0).await?;
917 /// let mut participant_a = network.connected().await?;
918 /// let mut stream_a = participant_a.opened().await?;
919 /// //Send Message
920 /// stream_a.send("Hello World")?;
921 /// drop(network);
922 /// # drop(remote);
923 /// # Ok(())
924 /// })
925 /// # }
926 /// ```
927 ///
928 /// [`send_raw`]: Stream::send_raw
929 /// [`recv`]: Stream::recv
930 /// [`Serialized`]: Serialize
931 #[inline]
932 pub fn send<M: Serialize>(&self, msg: M) -> Result<(), StreamError> {
933 self.send_raw_move(Message::serialize(&msg, self.params()))
934 }
935
936 /// This methods give the option to skip multiple calls of [`bincode`] and
937 /// [`compress`], e.g. in case the same Message needs to send on
938 /// multiple `Streams` to multiple [`Participants`]. Other then that,
939 /// the same rules apply than for [`send`].
940 /// You need to create a Message via [`Message::serialize`].
941 ///
942 /// # Example
943 /// ```rust
944 /// # use veloren_network::Promises;
945 /// use tokio::runtime::Runtime;
946 /// use bincode;
947 /// use veloren_network::{Network, ListenAddr, ConnectAddr, Pid, Message};
948 ///
949 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
950 /// let runtime = Runtime::new().unwrap();
951 /// let mut network = Network::new(Pid::new(), &runtime);
952 /// # let remote1 = Network::new(Pid::new(), &runtime);
953 /// # let remote2 = Network::new(Pid::new(), &runtime);
954 /// runtime.block_on(async {
955 /// network.listen(ListenAddr::Tcp("127.0.0.1:2210".parse().unwrap())).await?;
956 /// # let remote1_p = remote1.connect(ConnectAddr::Tcp("127.0.0.1:2210".parse().unwrap())).await?;
957 /// # let remote2_p = remote2.connect(ConnectAddr::Tcp("127.0.0.1:2210".parse().unwrap())).await?;
958 /// # assert_eq!(remote1_p.remote_pid(), remote2_p.remote_pid());
959 /// # remote1_p.open(4, Promises::ORDERED | Promises::CONSISTENCY, 0).await?;
960 /// # remote2_p.open(4, Promises::ORDERED | Promises::CONSISTENCY, 0).await?;
961 /// let mut participant_a = network.connected().await?;
962 /// let mut participant_b = network.connected().await?;
963 /// let mut stream_a = participant_a.opened().await?;
964 /// let mut stream_b = participant_b.opened().await?;
965 ///
966 /// //Prepare Message and decode it
967 /// let msg = Message::serialize("Hello World", stream_a.params());
968 /// //Send same Message to multiple Streams
969 /// stream_a.send_raw(&msg);
970 /// stream_b.send_raw(&msg);
971 /// drop(network);
972 /// # drop(remote1);
973 /// # drop(remote2);
974 /// # Ok(())
975 /// })
976 /// # }
977 /// ```
978 ///
979 /// [`send`]: Stream::send
980 /// [`Participants`]: crate::api::Participant
981 /// [`compress`]: lz_fear::raw::compress2
982 /// [`Message::serialize`]: crate::message::Message::serialize
983 #[inline]
984 pub fn send_raw(&self, message: &Message) -> Result<(), StreamError> {
985 self.send_raw_move(Message {
986 data: message.data.clone(),
987 #[cfg(feature = "compression")]
988 compressed: message.compressed,
989 })
990 }
991
992 fn send_raw_move(&self, message: Message) -> Result<(), StreamError> {
993 if self.send_closed.load(Ordering::Relaxed) {
994 return Err(StreamError::StreamClosed);
995 }
996 #[cfg(debug_assertions)]
997 message.verify(self.params());
998 self.a2b_msg_s.send((self.sid, message.data))?;
999 Ok(())
1000 }
1001
1002 /// use `recv` to wait on a Message send from the remote side by their
1003 /// `Stream`. The Message needs to implement [`DeserializeOwned`] and
1004 /// thus, the resulting type must already be known by the receiving side.
1005 /// If this is not know from the Application logic, one could use a `Enum`
1006 /// and then handle the received message via a `match` state.
1007 ///
1008 /// A [`StreamError`] will be returned in the error case, e.g. when the
1009 /// `Stream` got closed already.
1010 ///
1011 /// # Example
1012 /// ```
1013 /// # use veloren_network::Promises;
1014 /// use tokio::runtime::Runtime;
1015 /// use veloren_network::{Network, ListenAddr, ConnectAddr, Pid};
1016 ///
1017 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1018 /// // Create a Network, listen on Port `2220` and wait for a Stream to be opened, then listen on it
1019 /// let runtime = Runtime::new().unwrap();
1020 /// let mut network = Network::new(Pid::new(), &runtime);
1021 /// # let remote = Network::new(Pid::new(), &runtime);
1022 /// runtime.block_on(async {
1023 /// network.listen(ListenAddr::Tcp("127.0.0.1:2220".parse().unwrap())).await?;
1024 /// # let remote_p = remote.connect(ConnectAddr::Tcp("127.0.0.1:2220".parse().unwrap())).await?;
1025 /// # let mut stream_p = remote_p.open(4, Promises::ORDERED | Promises::CONSISTENCY, 0).await?;
1026 /// # stream_p.send("Hello World");
1027 /// let mut participant_a = network.connected().await?;
1028 /// let mut stream_a = participant_a.opened().await?;
1029 /// //Recv Message
1030 /// println!("{}", stream_a.recv::<String>().await?);
1031 /// drop(network);
1032 /// # drop(remote);
1033 /// # Ok(())
1034 /// })
1035 /// # }
1036 /// ```
1037 #[inline]
1038 pub async fn recv<M: DeserializeOwned>(&mut self) -> Result<M, StreamError> {
1039 self.recv_raw().await?.deserialize(self.output_limit)
1040 }
1041
1042 /// the equivalent like [`send_raw`] but for [`recv`], no [`bincode`] or
1043 /// [`decompress`] is executed for performance reasons.
1044 ///
1045 /// # Example
1046 /// ```
1047 /// # use veloren_network::Promises;
1048 /// use tokio::runtime::Runtime;
1049 /// use veloren_network::{Network, ListenAddr, ConnectAddr, Pid};
1050 ///
1051 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1052 /// // Create a Network, listen on Port `2230` and wait for a Stream to be opened, then listen on it
1053 /// let runtime = Runtime::new().unwrap();
1054 /// let mut network = Network::new(Pid::new(), &runtime);
1055 /// # let remote = Network::new(Pid::new(), &runtime);
1056 /// runtime.block_on(async {
1057 /// network.listen(ListenAddr::Tcp("127.0.0.1:2230".parse().unwrap())).await?;
1058 /// # let remote_p = remote.connect(ConnectAddr::Tcp("127.0.0.1:2230".parse().unwrap())).await?;
1059 /// # let mut stream_p = remote_p.open(4, Promises::ORDERED | Promises::CONSISTENCY, 0).await?;
1060 /// # stream_p.send("Hello World");
1061 /// let mut participant_a = network.connected().await?;
1062 /// let mut stream_a = participant_a.opened().await?;
1063 /// //Recv Message
1064 /// let msg = stream_a.recv_raw().await?;
1065 /// //Resend Message, without deserializing
1066 /// stream_a.send_raw(&msg)?;
1067 /// drop(network);
1068 /// # drop(remote);
1069 /// # Ok(())
1070 /// })
1071 /// # }
1072 /// ```
1073 ///
1074 /// [`send_raw`]: Stream::send_raw
1075 /// [`recv`]: Stream::recv
1076 /// [`decompress`]: lz_fear::raw::decompress_raw
1077 pub async fn recv_raw(&mut self) -> Result<Message, StreamError> {
1078 match &mut self.b2a_msg_recv_r {
1079 Some(b2a_msg_recv_r) => {
1080 match b2a_msg_recv_r.recv().await {
1081 Ok(data) => Ok(Message {
1082 data,
1083 #[cfg(feature = "compression")]
1084 compressed: self.promises.contains(Promises::COMPRESSED),
1085 }),
1086 Err(_) => {
1087 self.b2a_msg_recv_r = None; //prevent panic
1088 Err(StreamError::StreamClosed)
1089 },
1090 }
1091 },
1092 None => Err(StreamError::StreamClosed),
1093 }
1094 }
1095
1096 /// use `try_recv` to check for a Message send from the remote side by their
1097 /// `Stream`. This function does not block and returns immediately. It's
1098 /// intended for use in non-async context only. Other then that, the
1099 /// same rules apply than for [`recv`].
1100 ///
1101 /// # Example
1102 /// ```
1103 /// # use veloren_network::Promises;
1104 /// use tokio::runtime::Runtime;
1105 /// use veloren_network::{Network, ListenAddr, ConnectAddr, Pid};
1106 ///
1107 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1108 /// // Create a Network, listen on Port `2240` and wait for a Stream to be opened, then listen on it
1109 /// let runtime = Runtime::new().unwrap();
1110 /// let mut network = Network::new(Pid::new(), &runtime);
1111 /// # let remote = Network::new(Pid::new(), &runtime);
1112 /// runtime.block_on(async {
1113 /// network.listen(ListenAddr::Tcp("127.0.0.1:2240".parse().unwrap())).await?;
1114 /// # let remote_p = remote.connect(ConnectAddr::Tcp("127.0.0.1:2240".parse().unwrap())).await?;
1115 /// # let mut stream_p = remote_p.open(4, Promises::ORDERED | Promises::CONSISTENCY, 0).await?;
1116 /// # stream_p.send("Hello World");
1117 /// # std::thread::sleep(std::time::Duration::from_secs(1));
1118 /// let mut participant_a = network.connected().await?;
1119 /// let mut stream_a = participant_a.opened().await?;
1120 /// //Try Recv Message
1121 /// println!("{:?}", stream_a.try_recv::<String>()?);
1122 /// drop(network);
1123 /// # drop(remote);
1124 /// # Ok(())
1125 /// })
1126 /// # }
1127 /// ```
1128 ///
1129 /// [`recv`]: Stream::recv
1130 #[inline]
1131 pub fn try_recv<M: DeserializeOwned>(&mut self) -> Result<Option<M>, StreamError> {
1132 match &mut self.b2a_msg_recv_r {
1133 Some(b2a_msg_recv_r) => match b2a_msg_recv_r.try_recv() {
1134 Ok(data) => Ok(Some(
1135 Message {
1136 data,
1137 #[cfg(feature = "compression")]
1138 compressed: self.promises.contains(Promises::COMPRESSED),
1139 }
1140 .deserialize(self.output_limit)?,
1141 )),
1142 Err(async_channel::TryRecvError::Empty) => Ok(None),
1143 Err(async_channel::TryRecvError::Closed) => {
1144 self.b2a_msg_recv_r = None; //prevent panic
1145 Err(StreamError::StreamClosed)
1146 },
1147 },
1148 None => Err(StreamError::StreamClosed),
1149 }
1150 }
1151
1152 pub fn params(&self) -> StreamParams {
1153 StreamParams {
1154 promises: self.promises,
1155 }
1156 }
1157}
1158
1159impl PartialEq for Participant {
1160 fn eq(&self, other: &Self) -> bool {
1161 //don't check local_pid, 2 Participant from different network should match if
1162 // they are the "same"
1163 self.remote_pid == other.remote_pid
1164 }
1165}
1166
1167fn actively_wait<T, F>(name: &'static str, mut finished_receiver: oneshot::Receiver<T>, f: F)
1168where
1169 F: FnOnce(T) + Send + 'static,
1170 T: Send + 'static,
1171{
1172 const CHANNEL_ERR: &str = "Something is wrong in internal scheduler/participant coding";
1173
1174 if let Ok(handle) = tokio::runtime::Handle::try_current() {
1175 // When in Async Context WE MUST NOT SYNC BLOCK (as a deadlock might occur as
1176 // other is queued behind). And we CANNOT join our Future_Handle
1177 trace!("async context detected, defer shutdown");
1178 handle.spawn(async move {
1179 match finished_receiver.await {
1180 Ok(data) => f(data),
1181 Err(e) => error!("{}{}: {}", name, CHANNEL_ERR, e),
1182 }
1183 });
1184 } else {
1185 let mut cnt = 0;
1186 loop {
1187 use tokio::sync::oneshot::error::TryRecvError;
1188 match finished_receiver.try_recv() {
1189 Ok(data) => {
1190 f(data);
1191 break;
1192 },
1193 Err(TryRecvError::Closed) => panic!("{}{}", name, CHANNEL_ERR),
1194 Err(TryRecvError::Empty) => {
1195 trace!("actively sleeping");
1196 cnt += 1;
1197 if cnt > 10 {
1198 error!("Timeout waiting for shutdown, dropping");
1199 break;
1200 }
1201 std::thread::sleep(Duration::from_millis(100) * cnt);
1202 },
1203 }
1204 }
1205 };
1206}
1207
1208impl Drop for Network {
1209 #[instrument(name="network", skip(self), fields(p = %self.local_pid))]
1210 fn drop(&mut self) {
1211 trace!("Dropping Network");
1212 let (finished_sender, finished_receiver) = oneshot::channel();
1213 match self
1214 .shutdown_network_s
1215 .take()
1216 .unwrap()
1217 .send(finished_sender)
1218 {
1219 Err(e) => warn!(?e, "Runtime seems to be dropped already"),
1220 Ok(()) => actively_wait("network", finished_receiver, |()| {
1221 info!("Network dropped gracefully")
1222 }),
1223 };
1224 }
1225}
1226
1227impl Drop for Participant {
1228 #[instrument(name="remote", skip(self), fields(p = %self.remote_pid))]
1229 #[instrument(name="network", skip(self), fields(p = %self.local_pid))]
1230 fn drop(&mut self) {
1231 const SHUTDOWN_ERR: &str = "Error while dropping the participant, couldn't send all \
1232 outgoing messages, dropping remaining";
1233 const SCHEDULER_ERR: &str =
1234 "Something is wrong in internal scheduler coding or you dropped the runtime to early";
1235 // ignore closed, as we need to send it even though we disconnected the
1236 // participant from network
1237 debug!("Shutting down Participant");
1238
1239 match self.a2s_disconnect_s.try_lock() {
1240 Err(e) => debug!(?e, "Participant is being dropped by Network right now"),
1241 Ok(mut s) => match s.take() {
1242 None => info!("Participant already has been shutdown gracefully"),
1243 Some(a2s_disconnect_s) => {
1244 debug!("Disconnect from Scheduler");
1245 let (finished_sender, finished_receiver) = oneshot::channel();
1246 match a2s_disconnect_s
1247 .send((self.remote_pid, (Duration::from_secs(10), finished_sender)))
1248 {
1249 Err(e) => warn!(?e, SCHEDULER_ERR),
1250 Ok(()) => {
1251 actively_wait("participant", finished_receiver, |d| match d {
1252 Ok(()) => info!("Participant dropped gracefully"),
1253 Err(e) => error!(?e, SHUTDOWN_ERR),
1254 });
1255 },
1256 }
1257 },
1258 },
1259 }
1260 }
1261}
1262
1263impl Drop for Stream {
1264 #[instrument(name="remote", skip(self), fields(p = %self.remote_pid))]
1265 #[instrument(name="network", skip(self), fields(p = %self.local_pid))]
1266
1267 fn drop(&mut self) {
1268 // send if closed is unnecessary but doesn't hurt, we must not crash
1269 let sid = self.sid;
1270 if !self.send_closed.load(Ordering::Relaxed) {
1271 debug!(?sid, "Shutting down Stream");
1272 if let Err(e) = self.a2b_close_stream_s.take().unwrap().send(self.sid) {
1273 debug!(
1274 ?e,
1275 "bparticipant part of a gracefully shutdown was already closed"
1276 );
1277 }
1278 } else {
1279 trace!(?sid, "Stream Drop not needed");
1280 }
1281 }
1282}
1283
1284impl std::fmt::Debug for Participant {
1285 #[inline]
1286 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1287 write!(
1288 f,
1289 "Participant {{ local_pid: {:?}, remote_pid: {:?} }}",
1290 self.local_pid, self.remote_pid,
1291 )
1292 }
1293}
1294
1295impl<T> From<crossbeam_channel::SendError<T>> for StreamError {
1296 fn from(_err: crossbeam_channel::SendError<T>) -> Self { StreamError::StreamClosed }
1297}
1298
1299impl<T> From<crossbeam_channel::SendError<T>> for NetworkError {
1300 fn from(_err: crossbeam_channel::SendError<T>) -> Self { NetworkError::NetworkClosed }
1301}
1302
1303impl<T> From<mpsc::error::SendError<T>> for NetworkError {
1304 fn from(_err: mpsc::error::SendError<T>) -> Self { NetworkError::NetworkClosed }
1305}
1306
1307impl From<oneshot::error::RecvError> for NetworkError {
1308 fn from(_err: oneshot::error::RecvError) -> Self { NetworkError::NetworkClosed }
1309}
1310
1311impl From<io::Error> for NetworkError {
1312 fn from(_err: io::Error) -> Self { NetworkError::NetworkClosed }
1313}
1314
1315impl From<Box<bincode::error::DecodeError>> for StreamError {
1316 fn from(err: Box<bincode::error::DecodeError>) -> Self { StreamError::Deserialize(err) }
1317}
1318
1319impl core::fmt::Display for StreamError {
1320 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1321 match self {
1322 StreamError::StreamClosed => write!(f, "stream closed"),
1323 #[cfg(feature = "compression")]
1324 StreamError::Compression(err) => write!(f, "compression error on message: {}", err),
1325 StreamError::Deserialize(err) => write!(f, "deserialize error on message: {}", err),
1326 }
1327 }
1328}
1329
1330impl core::fmt::Display for ParticipantError {
1331 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1332 match self {
1333 ParticipantError::ParticipantDisconnected => write!(f, "Participant disconnect"),
1334 ParticipantError::ProtocolFailedUnrecoverable => {
1335 write!(f, "underlying protocol failed unrecoverable")
1336 },
1337 }
1338 }
1339}
1340
1341impl core::fmt::Display for NetworkError {
1342 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1343 match self {
1344 NetworkError::NetworkClosed => write!(f, "Network closed"),
1345 NetworkError::ListenFailed(_) => write!(f, "Listening failed"),
1346 NetworkError::ConnectFailed(_) => write!(f, "Connecting failed"),
1347 }
1348 }
1349}
1350
1351impl core::fmt::Display for NetworkConnectError {
1352 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1353 match self {
1354 NetworkConnectError::Io(e) => write!(f, "Io error: {}", e),
1355 NetworkConnectError::Handshake(e) => write!(f, "Handshake error: {}", e),
1356 NetworkConnectError::InvalidSecret => {
1357 write!(f, "You specified the wrong secret on your second channel")
1358 },
1359 }
1360 }
1361}
1362
1363/// implementing PartialEq as it's super convenient in tests
1364impl PartialEq for StreamError {
1365 fn eq(&self, other: &Self) -> bool {
1366 match self {
1367 StreamError::StreamClosed => match other {
1368 StreamError::StreamClosed => true,
1369 #[cfg(feature = "compression")]
1370 StreamError::Compression(_) => false,
1371 StreamError::Deserialize(_) => false,
1372 },
1373 #[cfg(feature = "compression")]
1374 StreamError::Compression(err) => match other {
1375 StreamError::StreamClosed => false,
1376 #[cfg(feature = "compression")]
1377 StreamError::Compression(other_err) => err == other_err,
1378 StreamError::Deserialize(_) => false,
1379 },
1380 StreamError::Deserialize(err) => match other {
1381 StreamError::StreamClosed => false,
1382 #[cfg(feature = "compression")]
1383 StreamError::Compression(_) => false,
1384 StreamError::Deserialize(other_err) => partial_eq_bincode(err, other_err),
1385 },
1386 }
1387 }
1388}
1389
1390impl std::error::Error for StreamError {}
1391impl std::error::Error for ParticipantError {}
1392impl std::error::Error for NetworkError {}
1393impl std::error::Error for NetworkConnectError {}