Source file second_pass.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
let src = Logs.Src.create "git.second-pass" ~doc:"logs git's second pass event"
module Log = (val Logs.src_log src : Logs.LOG)
module type S = sig
module FS : S.FS
module Hash : S.HASH
module Inflate : S.INFLATE
module Deflate : S.DEFLATE
module HDec : Unpack.H with module Hash := Hash
module PDec :
Unpack.P
with module Hash := Hash
and module Inflate := Inflate
and module Hunk := HDec
module PInfo :
Pack_info.S
with module Hash := Hash
and module Inflate := Inflate
and module HDec := HDec
and module PDec := PDec
module RPDec :
Unpack.D
with module Hash := Hash
and module Inflate := Inflate
and module Hunk := HDec
and module Pack := PDec
and module Mapper := FS.Mapper
type status = Resolved of Checkseum.Crc32.t * Hash.t | Root | Unresolved
val pp_status : status Fmt.t
val second_pass :
RPDec.pack
-> [`Normalized of PInfo.path] PInfo.t
-> (int64 * (PInfo.delta * status)) array Lwt.t
end
module Make
(Hash : S.HASH)
(FS : S.FS)
(Inflate : S.INFLATE)
(Deflate : S.DEFLATE)
(HDec : Unpack.H with module Hash := Hash)
(PDec : Unpack.P
with module Hash := Hash
and module Inflate := Inflate
and module Hunk := HDec)
(PInfo : Pack_info.S
with module Hash := Hash
and module Inflate := Inflate
and module HDec := HDec
and module PDec := PDec)
(RPDec : Unpack.D
with module Hash := Hash
and module Inflate := Inflate
and module Hunk := HDec
and module Pack := PDec
and module Mapper := FS.Mapper) =
struct
module FS = Helper.FS (FS)
module HDec = HDec
module PDec = PDec
module PInfo = PInfo
module RPDec = RPDec
open Lwt.Infix
type status = Resolved of Checkseum.Crc32.t * Hash.t | Root | Unresolved
let pp_status ppf = function
| Resolved (crc, hash) ->
Fmt.pf ppf "(Resolved (%a, %a))" Crc32.pp crc Hash.pp hash
| Root -> Fmt.pf ppf "Root"
| Unresolved -> Fmt.pf ppf "Unresolved"
type 'a protected = {mutable value: 'a; mutex: Lwt_mutex.t}
let is_not_root (_k, (_v, status)) =
match status with Root -> false | _ -> true
module RefMap = Map.Make (struct
type t = Hash.t
let compare = Hash.unsafe_compare
end)
module OfsMap = Map.Make (Int64)
type context =
{ protected_idx: int protected
; queue: (int64 * (PInfo.delta * status)) array
; ofs_deltas: int list OfsMap.t
; ref_deltas: int list RefMap.t
; cache_needed: (int64, int) RPDec.Cache.t
; cache_object:
(int64, RPDec.kind * Cstruct.t * int * RPDec.Ascendant.s) RPDec.Cache.t
; decoder: RPDec.pack }
let find abs_off array =
let rec go off len =
if len = 1 then off
else
let len' = len / 2 in
let off' = off + len' in
let abs_off', _ = array.(off') in
if abs_off = abs_off' then off'
else if abs_off < abs_off' then go off len'
else go off' (len - len')
in
go 0 (Array.length array)
let resolver ~thread:_ ~ztmp ~zwin context (_, (value, _)) =
match value with
| PInfo.Unresolved _ | PInfo.Delta _ -> assert false
| PInfo.Internal {length; abs_off; hash} -> (
Log.debug (fun l -> l "Start to resolve children of %a." Hash.pp hash) ;
let base = Cstruct.create length in
let base_hash = hash in
let children (abs_off, hash) =
List.map
(fun idx ->
let abs_off, _ = context.queue.(idx) in
abs_off )
( (try RefMap.find hash context.ref_deltas with _ -> [])
@ try OfsMap.find abs_off context.ofs_deltas with _ -> [] )
in
RPDec.Descendant.get_from_absolute_offset ~ztmp ~zwin
~cache:(context.cache_needed, context.cache_object)
~children base context.decoder abs_off
>>= function
| Ok (RPDec.Descendant.Root {children; _}) ->
let rec go parent depth = function
| RPDec.Descendant.Node {patch; children} :: rest ->
let idx = find patch.RPDec.Descendant.offset context.queue in
let value =
PInfo.Delta
{ hunks_descr= patch.RPDec.Descendant.descr
; inserts= patch.RPDec.Descendant.inserts
; depth
; from= parent }
in
Log.debug (fun l ->
l "Resolve children %a from base:%a." Hash.pp
patch.RPDec.Descendant.hash Hash.pp base_hash ) ;
context.queue.(idx)
<- ( patch.RPDec.Descendant.offset
, ( value
, Resolved
( patch.RPDec.Descendant.crc
, patch.RPDec.Descendant.hash ) ) ) ;
go value (succ depth) children ;
go parent depth rest
| RPDec.Descendant.Leaf patch :: rest ->
let idx = find patch.RPDec.Descendant.offset context.queue in
let value =
PInfo.Delta
{ hunks_descr= patch.RPDec.Descendant.descr
; inserts= patch.RPDec.Descendant.inserts
; depth
; from= parent }
in
Log.debug (fun l ->
l "Resolve children (leaf) %a from base:%a." Hash.pp
patch.RPDec.Descendant.hash Hash.pp base_hash ) ;
context.queue.(idx)
<- ( patch.RPDec.Descendant.offset
, ( value
, Resolved
( patch.RPDec.Descendant.crc
, patch.RPDec.Descendant.hash ) ) ) ;
go parent depth rest
| [] -> ()
in
go value 1 children ; Lwt.return_unit
| Error _ -> Lwt.return_unit )
let rec dispatcher ~thread ~ztmp ~zwin context =
Lwt_mutex.lock context.protected_idx.mutex
>>= fun () ->
while
context.protected_idx.value < Array.length context.queue
&& is_not_root context.queue.(context.protected_idx.value)
do
(context.protected_idx).value <- context.protected_idx.value + 1
done ;
Log.debug (fun l ->
l "Dispatcher will resolve (or terminate) at: %d."
context.protected_idx.value ) ;
if context.protected_idx.value >= Array.length context.queue then (
Lwt_mutex.unlock context.protected_idx.mutex ;
Lwt.return_unit )
else
let root = context.protected_idx.value in
(context.protected_idx).value <- context.protected_idx.value + 1 ;
Lwt_mutex.unlock context.protected_idx.mutex ;
resolver ~thread ~ztmp ~zwin context context.queue.(root)
>>= fun () -> dispatcher ~thread ~ztmp ~zwin context
let second_pass decoder info =
let matrix =
Hashtbl.fold
(fun k v acc ->
let status =
match v with
| PInfo.Delta _ | PInfo.Unresolved _ -> Unresolved
| PInfo.Internal _ -> Root
in
(k, (v, status)) :: acc )
info.PInfo.delta []
|> List.sort (fun (ka, _) (kb, _) -> Int64.compare ka kb)
|> Array.of_list
in
let cache_needed =
{RPDec.Cache.find= (fun _ -> None); promote= (fun _ _ -> ())}
in
let cache_object = RPDec.Ascendant.apply_cache (1024 * 1024) in
let ofs_deltas, ref_deltas =
let add_ofs ofs idx map =
OfsMap.add ofs (idx :: (try OfsMap.find ofs map with _ -> [])) map
in
let add_ref hash idx map =
RefMap.add hash (idx :: (try RefMap.find hash map with _ -> [])) map
in
Array.fold_left
(fun (ofs_deltas, ref_deltas, idx) (abs_off, (value, _)) ->
match value with
| PInfo.Delta {hunks_descr= {HDec.reference= HDec.Hash hash; _}; _}
|PInfo.Unresolved {hash; _} ->
ofs_deltas, add_ref hash idx ref_deltas, idx + 1
| PInfo.Delta
{hunks_descr= {HDec.reference= HDec.Offset rel_off; _}; _} ->
( add_ofs Int64.(sub abs_off rel_off) idx ofs_deltas
, ref_deltas
, idx + 1 )
| _ -> ofs_deltas, ref_deltas, idx + 1 )
(OfsMap.empty, RefMap.empty, 0)
matrix
|> fun (ofs_deltas, ref_deltas, _) -> ofs_deltas, ref_deltas
in
let context =
{ protected_idx= {value= 0; mutex= Lwt_mutex.create ()}
; queue= matrix
; ofs_deltas
; ref_deltas
; cache_needed
; cache_object
; decoder }
in
let pool =
Lwt_pool.create 4 (fun () ->
let ztmp = Cstruct.create 0x8000 in
let zwin = Inflate.window () in
Lwt.return (ztmp, zwin) )
in
Lwt.join
[ Lwt_pool.use pool (fun (ztmp, zwin) ->
dispatcher ~thread:0 ~ztmp ~zwin context )
; Lwt_pool.use pool (fun (ztmp, zwin) ->
dispatcher ~thread:1 ~ztmp ~zwin context )
; Lwt_pool.use pool (fun (ztmp, zwin) ->
dispatcher ~thread:2 ~ztmp ~zwin context )
; Lwt_pool.use pool (fun (ztmp, zwin) ->
dispatcher ~thread:3 ~ztmp ~zwin context ) ]
>>= fun () -> Lwt.return matrix
end