Optimise Flow.copy with Buf_read.as_flow

By default, Flow.copy creates a 4KB buffer and copies the data through
that. However, if the source of the copy is a buffered reader then it is
much more efficient to use its buffer directly.

This updates the flow you get from `Buf_read.as_flow` to offer this
optimisation, and also updates the eio_posix backend's flow to use this
optimisation when available (eio_linux already supported this).

Detected in a benchmark by Leandro Ostera.

bench/bench_copy.ml

@@ -1,3 +1,5 @@
+(* A client opens a connection to an echo service and sends a load of data via it. *)
+
 open Eio.Std
 
 let chunk_size = 1 lsl 16
@@ -31,7 +33,7 @@ let time name service =
   let time = t1 -. t0 in
   let bytes_per_second = float n_bytes /. time in
   traceln "%s: %.2f MB/s" name (bytes_per_second /. 1024. /. 1024.);
-  (Metric.create name (`Float bytes_per_second) "bytes/s" (name ^ " Flow.copy"))
+  Metric.create name (`Float bytes_per_second) "bytes/s" (name ^ " Flow.copy")
 
 let run _env =
   [

lib_eio/buf_read.ml

@@ -140,7 +140,13 @@ module F = struct
     consume t len;
     len
 
-  let read_methods = []
+  let rsb t fn =
+    ensure t 1;
+    let data = peek t in
+    let sent = fn [data] in
+    consume t sent
+
+  let read_methods = [Flow.Read_source_buffer rsb]
 end
 
 let as_flow =

lib_eio_posix/flow.ml

@@ -64,7 +64,22 @@ module Impl = struct
     with Unix.Unix_error (code, name, arg) ->
       raise (Err.wrap code name arg)
 
-  let copy t ~src = Eio.Flow.Pi.simple_copy ~single_write t ~src
+  (* Copy using the [Read_source_buffer] optimisation.
+     Avoids a copy if the source already has the data. *)
+  let copy_with_rsb rsb dst =
+    try
+      while true do rsb (single_write dst) done
+    with End_of_file -> ()
+
+  let copy t ~src =
+    let Eio.Resource.T (src_t, ops) = src in
+    let module Src = (val (Eio.Resource.get ops Eio.Flow.Pi.Source)) in
+    let rec aux = function
+      | Eio.Flow.Read_source_buffer rsb :: _ -> copy_with_rsb (rsb src_t) t
+      | _ :: xs -> aux xs
+      | [] -> Eio.Flow.Pi.simple_copy ~single_write t ~src
+    in
+    aux Src.read_methods
 
   let single_read t buf =
     match Low_level.readv t [| buf |] with