File: client.ml

package info (click to toggle)
sks 1.1.6-14
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, buster, sid
  • size: 2,296 kB
  • sloc: ml: 15,228; ansic: 1,069; sh: 358; makefile: 347
file content (206 lines) | stat: -rw-r--r-- 7,664 bytes parent folder | download | duplicates (5)
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
(***********************************************************************)
(* client.ml - Client side of set-reconciliation algorithm             *)
(*                                                                     *)
(* Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, *)
(*               2011, 2012, 2013  Yaron Minsky and Contributors       *)
(*                                                                     *)
(* 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 or see <http://www.gnu.org/licenses/>.                          *)
(***********************************************************************)

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 := (ZZp.Set.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 = ZZp.Set.diff local elements in
        let remotediff = ZZp.Set.diff elements local in
        marshal_noflush cout (Elements localdiff);
        setref := ZZp.Set.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 ZZp.Set.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)