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
|
(************************************************************************)
(* This file is part of SKS. SKS is free software; you can
redistribute it and/or modify it under the terms of the GNU General
Public License as published by the Free Software Foundation; either
version 2 of the License, or (at your option) any later version.
This program is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
USA *)
(***********************************************************************)
(** Client side of set-reconciliation algorithm *)
open StdLabels
open MoreLabels
module Unix=UnixLabels
open Common
open Printf
open ReconMessages
module Set = PSet.Set
module Map = PMap.Map
module PTree = PrefixTree
(* module ZZp = RMisc.ZZp *)
exception Bug of string
(***************************************************************)
(* Diagnostic Timers *****************************************)
(***************************************************************)
let flushcount = ref 0
let timer = MTimer.create ()
let tstart () =
MTimer.start timer
let tstop accum =
MTimer.stop timer;
accum := !accum +. MTimer.read_ms timer
let get_flushcount () = !flushcount
(***************************************************************)
(***************************************************************)
(***************************************************************)
type 'a bottomQ_entry = FlushEnded | Bottom of 'a
type reconbound = { num_completed: int;
verified_partitions: Bitstring.t Set.t;
}
(*
let reconbound_exceeded rb =
!Settings.mbar * (Set.cardinal rb.verified_partitions)
+ rb.num_recovered
> Settings.max_recover
*)
exception Continue
(** Send request and update [bottomQ] appropriately *)
let send_request cout tree ~bottomQ (node,key) =
let request =
if PTree.is_leaf node ||
PTree.num_elements tree node <
!Settings.recon_thresh_mult * !Settings.mbar
then ReconRqst_Full
{ rf_prefix = key;
rf_elements = PTree.elements tree node;
}
else ReconRqst_Poly
{ rp_prefix = key;
rp_size = PTree.size node;
rp_samples = PTree.svalues node;
}
in
marshal_noflush cout request;
Queue.push (Bottom (node,key)) bottomQ
(** Handle reply message and update [requestQ] appropriately *)
let handle_reply cout tree ~requestQ reply (node,key) setref =
match reply.msg with
| SyncFail ->
if PTree.is_leaf node then
raise (Bug ("Unexpected error. Syncfail received" ^
"at leaf node"));
let children = PTree.child_keys tree key in
let nodes =
List.map
~f:(fun key -> try PTree.get_node_key tree key
with Not_found ->
raise (Bug ("Client.read: PTree.get_node_key " ^
"should not fail")))
children in
(* update requestQ with requests corresponding to
children of present node *)
List.iter ~f:(fun req -> Queue.push req requestQ)
(List.combine nodes children)
| Elements elements -> setref := (ZSet.union !setref elements)
(* required for case where reconciliation terminates for due to the end
of the prefix tree *)
| FullElements elements ->
let local = PTree.get_zzp_elements tree node in
let localdiff = ZSet.diff local elements in
let remotediff = ZSet.diff elements local in
marshal_noflush cout (Elements localdiff);
setref := ZSet.union !setref remotediff
| _ -> failwith ( "Unexpected message: " ^
msg_to_string reply.msg )
(* after a timeout, give an extra 10 seconds to actually extract the data built up so far *)
let recover_timeout = 10
(** manages reconciliation connection, determining when messages are sent and
received on the channel. *)
let connection_manager cin cout tree initial_request =
let set = ref ZSet.empty in
let requestQ = Queue.create ()
and bottomQ = Queue.create () in
Queue.push initial_request requestQ;
(* state variables *)
let flushing = ref false (* whether a flush has been sent and not
yet bounced back. *)
in
let flush_queue () =
marshal_noflush cout Flush;
cout#flush;
Queue.push FlushEnded bottomQ;
flushing := true
in
try
(* Once both queues are empty, the reconciliation is done *)
while not (Queue.is_empty requestQ && Queue.is_empty bottomQ) do
match (try Some (Queue.top bottomQ) with Queue.Empty -> None) with
| None ->
(* following pop is safe, because requestQ can't be empty *)
let (node,key) = Queue.pop requestQ in
send_request cout tree ~bottomQ (node,key)
| Some FlushEnded ->
ignore (Queue.pop bottomQ);
flushing := false
| Some (Bottom (node,key)) ->
plerror 10 "Queue length: %d" (Queue.length bottomQ);
match try_unmarshal cin with
| Some reply ->
ignore (Queue.pop bottomQ);
handle_reply cout tree ~requestQ reply (node,key) set
| None ->
match (
if Queue.length bottomQ > !Settings.max_outstanding_recon_requests
then None
else
try Some (Queue.pop requestQ)
with Queue.Empty -> None
)
with
| None ->
if not !flushing then flush_queue ()
else (
ignore (Queue.pop bottomQ);
let reply = unmarshal cin in
handle_reply cout tree ~requestQ reply (node,key) set
)
| Some (node,key) ->
send_request cout tree ~bottomQ (node,key)
done;
marshal cout Done;
!set
with
| Eventloop.SigAlarm ->
ignore (Unix.alarm recover_timeout);
plerror 2 "%s" ("Reconciliation failed due to timeout. " ^
"Returning elements returned so far");
!set
| End_of_file | Sys_error _ as e ->
ignore (Unix.alarm recover_timeout);
eplerror 2 e "%s" ("Reconciliation failed. " ^
"Returning elements returned so far");
!set
(* Main reconciliation code *)
let handle tree cin cout =
flushcount := 0; (* number of round-trips *)
let startkey = Bitstring.create 0 in
connection_manager cin cout tree (PTree.root tree, startkey)
|