rw test

lib/xapi-fd-test/generate.ml

@@ -105,7 +105,6 @@ let run_ro t data ~f =
   observe_ro write ~f t.kind data
 
 let run_wo t ~f =
-  (* we can only implement delays on write, skip *)
   CancellableSleep.with_ @@ fun cancel ->
   let finally () = CancellableSleep.cancel cancel in
   let f arg = Fun.protect ~finally (fun () -> f arg) in
@@ -117,3 +116,23 @@ let run_wo t ~f =
         read
   in
   observe_wo read ~f t.kind ~size:t.size
+
+let run_rw t data ~f =
+  CancellableSleep.with_ @@ fun cancel ->
+  let finally () = CancellableSleep.cancel cancel in
+  let f arg = Fun.protect ~finally (fun () -> f arg) in
+  let read =
+    match t.delay_read with
+    | Some delay ->
+        Delay.apply_read cancel delay read
+    | None ->
+        read
+  in
+  let write =
+    match t.delay_write with
+    | Some delay ->
+        Delay.apply_write cancel delay single_write_substring
+    | None ->
+        single_write_substring
+  in
+  observe_rw read write ~f t.kind ~size:t.size data

lib/xapi-fd-test/generate.mli

@@ -73,3 +73,15 @@ val run_wo :
 
  @returns observations about [f]'s actions on the file descriptor
 *)
+
+val run_rw :
+     t
+  -> string
+  -> f:((rdwr, kind) make -> 'a)
+  -> ([> rdonly] observation option, [> wronly] observation option) observations * 'a or_exn
+(** [run_rw t data ~f] creates a file descriptor according to [t] and calls the function under test [f].
+ The file descriptor should be treated as read-write.
+
+ @returns observations about [f]'s actions the file descriptor
+*)
+

lib/xapi-fd-test/observations.ml

@@ -282,3 +282,25 @@ let observe_wo read ~f ~size kind =
     read |> Option.map @@ fun read -> {read with data= Buffer.contents rd_buf}
   in
   ({write= (); read; elapsed}, res)
+
+let observe_rw read write ~f ~size kind expected =
+  with_kind_rw kind @@ fun rw1 rw2 ->
+  let written = Buffer.create 0 in
+  let rd_buf = Buffer.create 0 in
+  let gw fd =
+     let r = do_write write written expected 0 (as_writable_opt fd) in
+     close fd;
+     r
+  and gr fd =
+     do_read ~size read rd_buf (as_readable_opt fd)
+  in
+  let res, thread_result = concurrently (wrap_measure f, concurrently (gr, gw)) (rw1, (rw2,rw2)) in
+  let elapsed, res = unwrap_exn res in
+  let read, write = unwrap_exn thread_result in
+  let read =
+    read |> unwrap_exn |> Option.map @@ fun read -> {read with data= Buffer.contents rd_buf}
+  and write =
+    write |> unwrap_exn
+    |> Option.map @@ fun write -> {write with data= Buffer.contents written}
+  in
+  ({read; write; elapsed}, res)

lib/xapi-fd-test/observations.mli

@@ -180,3 +180,23 @@ val observe_wo :
 
   @returns an observation of [f]'s actions on the file descriptor and [f]'s result
  *)
+
+val observe_rw :
+   (([> readable], kind) Properties.t, bytes) operation
+  -> (([> writable], kind) Properties.t, string) operation
+  -> f:((rdwr, kind) make -> 'a)
+  -> size:int
+  -> Unix.file_kind
+  -> string
+  -> ([> rdonly] observation option, [> wronly] observation option) observations * 'a or_exn
+(** [observe_rw read write ~f ~size kind expected] generates a file descriptor of [kind] type,
+  and calls [f] with it.
+  It observes [f]'s actions from the other side of a pipe, socket, file descriptor,
+  or block device if possible.
+
+  @param read the operation used for reading, allows insertion of delays
+  @param write the operation used for writing, allows insertion of delays
+  @param expected the string to write to the file descriptor
+  @returns an observation of [f]'s actions on the file descriptor and [f]'s result
+ *)
+