Add eio backend

CHANGES.md

@@ -6,6 +6,10 @@
 - Update to cmdliner.1.1.0 (#382, @MisterDA)
 - Mirage support: optional dependency to unix (#396, @art-w)
 
+## Added
+
+- Add `index.eio` with Eio backend (#397, @art-w)
+
 # 1.6.2 (2023-06-06)
 
 ## Changed

bench/bench.ml

@@ -214,7 +214,7 @@ module Index = struct
 
   let run ~with_metrics ~nb_entries ~log_size ~root ~name ~fresh ~readonly b =
     let index =
-      Index.v ~cache:(Index.empty_cache ()) ~fresh ~readonly ~log_size
+      Index.v ~io:() ~cache:(Index.empty_cache ()) ~fresh ~readonly ~log_size
         (root // name)
     in
     let result = Benchmark.run ~nb_entries (b ~with_metrics index) in

bench/replay.ml

@@ -164,7 +164,10 @@ module Index = struct
   include Index
 
   let cache = Index.empty_cache ()
-  let v root = Index.v ~cache ~readonly:false ~fresh:true ~log_size:500_000 root
+
+  let v root =
+    Index.v ~io:() ~cache ~readonly:false ~fresh:true ~log_size:500_000 root
+
   let close t = Index.close t
 end
 

index.opam

@@ -36,7 +36,15 @@ depends: [
   "crowbar"  {with-test & >= "0.2"}
   "re"       {with-test}
   "lru"      {>= "0.3.0"}
+  "eio"      {with-test & >= "0.14" }
+  "eio_main" {with-test & >= "0.14" }
 ]
+
+depopts: [
+  "eio"      {>= "0.14" }
+  "eio_main" {>= "0.14" }
+]
+
 synopsis: "A platform-agnostic multi-level index for OCaml"
 description:"""
 Index is a scalable implementation of persistent indices in OCaml.

src/checks.ml

@@ -42,6 +42,8 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
     @@ pos 0 (some string) None
     @@ info ~doc:"Path to the Index store on disk" ~docv:"PATH" []
 
+  type io = Platform.IO.io
+
   module Stat = struct
     type io = {
       size : size;
@@ -62,16 +64,17 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
 
     type t = { entry_size : size; files : files } [@@deriving repr]
 
-    let with_io : type a. string -> (IO.t -> a) -> a option =
-     fun path f ->
-      match IO.v path with
+    let with_io : type a. io:Platform.IO.io -> string -> (IO.t -> a) -> a option
+        =
+     fun ~io path f ->
+      match IO.v ~io path with
       | Error `No_file_on_disk -> None
       | Ok io ->
           let a = f io in
           IO.close io;
           Some a
 
-    let io path =
+    let run_io path =
       with_io path @@ fun io ->
       let IO.Header.{ offset; generation } = IO.Header.get io in
       let fanout_size = Bytes (IO.get_fanout_size io) in
@@ -80,12 +83,12 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
       let generation = Int63.to_int64 generation in
       { size; offset; generation; fanout_size }
 
-    let run ~root =
+    let run ~io ~root =
       Logs.app (fun f -> f "Getting statistics for store: `%s'@," root);
-      let data = io (Layout.data ~root) in
-      let log = io (Layout.log ~root) in
-      let log_async = io (Layout.log_async ~root) in
-      let merge = io (Layout.merge ~root) in
+      let data = run_io ~io (Layout.data ~root) in
+      let log = run_io ~io (Layout.log ~root) in
+      let log_async = run_io ~io (Layout.log_async ~root) in
+      let merge = run_io ~io (Layout.merge ~root) in
       let lock =
         IO.Lock.pp_dump (Layout.lock ~root)
         |> Option.map (fun f ->
@@ -99,7 +102,7 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
       }
       |> Repr.pp_json ~minify:false t Fmt.stdout
 
-    let term = Cmdliner.Term.(const (fun root () -> run ~root) $ path)
+    let term ~io = Cmdliner.Term.(const (fun root () -> run ~io ~root) $ path)
   end
 
   module Integrity_check = struct
@@ -120,9 +123,9 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
              highlight (fun ppf () -> (Repr.pp Entry.t) ppf entry))
       |> Fmt.(concat ~sep:cut)
 
-    let run ~root =
+    let run ~io ~root =
       let context = 2 in
-      match IO.v (Layout.data ~root) with
+      match IO.v ~io (Layout.data ~root) with
       | Error `No_file_on_disk -> Fmt.failwith "No data file in %s" root
       | Ok io ->
           let io_offset = IO.offset io in
@@ -151,7 +154,7 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
                            ());
                    previous := e)
 
-    let term = Cmdliner.Term.(const (fun root () -> run ~root) $ path)
+    let term ~io = Cmdliner.Term.(const (fun root () -> run ~io ~root) $ path)
   end
 
   module Cli = struct
@@ -166,7 +169,7 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
       in
       Logs_fmt.reporter ~pp_header ()
 
-    let main () : empty =
+    let main ~io () : empty =
       let default = Term.(ret (const (`Help (`Auto, None)))) in
       let info =
         let doc = "Check and repair Index data-stores." in
@@ -175,12 +178,12 @@ module Make (K : Data.Key) (V : Data.Value) (Platform : Platform_args) = struct
       let commands =
         [
           ( Term.(
-              Stat.term
+              Stat.term ~io
               $ Log.setup_term ~reporter (module Clock) (module Fmt_tty)),
             Cmd.info ~doc:"Print high-level statistics about the store." "stat"
           );
           ( Term.(
-              Integrity_check.term
+              Integrity_check.term ~io
               $ Log.setup_term ~reporter (module Clock) (module Fmt_tty)),
             Cmd.info
               ~doc:"Search the store for integrity faults and corruption."

src/checks_intf.ml

@@ -3,24 +3,26 @@ open! Import
 type empty = |
 
 module type S = sig
+  type io
+
   module Stat : sig
-    val run : root:string -> unit
+    val run : io:io -> root:string -> unit
     (** Read basic metrics from an existing store. *)
 
-    val term : (unit -> unit) Cmdliner.Term.t
+    val term : io:io -> (unit -> unit) Cmdliner.Term.t
     (** A pre-packaged [Cmdliner] term for executing {!run}. *)
   end
 
   module Integrity_check : sig
-    val run : root:string -> unit
+    val run : io:io -> root:string -> unit
     (** Check that the integrity invariants of a store are preserved, and
         display any broken invariants. *)
 
-    val term : (unit -> unit) Cmdliner.Term.t
+    val term : io:io -> (unit -> unit) Cmdliner.Term.t
     (** A pre-packaged [Cmdliner] term for executing {!run}. *)
   end
 
-  val cli : unit -> empty
+  val cli : io:io -> unit -> empty
   (** Run a [Cmdliner] binary containing tools for running offline integrity
       checks. *)
 end
@@ -38,5 +40,6 @@ module type Checks = sig
   module type S = S
   module type Platform_args = Platform_args
 
-  module Make (K : Data.Key) (V : Data.Value) (_ : Platform_args) : S
+  module Make (K : Data.Key) (V : Data.Value) (P : Platform_args) :
+    S with type io = P.IO.io
 end

src/eio/buffer.ml

@@ -0,0 +1,54 @@
+(* The MIT License
+
+   Copyright (c) 2021 Clément Pascutto <clement@tarides.com>
+
+   Permission is hereby granted, free of charge, to any person obtaining a copy
+   of this software and associated documentation files (the "Software"), to deal
+   in the Software without restriction, including without limitation the rights
+   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+   copies of the Software, and to permit persons to whom the Software is
+   furnished to do so, subject to the following conditions:
+
+   The above copyright notice and this permission notice shall be included in
+   all copies or substantial portions of the Software. *)
+
+open! Import
+
+type t = { mutable buffer : bytes; mutable position : int }
+
+external unsafe_blit_string : string -> int -> bytes -> int -> int -> unit
+  = "caml_blit_string"
+[@@noalloc]
+(** Bytes.unsafe_blit_string not available in OCaml 4.08. *)
+
+let create n = { buffer = Bytes.create n; position = 0 }
+
+let write_with (write : string -> int -> int -> unit) b =
+  write (Bytes.unsafe_to_string b.buffer) 0 b.position
+
+let length b = b.position
+let is_empty b = b.position = 0
+let clear b = b.position <- 0
+
+let resize b more =
+  let old_pos = b.position in
+  let old_len = Bytes.length b.buffer in
+  let new_len = ref old_len in
+  while old_pos + more > !new_len do
+    new_len := 2 * !new_len
+  done;
+  let new_buffer = Bytes.create !new_len in
+  Bytes.blit b.buffer 0 new_buffer 0 b.position;
+  b.buffer <- new_buffer
+
+let add_substring b s ~off ~len =
+  let new_position = b.position + len in
+  if new_position > Bytes.length b.buffer then resize b len;
+  unsafe_blit_string s off b.buffer b.position len;
+  b.position <- new_position
+
+let blit ~src ~src_off ~dst ~dst_off ~len =
+  assert (src_off + len <= src.position);
+  Bytes.blit src.buffer src_off dst dst_off len
+
+let add_string b s = add_substring b s ~off:0 ~len:(String.length s)

src/eio/buffer.mli

@@ -0,0 +1,46 @@
+(* The MIT License
+
+   Copyright (c) 2021 Clément Pascutto <clement@tarides.com>
+
+   Permission is hereby granted, free of charge, to any person obtaining a copy
+   of this software and associated documentation files (the "Software"), to deal
+   in the Software without restriction, including without limitation the rights
+   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+   copies of the Software, and to permit persons to whom the Software is
+   furnished to do so, subject to the following conditions:
+
+   The above copyright notice and this permission notice shall be included in
+   all copies or substantial portions of the Software. *)
+
+(** Extensible buffers with non-allocating access to the buffer's contents. *)
+
+type t
+(** The type of buffers. *)
+
+val create : int -> t
+(** [create n] is a fresh buffer with initial size [n]. *)
+
+val length : t -> int
+(** [length b] is the number of bytes contained in the buffer. *)
+
+val is_empty : t -> bool
+(** [is_empty t] iff [t] contains 0 characters. *)
+
+val clear : t -> unit
+(** [clear t] clears the data contained in [t]. It does not reset the buffer to
+    its initial size. *)
+
+val add_substring : t -> string -> off:int -> len:int -> unit
+(** [add_substring t s ~off ~len] appends the substring
+    [s.(off) .. s.(off + len - 1)] at the end of [t], resizing [t] if necessary. *)
+
+val add_string : t -> string -> unit
+(** [add_string t s] appends [s] at the end of [t], resizing [t] if necessary. *)
+
+val write_with : (string -> int -> int -> unit) -> t -> unit
+(** [write_with writer t] uses [writer] to write the contents of [t]. [writer]
+    takes a string to write, an offset and a length. *)
+
+val blit : src:t -> src_off:int -> dst:bytes -> dst_off:int -> len:int -> unit
+(** [blit] copies [len] bytes from the buffer [src], starting at offset
+    [src_off], into the supplied bytes [dst], starting at offset [dst_off]. *)

src/eio/dune

@@ -0,0 +1,6 @@
+(library
+ (public_name index.eio)
+ (name index_eio)
+ (optional)
+ (libraries fmt fmt.tty index logs logs.threaded threads.posix unix eio
+   eio.core cstruct mtime mtime.clock.os optint progress))

src/eio/import.ml

@@ -0,0 +1,10 @@
+module Int63 = Optint.Int63
+
+type int63 = Int63.t
+
+module Mtime = struct
+  include Mtime
+
+  let span_to_s span = Mtime.Span.to_float_ns span *. 1e-9
+  let span_to_us span = Mtime.Span.to_float_ns span *. 1e-3
+end