Merge pull request #429 from samoht/merge

Fix merge for concurrent but idempotent updates

src/ir_contents.ml

@@ -16,18 +16,27 @@
 
 open Lwt.Infix
 
+let merge_state dt =
+  let (=) = Ir_type.equal dt in
+  let default = Ir_merge.default dt in
+  let f ~old x y =
+    if x = y then Ir_merge.ok x
+    else Ir_merge.f default ~old x y
+  in
+  Ir_merge.v dt f
+
 module String = struct
   type t = string
   let t = Ir_type.string
-  let merge = Ir_merge.default Ir_type.(option string)
+  let merge = merge_state Ir_type.(option string)
   let pp = Fmt.string
   let of_string s = Ok s
 end
 
 module Cstruct = struct
   type t = Cstruct.t
-  let t = Ir_type.(like string) (fun x -> Cstruct.of_string x) Cstruct.to_string
-  let merge = Ir_merge.default Ir_type.(option t)
+  let t = Ir_type.cstruct
+  let merge = merge_state Ir_type.(option t)
   let pp = Ir_type.dump t
   let of_string s = Ok (Cstruct.of_string s)
 end

src/ir_type.ml

@@ -78,6 +78,7 @@ and 'a prim =
   | Int64  : int64 prim
   | Float  : float prim
   | String : string prim
+  | Cstruct: Cstruct.t prim
 
 and 'a tuple =
   | Pair   : 'a t * 'b t -> ('a * 'b) tuple
@@ -178,6 +179,7 @@ let int32 = Prim Int32
 let int64 = Prim Int64
 let float = Prim Float
 let string = Prim String
+let cstruct = Prim Cstruct
 
 let list l = List l
 let array a = Array a
@@ -300,6 +302,7 @@ module Dump = struct
   let pair = Fmt.Dump.pair
   let triple a b c ppf (x, y, z) = Fmt.pf ppf "(%a, %a, %a)" a x b y c z
   let option = Fmt.Dump.option
+  let cstruct = Cstruct.hexdump_pp
 
   let rec t: type a. a t -> a Fmt.t = function
     | Self s    -> t s.self
@@ -328,6 +331,7 @@ module Dump = struct
     | Int64  -> int64
     | Float  -> float
     | String -> string
+    | Cstruct -> cstruct
 
   and record: type a. a record -> a Fmt.t = fun r ppf x ->
     let fields = fields r in
@@ -362,6 +366,7 @@ module Equal = struct
   let int32 (x:int32) (y:int32) = x = y
   let int64 (x:int64) (y:int64) = x = y
   let string x y = x == y || String.compare x y = 0
+  let cstruct = Cstruct.equal
 
   (* NOTE: equality is ill-defined on float *)
   let float (x:float) (y:float) =  x = y
@@ -417,6 +422,7 @@ module Equal = struct
     | Int64  -> int64
     | Float  -> float
     | String -> string
+    | Cstruct -> cstruct
 
   and record: type a. a record -> a equal = fun r x y ->
     List.for_all (function Field f -> field f x y) (fields r)
@@ -455,6 +461,7 @@ module Compare = struct
   let int64 = Int64.compare
   let float (x:float) (y:float) = Pervasives.compare x y
   let string x y = if x == y then 0 else String.compare x y
+  let cstruct = Cstruct.compare
 
   let list c x y =
     if x == y then 0 else
@@ -529,6 +536,7 @@ module Compare = struct
     | Int64  -> int64
     | Float  -> float
     | String -> string
+    | Cstruct -> cstruct
 
   and record: type a. a record -> a compare = fun r x y ->
     let rec aux = function
@@ -577,6 +585,10 @@ module Encode_json = struct
   let int e i = float e (float_of_int i)
   let bool e = function false -> float e 0. | _ -> float e 1.
 
+  let cstruct e c =
+    let `Hex hex = Hex.of_cstruct c in
+    lexeme e (`String hex)
+
   let list l e x =
     lexeme e `As;
     List.iter (l e) x;
@@ -631,6 +643,7 @@ module Encode_json = struct
     | Int64  -> int64
     | Float  -> float
     | String -> string
+    | Cstruct -> cstruct
 
   and record: type a. a record -> a encode_json = fun r e x ->
     let fields = fields r in
@@ -750,6 +763,11 @@ module Decode_json = struct
     | `String s -> Ok s
     | l         -> error e l "`String"
 
+  let cstruct e =
+    lexeme e >>= function
+    | `String s -> Ok (Hex.to_cstruct (`Hex s))
+    | l         -> error e l "`String"
+
   let float e =
     lexeme e >>= function
     | `Float f -> Ok f
@@ -828,6 +846,7 @@ module Decode_json = struct
     | Int64  -> int64
     | Float  -> float
     | String -> string
+    | Cstruct -> cstruct
 
   and record: type a. a record -> a decode = fun r e ->
     expect_lexeme e `Os >>= fun () ->

src/ir_type.mli

@@ -25,6 +25,7 @@ val int32: int32 t
 val int64: int64 t
 val float: float t
 val string: string t
+val cstruct: Cstruct.t t
 val list: 'a t -> 'a list t
 val array: 'a t -> 'a array t
 val option: 'a t -> 'a option t

src/irmin.mli

@@ -89,6 +89,9 @@ module Type: sig
   val string: string t
   (** [string] is a representation of the string type. *)
 
+  val cstruct: Cstruct.t t
+  (** [cstruct] is a representation of the [Cstruct.t] type. *)
+
   val list: 'a t -> 'a list t
   (** [list t] is a representation of list of values of type [t]. *)
 

test/test_store.ml

@@ -1352,15 +1352,18 @@ module Make (S: Test_S) = struct
     merge_exn "merge unrelated"
 
   let rec write fn = function
-    | 0 -> Lwt.return_unit
-    | i -> (fn i >>= Lwt_unix.yield) <&> write fn (i-1)
+    | 0 -> []
+    | i -> (fun () -> fn i >>= Lwt_unix.yield) :: write fn (i-1)
+
+  let perform l = Lwt_list.iter_p (fun f -> f ()) l
 
   let rec read fn check = function
-    | 0 -> Lwt.return_unit
+    | 0 -> []
     | i ->
-      fn i >>= fun v ->
-      check i v;
-      read fn check (i-1)
+      (fun () ->
+         fn i >|= fun v ->
+         check i v)
+      :: read fn check (i-1)
 
   let test_concurrent_low x () =
     let test_branches repo =
@@ -1372,8 +1375,8 @@ module Make (S: Test_S) = struct
           (fun _i -> S.Branch.find repo k >|= get)
           (fun i  -> check (S.commit_t repo) (Fmt.strf "tag %d" i) v)
       in
-      write 1 >>= fun () ->
-      Lwt.join [ write 10; read 10; write 10; read 10; ]
+      perform (write 1) >>= fun () ->
+      perform (write 10 @ read 10 @  write 10 @ read 10)
     in
     let test_contents repo =
       kv2 ~repo >>= fun k ->
@@ -1387,8 +1390,8 @@ module Make (S: Test_S) = struct
           (fun _i -> P.Contents.find t k >|= get)
           (fun i  -> check S.contents_t (Fmt.strf "contents %d" i) v)
       in
-      write 1 >>= fun () ->
-      Lwt.join [ write 10; read 10; write 10; read 10; ]
+      perform (write 1) >>= fun () ->
+      perform (write 10 @ read 10 @ write 10 @ read 10)
     in
     run x (fun repo -> Lwt.join [test_branches repo; test_contents repo])
 
@@ -1404,8 +1407,8 @@ module Make (S: Test_S) = struct
           (fun _ -> S.get t k)
           (fun i -> check S.contents_t (Fmt.strf "update: one %d" i) v)
       in
-      Lwt.join [ write t1 10; write t2 10 ] >>= fun () ->
-      Lwt.join [ read t1 10 ]
+      perform (write t1 10 @ write t2 10) >>= fun () ->
+      perform (read t1 10)
     in
     let test_multi repo =
       let k i = ["a";"b";"c"; string_of_int i ] in
@@ -1420,8 +1423,8 @@ module Make (S: Test_S) = struct
           (fun i -> S.get t (k i))
           (fun i -> check S.contents_t (Fmt.strf "update: multi %d" i) (v i))
       in
-      Lwt.join [ write t1 10; write t2 10 ] >>= fun () ->
-      Lwt.join [ read t1 10 ]
+      perform (write t1 10 @ write t2 10) >>= fun () ->
+      perform (read t1 10)
     in
     run x (fun repo ->
         test_one   repo >>= fun () ->
@@ -1451,8 +1454,8 @@ module Make (S: Test_S) = struct
           (fun i -> check S.contents_t (Fmt.strf "update: multi %d" i) (v i))
       in
       S.set t1 ~info:(infof "update") (k 0) (v 0) >>= fun () ->
-      Lwt.join [ write t1 1 10; write t2 2 10 ] >>= fun () ->
-      Lwt.join [ read t1 10 ]
+      perform (write t1 1 10 @ write t2 2 10) >>= fun () ->
+      perform (read t1 10)
     in
     run x test
 
@@ -1489,8 +1492,8 @@ module Make (S: Test_S) = struct
           (fun i -> check S.contents_t (Fmt.strf "update: multi %d" i) (v i))
       in
       S.set t1 ~info:(infof "update") (k 0) (v 0) >>= fun () ->
-      Lwt.join [ write t1 1 5; write t2 2 5 ] >>= fun () ->
-      Lwt.join [ read t1 5 ]
+      perform (write t1 1 5 @ write t2 2 5) >>= fun () ->
+      perform (read t1 5)
     in
     run x test