robur/git-kv
4
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Refactor test/dune and add a failing test #2

Merged
reynir merged 12 commits from batch-test into main 2024-10-29 11:21:16 +00:00
Conversation 15 Commits 12 Files changed 8 +16 -16
1 changed files with 16 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 3cae0f7765 - Show all commits

32
src/git_kv.ml
View file

@ -9,7 +9,9 @@ type t =
; branch : Git.Reference.t ; branch : Git.Reference.t
; store : Store.t ; store : Store.t
; mutable committed : Digestif.SHA1.t option ; mutable committed : Digestif.SHA1.t option
; mutable change_and_push_waiter : unit Lwt.t option ; mutable change_and_push_running : bool
; condition : unit Lwt_condition.t
; mutex : Lwt_mutex.t
reynir commented 2024-10-29 10:19:16 +00:00
Outdated
Review
Copy link

Could this be done with the mutex only instead of a bool, mutex and a condition?

Could this be done with the mutex only instead of a bool, mutex and a condition?
hannes commented 2024-10-29 10:24:21 +00:00
Outdated
Review
Copy link

Indeed, we can acquire the mutex at the begin of change_and_push -- which will wait (block) until the mutex is unused.

Indeed, we can acquire the mutex at the begin of change_and_push -- which will wait (block) until the mutex is unused.
dinosaure commented 2024-10-29 10:40:08 +00:00
Outdated
Review
Copy link

yes, basically protect the whole change_and_push with a mutex will solve all of our issues 😄

yes, basically protect the whole `change_and_push` with a mutex will solve all of our issues 😄
; mutable head : Store.hash option } ; mutable head : Store.hash option }
let init_store () = let init_store () =
@ -102,7 +104,7 @@ let connect ctx endpoint =
init_store () >>= fun store -> init_store () >>= fun store ->
let store = to_invalid store in let store = to_invalid store in
let edn, branch = split_url endpoint in let edn, branch = split_url endpoint in
let t = { ctx ; edn ; branch ; store ; committed= None; change_and_push_waiter= None; head= None } in let t = { ctx ; edn ; branch ; store ; committed= None; change_and_push_running= false; condition= Lwt_condition.create (); mutex= Lwt_mutex.create (); head= None } in
pull t >>= fun r -> pull t >>= fun r ->
let _r = to_invalid r in let _r = to_invalid r in
Lwt.return t Lwt.return t
@ -370,7 +372,7 @@ let of_octets ctx ~remote data =
Result.fold ~ok:Fun.id ~error:(function `Msg msg -> failwith msg) >>= fun store -> Result.fold ~ok:Fun.id ~error:(function `Msg msg -> failwith msg) >>= fun store ->
analyze store data >>= fun head -> analyze store data >>= fun head ->
let edn, branch = split_url remote in let edn, branch = split_url remote in
Lwt.return_ok { ctx ; edn ; branch ; store ; committed= None; change_and_push_waiter= None; head; }) Lwt.return_ok { ctx ; edn ; branch ; store ; committed= None; change_and_push_running= false; condition= Lwt_condition.create (); mutex= Lwt_mutex.create (); head; })
(fun _exn -> (fun _exn ->
Lwt.return_error (`Msg "Invalid PACK file")) Lwt.return_error (`Msg "Invalid PACK file"))
@ -684,12 +686,13 @@ module Make (Pclock : Mirage_clock.PCLOCK) = struct
[change_and_push_waiter] before we wait on the existing [change_and_push_waiter] before we wait on the existing
[change_and_push_waiter] task without any yield point in between to [change_and_push_waiter] task without any yield point in between to
ensure serializability. *) ensure serializability. *)
hannes commented 2024-10-29 08:53:01 +00:00
Outdated
Review
Copy link

the reason for this change is: we may have one change_and_push that is active, and when there are then multiple other change_and_push that should be executed, each needs to wait for the next one.

previously, the code had all other change_and_push wait for the same task -- and thus there may have been multiple waiting change_and_push waken up at the same time, leading to races.

the reason for this change is: we may have one change_and_push that is active, and when there are then multiple other change_and_push that should be executed, each needs to wait for the next one. previously, the code had all other change_and_push wait for the same task -- and thus there may have been multiple waiting change_and_push waken up at the same time, leading to races.
let th, wk = Lwt.wait () in let open Lwt.Syntax in
let th' = t.change_and_push_waiter in let* () = Lwt_mutex.with_lock t.mutex @@ fun () ->
t.change_and_push_waiter <- Some th; let rec await () =
hannes marked this conversation as resolved Outdated
dinosaure commented 2024-10-29 08:48:07 +00:00
Outdated
Review
Copy link

t.change_and_push is already set 3 lines before.

`t.change_and_push` is already set 3 lines before.
hannes commented 2024-10-29 08:51:09 +00:00
Outdated
Review
Copy link

indeed, removed in 750ec11

indeed, removed in 750ec11
( match th' with if t.change_and_push_running
| None -> Lwt.return_unit then Lwt_condition.wait ~mutex:t.mutex t.condition >>= await
| Some th -> th ) >>= fun () -> else begin t.change_and_push_running <- true; Lwt.return_unit end in
await () in
( let open Lwt_result.Infix in ( let open Lwt_result.Infix in
tree_root_hash_of_store t >>= fun tree_root_hash -> tree_root_hash_of_store t >>= fun tree_root_hash ->
let t' = { t with committed = Some tree_root_hash } in let t' = { t with committed = Some tree_root_hash } in
@ -722,12 +725,9 @@ module Make (Pclock : Mirage_clock.PCLOCK) = struct
>|= Result.map_error >|= Result.map_error
(fun err -> `Msg (Fmt.str "error pushing %a" Store.pp_error err)) (fun err -> `Msg (Fmt.str "error pushing %a" Store.pp_error err))
>>= fun res -> >>= fun res ->
Lwt.wakeup_later wk () ; let* () = Lwt_mutex.with_lock t.mutex @@ fun () ->
(* (hannes) since some other task may have mutated the t.change_and_push_running <- false;
change_and_push_waiter, we only reset it to None if there's a physical Lwt_condition.signal t.condition ();
equality between its value and our created task above. *) Lwt.return_unit in
(match t.change_and_push_waiter with
| Some th' -> if th' == th then t.change_and_push_waiter <- None
| None -> ());
Lwt.return res Lwt.return res
end end