package irmin-pack

  1. Overview
  2. Docs
package irmin-pack
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Source file gc.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
(*
 * Copyright (c) 2022-2022 Tarides <contact@tarides.com>
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 *)

open! Import
module Payload = Control_file.Latest_payload

let buffer_size = 8192

exception Pack_error = Errors.Pack_error

module type Args = sig
  (* The following [with module Io = Io.Unix] forces unix *)
  module Fm : File_manager.S with module Io = Io.Unix
  module Dict : Dict.S with module Fm = Fm
  module Errs : Io_errors.S with module Io = Fm.Io
  module Dispatcher : Dispatcher.S with module Fm = Fm

  type hash
  type key = hash Irmin_pack.Pack_key.t [@@deriving irmin]

  module Hash : sig
    val hash_size : int
  end

  module Node_value : sig
    type t
    type step

    val pred :
      t ->
      (step option * [ `Contents of key | `Inode of key | `Node of key ]) list
  end

  module Node_store : sig
    type 'a t

    val v :
      config:Irmin.Backend.Conf.t ->
      fm:Fm.t ->
      dict:Dict.t ->
      dispatcher:Dispatcher.t ->
      read t

    val unsafe_find :
      check_integrity:bool -> [< read ] t -> key -> Node_value.t option
  end

  module Commit_value : sig
    type t

    val node : t -> key
    val parents : t -> key list
  end

  module Commit_store :
    Pack_store.S
      with type value = Commit_value.t
       and type key = key
       and type file_manager = Fm.t
       and type dict = Dict.t
       and type dispatcher = Dispatcher.t
       and type hash = hash
end

module type S = sig
  module Args : Args

  val run_and_output_result : generation:int -> string -> Args.key -> int63

  val transfer_append_exn :
    read_exn:(off:int63 -> len:int -> bytes -> unit) ->
    append_exn:(string -> unit) ->
    off:int63 ->
    len:int63 ->
    bytes ->
    unit
end

module Make (Args : Args) : S with module Args := Args = struct
  open Args
  module Io = Fm.Io
  module Mapping_file = Fm.Mapping_file
  module Ao = Append_only_file.Make (Io)

  module X = struct
    type t = int63 [@@deriving irmin]

    let equal = Irmin.Type.(unstage (equal t))
    let hash = Irmin.Type.(unstage (short_hash t))
    let hash (t : t) : int = hash t
  end

  module Table = Hashtbl.Make (X)

  let string_of_key = Irmin.Type.to_string key_t

  let transfer_append_exn ~read_exn ~append_exn ~(off : int63) ~(len : int63)
      buffer =
    let buffer_size = Bytes.length buffer |> Int63.of_int in
    let rec aux off len_remaining =
      let open Int63.Syntax in
      let min a b = if a < b then a else b in
      let len = min buffer_size len_remaining in
      let len' = Int63.to_int len in
      read_exn ~off ~len:len' buffer;
      let () =
        if len = buffer_size then append_exn (Bytes.to_string buffer)
        else append_exn (Bytes.sub_string buffer 0 len')
      in
      let len_remaining = len_remaining - len in
      if len_remaining > Int63.zero then aux (off + len) len_remaining
    in
    aux off len

  (** [iter_from_node_key node_key _ _ ~f] calls [f] with the key of the node
      and iterates over its children.

      [f k] returns [Follow] or [No_follow], indicating the iteration algorithm
      if the children of [k] should be traversed or skiped. *)
  let iter node_key node_store ~f k =
    let marks = Table.create 1024 in
    let mark offset = Table.add marks offset () in
    let has_mark offset = Table.mem marks offset in
    let rec iter_from_node_key_exn node_key node_store ~f k =
      match
        Node_store.unsafe_find ~check_integrity:false node_store node_key
      with
      | None -> raise (Pack_error (`Dangling_key (string_of_key node_key)))
      | Some node ->
          iter_from_node_children_exn node_store ~f (Node_value.pred node) k
    and iter_from_node_children_exn node_store ~f children k =
      match children with
      | [] -> k ()
      | (_step, kinded_key) :: tl -> (
          let k () = iter_from_node_children_exn node_store ~f tl k in
          match kinded_key with
          | `Contents key ->
              let (_ : int63) = f key in
              k ()
          | `Inode key | `Node key ->
              let offset = f key in
              if has_mark offset then k ()
              else (
                mark offset;
                iter_from_node_key_exn key node_store ~f k))
    in
    iter_from_node_key_exn node_key node_store ~f k

  (* Dangling_parent_commit are the parents of the gced commit. They are kept on
     disk in order to correctly deserialised the gced commit. *)
  let magic_parent =
    Pack_value.Kind.to_magic Pack_value.Kind.Dangling_parent_commit

  (* Transfer the commit with a different magic. Note that this is modifying
     existing written data. *)
  let transfer_parent_commit_exn ~read_exn ~write_exn ~mapping key =
    let off, len =
      match Irmin_pack.Pack_key.inspect key with
      | Indexed _ ->
          (* As this is the second time we are reading this key, this case is
             unreachable. *)
          assert false
      | Direct { offset; length; _ } -> (offset, length)
    in
    let buffer = Bytes.create len in
    read_exn ~off ~len buffer;
    let entry = Mapping_file.find_nearest_leq mapping off in
    match entry with
    | None -> assert false
    | Some { poff; _ } ->
        Bytes.set buffer Hash.hash_size magic_parent;
        (* Bytes.unsafe_to_string usage: We assume read_exn returns unique ownership of buffer
           to this function. Then at the call to Bytes.unsafe_to_string we give up unique
           ownership to buffer (we do not modify it thereafter) in return for ownership of the
           resulting string, which we pass to write_exn. This usage is safe. *)
        write_exn ~off:poff ~len (Bytes.unsafe_to_string buffer)

  let create_new_suffix ~root ~generation =
    let open Result_syntax in
    let path = Irmin_pack.Layout.V3.suffix ~root ~generation in
    let auto_flush_threshold = 1_000_000 in
    let auto_flush_callback x = Ao.flush x |> Errs.raise_if_error in
    let* suffix =
      Ao.create_rw ~path ~overwrite:true ~auto_flush_threshold
        ~auto_flush_callback
    in
    Ok suffix

  let run ~generation root commit_key =
    let open Result_syntax in
    let config =
      Irmin_pack.Conf.init ~fresh:false ~readonly:true ~lru_size:0 root
    in

    (* Step 1. Open the files *)
    [%log.debug "GC: opening files in RO mode"];
    let* fm = Fm.open_ro config in
    Errors.finalise (fun _outcome ->
        Fm.close fm |> Errs.log_if_error "GC: Close File_manager")
    @@ fun () ->
    let* dict = Dict.v fm in
    let* dispatcher = Dispatcher.v ~root fm in
    let node_store = Node_store.v ~config ~fm ~dict ~dispatcher in
    let commit_store = Commit_store.v ~config ~fm ~dict ~dispatcher in

    (* Step 2. Load commit which will make [commit_key] [Direct] if it's not
       already the case. *)
    let* commit =
      match
        Commit_store.unsafe_find ~check_integrity:false commit_store commit_key
      with
      | None -> Error (`Commit_key_is_dangling (string_of_key commit_key))
      | Some commit -> Ok commit
    in
    let commit_offset, _ =
      let state : _ Irmin_pack.Pack_key.state =
        Irmin_pack.Pack_key.inspect commit_key
      in
      match state with
      | Indexed _ -> assert false
      | Direct x -> (x.offset, x.length)
    in

    (* Step 3. Create the new mapping. *)
    let* mapping =
      (* Step 3.1 Start [Mapping_file] routine which will create the
         reachable file. *)
      (fun f -> Mapping_file.create ~root ~generation ~register_entries:f)
      @@ fun ~register_entry ->
      (* Step 3.2 Put the commit parents in the reachable file.
         The parent(s) of [commit_key] must be included in the iteration
         because, when decoding the [Commit_value.t] at [commit_key], the
         parents will have to be read in order to produce a key for them. *)
      let register_object_exn key =
        match Irmin_pack.Pack_key.inspect key with
        | Indexed _ ->
            raise
              (Pack_error (`Commit_parent_key_is_indexed (string_of_key key)))
        | Direct { offset; length; _ } -> register_entry ~off:offset ~len:length
      in
      List.iter register_object_exn (Commit_value.parents commit);

      (* Step 3.3 Put the nodes and contents in the reachable file. *)
      let register_object_exn key =
        match Irmin_pack.Pack_key.inspect key with
        | Indexed _ ->
            raise
              (Pack_error (`Node_or_contents_key_is_indexed (string_of_key key)))
        | Direct { offset; length; _ } ->
            register_entry ~off:offset ~len:length;
            offset
      in
      let node_key = Commit_value.node commit in
      let (_ : int63) = register_object_exn node_key in
      iter node_key node_store ~f:register_object_exn (fun () -> ());

      (* Step 3.4 Return and let the [Mapping_file] routine create the mapping
         file. *)
      ()
    in

    let* () =
      (* Step 4. Create the new prefix. *)
      let auto_flush_callback x = Ao.flush x |> Errs.raise_if_error in
      let* prefix =
        let path = Irmin_pack.Layout.V3.prefix ~root ~generation in
        Ao.create_rw ~path ~overwrite:true ~auto_flush_threshold:1_000_000
          ~auto_flush_callback
      in
      let* () =
        Errors.finalise (fun _outcome ->
            Ao.close prefix |> Errs.log_if_error "GC: Close prefix")
        @@ fun () ->
        ();

        (* Step 5. Transfer to the new prefix, flush and close. *)
        [%log.debug "GC: transfering to the new prefix"];
        let buffer = Bytes.create buffer_size in
        (* Step 5.1. Transfer all. *)
        let read_exn = Dispatcher.read_in_prefix_and_suffix_exn dispatcher in
        let append_exn = Ao.append_exn prefix in
        let f ~off ~len =
          let len = Int63.of_int len in
          transfer_append_exn ~read_exn ~append_exn ~off ~len buffer
        in
        let* () = Mapping_file.iter mapping f in
        Ao.flush prefix
      in
      (* Step 5.2. Transfer again the parent commits but with a modified
         magic. Reopen the new prefix, this time _not_ in append-only
         as we have to modify data inside the file. *)
      let read_exn = Dispatcher.read_exn dispatcher in
      let* prefix =
        let path = Irmin_pack.Layout.V3.prefix ~root ~generation in
        Io.open_ ~path ~readonly:false
      in
      let* () =
        Errors.finalise (fun _outcome ->
            Io.close prefix
            |> Errs.log_if_error "GC: Close prefix after parent rewrite")
        @@ fun () ->
        let write_exn = Io.write_exn prefix in
        List.iter
          (fun key ->
            transfer_parent_commit_exn ~read_exn ~write_exn ~mapping key)
          (Commit_value.parents commit);
        Ok ()
      in
      Ok ()
    in

    (* Step 6. Create the new suffix and prepare 2 functions for read and write
       operations. *)
    let buffer = Bytes.create buffer_size in
    [%log.debug "GC: creating new suffix"];
    let* suffix = create_new_suffix ~root ~generation in
    Errors.finalise (fun _outcome ->
        Ao.close suffix |> Errs.log_if_error "GC: Close suffix")
    @@ fun () ->
    let read_exn = Dispatcher.read_exn dispatcher in
    let append_exn = Ao.append_exn suffix in
    let transfer_exn = transfer_append_exn ~read_exn ~append_exn buffer in

    (* Step 7. Transfer to the next suffix. *)
    [%log.debug "GC: transfering to the new suffix"];
    let num_iterations = 7 in
    (* [transfer_loop] is needed because after garbage collection there may be new objects
       at the end of the suffix file that need to be copied over *)
    let rec transfer_loop i ~off =
      if i = 0 then off
      else
        let () = Fm.reload fm |> Errs.raise_if_error in
        let pl : Payload.t = Fm.Control.payload (Fm.control fm) in
        let end_offset =
          Dispatcher.offset_of_suffix_off dispatcher pl.entry_offset_suffix_end
        in
        let len = Int63.Syntax.(end_offset - off) in
        [%log.debug
          "GC: transfer_loop iteration %d, offset %a, length %a"
            (num_iterations - i + 1)
            Int63.pp off Int63.pp len];
        let () = transfer_exn ~off ~len in
        (* Check how many bytes are left, [4096*5] is selected because it is roughly the
           number of bytes that requires a read from the block device on ext4 *)
        if Int63.to_int len < 4096 * 5 then end_offset
        else
          let off = Int63.Syntax.(off + len) in
          transfer_loop ~off (i - 1)
    in
    let flush_and_raise () = Ao.flush suffix |> Errs.raise_if_error in
    let* offs =
      Errs.catch (fun () ->
          let offs = transfer_loop ~off:commit_offset num_iterations in
          flush_and_raise ();
          offs)
    in
    (* Step 8. Inform the caller of the end_offset copied. *)
    Ok offs

  (* No one catches errors when this function terminates. Write the result in a
     file and terminate the process with an exception, if needed. *)
  let run_and_output_result ~generation root commit_key =
    let result = run ~generation root commit_key in
    let write_result = Fm.write_gc_output ~root ~generation result in
    write_result |> Errs.raise_if_error;
    result |> Errs.raise_if_error
end
OCaml

Innovation. Community. Security.

GitHub Discord Twitter Peertube RSS
About
Changelog Releases Industrial Users Academic Users Why OCaml
Ecosystem
Packages Community Events OCaml Planet Jobs
Resources
Install OCaml Get Started Platform Tools Language Manual Standard Library API Books Exercises Papers OCaml Playground Logo
Policies
Carbon Footprint Governance Privacy Code of Conduct