Struct Sftp

Source

pub struct Sftp<W, R> { /* private fields */ }

Implementations§

Source §

impl<W: AsyncWrite + Unpin, R: AsyncRead + Unpin> Sftp<W, R>

Source

pub async fn handshake(w: W, r: R) -> Result<Self>

Examples found in repository ?

examples/measure-roundtrips.rs (line 44)

36async fn main() -> Result<()> {
37    let mut args = std::env::args().skip(1);
38    let host = args
39        .next()
40        .context("usage: measure-roundtrips <ssh-host> [remote-dir]")?;
41    let dir = args.next().unwrap_or_else(|| "/usr/include".to_string());
42
43    let (_child, w, r) = transport::open(&host)?;
44    let mut s = Sftp::handshake(w, r).await?;
45    println!("host {host}  sftp v{}", s.version());
46
47    let tau = measure_tau(&mut s).await?;
48    println!("tau (one round trip) = {:.1} ms", ms(tau));
49
50    let entries = list(&mut s, &dir).await?;
51    let files: Vec<String> = entries
52        .iter()
53        .filter(|(name, a)| {
54            !a.is_dir() && !a.is_symlink() && a.size.unwrap_or(0) > 0 && name != "." && name != ".."
55        })
56        .map(|(name, _)| format!("{}/{}", dir.trim_end_matches('/'), name))
57        .collect();
58    ensure!(
59        !files.is_empty(),
60        "no regular non-empty files in {dir}; pass a different remote-dir"
61    );
62    println!(
63        "{} entries in {dir}, {} usable files",
64        entries.len(),
65        files.len()
66    );
67    println!();
68
69    let mut largest = 0usize;
70    let mut largest_open = 0.0f64;
71    let mut largest_read = 0.0f64;
72    let mut prev_read = 0.0f64;
73    let mut last_read = 0.0f64;
74    for n in BATCH_SIZES {
75        if n > files.len() {
76            continue;
77        }
78        let batch = &files[..n];
79        let (t_open, handles) = batch_open(&mut s, batch).await?;
80        let (t_read, bytes) = batch_read(&mut s, &handles).await?;
81        batch_close(&mut s, &handles).await?;
82
83        let open_tau = t_open.as_secs_f64() / tau.as_secs_f64();
84        let read_tau = t_read.as_secs_f64() / tau.as_secs_f64();
85        println!(
86            "n={n:<3} open {:>7.1} ms ({open_tau:>5.2} tau)   read {:>7.1} ms ({read_tau:>5.2} tau)   {bytes} B",
87            ms(t_open),
88            ms(t_read)
89        );
90        largest = n;
91        largest_open = open_tau;
92        largest_read = read_tau;
93        if n > 1 {
94            prev_read = last_read;
95        }
96        last_read = read_tau;
97    }
98
99    println!();
100    ensure!(
101        largest > 1,
102        "only one usable file; cannot distinguish pipelined from serial"
103    );
104    // The verdict rests on opens alone. An open carries a path and nothing else, so
105    // its cost is round trips and only round trips. A read also carries the file,
106    // and under an injected per-packet delay the transfer dominates: 400 KB reads
107    // as 15 tau however few round trips fetched it. Judging on reads made this
108    // check fail on how much data the chosen directory happens to hold, which is
109    // a property of the machine rather than of the code -- exactly the kind of
110    // flaky gate that teaches people to ignore a red check.
111    //
112    // Reads still say something, just not in absolute terms: if doubling the batch
113    // does not double the time, the round trips did not scale either.
114    if prev_read > 0.0 {
115        let growth = largest_read / prev_read;
116        println!(
117            "reads {largest_read:.2} tau at n={largest}, {growth:.2}x the previous batch (serial would be about 2x)"
118        );
119    }
120
121    let serial = largest as f64;
122    if largest_open < serial / PIPELINE_MARGIN {
123        println!(
124            "VERDICT pipelined: opens cost {largest_open:.2} tau at n={largest} (serial would cost about {serial:.0})"
125        );
126        Ok(())
127    } else {
128        bail!(
129            "VERDICT serial: opens cost {largest_open:.2} tau at n={largest}, near the serial cost {serial:.0}. Invariant 1 (O(1) round trips per page) is not reachable over this transport."
130        )
131    }
132}

Source

pub fn version(&self) -> u32

Examples found in repository ?

examples/measure-roundtrips.rs (line 45)

36async fn main() -> Result<()> {
37    let mut args = std::env::args().skip(1);
38    let host = args
39        .next()
40        .context("usage: measure-roundtrips <ssh-host> [remote-dir]")?;
41    let dir = args.next().unwrap_or_else(|| "/usr/include".to_string());
42
43    let (_child, w, r) = transport::open(&host)?;
44    let mut s = Sftp::handshake(w, r).await?;
45    println!("host {host}  sftp v{}", s.version());
46
47    let tau = measure_tau(&mut s).await?;
48    println!("tau (one round trip) = {:.1} ms", ms(tau));
49
50    let entries = list(&mut s, &dir).await?;
51    let files: Vec<String> = entries
52        .iter()
53        .filter(|(name, a)| {
54            !a.is_dir() && !a.is_symlink() && a.size.unwrap_or(0) > 0 && name != "." && name != ".."
55        })
56        .map(|(name, _)| format!("{}/{}", dir.trim_end_matches('/'), name))
57        .collect();
58    ensure!(
59        !files.is_empty(),
60        "no regular non-empty files in {dir}; pass a different remote-dir"
61    );
62    println!(
63        "{} entries in {dir}, {} usable files",
64        entries.len(),
65        files.len()
66    );
67    println!();
68
69    let mut largest = 0usize;
70    let mut largest_open = 0.0f64;
71    let mut largest_read = 0.0f64;
72    let mut prev_read = 0.0f64;
73    let mut last_read = 0.0f64;
74    for n in BATCH_SIZES {
75        if n > files.len() {
76            continue;
77        }
78        let batch = &files[..n];
79        let (t_open, handles) = batch_open(&mut s, batch).await?;
80        let (t_read, bytes) = batch_read(&mut s, &handles).await?;
81        batch_close(&mut s, &handles).await?;
82
83        let open_tau = t_open.as_secs_f64() / tau.as_secs_f64();
84        let read_tau = t_read.as_secs_f64() / tau.as_secs_f64();
85        println!(
86            "n={n:<3} open {:>7.1} ms ({open_tau:>5.2} tau)   read {:>7.1} ms ({read_tau:>5.2} tau)   {bytes} B",
87            ms(t_open),
88            ms(t_read)
89        );
90        largest = n;
91        largest_open = open_tau;
92        largest_read = read_tau;
93        if n > 1 {
94            prev_read = last_read;
95        }
96        last_read = read_tau;
97    }
98
99    println!();
100    ensure!(
101        largest > 1,
102        "only one usable file; cannot distinguish pipelined from serial"
103    );
104    // The verdict rests on opens alone. An open carries a path and nothing else, so
105    // its cost is round trips and only round trips. A read also carries the file,
106    // and under an injected per-packet delay the transfer dominates: 400 KB reads
107    // as 15 tau however few round trips fetched it. Judging on reads made this
108    // check fail on how much data the chosen directory happens to hold, which is
109    // a property of the machine rather than of the code -- exactly the kind of
110    // flaky gate that teaches people to ignore a red check.
111    //
112    // Reads still say something, just not in absolute terms: if doubling the batch
113    // does not double the time, the round trips did not scale either.
114    if prev_read > 0.0 {
115        let growth = largest_read / prev_read;
116        println!(
117            "reads {largest_read:.2} tau at n={largest}, {growth:.2}x the previous batch (serial would be about 2x)"
118        );
119    }
120
121    let serial = largest as f64;
122    if largest_open < serial / PIPELINE_MARGIN {
123        println!(
124            "VERDICT pipelined: opens cost {largest_open:.2} tau at n={largest} (serial would cost about {serial:.0})"
125        );
126        Ok(())
127    } else {
128        bail!(
129            "VERDICT serial: opens cost {largest_open:.2} tau at n={largest}, near the serial cost {serial:.0}. Invariant 1 (O(1) round trips per page) is not reachable over this transport."
130        )
131    }
132}

Source

pub fn alloc_id(&mut self) -> u32

Examples found in repository ?

examples/measure-roundtrips.rs (line 142)

139async fn measure_tau(s: &mut Session) -> Result<Duration> {
140    let mut samples = Vec::with_capacity(TAU_REPS);
141    for _ in 0..TAU_REPS {
142        let id = s.alloc_id();
143        let t = Instant::now();
144        s.queue(REALPATH, &Enc::new().u32(id).str(b".").done())
145            .await?;
146        s.flush().await?;
147        let r = s.recv().await?;
148        ensure!(r.id == id, "reply id {} does not match request {id}", r.id);
149        samples.push(t.elapsed());
150    }
151    samples.sort_unstable();
152    Ok(samples[samples.len() / 2])
153}
154
155/// One READDIR sweep carries every entry's attrs, which is what lets the origin
156/// layer skip per-file STAT entirely and keep invariant 1 within reach.
157async fn list(s: &mut Session, dir: &str) -> Result<Vec<(String, Attrs)>> {
158    let id = s.alloc_id();
159    s.queue(OPENDIR, &Enc::new().u32(id).str(dir.as_bytes()).done())
160        .await?;
161    s.flush().await?;
162    let r = s.recv().await?;
163    ensure!(
164        r.kind == HANDLE,
165        "opendir {dir} refused (reply type {})",
166        r.kind
167    );
168    let handle = Dec::new(r.payload())
169        .str()
170        .context("opendir handle")?
171        .to_vec();
172
173    let mut out = Vec::new();
174    loop {
175        let id = s.alloc_id();
176        s.queue(READDIR, &Enc::new().u32(id).str(&handle).done())
177            .await?;
178        s.flush().await?;
179        let r = s.recv().await?;
180        if r.kind == STATUS {
181            break;
182        }
183        ensure!(r.kind == NAME, "readdir gave reply type {}", r.kind);
184        let mut d = Dec::new(r.payload());
185        let count = d.u32().context("readdir count")?;
186        for _ in 0..count {
187            let name = String::from_utf8_lossy(d.str().context("filename")?).into_owned();
188            d.str().context("longname")?;
189            let attrs = Attrs::decode(&mut d).context("attrs")?;
190            out.push((name, attrs));
191        }
192    }
193
194    let id = s.alloc_id();
195    s.queue(CLOSE, &Enc::new().u32(id).str(&handle).done())
196        .await?;
197    s.flush().await?;
198    s.recv().await?;
199    Ok(out)
200}
201
202async fn batch_open(s: &mut Session, paths: &[String]) -> Result<(Duration, Vec<Vec<u8>>)> {
203    let t = Instant::now();
204    for p in paths {
205        let id = s.alloc_id();
206        s.queue(
207            OPEN,
208            &Enc::new()
209                .u32(id)
210                .str(p.as_bytes())
211                .u32(FXF_READ)
212                .u32(0)
213                .done(),
214        )
215        .await?;
216    }
217    s.flush().await?;
218
219    let mut handles = Vec::with_capacity(paths.len());
220    for _ in 0..paths.len() {
221        let r = s.recv().await?;
222        ensure!(r.kind == HANDLE, "open refused (reply type {})", r.kind);
223        handles.push(Dec::new(r.payload()).str().context("open handle")?.to_vec());
224    }
225    Ok((t.elapsed(), handles))
226}
227
228async fn batch_read(s: &mut Session, handles: &[Vec<u8>]) -> Result<(Duration, usize)> {
229    let t = Instant::now();
230    for h in handles {
231        let id = s.alloc_id();
232        s.queue(READ, &Enc::new().u32(id).str(h).u64(0).u32(READ_LEN).done())
233            .await?;
234    }
235    s.flush().await?;
236
237    let mut bytes = 0usize;
238    for _ in 0..handles.len() {
239        let r = s.recv().await?;
240        match r.kind {
241            DATA => bytes += Dec::new(r.payload()).str().map_or(0, |b| b.len()),
242            STATUS => {}
243            other => bail!("read gave reply type {other}"),
244        }
245    }
246    Ok((t.elapsed(), bytes))
247}
248
249async fn batch_close(s: &mut Session, handles: &[Vec<u8>]) -> Result<()> {
250    for h in handles {
251        let id = s.alloc_id();
252        s.queue(CLOSE, &Enc::new().u32(id).str(h).done()).await?;
253    }
254    s.flush().await?;
255    for _ in 0..handles.len() {
256        s.recv().await?;
257    }
258    Ok(())
259}

Source

pub async fn queue(&mut self, kind: u8, payload: &[u8]) -> Result<()>

Examples found in repository ?

examples/measure-roundtrips.rs (line 144)

139async fn measure_tau(s: &mut Session) -> Result<Duration> {
140    let mut samples = Vec::with_capacity(TAU_REPS);
141    for _ in 0..TAU_REPS {
142        let id = s.alloc_id();
143        let t = Instant::now();
144        s.queue(REALPATH, &Enc::new().u32(id).str(b".").done())
145            .await?;
146        s.flush().await?;
147        let r = s.recv().await?;
148        ensure!(r.id == id, "reply id {} does not match request {id}", r.id);
149        samples.push(t.elapsed());
150    }
151    samples.sort_unstable();
152    Ok(samples[samples.len() / 2])
153}
154
155/// One READDIR sweep carries every entry's attrs, which is what lets the origin
156/// layer skip per-file STAT entirely and keep invariant 1 within reach.
157async fn list(s: &mut Session, dir: &str) -> Result<Vec<(String, Attrs)>> {
158    let id = s.alloc_id();
159    s.queue(OPENDIR, &Enc::new().u32(id).str(dir.as_bytes()).done())
160        .await?;
161    s.flush().await?;
162    let r = s.recv().await?;
163    ensure!(
164        r.kind == HANDLE,
165        "opendir {dir} refused (reply type {})",
166        r.kind
167    );
168    let handle = Dec::new(r.payload())
169        .str()
170        .context("opendir handle")?
171        .to_vec();
172
173    let mut out = Vec::new();
174    loop {
175        let id = s.alloc_id();
176        s.queue(READDIR, &Enc::new().u32(id).str(&handle).done())
177            .await?;
178        s.flush().await?;
179        let r = s.recv().await?;
180        if r.kind == STATUS {
181            break;
182        }
183        ensure!(r.kind == NAME, "readdir gave reply type {}", r.kind);
184        let mut d = Dec::new(r.payload());
185        let count = d.u32().context("readdir count")?;
186        for _ in 0..count {
187            let name = String::from_utf8_lossy(d.str().context("filename")?).into_owned();
188            d.str().context("longname")?;
189            let attrs = Attrs::decode(&mut d).context("attrs")?;
190            out.push((name, attrs));
191        }
192    }
193
194    let id = s.alloc_id();
195    s.queue(CLOSE, &Enc::new().u32(id).str(&handle).done())
196        .await?;
197    s.flush().await?;
198    s.recv().await?;
199    Ok(out)
200}
201
202async fn batch_open(s: &mut Session, paths: &[String]) -> Result<(Duration, Vec<Vec<u8>>)> {
203    let t = Instant::now();
204    for p in paths {
205        let id = s.alloc_id();
206        s.queue(
207            OPEN,
208            &Enc::new()
209                .u32(id)
210                .str(p.as_bytes())
211                .u32(FXF_READ)
212                .u32(0)
213                .done(),
214        )
215        .await?;
216    }
217    s.flush().await?;
218
219    let mut handles = Vec::with_capacity(paths.len());
220    for _ in 0..paths.len() {
221        let r = s.recv().await?;
222        ensure!(r.kind == HANDLE, "open refused (reply type {})", r.kind);
223        handles.push(Dec::new(r.payload()).str().context("open handle")?.to_vec());
224    }
225    Ok((t.elapsed(), handles))
226}
227
228async fn batch_read(s: &mut Session, handles: &[Vec<u8>]) -> Result<(Duration, usize)> {
229    let t = Instant::now();
230    for h in handles {
231        let id = s.alloc_id();
232        s.queue(READ, &Enc::new().u32(id).str(h).u64(0).u32(READ_LEN).done())
233            .await?;
234    }
235    s.flush().await?;
236
237    let mut bytes = 0usize;
238    for _ in 0..handles.len() {
239        let r = s.recv().await?;
240        match r.kind {
241            DATA => bytes += Dec::new(r.payload()).str().map_or(0, |b| b.len()),
242            STATUS => {}
243            other => bail!("read gave reply type {other}"),
244        }
245    }
246    Ok((t.elapsed(), bytes))
247}
248
249async fn batch_close(s: &mut Session, handles: &[Vec<u8>]) -> Result<()> {
250    for h in handles {
251        let id = s.alloc_id();
252        s.queue(CLOSE, &Enc::new().u32(id).str(h).done()).await?;
253    }
254    s.flush().await?;
255    for _ in 0..handles.len() {
256        s.recv().await?;
257    }
258    Ok(())
259}

Source

pub async fn flush(&mut self) -> Result<()>

Examples found in repository ?

examples/measure-roundtrips.rs (line 146)

139async fn measure_tau(s: &mut Session) -> Result<Duration> {
140    let mut samples = Vec::with_capacity(TAU_REPS);
141    for _ in 0..TAU_REPS {
142        let id = s.alloc_id();
143        let t = Instant::now();
144        s.queue(REALPATH, &Enc::new().u32(id).str(b".").done())
145            .await?;
146        s.flush().await?;
147        let r = s.recv().await?;
148        ensure!(r.id == id, "reply id {} does not match request {id}", r.id);
149        samples.push(t.elapsed());
150    }
151    samples.sort_unstable();
152    Ok(samples[samples.len() / 2])
153}
154
155/// One READDIR sweep carries every entry's attrs, which is what lets the origin
156/// layer skip per-file STAT entirely and keep invariant 1 within reach.
157async fn list(s: &mut Session, dir: &str) -> Result<Vec<(String, Attrs)>> {
158    let id = s.alloc_id();
159    s.queue(OPENDIR, &Enc::new().u32(id).str(dir.as_bytes()).done())
160        .await?;
161    s.flush().await?;
162    let r = s.recv().await?;
163    ensure!(
164        r.kind == HANDLE,
165        "opendir {dir} refused (reply type {})",
166        r.kind
167    );
168    let handle = Dec::new(r.payload())
169        .str()
170        .context("opendir handle")?
171        .to_vec();
172
173    let mut out = Vec::new();
174    loop {
175        let id = s.alloc_id();
176        s.queue(READDIR, &Enc::new().u32(id).str(&handle).done())
177            .await?;
178        s.flush().await?;
179        let r = s.recv().await?;
180        if r.kind == STATUS {
181            break;
182        }
183        ensure!(r.kind == NAME, "readdir gave reply type {}", r.kind);
184        let mut d = Dec::new(r.payload());
185        let count = d.u32().context("readdir count")?;
186        for _ in 0..count {
187            let name = String::from_utf8_lossy(d.str().context("filename")?).into_owned();
188            d.str().context("longname")?;
189            let attrs = Attrs::decode(&mut d).context("attrs")?;
190            out.push((name, attrs));
191        }
192    }
193
194    let id = s.alloc_id();
195    s.queue(CLOSE, &Enc::new().u32(id).str(&handle).done())
196        .await?;
197    s.flush().await?;
198    s.recv().await?;
199    Ok(out)
200}
201
202async fn batch_open(s: &mut Session, paths: &[String]) -> Result<(Duration, Vec<Vec<u8>>)> {
203    let t = Instant::now();
204    for p in paths {
205        let id = s.alloc_id();
206        s.queue(
207            OPEN,
208            &Enc::new()
209                .u32(id)
210                .str(p.as_bytes())
211                .u32(FXF_READ)
212                .u32(0)
213                .done(),
214        )
215        .await?;
216    }
217    s.flush().await?;
218
219    let mut handles = Vec::with_capacity(paths.len());
220    for _ in 0..paths.len() {
221        let r = s.recv().await?;
222        ensure!(r.kind == HANDLE, "open refused (reply type {})", r.kind);
223        handles.push(Dec::new(r.payload()).str().context("open handle")?.to_vec());
224    }
225    Ok((t.elapsed(), handles))
226}
227
228async fn batch_read(s: &mut Session, handles: &[Vec<u8>]) -> Result<(Duration, usize)> {
229    let t = Instant::now();
230    for h in handles {
231        let id = s.alloc_id();
232        s.queue(READ, &Enc::new().u32(id).str(h).u64(0).u32(READ_LEN).done())
233            .await?;
234    }
235    s.flush().await?;
236
237    let mut bytes = 0usize;
238    for _ in 0..handles.len() {
239        let r = s.recv().await?;
240        match r.kind {
241            DATA => bytes += Dec::new(r.payload()).str().map_or(0, |b| b.len()),
242            STATUS => {}
243            other => bail!("read gave reply type {other}"),
244        }
245    }
246    Ok((t.elapsed(), bytes))
247}
248
249async fn batch_close(s: &mut Session, handles: &[Vec<u8>]) -> Result<()> {
250    for h in handles {
251        let id = s.alloc_id();
252        s.queue(CLOSE, &Enc::new().u32(id).str(h).done()).await?;
253    }
254    s.flush().await?;
255    for _ in 0..handles.len() {
256        s.recv().await?;
257    }
258    Ok(())
259}

Source

pub async fn recv(&mut self) -> Result<Reply>

Examples found in repository ?

examples/measure-roundtrips.rs (line 147)

139async fn measure_tau(s: &mut Session) -> Result<Duration> {
140    let mut samples = Vec::with_capacity(TAU_REPS);
141    for _ in 0..TAU_REPS {
142        let id = s.alloc_id();
143        let t = Instant::now();
144        s.queue(REALPATH, &Enc::new().u32(id).str(b".").done())
145            .await?;
146        s.flush().await?;
147        let r = s.recv().await?;
148        ensure!(r.id == id, "reply id {} does not match request {id}", r.id);
149        samples.push(t.elapsed());
150    }
151    samples.sort_unstable();
152    Ok(samples[samples.len() / 2])
153}
154
155/// One READDIR sweep carries every entry's attrs, which is what lets the origin
156/// layer skip per-file STAT entirely and keep invariant 1 within reach.
157async fn list(s: &mut Session, dir: &str) -> Result<Vec<(String, Attrs)>> {
158    let id = s.alloc_id();
159    s.queue(OPENDIR, &Enc::new().u32(id).str(dir.as_bytes()).done())
160        .await?;
161    s.flush().await?;
162    let r = s.recv().await?;
163    ensure!(
164        r.kind == HANDLE,
165        "opendir {dir} refused (reply type {})",
166        r.kind
167    );
168    let handle = Dec::new(r.payload())
169        .str()
170        .context("opendir handle")?
171        .to_vec();
172
173    let mut out = Vec::new();
174    loop {
175        let id = s.alloc_id();
176        s.queue(READDIR, &Enc::new().u32(id).str(&handle).done())
177            .await?;
178        s.flush().await?;
179        let r = s.recv().await?;
180        if r.kind == STATUS {
181            break;
182        }
183        ensure!(r.kind == NAME, "readdir gave reply type {}", r.kind);
184        let mut d = Dec::new(r.payload());
185        let count = d.u32().context("readdir count")?;
186        for _ in 0..count {
187            let name = String::from_utf8_lossy(d.str().context("filename")?).into_owned();
188            d.str().context("longname")?;
189            let attrs = Attrs::decode(&mut d).context("attrs")?;
190            out.push((name, attrs));
191        }
192    }
193
194    let id = s.alloc_id();
195    s.queue(CLOSE, &Enc::new().u32(id).str(&handle).done())
196        .await?;
197    s.flush().await?;
198    s.recv().await?;
199    Ok(out)
200}
201
202async fn batch_open(s: &mut Session, paths: &[String]) -> Result<(Duration, Vec<Vec<u8>>)> {
203    let t = Instant::now();
204    for p in paths {
205        let id = s.alloc_id();
206        s.queue(
207            OPEN,
208            &Enc::new()
209                .u32(id)
210                .str(p.as_bytes())
211                .u32(FXF_READ)
212                .u32(0)
213                .done(),
214        )
215        .await?;
216    }
217    s.flush().await?;
218
219    let mut handles = Vec::with_capacity(paths.len());
220    for _ in 0..paths.len() {
221        let r = s.recv().await?;
222        ensure!(r.kind == HANDLE, "open refused (reply type {})", r.kind);
223        handles.push(Dec::new(r.payload()).str().context("open handle")?.to_vec());
224    }
225    Ok((t.elapsed(), handles))
226}
227
228async fn batch_read(s: &mut Session, handles: &[Vec<u8>]) -> Result<(Duration, usize)> {
229    let t = Instant::now();
230    for h in handles {
231        let id = s.alloc_id();
232        s.queue(READ, &Enc::new().u32(id).str(h).u64(0).u32(READ_LEN).done())
233            .await?;
234    }
235    s.flush().await?;
236
237    let mut bytes = 0usize;
238    for _ in 0..handles.len() {
239        let r = s.recv().await?;
240        match r.kind {
241            DATA => bytes += Dec::new(r.payload()).str().map_or(0, |b| b.len()),
242            STATUS => {}
243            other => bail!("read gave reply type {other}"),
244        }
245    }
246    Ok((t.elapsed(), bytes))
247}
248
249async fn batch_close(s: &mut Session, handles: &[Vec<u8>]) -> Result<()> {
250    for h in handles {
251        let id = s.alloc_id();
252        s.queue(CLOSE, &Enc::new().u32(id).str(h).done()).await?;
253    }
254    s.flush().await?;
255    for _ in 0..handles.len() {
256        s.recv().await?;
257    }
258    Ok(())
259}