Do not terminate if single packet is too big to write

Instead, indicate in statistics that we were not able to write all
packets.

Author: jtt

src/output.rs

@@ -38,7 +38,28 @@ struct Interface(Pcap);
 
 impl PacketWriter for Interface {
     fn write_raw(&mut self, buf: &[u8]) -> Result<usize> {
-        Ok(self.0.inject(buf)?)
+        match self.0.inject(buf) {
+            Ok(ret) => Ok(ret),
+            Err(err) => {
+                tracing::warn!(?err, len = ?buf.len(), "error while trying to write");
+                // we do not want to stop writing if we get error indicating that
+                // packet was too large to write.
+                if let luomu_libpcap::Error::PcapError(ref msg) = err {
+                    if msg.contains("Message too") && buf.len() > 1500 {
+                        // this is a stupid way to detect such errors, but there
+                        // is no other way currently, as the only thing we get
+                        // is error message from libpcap and it can contain at
+                        // least "Message too long" and "Message too large"
+                        // depending on the Linux distribution of choice.
+                        Ok(0)
+                    } else {
+                        Err(err.into())
+                    }
+                } else {
+                    Err(err.into())
+                }
+            }
+        }
     }
 }
 

src/pipe.rs

@@ -19,6 +19,8 @@ pub struct Stats {
     packets: u64,
     /// Number of bytes processed since start or last reset.
     bytes: u64,
+    /// Number of packets which we were not able to send.
+    invalid: u64,
     /// When packet processing has started.
     start: Instant,
     /// Interval for producing stats
@@ -36,18 +38,25 @@ impl Default for Stats {
             last_stat: Instant::now(),
             packets: Default::default(),
             bytes: Default::default(),
+            invalid: Default::default(),
             sender: None,
             interval: None,
         }
     }
 }
 
 impl Stats {
-    /// Updates the statistics with a packet containing given number of bytes
+    /// Updates the statistics with a packet containing given number of bytes.
+    /// If `bytes` is 0, this is to indicate that packet was not sent and
+    /// should increase the "invalid" packet count.
     ///
     /// Sends summary of statistics if it is time to send them.
     fn update(&mut self, bytes: u64) {
-        self.packets += 1;
+        if bytes == 0 {
+            self.invalid += 1
+        } else {
+            self.packets += 1;
+        }
         self.bytes += bytes;
         if let Some(val) = self.interval {
             if self.last_stat.elapsed() > val {
@@ -71,9 +80,14 @@ impl Stats {
         let bps = (self.bytes as f64 * 8_f64) / elapsed.as_secs_f64();
         let mbps = (self.bytes as f64 / (1024 * 1024) as f64) / elapsed.as_secs_f64();
 
+        let packet_count = match self.invalid {
+            0 => format!("{} packets", self.packets),
+            _ => format!("{} packets ({} not sent)", self.packets, self.invalid),
+        };
+
         format!(
-            "{} packets, {} bytes in {}ms / {:.3}pps, {:.3}bps ({:.3} MBps)",
-            self.packets,
+            "{}, {} bytes in {}ms / {:.3}pps, {:.3}bps ({:.3} MBps)",
+            packet_count,
             self.bytes,
             elapsed.as_millis(),
             pps,
@@ -86,6 +100,7 @@ impl Stats {
     fn reset(&mut self) {
         self.bytes = 0;
         self.packets = 0;
+        self.invalid = 0;
         self.start = Instant::now();
     }