feat: remove task spawn in session-sender

controller already creates a task for session-sender's recv
functionality

removed extra spawn inside session-sender's recv function

Author: thatdevsherry

examples/controller/controller.rs

@@ -57,7 +57,7 @@ impl Controller {
         let (start_session_tx, start_session_rx) = oneshot::channel::<()>();
         let (twamp_test_complete_tx, twamp_test_complete_rx) = oneshot::channel::<()>();
         let (reflector_port_tx, reflector_port_rx) = oneshot::channel::<u16>();
-        let control_client_handle = spawn(async move {
+        let task_control_client = spawn(async move {
             self.control_client
                 .do_twamp_control(
                     twamp_control,
@@ -71,10 +71,9 @@ impl Controller {
                 .await
                 .unwrap();
         });
-        let reflected_pkts_vec: Arc<Mutex<Vec<(TwampTestPacketUnauthReflected, TimeStamp)>>> =
-            Arc::new(Mutex::new(Vec::new()));
+        let reflected_pkts_vec = Arc::new(Mutex::new(Vec::new()));
         let reflected_pkts_vec_cloned = Arc::clone(&reflected_pkts_vec);
-        let session_sender_handle = spawn(async move {
+        let task_session_sender = spawn(async move {
             // Wait until we get the Accept-Session's port.
             let final_port = reflector_port_rx.await.unwrap();
             debug!("Received reflector port: {}", final_port);
@@ -94,31 +93,31 @@ impl Controller {
             ));
             let session_sender_send = Arc::clone(self.session_sender.as_ref().unwrap());
             let session_sender_recv = Arc::clone(self.session_sender.as_ref().unwrap());
-            let send_task = spawn(async move {
+            let task_session_sender_send = spawn(async move {
                 let _ = session_sender_send.send_it(number_of_test_packets).await;
                 info!("Sent all test packets");
             });
-            let recv_task = spawn(async move {
+            let task_session_sender_recv = spawn(async move {
                 let _ = session_sender_recv
                     .recv(number_of_test_packets, reflected_pkts_vec_cloned)
                     .await;
                 info!("Got back all test packets");
             });
             // wait for all test pkts to be sent.
-            send_task.await.unwrap();
+            task_session_sender_send.await.unwrap();
 
             select! {
                 // If stop-session-sleep duration finishes before all pkts are received, drop
                 // recv task and conclude.
                 _ = sleep(Duration::from_secs(stop_session_sleep)) => (),
                 // Ignore stop-session-sleep duration if session-sender got all test pkts before
                 // duration.
-                _ = recv_task => ()
+                _ = task_session_sender_recv => ()
             }
             // Inform Control-Client to send Stop-Sessions
             twamp_test_complete_tx.send(()).unwrap();
         });
-        try_join!(control_client_handle, session_sender_handle).unwrap();
+        try_join!(task_control_client, task_session_sender).unwrap();
         debug!("Control-Client & Session-Sender tasks completed.");
         let acquired_vec = reflected_pkts_vec.lock().await;
         debug!("Reflected pkts len: {}", acquired_vec.len());

src/session_reflector/mod.rs

@@ -9,6 +9,7 @@ use thiserror::Error;
 use tokio::{net::UdpSocket, time::timeout};
 use tracing::*;
 
+/// Errors that can be raised by [SessionReflector].
 #[derive(Error, Debug)]
 pub enum SessionReflectorError {
     /// Raised when "connect" call to a Session-Sender fails.

src/session_sender/mod.rs

@@ -1,14 +1,53 @@
 use crate::timestamp::TimeStamp;
 use crate::twamp_test::{TwampTestPacketUnauth, TwampTestPacketUnauthReflected};
-use anyhow::Result;
 use deku::prelude::*;
 use std::{
+    io,
     net::{SocketAddr, SocketAddrV4},
     sync::Arc,
 };
-use tokio::{net::UdpSocket, spawn, sync::Mutex};
+use thiserror::Error;
+use tokio::{net::UdpSocket, sync::Mutex};
 use tracing::*;
 
+/// Errors that can be raised by [SessionSender].
+#[derive(Error, Debug)]
+pub enum SessionSenderError {
+    /// Raised when "connect" call to a Session-Reflector fails.
+    ///
+    /// - First field is the socket address of the Session-Reflector it was trying to connect to.
+    #[error("failed to connect to Session-Reflector: {0}")]
+    SessionReflectorConnectError(SocketAddrV4, #[source] io::Error),
+
+    /// Indicates read error on TWAMP-Test **after** it was initiated.
+    ///
+    /// - First field is the number of packets that were reflected back to Session-Sender before
+    /// this error.
+    #[error("failed to read UDP datagram from socket after {0} packets were reflected")]
+    SessionReflectorReadError(u32, #[source] io::Error),
+
+    #[error("failed to send UDP datagram to Session-Reflector after {0} packets were sent")]
+    SessionReflectorWriteError(u32, #[source] io::Error),
+
+    /// Indicates Session-Sender not being bound to a socket.
+    ///
+    /// This usually indicates an OS level error, since UdpSocket is made from
+    /// "bind", and this error indicates something unbinded the UdpSocket after it's creation.
+    #[error("Session-Sender was not not bound to a socket")]
+    NotBound(#[source] io::Error),
+
+    /// Indicates Session-Sender was not "connected" to Session-Reflector's UdpSocket.
+    ///
+    /// TODO: This can also indicate an OS level issue, since "connection" is established in
+    /// [SessionSender::new].
+    #[error("Session-Reflector was not connected to socket")]
+    SessionReflectorNotConnected(#[source] io::Error),
+
+    /// Indicates underlying (de)serialization failure.
+    #[error("failed to convert between rust and wire format")]
+    WireConversionError(#[source] deku::error::DekuError),
+}
+
 #[derive(Debug)]
 pub struct SessionSender {
     pub socket: Arc<UdpSocket>,
@@ -23,16 +62,28 @@ impl SessionSender {
         }
     }
 
-    pub async fn send_it(&self, number_of_packets: u32) -> Result<()> {
+    pub async fn send_it(&self, number_of_packets: u32) -> Result<(), SessionSenderError> {
         info!("Sending Twamp-Test packets to {}", self.dest);
         for i in 0..number_of_packets {
             let twamp_test = TwampTestPacketUnauth::new(i, 0, true);
             trace!("Twamp-Test: {:?}", twamp_test);
-            let encoded = twamp_test.to_bytes().unwrap();
-            let l = self.socket.local_addr().unwrap();
-            let p = self.socket.peer_addr().unwrap();
-            trace!("Sending pkt from {} to {}", l, p);
-            let len = self.socket.send(&encoded[..]).await?;
+            let encoded = twamp_test
+                .to_bytes()
+                .map_err(|err| SessionSenderError::WireConversionError(err))?;
+            let local_addr = self
+                .socket
+                .local_addr()
+                .map_err(|err| SessionSenderError::NotBound(err))?;
+            let peer_addr = self
+                .socket
+                .peer_addr()
+                .map_err(|err| SessionSenderError::SessionReflectorNotConnected(err))?;
+            trace!("Sending pkt from {} to {}", local_addr, peer_addr);
+            let len = self
+                .socket
+                .send(&encoded[..])
+                .await
+                .map_err(|err| SessionSenderError::SessionReflectorWriteError(i, err))?;
             trace!("Twamp-Test sent of bytes: {}", len);
         }
         Ok(())
@@ -42,28 +93,27 @@ impl SessionSender {
         &self,
         number_of_packets: u32,
         reflected_pkts_shared: Arc<Mutex<Vec<(TwampTestPacketUnauthReflected, TimeStamp)>>>,
-    ) {
-        let sock_clone = Arc::clone(&self.socket);
-        let reflect_task = spawn(async move {
-            let mut count: u32 = 1;
-            loop {
-                let mut buf = [0u8; 1024]; // Buffer to hold incoming packets
-                let bytes_read = sock_clone.recv(&mut buf).await.unwrap();
-                trace!("Bytes read: {}", bytes_read);
-                let (_rest, reflected_pkt) =
-                    TwampTestPacketUnauthReflected::from_bytes((&buf, 0)).unwrap();
-                trace!("Received reflected pkt: {:?}", reflected_pkt);
-                //debug!("Adding reflector pkt to vec");
-                let mut acquired_vec = reflected_pkts_shared.lock().await;
-                //debug!("Added reflector pkt to vec");
-                acquired_vec.push((reflected_pkt, TimeStamp::default()));
-                if count == number_of_packets {
-                    break;
-                }
-                count += 1;
+    ) -> Result<(), SessionSenderError> {
+        let mut count: u32 = 0;
+        loop {
+            let mut buf = [0u8; 1024]; // Buffer to hold incoming packets
+            let bytes_read = self
+                .socket
+                .recv(&mut buf)
+                .await
+                .map_err(|err| SessionSenderError::SessionReflectorReadError(count, err))?;
+            trace!("Bytes read: {}", bytes_read);
+            let (_rest, reflected_pkt) = TwampTestPacketUnauthReflected::from_bytes((&buf, 0))
+                .map_err(|err| SessionSenderError::WireConversionError(err))?;
+            trace!("Received reflected pkt: {:?}", reflected_pkt);
+            let mut acquired_vec = reflected_pkts_shared.lock().await;
+            acquired_vec.push((reflected_pkt, TimeStamp::default()));
+            count += 1;
+            if count == number_of_packets {
+                break;
             }
-        });
-        reflect_task.await.unwrap()
+        }
+        Ok(())
     }
 }