new parameter to compute transfer hashes

Author: github-af

src/aux/file/receive.rs

@@ -1,9 +1,9 @@
-use fasthash::HasherExt;
-
 use crate::aux::{self, file};
+use fasthash::HasherExt;
 use std::{
     fs,
     hash::Hash,
+    hash::Hasher,
     io::{Read, Write},
     net,
     os::unix::{self, fs::PermissionsExt},
@@ -156,7 +156,7 @@ where
             0 => {
                 if 0 < cursor {
                     if let Some(hasher) = hasher.as_mut() {
-                        buffer[..cursor].hash(hasher);
+                        hasher.write(&buffer[..cursor]);
                     }
                     file.write_all(&buffer[..cursor])?;
                 }

src/aux/file/send.rs

@@ -1,9 +1,8 @@
-use fasthash::HasherExt;
-
 use crate::aux::{self, file};
+use fasthash::HasherExt;
 use std::{
     fs,
-    hash::Hash,
+    hash::{Hash, Hasher},
     io::{Read, Write},
     net,
     os::unix::{self, fs::PermissionsExt},
@@ -108,7 +107,7 @@ where
                 if 0 < cursor {
                     total += cursor;
                     if let Some(hasher) = hasher.as_mut() {
-                        buffer[..cursor].hash(hasher);
+                        hasher.write(&buffer[..cursor]);
                     }
                     diode.write_all(&buffer[..cursor])?;
                 }

src/bin/diode-oneshot-receive.rs

@@ -83,6 +83,8 @@ struct Args {
     min_repair: u32,
     #[clap(long, help = "Set CPU affinity for threads")]
     cpu_affinity: bool,
+    #[clap(long, help = "Hash each client transfered data")]
+    hash: bool,
 }
 
 fn main() {
@@ -120,6 +122,7 @@ fn main() {
             heartbeat_interval: None,
             batch_receive: args.batch,
             cpu_affinity: args.cpu_affinity,
+            hash: args.hash,
         },
         raptorq,
         |_| Ok::<_, io::Error>(io::stdout()),

src/bin/diode-oneshot-send.rs

@@ -65,6 +65,8 @@ struct Args {
     repair: u32,
     #[clap(long, help = "Set CPU affinity for threads")]
     cpu_affinity: bool,
+    #[clap(long, help = "Hash each client transfered data")]
+    hash: bool,
 }
 
 fn main() {
@@ -100,6 +102,7 @@ fn main() {
             to_mtu: args.to_mtu,
             batch_send: args.batch,
             cpu_affinity: args.cpu_affinity,
+            hash: args.hash,
         },
         raptorq,
     ) {

src/bin/diode-receive.rs

@@ -127,6 +127,8 @@ struct Args {
     heartbeat: Option<time::Duration>,
     #[clap(long, help = "Set CPU affinity for threads")]
     cpu_affinity: bool,
+    #[clap(long, help = "Hash each client transfered data")]
+    hash: bool,
 }
 
 enum Client {
@@ -210,6 +212,7 @@ fn main() {
             heartbeat_interval: args.heartbeat,
             batch_receive: args.batch,
             cpu_affinity: args.cpu_affinity,
+            hash: args.hash,
         },
         raptorq,
         |_| Client::try_from(&args.to),

src/bin/diode-send.rs

@@ -115,6 +115,8 @@ struct Args {
     repair: u32,
     #[clap(long, help = "Set CPU affinity for threads")]
     cpu_affinity: bool,
+    #[clap(long, help = "Hash each client transfered data")]
+    hash: bool,
 }
 
 enum Client {
@@ -205,6 +207,7 @@ fn main() {
             to_mtu: args.to_mtu,
             batch_send: args.batch,
             cpu_affinity: args.cpu_affinity,
+            hash: args.hash,
         },
         raptorq,
     ) {

src/receive/client.rs

@@ -1,7 +1,9 @@
 //! Worker that writes decoded and reordered messages to client
 
 use crate::{protocol, receive};
+use fasthash::HasherExt;
 use std::{
+    hash::Hasher,
     io::{self, Write},
     os::fd::AsRawFd,
     thread,
@@ -26,6 +28,12 @@ where
 
     let mut transmitted = 0;
 
+    let mut hasher = if receiver.config.hash {
+        Some(fasthash::SpookyHasherExt::default())
+    } else {
+        None
+    };
+
     loop {
         let block = if let Some(timeout) = receiver.config.abort_timeout {
             recvq.recv_timeout(timeout).map_err(receive::Error::from)?
@@ -39,7 +47,13 @@ where
 
         if !payload.is_empty() {
             log::trace!("client {client_id:x}: payload {} bytes", payload.len());
+
+            if let Some(hasher) = hasher.as_mut() {
+                hasher.write(payload);
+            }
+
             transmitted += payload.len();
+
             client.write_all(payload)?;
             if receiver.config.flush {
                 client.flush()?;
@@ -58,9 +72,16 @@ where
                 return Ok(());
             }
             protocol::BlockType::End => {
-                log::info!(
-                    "client {client_id:x}: finished transfer, {transmitted} bytes transmitted"
-                );
+                if let Some(hasher) = hasher {
+                    let hash = hasher.finish_ext();
+                    log::info!(
+                        "client {client_id:x}: finished transfer, {transmitted} bytes transmitted, hash is {hash:x}"
+                    );
+                } else {
+                    log::info!(
+                        "client {client_id:x}: finished transfer, {transmitted} bytes transmitted"
+                    );
+                }
                 client.flush()?;
                 (receiver.client_end)(
                     client.into_inner().map_err(|e| {

src/receive/mod.rs

@@ -43,6 +43,7 @@ pub struct Config {
     pub abort_timeout: Option<time::Duration>,
     pub heartbeat_interval: Option<time::Duration>,
     pub cpu_affinity: bool,
+    pub hash: bool,
 }
 
 pub enum Error {

src/send/client.rs

@@ -1,7 +1,8 @@
 //! Worker that reads data from a client socket and split it into [`crate::protocol`] blocks
 
 use crate::{protocol, send};
-use std::{io, os::fd::AsRawFd, thread};
+use fasthash::HasherExt;
+use std::{hash::Hasher, io, os::fd::AsRawFd, thread};
 
 pub fn start<C>(
     sender: &send::Sender<C>,
@@ -24,6 +25,12 @@ where
     let mut cursor = 0;
     let mut transmitted = 0;
 
+    let mut hasher = if sender.config.hash {
+        Some(fasthash::SpookyHasherExt::default())
+    } else {
+        None
+    };
+
     loop {
         log::trace!("client {client_id:x}: read...");
 
@@ -46,6 +53,10 @@ where
 
         log::trace!("client {client_id:x}: send {cursor} bytes");
 
+        if let Some(hasher) = hasher.as_mut() {
+            hasher.write(&buffer[..cursor]);
+        }
+
         sender.to_encoding.send(Some(protocol::Block::new(
             block_type,
             &sender.raptorq,
@@ -57,7 +68,14 @@ where
         cursor = 0;
 
         if 0 == read {
-            log::info!("client {client_id:x}: disconnect, {transmitted} bytes sent");
+            if let Some(hasher) = hasher {
+                let hash = hasher.finish_ext();
+                log::info!(
+                    "client {client_id:x}: disconnect, {transmitted} bytes sent, hash is {hash:x}"
+                );
+            } else {
+                log::info!("client {client_id:x}: disconnect, {transmitted} bytes sent");
+            }
             return Ok(());
         }
 

src/send/mod.rs

@@ -44,6 +44,7 @@ pub struct Config {
     pub to_mtu: u16,
     pub batch_send: Option<u32>,
     pub cpu_affinity: bool,
+    pub hash: bool,
 }
 
 pub enum Error {