Merge pull request #609 from samoht/batch

backends: distinguish between read and writes operations

src/irmin-chunk/irmin_chunk.ml

@@ -122,9 +122,9 @@ struct
   type key = CA.key
   type value = V.t
 
-  type t = {
+  type 'a t = {
     chunking    : [`Max | `Best_fit];
-    db          : CA.t;             (* An handler to the underlying database. *)
+    db          : 'a CA.t;          (* An handler to the underlying database. *)
     chunk_size  : int;                                 (* the size of chunks. *)
     max_children: int;     (* the maximum number of children a node can have. *)
     max_data    : int; (* the maximum length (in bytes) of data stored in one
@@ -205,6 +205,8 @@ struct
     CA.v config >|= fun db ->
     { chunking; db; chunk_size; max_children; max_data }
 
+  let batch t f = CA.batch t.db (fun db -> f { t with db })
+
   let find_leaves t key =
     AO.find t.db key >>= function
     | None    -> Lwt.return []

src/irmin-fs/irmin_fs.ml

@@ -56,7 +56,7 @@ struct
 
   type value = V.t
 
-  type t = {
+  type 'a t = {
     path: string;
   }
 
@@ -69,6 +69,9 @@ struct
     IO.mkdir path >|= fun () ->
     { path }
 
+  let cast t = (t :> [`Read | `Write] t)
+  let batch t f = f (cast t)
+
   let file_of_key { path; _ } key =
     path / S.file_of_key (Irmin.Type.to_string K.t key)
 
@@ -149,7 +152,7 @@ struct
   module RO = Read_only_ext(IO)(S)(K)(V)
   module W = Irmin.Private.Watch.Make(K)(V)
 
-  type t = { t: RO.t; w: W.t }
+  type t = { t: unit RO.t; w: W.t }
   type key = RO.key
   type value = RO.value
   type watch = W.watch * (unit -> unit Lwt.t)

src/irmin-git/irmin_git.ml

@@ -101,7 +101,7 @@ module Make_private
 
   module Content_addressable (V: V) = struct
 
-    type t = G.t
+    type 'a t = G.t
     type key = H.t
     type value = V.t
 
@@ -847,6 +847,8 @@ module Make_ext
       let node_t t = contents_t t, t.g
       let commit_t t = node_t t, t.g
 
+      let batch t f = f (contents_t t) (node_t t) (commit_t t)
+
       type config = {
         root   : string;
         dot_git: string option;
@@ -984,7 +986,7 @@ module Content_addressable (G: Git.S) (V: Irmin.Type.S) = struct
   let state t =
     M.repo_of_git t >|= fun r ->
     M.Private.Repo.contents_t r
-  type t = G.t
+  type 'a t = G.t
   type key = X.key
   type value = X.value
   let with_state f t x = state t >>= fun t -> f t x

src/irmin-git/irmin_git.mli

@@ -33,7 +33,7 @@ val dot_git: string option Irmin.Private.Conf.key
 
 module Content_addressable (G: Git.S) (V: Irmin.Type.S)
   : Irmin.CONTENT_ADDRESSABLE_STORE
-    with type t = G.t
+    with type 'a t = G.t
      and type key = G.Hash.t
      and type value = V.t
 

src/irmin-http/irmin_http.ml

@@ -193,7 +193,7 @@ struct
 
   module HTTP = Helper (Client)
 
-  type t = { uri: Uri.t; item: string; items: string; ctx: Client.ctx option }
+  type 'a t = { uri: Uri.t; item: string; items: string; ctx: Client.ctx option }
   let uri t = t.uri
   let item t = t.item
   let items t = t.items
@@ -216,6 +216,12 @@ struct
         if Cohttp.Response.status r = `Not_found then Lwt.return_false
         else Lwt.return_true)
 
+  let cast t = (t :> [`Read|`Write] t)
+
+  let batch t f =
+    (* TODO:cache the writes locally and send everything in one batch *)
+    f (cast t)
+
   let v ?ctx uri item items =
     Lwt.return { uri; item; items; ctx }
 
@@ -250,7 +256,7 @@ module RW (Client: Cohttp_lwt.S.Client)
 
   let empty_cache () = { stop = fun () -> (); }
 
-  type t = { t: RO.t; w: W.t; keys: cache; glob: cache }
+  type t = { t: unit RO.t; w: W.t; keys: cache; glob: cache }
 
   let get t = HTTP.call `GET (RO.uri t.t) t.t.ctx
   let put t = HTTP.call `PUT (RO.uri t.t) t.t.ctx
@@ -385,13 +391,15 @@ module Make
 struct
   module X = struct
     module Hash = H
-    module XContents = struct
-      module Key = H
-      module Val = C
-      include AO(Client)(H)(C)
-      let v ?ctx config = v ?ctx config "blob" "blobs"
+    module Contents = struct
+      module X = struct
+        module Key = H
+        module Val = C
+        include AO(Client)(H)(C)
+      end
+      include Irmin.Contents.Store(X)
+      let v ?ctx config = X.v ?ctx config "blob" "blobs"
     end
-    module Contents = Irmin.Contents.Store(XContents)
     module Node = struct
       module X = struct
         module Key = H
@@ -421,20 +429,28 @@ struct
     module Repo = struct
       type t = {
         config: Irmin.config;
-        contents: Contents.t;
-        node: Node.t;
-        commit: Commit.t;
+        contents: [`Read] Contents.t;
+        node: [`Read] Node.t;
+        commit: [`Read] Commit.t;
         branch: Branch.t;
       }
       let branch_t t = t.branch
       let commit_t t = t.commit
       let node_t t = t.node
       let contents_t t = t.contents
 
+      let batch t f =
+        Contents.X.batch t.contents @@ fun contents_t ->
+        Node.X.batch (snd t.node) @@ fun node_t ->
+        Commit.X.batch (snd t.commit) @@ fun commit_t ->
+        let node_t = contents_t, node_t in
+        let commit_t = node_t, commit_t in
+        f contents_t node_t commit_t
+
       let v config =
         let uri = get_uri config in
         let ctx = Client.ctx () in
-        XContents.v ?ctx uri >>= fun contents ->
+        Contents.v ?ctx uri >>= fun contents ->
         Node.v ?ctx uri      >>= fun node ->
         Commit.v ?ctx uri    >>= fun commit ->
         Branch.v ?ctx uri    >|= fun branch ->

src/irmin-http/irmin_http_server.ml

@@ -56,7 +56,11 @@ module Make (HTTP: Cohttp_lwt.S.Server) (S: Irmin.S) = struct
     let err = Fmt.strf "Parse error %S: %s" str e in
     Wm.respond ~body:(`String err) 400 rd
 
-  module Content_addressable (S: Irmin.CONTENT_ADDRESSABLE_STORE)
+  module Content_addressable
+      (S: sig
+         include Irmin.CONTENT_ADDRESSABLE_STORE
+         val batch: P.Repo.t -> ([`Read|`Write] t -> 'a Lwt.t) -> 'a Lwt.t
+       end)
       (K: Irmin.Type.S with type t = S.key)
       (V: Irmin.Type.S with type t = S.value) =
   struct
@@ -66,7 +70,7 @@ module Make (HTTP: Cohttp_lwt.S.Server) (S: Irmin.S) = struct
       | Ok key  -> f key
       | Error _ -> Wm.respond 404 rd
 
-    class items db = object
+    class items repo = object
       inherit resource
       method! allowed_methods rd = Wm.continue [`POST] rd
 
@@ -82,6 +86,7 @@ module Make (HTTP: Cohttp_lwt.S.Server) (S: Irmin.S) = struct
         match Irmin.Type.of_string V.t body with
         | Error e -> parse_error rd body e
         | Ok body ->
+          S.batch repo @@ fun db ->
           S.add db body >>= fun new_id ->
           let resp_body = `String (Irmin.Type.to_string K.t new_id) in
           Wm.continue true { rd with Wm.Rd.resp_body }
@@ -289,9 +294,21 @@ module Make (HTTP: Cohttp_lwt.S.Server) (S: Irmin.S) = struct
 
   end
 
-  module Blob = Content_addressable(P.Contents)(P.Contents.Key)(P.Contents.Val)
-  module Tree = Content_addressable(P.Node)(P.Node.Key)(P.Node.Val)
-  module Commit = Content_addressable(P.Commit)(P.Commit.Key)(P.Commit.Val)
+  module Blob = Content_addressable(struct
+      include P.Contents
+      let batch t f = P.Repo.batch t @@ fun x _ _ -> f x
+    end)(P.Contents.Key)(P.Contents.Val)
+
+  module Tree = Content_addressable(struct
+      include P.Node
+      let batch t f = P.Repo.batch t @@ fun _ x _ -> f x
+    end)(P.Node.Key)(P.Node.Val)
+
+  module Commit = Content_addressable(struct
+      include P.Commit
+      let batch t f = P.Repo.batch t @@ fun _ _ x -> f x
+    end)(P.Commit.Key)(P.Commit.Val)
+
   module Branch = Atomic_write(P.Branch)(P.Branch.Key)(P.Branch.Val)
 
   type repo = S.Repo.t
@@ -303,11 +320,11 @@ module Make (HTTP: Cohttp_lwt.S.Server) (S: Irmin.S) = struct
     let commit = P.Repo.commit_t db in
     let branch = P.Repo.branch_t db in
     let routes = [
-      ("/blobs"     , fun () -> new Blob.items     blob);
+      ("/blobs"     , fun () -> new Blob.items     db);
       ("/blob/:id"  , fun () -> new Blob.item      blob);
-      ("/trees"     , fun () -> new Tree.items     tree);
+      ("/trees"     , fun () -> new Tree.items     db);
       ("/tree/:id"  , fun () -> new Tree.item      tree);
-      ("/commits"   , fun () -> new Commit.items   commit);
+      ("/commits"   , fun () -> new Commit.items   db);
       ("/commit/:id", fun () -> new Commit.item    commit);
       ("/branches"  , fun () -> new Branch.items   branch);
       ("/branch/*"  , fun () -> new Branch.item    branch);

src/irmin-mem/irmin_mem.ml

@@ -28,10 +28,13 @@ module Read_only (K: Irmin.Type.S) (V: Irmin.Type.S) = struct
 
   type key = K.t
   type value = V.t
-  type t = { mutable t: value KMap.t }
+  type 'a t = { mutable t: value KMap.t }
   let map = { t = KMap.empty }
   let v _config = Lwt.return map
 
+  let cast t = (t :> [`Read | `Write] t)
+  let batch t f = f (cast t)
+
   let pp_key = Irmin.Type.pp K.t
 
   let find { t; _ } key =
@@ -62,7 +65,7 @@ module Atomic_write (K: Irmin.Type.S) (V: Irmin.Type.S) = struct
   module W = Irmin.Private.Watch.Make(K)(V)
   module L = Irmin.Private.Lock.Make(K)
 
-  type t = { t: RO.t; w: W.t; lock: L.t }
+  type t = { t: unit RO.t; w: W.t; lock: L.t }
   type key = RO.key
   type value = RO.value
   type watch = W.watch