Bring back snark persistence: iteration 1

src/lib/disk_cache/filesystem/disk_cache.ml

@@ -10,36 +10,96 @@ module Make (B : sig
   include Binable.S
 end) =
 struct
-  type t = string * int ref
+  type t =
+    { root : string
+    ; next_idx : int ref
+    ; eviction_freezed : bool ref
+    ; disk_meta_location : string option
+    }
 
-  type id = { idx : int }
+  type persistence = int [@@deriving bin_io_unversioned]
 
-  let initialize path ~logger =
+  type id = { idx : int } [@@deriving bin_io_unversioned]
+
+  let freeze_eviction_and_snapshot ~logger
+      { eviction_freezed; next_idx; disk_meta_location; _ } =
+    eviction_freezed := true ;
+    match disk_meta_location with
+    | None ->
+        [%log info]
+          "No metadata location is set for FS disk cache, not saving disk \
+           cache persistence information" ;
+        Async.Deferred.unit
+    | Some disk_meta_location ->
+        Async_unix.Writer.save_bin_prot disk_meta_location
+          bin_writer_persistence !next_idx
+
+  let initialize path ~logger ?disk_meta_location () =
+    let open Async in
+    let open Deferred.Let_syntax in
+    let%bind next_idx =
+      match disk_meta_location with
+      | None ->
+          return 0
+      | Some disk_meta_location -> (
+          match%map
+            Async.Reader.load_bin_prot disk_meta_location bin_reader_persistence
+          with
+          | Error e ->
+              [%log warn]
+                "Failed to read FS disk cache persistence information from \
+                 disk, initializing a fresh cache"
+                ~metadata:
+                  [ ("location", `String disk_meta_location)
+                  ; ("reason", `String (Error.to_string_hum e))
+                  ] ;
+              0
+          | Ok idx ->
+              idx )
+    in
     Async.Deferred.Result.map (Disk_cache_utils.initialize_dir path ~logger)
-      ~f:(fun path -> (path, ref 0))
+      ~f:(fun root ->
+        let cache =
+          { root
+          ; next_idx = ref next_idx
+          ; eviction_freezed = ref false
+          ; disk_meta_location
+          }
+        in
+
+        Option.iter disk_meta_location ~f:(fun _ ->
+            Mina_stdlib_unix.Exit_handlers.register_async_shutdown_handler
+              ~logger
+              ~description:
+                "Shutting down LMDB Disk Cache GC Eviction and store \
+                 persistence info needed to reload from disk" (fun () ->
+                freeze_eviction_and_snapshot ~logger cache ) ) ;
+
+        cache )
 
   let path root i = root ^ Filename.dir_sep ^ Int.to_string i
 
-  let get ((root, _) : t) (id : id) : B.t =
+  let get ({ root; _ } : t) (id : id) : B.t =
     (* Read from the file. *)
     In_channel.with_file ~binary:true (path root id.idx) ~f:(fun chan ->
         let str = In_channel.input_all chan in
         Binable.of_string (module B) str )
 
-  let put ((root, next_idx) : t) x : id =
+  let put ({ root; next_idx; eviction_freezed; _ } : t) x : id =
     let idx = !next_idx in
     incr next_idx ;
     let res = { idx } in
     (* When this reference is GC'd, delete the file. *)
     Core.Gc.Expert.add_finalizer_last_exn res (fun () ->
-        (* Ignore errors: if a directory is deleted, it's ok. *)
-        try Core.Unix.unlink (path root idx) with _ -> () ) ;
+        if not !eviction_freezed then
+          (* Ignore errors: if a directory is deleted, it's ok. *)
+          try Core.Unix.unlink (path root idx) with _ -> () ) ;
     (* Write the proof to the file. *)
     Out_channel.with_file ~binary:true (path root idx) ~f:(fun chan ->
         Out_channel.output_string chan @@ Binable.to_string (module B) x ) ;
     res
 
-  let count ((path, _) : t) = Sys.ls_dir path |> List.length
+  let count ({ root; _ } : t) = Sys.ls_dir root |> List.length
 end
 
 let%test_module "disk_cache filesystem" =

src/lib/disk_cache/identity/disk_cache.ml

@@ -2,14 +2,15 @@ open Async_kernel
 open Core_kernel
 
 module Make (Data : sig
-  type t
+  type t [@@deriving bin_io]
 end) =
 struct
   type t = unit
 
-  type id = Data.t
+  type id = Data.t [@@deriving bin_io_unversioned]
 
-  let initialize _path ~logger:_ = Deferred.Result.return ()
+  let initialize _path ~logger:_ ?disk_meta_location:_ () =
+    Deferred.Result.return ()
 
   let get () = ident
 

src/lib/disk_cache/intf/disk_cache_intf.ml

@@ -6,13 +6,17 @@ module type S = sig
 
   type t
 
-  (** Initialize the on-disk cache explicitly before interactions with it take place. *)
+  (** Initialize the on-disk cache explicitly before interactions with it take
+      place. If disk_meta_location is set, will try to read from metadata at 
+      that location, so to reuse disk cache from a previous run. *)
   val initialize :
        string
     -> logger:Logger.t
+    -> ?disk_meta_location:string
+    -> unit
     -> (t, [> `Initialization_error of Error.t ]) Deferred.Result.t
 
-  type id
+  type id [@@deriving bin_io]
 
   (** Put the value to disk, return an identifier that is associated with a special handler in GC. *)
   val put : t -> Data.t -> id

src/lib/disk_cache/lmdb/disk_cache.ml

@@ -24,29 +24,86 @@ module Make (Data : Binable.S) = struct
           (** A list of ids that are no longer reachable from OCaml runtime, but
               haven't been cleared inside the LMDB disk cache *)
     ; queue_guard : Error_checking_mutex.t
+    ; eviction_freezed : bool ref
+    ; disk_meta_location : string option
     }
 
+  type persistence = int [@@deriving bin_io_unversioned]
+
   (** How big can the queue [reusable_keys] be before we do a cleanup *)
   let reuse_size_limit = 512
 
-  let initialize path ~logger =
+  let freeze_eviction_and_snapshot ~logger
+      ({ eviction_freezed; counter; disk_meta_location; _ } : t) =
+    eviction_freezed := true ;
+    match disk_meta_location with
+    | None ->
+        [%log info]
+          "No metadata location is set for LMDB disk cache, not saving disk \
+           cache persistence information" ;
+        Async.Deferred.unit
+    | Some disk_meta_location ->
+        Async_unix.Writer.save_bin_prot disk_meta_location
+          bin_writer_persistence !counter
+
+  let initialize path ~logger ?disk_meta_location () =
+    let open Async in
+    let open Deferred.Let_syntax in
+    let%bind counter =
+      match disk_meta_location with
+      | None ->
+          return 0
+      | Some disk_meta_location -> (
+          match%map
+            Async.Reader.load_bin_prot disk_meta_location bin_reader_persistence
+          with
+          | Error e ->
+              [%log warn]
+                "Failed to read LMDB disk cache persistence information from \
+                 disk $location, initializing a fresh cache"
+                ~metadata:
+                  [ ("location", `String disk_meta_location)
+                  ; ("reason", `String (Error.to_string_hum e))
+                  ] ;
+              0
+          | Ok idx ->
+              [%log info]
+                "Successfully restored LMDB disk cacahe persistence from disk \
+                 $location"
+                ~metadata:[ ("location", `String disk_meta_location) ] ;
+              idx )
+    in
     Async.Deferred.Result.map (Disk_cache_utils.initialize_dir path ~logger)
       ~f:(fun path ->
         let env, db = Rw.create path in
-        { env
-        ; db
-        ; counter = ref 0
-        ; reusable_keys = Queue.create ()
-        ; queue_guard = Error_checking_mutex.create ()
-        } )
-
-  type id = { idx : int }
+        let cache =
+          { env
+          ; db
+          ; counter = ref counter
+          ; reusable_keys = Queue.create ()
+          ; queue_guard = Error_checking_mutex.create ()
+          ; eviction_freezed = ref false
+          ; disk_meta_location
+          }
+        in
+
+        Option.iter disk_meta_location ~f:(fun _ ->
+            Mina_stdlib_unix.Exit_handlers.register_async_shutdown_handler
+              ~logger
+              ~description:
+                "Shutting down LMDB Disk Cache GC Eviction and store \
+                 persistence info needed to reload from disk" (fun () ->
+                freeze_eviction_and_snapshot ~logger cache ) ) ;
+        cache )
+
+  type id = { idx : int } [@@deriving bin_io_unversioned]
 
   let get ({ env; db; _ } : t) ({ idx } : id) : Data.t =
     Rw.get ~env db idx |> Option.value_exn
 
-  let put ({ env; db; counter; reusable_keys; queue_guard } : t) (x : Data.t) :
-      id =
+  let put
+      ({ env; db; counter; reusable_keys; queue_guard; eviction_freezed; _ } : t)
+      (x : Data.t) : id =
     let idx =
       match
         Error_checking_mutex.critical_section queue_guard ~f:(fun () ->
@@ -63,12 +120,13 @@ module Make (Data : Binable.S) = struct
     let res = { idx } in
     (* When this reference is GC'd, delete the file. *)
     Gc.Expert.add_finalizer_last_exn res (fun () ->
-        (* The actual deletion is delayed, as GC maybe triggered in LMDB's
-           critical section. LMDB critical section then will be re-entered if
-           it's invoked directly in a GC hook.
-           This causes mutex double-acquiring and node freezes. *)
-        Error_checking_mutex.critical_section queue_guard ~f:(fun () ->
-            Queue.enqueue reusable_keys idx ) ) ;
+        if not !eviction_freezed then
+          (* The actual deletion is delayed, as GC maybe triggered in LMDB's
+             critical section. LMDB critical section then will be re-entered if
+             it's invoked directly in a GC hook.
+             This causes mutex double-acquiring and node freezes. *)
+          Error_checking_mutex.critical_section queue_guard ~f:(fun () ->
+              Queue.enqueue reusable_keys idx ) ) ;
 
     Error_checking_mutex.critical_section queue_guard ~f:(fun () ->
         if Queue.length reusable_keys >= reuse_size_limit then (

src/lib/disk_cache/test/test_cache_deadlock.ml

@@ -31,7 +31,7 @@ end
 let run_test_with_cache (module Cache_impl : Cache_intf) ~timeout_seconds
     ~tmpdir =
   let module Cache = Cache_impl.Make (Evil_data) in
-  let%bind cache_result = Cache.initialize tmpdir ~logger:(Logger.null ()) in
+  let%bind cache_result = Cache.initialize tmpdir ~logger:(Logger.null ()) () in
   let cache =
     match cache_result with
     | Ok cache ->

src/lib/disk_cache/test_lib/disk_cache_test_lib.ml

@@ -46,7 +46,7 @@ module Make_impl (Cache : Disk_cache_intf.S_with_count with module Data := Mock)
 
   let initialize_cache_or_fail tmpd ~logger =
     let open Deferred.Let_syntax in
-    let%bind cache_res = Cache.initialize tmpd ~logger in
+    let%bind cache_res = Cache.initialize tmpd ~logger () in
     match cache_res with
     | Ok cache ->
         return cache
@@ -107,7 +107,7 @@ module Make_impl (Cache : Disk_cache_intf.S_with_count with module Data := Mock)
       ~f:(remove_data_on_gc_impl ~gc_strict)
 
   let initialize_and_expect_failure path ~logger =
-    let%bind cache_res = Cache.initialize path ~logger in
+    let%bind cache_res = Cache.initialize path ~logger () in
     match cache_res with
     | Ok _ ->
         failwith "unexpected initialization success"
@@ -130,7 +130,7 @@ module Make_impl (Cache : Disk_cache_intf.S_with_count with module Data := Mock)
       Core.Unix.mkdir some_dir ;
       let dir_symlink = tmp_dir ^/ "dir_link" in
       Core.Unix.symlink ~target:some_dir_name ~link_name:dir_symlink ;
-      Cache.initialize dir_symlink ~logger
+      Cache.initialize dir_symlink ~logger ()
       >>| function
       | Ok _ ->
           ()

src/lib/mina_lib/conf_dir.ml

@@ -32,13 +32,14 @@ let rec check_and_set_lockfile ~logger conf_dir =
       | Ok () ->
           [%log debug] "Created daemon lockfile $lockfile"
             ~metadata:[ ("lockfile", `String lockfile) ] ;
-          Mina_stdlib_unix.Exit_handlers.register_async_shutdown_handler ~logger
-            ~description:"Remove daemon lockfile" (fun () ->
-              match%bind Sys.file_exists lockfile with
-              | `Yes ->
-                  Unix.unlink lockfile
-              | _ ->
-                  return () )
+          Mina_stdlib_unix.Exit_handlers.(
+            register_async_shutdown_handler ~priority:Priority.Exclusion ~logger
+              ~description:"Remove daemon lockfile" (fun () ->
+                match%bind Sys.file_exists lockfile with
+                | `Yes ->
+                    Unix.unlink lockfile
+                | _ ->
+                    return () ))
       | Error exn ->
           Error.tag_arg (Error.of_exn exn)
             "Could not create the daemon lockfile" ("lockfile", lockfile)

src/lib/mina_lib/mina_lib.ml

@@ -1693,6 +1693,7 @@ let raise_on_initialization_error (`Initialization_error e) =
 
 let initialize_proof_cache_db (config : Config.t) =
   Proof_cache_tag.create_db ~logger:config.logger
+    ~disk_meta_location:(config.conf_dir ^/ "proof_cache_metadata")
     (config.conf_dir ^/ "proof_cache")
   >>| function Error e -> raise_on_initialization_error e | Ok db -> db
 
@@ -1885,14 +1886,15 @@ let create ~commit_id ?wallets (config : Config.t) =
           let get_current_frontier () =
             Broadcast_pipe.Reader.peek frontier_broadcast_pipe_r
           in
-          Mina_stdlib_unix.Exit_handlers.register_async_shutdown_handler
-            ~logger:config.logger
-            ~description:"Close transition frontier, if exists" (fun () ->
-              match get_current_frontier () with
-              | None ->
-                  Deferred.unit
-              | Some frontier ->
-                  Transition_frontier.close ~loc:__LOC__ frontier ) ;
+          Mina_stdlib_unix.Exit_handlers.(
+            register_async_shutdown_handler
+              ~priority:Priority.Transition_frontier ~logger:config.logger
+              ~description:"Close transition frontier, if exists" (fun () ->
+                match get_current_frontier () with
+                | None ->
+                    Deferred.unit
+                | Some frontier ->
+                    Transition_frontier.close ~loc:__LOC__ frontier )) ;
           (* knot-tying hacks so we can pass a get_node_status function before net, Mina_lib.t created *)
           let sync_status_ref = ref None in
           let get_node_status (net : Mina_networking.t) =
@@ -2036,7 +2038,7 @@ let create ~commit_id ?wallets (config : Config.t) =
             notify_online_impl () |> don't_wait_for ;
             Deferred.unit
           in
-          let transaction_pool, tx_remote_sink, tx_local_sink =
+          let%bind transaction_pool, tx_remote_sink, tx_local_sink =
             (* make transaction pool return writer for local and incoming diffs *)
             Network_pool.Transaction_pool.create ~config:txn_pool_config
               ~constraint_constants ~consensus_constants
@@ -2049,10 +2051,13 @@ let create ~commit_id ?wallets (config : Config.t) =
           in
           let snark_pool_config =
             Network_pool.Snark_pool.Resource_pool.make_config ~verifier
-              ~trust_system:config.trust_system
-              ~disk_location:config.snark_pool_disk_location ~proof_cache_db
+              ~trust_system:config.trust_system ~proof_cache_db
+              ~persistence:
+                ( `Disk_location config.snark_pool_disk_location
+                , `Store_every (Time.Span.of_min 5.0) )
+              ()
           in
-          let snark_pool, snark_remote_sink, snark_local_sink =
+          let%bind snark_pool, snark_remote_sink, snark_local_sink =
             Network_pool.Snark_pool.create ~config:snark_pool_config
               ~constraint_constants ~consensus_constants
               ~time_controller:config.time_controller ~logger:config.logger