Source file vc.ml

(*****************************************************************************)
(*                                                                           *)
(* Open Source License                                                       *)
(* Copyright (c) 2019,2020 DaiLambda, Inc. <contact@dailambda.jp>            *)
(*                                                                           *)
(* 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.                    *)
(*                                                                           *)
(* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR*)
(* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,  *)
(* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL   *)
(* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER*)
(* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING   *)
(* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER       *)
(* DEALINGS IN THE SOFTWARE.                                                 *)
(*                                                                           *)
(*****************************************************************************)

open Utils
open Node
open Cursor
open Result_lwt.Syntax

type t =
  { commit_db : Commit_db.t
  ; context : Context.t
  ; auto_flush_seconds : int
  }

let commit_db { commit_db ; _ } = commit_db
let context { context ; _ } = context

let empty t = Cursor.empty t.context

let check_prefix s =
  if s = "" then failwith "Path prefix cannot be empty"
  else if s.[String.length s - 1] = '/' then
    failwith "Path prefix cannot be a directory"
  else ()

let default_auto_flush_seconds =
  try
    let n = int_of_string @@ Sys.getenv "PLEBEIA_SYNC_SECONDS" in
    Log.fatal "Configuration: synchronize_s=%d" n;
    n
  with
  | Not_found -> 20
  | _ -> Stdlib.failwith "PLEBEIA_SYNC_SECONDS=<int>"

let create ?hashcons ?node_cache ?(lock=true)
    ?bytes_per_cell ?hash_func ?bytes_per_hash ?resize_step_bytes
    ?(auto_flush_seconds=default_auto_flush_seconds)
    path_prefix =
  check_prefix path_prefix;
  let*= () =
    if lock then
      let+= _ = Lock.lock @@ path_prefix ^ ".lock" in ()
    else Lwt.return_unit
  in
  let*= context =
    Context.create ?hashcons ?node_cache
      ?bytes_per_cell ?hash_func ?bytes_per_hash ?resize_step_bytes
      (path_prefix ^ ".context")
  in
  let storage_context = Context.get_storage context in
  let*= commit_tree = Commit_tree.create ?resize_step_bytes (path_prefix ^ ".commit_tree") in
  let+=? commit_db = Commit_db.create ~hash_func:context.hash.hash_func ~storage_context ~commit_tree in
  { commit_db ; context; auto_flush_seconds }

let open_ ~mode ?hashcons ?node_cache
    ?bytes_per_cell ?hash_func ?bytes_per_hash ?resize_step_bytes
    ?(auto_flush_seconds=default_auto_flush_seconds)
    path_prefix =
  check_prefix path_prefix;
  let*= () =
    if mode = Storage.Writer then
      let+= _ = Lock.lock @@ path_prefix ^ ".lock" in ()
    else Lwt.return_unit in
  let*= context =
    Context.open_ ~mode
      ?hashcons ?node_cache
      ?bytes_per_cell ?hash_func ?bytes_per_hash ?resize_step_bytes
      (path_prefix ^ ".context")
  in
  let storage_context = Context.get_storage context in
  let*= commit_tree =
    let fn = path_prefix ^ ".commit_tree" in
    let*= res = Commit_tree.open_ ~mode fn ?resize_step_bytes in
    match res with
    | Some ct -> Lwt.return ct
    | None ->
        if mode = Storage.Writer then Commit_tree.create fn
        else failwithf "%s does not exist" fn
  in
  let+=? commit_db = Commit_db.create ~hash_func:context.hash.hash_func ~storage_context ~commit_tree in
  { commit_db ; context; auto_flush_seconds }

let enable_process_sync t =
  Storage.enable_process_sync t.context.storage;
  (* XXX Internal... *)
  Storage.enable_process_sync (Commit_tree.Internal.get_storage (Commit_db.commit_tree t.commit_db))

let close { commit_db ; context; _ } =
  let*= () = Context.close context in
  Commit_tree.close (Commit_db.commit_tree commit_db)

let compute_commit_hash { commit_db ; _ } ~parent c =
  let c = Cursor.go_top c in
  let (c, (hp, s)) = Cursor.compute_hash c in
  assert (s = "");
  c, Commit_db.compute_hash commit_db ~parent hp

let since_last_sync = ref (Mtime_clock.counter ())

let commit vc =
  let context = context vc in
  let*= () = Storage.commit context.storage in
  Commit_db.commit vc.commit_db

let flush vc =
  let context = context vc in
  let*= () = Storage.flush context.storage in
  Commit_db.flush vc.commit_db

let commit
    ?(allow_missing_parent=false)
    ({ commit_db ; context ; auto_flush_seconds } as vc)
    ~parent
    ~hash_override
    c =
  let c = Cursor.go_top c in
  (* make sure contexts are the same *)
  let () =
    let Cursor (_, _, context', _) = c in
    assert (context == context'); (* XXX should not use pointer equality *)
  in
  (* check of parent *)
  begin match parent with
    | None -> ()
    | Some ph ->
        match Commit_db.find commit_db ph with
        | None when allow_missing_parent ->
            Log.warning "Vc.commit: parent:%a : parent is not in the storage"
              (Format.option Commit_hash.pp) parent
        | None ->
            failwithf "parent does not exist %a" Commit_hash.pp ph
        | Some _ -> ()
  end;
  (* write the Plebeia tree *)
  let*=? (c, i, hp) = Lwt.return @@ Cursor_storage.write_top_cursor c in
  (* record the Info *)
  let*=? iinfo =
    let Cursor.Cursor (_,_,_,info) = c in
    if List.length info.Info.copies <> 0 then
      Log.debug "Vc: COPIES=%d" (List.length info.Info.copies);
    let storage = context.Context.storage in
    Lwt.return @@ Info.write storage info
  in
  let commit_entry = Commit_db.make_commit commit_db
      ~parent
      ~index: i
      ~info: iinfo
      ?hash_override
      hp
  in
  Log.notice "Vc.commit: %a" Commit.pp commit_entry;
  begin match Commit_db.find commit_db commit_entry.hash with
    | Some commit_entry' ->
        if commit_entry'.parent = commit_entry.parent then
          Log.notice "Vc.commit: Overriding hash %a (old index %a, new index %a)"
            Commit_hash.pp commit_entry.hash
            Index.pp commit_entry'.index
            Index.pp commit_entry.index
        else
          failwithf "hash collision %a" Commit_hash.pp commit_entry.hash
    | None -> ()
  end;
  (* XXX We have 2 possible Index_overflow errors:
     - storage
     - commit_db
  *)
  let*=? () = Commit_db.add commit_db commit_entry in
  (* Synching is not always necessary *)
  let+= () =
    if
      auto_flush_seconds = 0
      || Mtime.Span.to_s (Mtime_clock.count !since_last_sync) > float auto_flush_seconds
    then begin
      if auto_flush_seconds <> 0 then Log.notice "Vc: flushing...";
      let+= () = flush vc in
      if auto_flush_seconds <> 0 then Log.notice "Vc: flushed";
      since_last_sync := Mtime_clock.counter ()
    end else begin
      commit vc
    end
  in
  Ok (c, hp, commit_entry)

let update_reader t =
  Log.debug "Vc: updating storage and commit_db";
  (* Commit_db.update_reader updates both the commit_tree and storage *)
  let+= (), secs = with_time_lwt (fun () -> Commit_db.update_reader t.commit_db) in
  Log.debug "Vc: updated storage and commit_db in %a" Mtime.Span.pp secs

let load_info context i_info =
  let storage = context.Context.storage in
  Info.read storage i_info

let checkout' ?(keep_info=false)  ({ commit_db ; context ; _ } as t) hash =
  (* XXX We need to get the index of commit to load the info *)
  let find () =
    Option.map
      (fun ({ Commit.index; info; _ } as root) ->
         let info = Result.from_Ok @@ load_info context info in
         (root,
          _Cursor (_Top,
                   Disk (index, Not_Extender),
                   context,
                   if keep_info then info else Info.empty)))
      (Commit_db.find commit_db hash)
  in
  match find () with
  | Some _ as res -> Lwt.return res
  | None when Context.mode context = Writer ->
      (* Writer knows everything.  hash really does not exist. *)
      Lwt.return_none
  | None ->
      (* if it is a reader, it may not know the new roots written by
         the writer.  The reader should update the roots and retry
         the checkout *)
      Log.notice "checkout: update_reader...";
      let+= () = update_reader t in
      Log.notice "retry checkout...";
      find ()

let checkout ?keep_info roots hash =
  let+= res = checkout' ?keep_info roots hash in
  Option.map snd res

let mem ({ commit_db ; context ; _ } as t) hash =
  (* XXX We need to get the index of commit to load the info *)
  match Commit_db.find commit_db hash with
  | Some _ -> Lwt.return_true
  | None when Context.mode context = Writer ->
      (* Writer knows everything.  hash really does not exist. *)
      Lwt.return_false
  | None ->
      (* if it is a reader, it may not know the new roots written by
         the writer.  The reader should update the roots and retry
         the checkout *)
      let+= () = update_reader t in
      (* XXX dupe *)
      match Commit_db.find commit_db hash with
      | Some _ -> true
      | None -> false (* Really not found *)