send acks on timeout

hptp-recv/src/main.rs

@@ -95,10 +95,10 @@ where
         }
     }
 
-    fn ack_msg(&self) -> DownMsg {
-        DownMsg::Ack {
-            idxs: self.segs.keys().cloned().collect(),
-        }
+    fn ack_idxs(&self) -> Vec<SegIdx> {
+        let mut idxs: Vec<_> = self.segs.keys().cloned().collect();
+        idxs.sort();
+        idxs
     }
 
     fn is_file_complete(&self) -> bool {
@@ -110,17 +110,25 @@ async fn download<OUT>(log: &mut Logger, peer: &mut DownPeer, out: &mut OUT) ->
 where
     OUT: AsyncWrite + Unpin,
 {
-    let mut sink = SegmentSink::new(out);
-    let mut to_send = vec![];
+    use tokio::time::{Duration, Instant};
 
-    enum Action {
-        Continue,
-        Quit,
+    enum Evt {
+        Recv(UpMsg),
+        Timer,
     }
 
+    let mut sink = SegmentSink::new(out);
+
+    const DELAY_MS: u64 = 1000;
+    let mut deadline = Instant::now();
+    let mut to_send = vec![];
+
     loop {
-        let msg = match peer.recv().await {
-            Ok(m) => m,
+        let timer = tokio::time::delay_until(deadline);
+        let evt = tokio::select!(
+            _ = timer => Evt::Timer,
+            r = peer.recv() => match r {
+                Ok(m) => Evt::Recv(m),
                 Err(peer::RecvError::InvalidMessage { .. }) => {
                     log.recv_corrupt().await;
                     continue;
@@ -128,35 +136,35 @@ where
                 Err(peer::RecvError::Io { source }) => {
                     return Err(source.into());
                 }
-        };
+            }
+        );
 
-        let act;
-        match msg {
-            UpMsg::Data { payload, seg_idx } => {
+        match evt {
+            Evt::Timer => {
+                deadline += Duration::from_millis(DELAY_MS);
+                log.debug_msg("timeout").await;
+            }
+
+            Evt::Recv(UpMsg::Data { payload, seg_idx }) => {
                 let len = payload.len();
                 match sink.put(seg_idx, payload).await {
                     Put::Duplicate => {
                         log.recv_data_ignored(seg_idx as usize, len).await;
                     }
-
                     Put::Fresh => {
-                        log.recv_data_accepted(seg_idx as usize, len, hptp::logger::OutOfOrder)
+                        let ord = hptp::logger::OutOfOrder;
+                        log.recv_data_accepted(seg_idx as usize, len, ord).await;
+                    }
+                }
+            }
+        };
+
+        {
+            let idxs = sink.ack_idxs();
+            log.debug_msg(format!("send acks {:?}", idxs))
                 .await;
-                        log.debug_msg(format!("sending acks: {:?}", {
-                            let mut idxs = sink.segs.keys().cloned().collect::<Vec<_>>();
-                            idxs.sort();
-                            idxs
-                        }))
-                        .await;
-                        to_send.push(sink.ack_msg());
-                    }
-                }
-                act = if sink.is_file_complete() {
-                    Action::Quit
-                } else {
-                    Action::Continue
-                }
-            }
+            to_send.push(DownMsg::Ack { idxs: idxs.clone() });
+            to_send.push(DownMsg::Ack { idxs });
         }
 
         for m in to_send.drain(..) {
@@ -167,9 +175,8 @@ where
             }
         }
 
-        match act {
-            Action::Continue => (),
-            Action::Quit => break,
+        if sink.is_file_complete() {
+            break
         }
     }
 

hptp-send/src/main.rs

@@ -148,9 +148,9 @@ where
     }
 
     let mut src = SegmentSource::new(inp);
-    let mut deadline = Instant::now();
-    const DELAY_MS: u64 = 100;
 
+    const DELAY_MS: u64 = 100;
+    let mut deadline = Instant::now();
     let mut to_send = vec![];
 
     loop {