From: Ian Jackson <ijackson@chiark.greenend.org.uk>
Date: Sat, 31 Jul 2021 11:20:01 +0000 (+0100)
Subject: Report when http req fails, not after retry delay
X-Git-Tag: hippotat/1.0.0~420
X-Git-Url: https://www.chiark.greenend.org.uk/ucgi/~ian/git?a=commitdiff_plain;h=39c8c8a21bdf4d9cb288aea184485be38a861e28;p=hippotat.git

Report when http req fails, not after retry delay

Signed-off-by: Ian Jackson <ijackson@chiark.greenend.org.uk>
---

diff --git a/src/bin/client.rs b/src/bin/client.rs
index 2bbde41..c6922df 100644
--- a/src/bin/client.rs
+++ b/src/bin/client.rs
@@ -5,7 +5,7 @@
 use hippotat::prelude::*;
 
 type OutstandingRequest<'r> = Pin<Box<
-    dyn Future<Output=Result<Bytes,AE>> + Send + 'r
+    dyn Future<Output=Option<Bytes>> + Send + 'r
     >>;
 
 impl<T> HCC for T where
@@ -13,9 +13,10 @@ impl<T> HCC for T where
 trait HCC: hyper::client::connect::Connect + Clone + Send + Sync + 'static { }
 
 #[throws(AE)]
-fn submit_request<'r, 'ic:'r, 'hc:'r, C:HCC>(
+fn submit_request<'r, 'ic:'r, 'hc:'r, 'rep:'r, C:HCC>(
   ic: &'ic InstanceConfig,
   hclient: &'hc Arc<hyper::Client<C>>,
+  reporter: &'rep parking_lot::Mutex<Reporter>,
   reqs: &mut Vec<OutstandingRequest<'r>>,
   upbound: Vec<Vec<u8>>
 ) {
@@ -42,9 +43,13 @@ fn submit_request<'r, 'ic:'r, 'hc:'r, C:HCC>(
       // xxx: some size limit to avoid mallocing the universe
       let resp = hyper::body::to_bytes(resp).await
         .context("HTTP error fetching response body")?;
+
       Ok::<_,AE>(resp)
     }).await? }.await;
-    if r.is_err() {
+
+    let r = reporter.lock().report(r);
+
+    if r.is_none() {
       tokio::time::sleep(ic.http_retry).await;
     }
     r
@@ -60,7 +65,7 @@ async fn run_client<C:HCC>(
 {
   debug!("{}: config: {:?}", &ic, &ic);
 
-  let mut reporter = Reporter { };
+  let reporter = parking_lot::Mutex::new(Reporter { });
 
   let mut ipif = tokio::process::Command::new("sh")
     .args(&["-c", &ic.ipif])
@@ -136,7 +141,7 @@ async fn run_client<C:HCC>(
           dbg!(&reqs.len(), &upbound_total, &upbound.len());
 
           //: impl futures::Stream<Cow<&[u8]>>
-          submit_request(&ic, &hclient, &mut reqs, upbound)?;
+          submit_request(&ic, &hclient, &reporter, &mut reqs, upbound)?;
         }
 
         () = async { },
@@ -145,14 +150,14 @@ async fn run_client<C:HCC>(
             reqs.len() < ic.max_requests_outstanding.sat()) =>
         {
           let upbound = tx_defer.take().into_iter().collect_vec();
-          submit_request(&ic, &hclient, &mut reqs, upbound)?;
+          submit_request(&ic, &hclient, &reporter, &mut reqs, upbound)?;
         }
 
         (got, goti, _) = async { future::select_all(&mut reqs).await },
           if ! reqs.is_empty() =>
         {
           reqs.swap_remove(goti);
-          if let Some(got) = reporter.report(got) {
+          if let Some(got) = got {
             dbg!(&got.remaining()); // xxx
           }
         }