diff --git a/doc/design/rolling-checksums.rst b/doc/design/rolling-checksums.rst new file mode 100644 index 00000000..1182b8ec --- /dev/null +++ b/doc/design/rolling-checksums.rst @@ -0,0 +1,97 @@ +================= +Rolling Checksums +================= + +Problem +======= +If a node crashed, and failed to write some tlog entries to disk, this is not detected by Arakoon. The node announces it's in sync up to the last entry in the tlogs, even if other nodes diverged while the node was offline. Note that this should not happen if fsync is set to true (which is the default) and none of the layers below (file system, hardware) lie about fsync behaviour. + +Another problematic situation occurs when for some reason tlog files, databases or even nodes end up in the wrong cluster. The nodes will continue as if nothing happened, and don't know they are diverged. + +Example +------- +Consider this situation: + ++----------------------------------+------------------------------------+------------------------------------+ +| node0 | node1 | node 2 | ++==================================+====================================+====================================+ +| 0:(Vm (node0,0.000000)) | 0:(Vm (node0,0.000000)) | 0:(Vm (node0,0.000000)) | ++----------------------------------+------------------------------------+------------------------------------+ +| 1:(Vc ([Set;"a";1;"...";],false) | 1:(Vc ([Set;"a";1;"...";],false) | 1:(Vc ([Set;"a";1;"...";],false) | ++----------------------------------+------------------------------------+------------------------------------+ +| 2:(Vc ([Set;"b";1;"...";],false) | *2:(Vc ([Set;"b";1;"...";],false)* | *2:(Vc ([Set;"b";1;"...";],false)* | ++----------------------------------+------------------------------------+------------------------------------+ +| 3:(Vc ([Set;"c";1;"...";],false) | *3:(Vc ([Set;"c";1;"...";],false)* | | ++----------------------------------+------------------------------------+------------------------------------+ + +Node1 and node2 crashed, and the last tlog entries were lost. They are restarted, while node0 is still offline. When node0 comes back, this will result in the following situation: + ++--------------------------------------+----------------------------------+----------------------------------+ +| node0 | node1 | node 2 | ++======================================+==================================+==================================+ +| 0:(Vm (node0,0.000000)) | 0:(Vm (node0,0.000000)) | 0:(Vm (node0,0.000000)) | ++--------------------------------------+----------------------------------+----------------------------------+ +| 1:(Vc ([Set;"a";1;"...";],false) | 1:(Vc ([Set;"a";1;"...";],false) | 1:(Vc ([Set;"a";1;"...";],false) | ++--------------------------------------+----------------------------------+----------------------------------+ +| **2:(Vc ([Set;"b";1;"...";],false)** | **2:(Vm (node1,0.000000))** | **2:(Vm (node1,0.000000))** | ++--------------------------------------+----------------------------------+----------------------------------+ +| 3:(Vc ([Set;"d";1;"...";],false) | 3:(Vc ([Set;"d";1;"...";],false) | 3:(Vc ([Set;"d";1;"...";],false) | ++--------------------------------------+----------------------------------+----------------------------------+ + +Checksums +========= +This problem can be solved by using a rolling checksum, computed over all the entries in the tlogs. This checksum should be the same for all nodes. The checksum is part of the value that is synced with multi-paxos. + +1. The client sends a request to the master node. +2. The master computes the rolling checksum, and makes a value of this checksum and the update commands. +3. This value is sent to the slaves in an accept request. +4. The slaves compute the rolling checksum, and compare it with the checksum in the value. +5. If the checksums are equal, the tlogs are in sync, the value is written to the tlogs, and the algorithm proceeds as usual. +6. If the checksums are different, something bad happened. The node halts, and the tlogs need to be inspected manually. + +The catchup consists of two phases. In the first phase, the missing tlog entries are received from another node. The checksum of the first of these entries will be validated, to prevent a catchup from a diverged node. During the second phase, the tlog entries are replayed to the store, and all checksums are validated. + +Remark +------ +When several consecutive entries in the tlogs have the same number, it is only the last one that is agreed upon by multi-paxos, and thus only this entry is used in the computation of the checksum. + +Tlog Specification +================== +* Serial number (int64) +* Crc-32 checksum of Cmd (int32) +* Cmd + - Value + - Marker, optional (string option) + +Older value format +------------------ +* Update + - Update type (int32 between 1 and 16) + - Update details (depends on type) +* Synced (bool) + +Old value format +---------------- +* 0xff (int32) +* Value type (char 'c' or 'm') +* Value details (depends on type) + +New value format +---------------- +* 0x100 (int32) +* Checksum (int32 if crc-32 is used) +* Value type (char 'c' or 'm') +* Value details (depends on type) + +Checksums in store +================== +The current tlog index and checksum are stored in the local store. When a node starts, they are compared with the values in the tlog. +Thus, after a collapse, the checksum of the last collapsed value is saved in the head database. Therefore the rolling tlog checksum is a continuation of the checksum in the head database. + +Upgrade Path +============ +To upgrade Arakoon to the new version, with a new tlog format, the nodes need to be restarted. A node that restarts after the upgrade can still read the old tlogs, and the checksums of these values will be set to None. Values with checksum None will never be valuated. + +Nodes that need to do a catchup will do this as usual. The first received values will have checksum None, and are written to the tlogs in the old format, until all values from before the upgrade are synced. All values that are created after the upgrade will get a checksum. The checksum of the first value is a normal checksum (not depending on previous values), and all following checksums are rolling. + +New and old nodes can not and will not communicate (they have a different magic). If the nodes are restarted one by one, the old nodes will keep going as long as possible, while the new nodes can't make progress because they don't have a majority. When the critical point is reached, the new nodes will do a catchup and take over. diff --git a/pylabs/test/server/left/system_tests_long_left.py b/pylabs/test/server/left/system_tests_long_left.py index 8d78e79c..ecb8159d 100644 --- a/pylabs/test/server/left/system_tests_long_left.py +++ b/pylabs/test/server/left/system_tests_long_left.py @@ -49,6 +49,7 @@ def test_catchup_only(): C.stopOne(n0) C.compare_stores(n1,n0) C.start_all() + time.sleep(1.0) C.assert_running_nodes(2) diff --git a/pylabs/test/server/quick/test_checksums.py b/pylabs/test/server/quick/test_checksums.py new file mode 100644 index 00000000..5920aeef --- /dev/null +++ b/pylabs/test/server/quick/test_checksums.py @@ -0,0 +1,80 @@ +""" +Copyright (2010-2014) INCUBAID BVBA + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + + + +import time +import shutil + +from .. import system_tests_common as C +from nose.tools import * + +from Compat import X + +""" +@C.with_custom_setup(C.setup_3_nodes, C.basic_teardown) +def test_diverge (): + C.iterate_n_times(100, C.simple_set) + C.stop_all() + + C.remove_node(1) + C.remove_node(2) + C.regenerateClientConfig(C.cluster_id) + C.start_all() + C.iterate_n_times(5, C.simple_set, 100) + C.stop_all() + + C.remove_node(0) + C.add_node(1) + C.add_node(2) + C.regenerateClientConfig(C.cluster_id) + C.start_all() + C.iterate_n_times(10, C.simple_set) + C.stop_all() + + C.add_node(0) + C.regenerateClientConfig(C.cluster_id) + C.start_all() + time.sleep(3.0) +""" + +@C.with_custom_setup(C.setup_2_nodes_forced_master ,C.basic_teardown) +def test_power_failure (): + cluster = C._getCluster() + logging.info("") + C.iterate_n_times(50, C.simple_set) + + for i in range(2): + node_id = C.node_names[i] + C.stopOne(node_id) + home = cluster.getNodeConfig(node_id)['home'] + backup = '/'.join([X.tmpDir, 'backup_' + node_id]) + shutil.copytree(home, backup) + C.startOne(node_id) + + C.iterate_n_times(10, C.simple_set) + C.stop_all() + + for i in range(2): + node_id = C.node_names[i] + home = cluster.getNodeConfig(node_id)['home'] + backup = '/'.join([X.tmpDir, 'backup_' + node_id]) + shutil.rmtree(home) + shutil.move(backup, home) + C.startOne(node_id) + + C.iterate_n_times(10, C.simple_set) + C.startOne(C.node_names[2]) diff --git a/pylabs/test/server/right/system_tests_long_right.py b/pylabs/test/server/right/system_tests_long_right.py index 23877155..e9f3a91c 100644 --- a/pylabs/test/server/right/system_tests_long_right.py +++ b/pylabs/test/server/right/system_tests_long_right.py @@ -508,34 +508,6 @@ def test_sabotage(): returncode = X.subprocess.call(cmd) assert_equals(returncode, 50) -@Common.with_custom_setup( Common.setup_3_nodes_forced_master, Common.basic_teardown ) -def test_large_catchup_while_running(): - """ make sure catchup does not interphere with normal operation (eta : 720s) """ - cli = Common.get_client() - cluster = Common._getCluster() - - cli.set('k','v') - m = cli.whoMaster() - - nod1 = Common.node_names[0] - nod2 = Common.node_names[1] - nod3 = Common.node_names[2] - - n_name,others = (nod1, [nod2,nod3]) if nod1 != m else (nod2, [nod1, nod3]) - node_pid = cluster._getPid(n_name) - - time.sleep(0.1) - X.subprocess.call(["kill","-STOP",str(node_pid)]) - Common.iterate_n_times( 200000, Common.simple_set ) - for n in others: - Common.collapse(n) - - time.sleep(1.0) - X.subprocess.call(["kill","-CONT", str(node_pid) ]) - cli.delete('k') - time.sleep(10.0) - Common.assert_running_nodes(3) - @Common.with_custom_setup(Common.setup_1_node, Common.basic_teardown) def test_log_rotation(): diff --git a/pylabs/test/server/system_tests_common.py b/pylabs/test/server/system_tests_common.py index 407ad728..b0fcce6f 100644 --- a/pylabs/test/server/system_tests_common.py +++ b/pylabs/test/server/system_tests_common.py @@ -537,6 +537,11 @@ def add_node ( i ): cluster.addLocalNode (ni ) cluster.createDirs(ni) +def remove_node(i): + ni = node_names[i] + cluster = _getCluster() + cluster.removeNode(ni) + def start_all(clusterId = None) : cluster = _getCluster(clusterId ) cluster.start() diff --git a/src/client/arakoon_client.mli b/src/client/arakoon_client.mli index 24ad6bf0..362bd926 100644 --- a/src/client/arakoon_client.mli +++ b/src/client/arakoon_client.mli @@ -85,7 +85,7 @@ class type client = object method nop : unit -> unit Lwt.t (** [nop ()] is a paxos no-operation. - *) + *) method confirm: key -> value -> unit Lwt.t (** [confirm key value] does nothing if this value was already associated to the key, otherwise, it behaves as [set key value] @@ -118,7 +118,7 @@ class type client = object [replace key wanted] assigns the wanted value to the key, and returns the previous assignment (if any) for that key. If wanted is None, the binding is deleted. - *) + *) method ping: string -> string -> string Lwt.t diff --git a/src/client/client_protocol.ml b/src/client/client_protocol.ml index 7a6d8b24..7fe9089a 100644 --- a/src/client/client_protocol.ml +++ b/src/client/client_protocol.ml @@ -312,18 +312,24 @@ let one_command stop (ic,oc,id as conn) (backend:Backend.backend) = end | LAST_ENTRIES -> begin - Sn.input_sn ic >>= fun i -> - Logger.debug_f_ "connection=%s LAST_ENTRIES: i=%Li" id i >>= fun () -> - response_ok oc >>= fun () -> - backend # last_entries i oc >>= fun () -> - Lwt.return false + wrap_exception + (fun () -> + let msg = "Operation LAST_ENTRIES is not supported" in + Lwt.fail (XException(Arakoon_exc.E_NOT_SUPPORTED, msg))) end | LAST_ENTRIES2 -> + begin + wrap_exception + (fun () -> + let msg = "Operation LAST_ENTRIES2 is not supported" in + Lwt.fail (XException(Arakoon_exc.E_NOT_SUPPORTED, msg))) + end + | LAST_ENTRIES3 -> begin Sn.input_sn ic >>= fun i -> - Logger.debug_f_ "connection=%s LAST_ENTRIES2: i=%Li" id i >>= fun () -> + Logger.debug_f_ "connection=%s LAST_ENTRIES3: i=%Li" id i >>= fun () -> response_ok oc >>= fun () -> - backend # last_entries2 i oc >>= fun () -> + backend # last_entries i oc >>= fun () -> Lwt.return false end | WHO_MASTER -> diff --git a/src/client/common.ml b/src/client/common.ml index 0eb41dcc..bb4b9850 100644 --- a/src/client/common.ml +++ b/src/client/common.ml @@ -49,6 +49,7 @@ type client_command = | TEST_AND_SET | LAST_ENTRIES | LAST_ENTRIES2 + | LAST_ENTRIES3 | RANGE_ENTRIES | MIGRATE_RANGE | SEQUENCE @@ -133,6 +134,7 @@ let code2int = [ GET_TXID , 0x43l; COPY_DB_TO_HEAD , 0x44l; USER_HOOK , 0x45l; + LAST_ENTRIES3 , 0x46l; ] let int2code = diff --git a/src/client/remote_nodestream.ml b/src/client/remote_nodestream.ml index cd2ec120..734ee2a8 100644 --- a/src/client/remote_nodestream.ml +++ b/src/client/remote_nodestream.ml @@ -25,9 +25,9 @@ let section = Logger.Section.main class type nodestream = object method iterate: - Sn.t -> (Sn.t * Value.t -> unit Lwt.t) -> - Tlogcollection.tlog_collection -> - head_saved_cb:(string -> unit Lwt.t) -> unit Lwt.t + Sn.t -> f_entry:(Sn.t * Value.t -> unit Lwt.t) -> + f_head:(Lwt_io.input_channel -> unit Lwt.t) -> + f_file:(string -> int64 -> Lwt_io.input_channel -> unit Lwt.t) -> unit Lwt.t method collapse: int -> unit Lwt.t @@ -50,18 +50,13 @@ class type nodestream = object method drop_master: unit -> unit Lwt.t end -class remote_nodestream ((ic,oc) as conn) = - (object - method iterate (i:Sn.t) (f: Sn.t * Value.t -> unit Lwt.t) - (tlog_coll: Tlogcollection.tlog_collection) - ~head_saved_cb - = +class remote_nodestream ((ic,oc) as conn) = (object + method iterate i ~f_entry ~f_head ~f_file = let outgoing buf = - command_to buf LAST_ENTRIES2; + command_to buf LAST_ENTRIES3; Sn.sn_to buf i in let incoming ic = - let save_head () = tlog_coll # save_head ic in let last_seen = ref None in let rec loop_entries () = Sn.input_sn ic >>= fun i2 -> @@ -78,7 +73,7 @@ class remote_nodestream ((ic,oc) as conn) = Llio.input_int32 ic >>= fun _chksum -> Llio.input_string ic >>= fun entry -> let value = Value.value_from (Llio.make_buffer entry 0) in - f (i2, value) >>= fun () -> + f_entry (i2, value) >>= fun () -> loop_entries () end end @@ -94,10 +89,8 @@ class remote_nodestream ((ic,oc) as conn) = end | 2 -> begin - Logger.info_f_ "save_head" >>= fun ()-> - save_head () >>= fun () -> - let hf_name = tlog_coll # get_head_name () in - head_saved_cb hf_name >>= fun () -> + Logger.info_f_ "save_head" >>= fun () -> + f_head ic >>= fun () -> loop_parts () end | 3 -> @@ -106,7 +99,7 @@ class remote_nodestream ((ic,oc) as conn) = Llio.input_string ic >>= fun name -> Llio.input_int64 ic >>= fun length -> Logger.info_f_ "got %s (%Li bytes)" name length >>= fun () -> - tlog_coll # save_tlog_file name length ic >>= fun () -> + f_file name length ic >>= fun () -> loop_parts () end | x -> Llio.lwt_failfmt "don't know what %i means" x diff --git a/src/client/remote_nodestream.mli b/src/client/remote_nodestream.mli index 555085f4..86ed14d0 100644 --- a/src/client/remote_nodestream.mli +++ b/src/client/remote_nodestream.mli @@ -21,9 +21,9 @@ open Ncfg class type nodestream = object method iterate: - Sn.t -> (Sn.t * Value.t -> unit Lwt.t) -> - Tlogcollection.tlog_collection -> - head_saved_cb:(string -> unit Lwt.t) -> unit Lwt.t + Sn.t -> f_entry:(Sn.t * Value.t -> unit Lwt.t) -> + f_head:(Lwt_io.input_channel -> unit Lwt.t) -> + f_file:(string -> int64 -> Lwt_io.input_channel -> unit Lwt.t) -> unit Lwt.t method collapse: int -> unit Lwt.t diff --git a/src/main/replay_main.ml b/src/main/replay_main.ml index da1cf4d4..dd6dd7f9 100644 --- a/src/main/replay_main.ml +++ b/src/main/replay_main.ml @@ -49,7 +49,7 @@ let replay_tlogs tlog_dir tlf_dir db_name end_i = | None -> begin Tlc2.get_last_tlog tlog_dir tlf_dir >>= fun (_new_c,fn) -> - Tlc2._validate_one fn "" ~check_marker:false >>= fun (last, _index) -> + Tlc2._validate_one fn "" ~check_marker:false >>= fun (last, _, _index) -> let i = match last with | None -> Sn.start diff --git a/src/msg/tcp_messaging.ml b/src/msg/tcp_messaging.ml index 82fd6432..b2245ead 100644 --- a/src/msg/tcp_messaging.ml +++ b/src/msg/tcp_messaging.ml @@ -69,7 +69,8 @@ class tcp_messaging my_addresses my_cookie (drop_it: drop_function) max_buffer_size ~stop = - let _MAGIC = 0xB0BAFE7L in + (* previous magic: 0xB0BAFE7L *) + let _MAGIC = 0x20140820L in let _VERSION = 1 in let my_ips, my_port = my_addresses in let my_ip = List.hd my_ips in diff --git a/src/node/backend.ml b/src/node/backend.ml index ed3b0454..44c82f74 100644 --- a/src/node/backend.ml +++ b/src/node/backend.ml @@ -52,7 +52,6 @@ class type backend = object method prefix_keys: consistency:consistency -> string -> int -> (Key.t counted_list) method last_entries : Sn.t ->Lwt_io.output_channel -> unit Lwt.t - method last_entries2: Sn.t ->Lwt_io.output_channel -> unit Lwt.t method multi_get: consistency:consistency -> diff --git a/src/node/catchup.ml b/src/node/catchup.ml index 463298b1..f21c3487 100644 --- a/src/node/catchup.ml +++ b/src/node/catchup.ml @@ -86,18 +86,22 @@ let head_saved_epilogue hfn tlog_coll = S.make_store ~lcnum:default_lcnum ~ncnum:default_ncnum ~read_only:true hfn >>= fun store -> let hio = S.consensus_i store in + let hcso = S.get_checksum store in S.close store ~flush:false ~sync:false >>= fun () -> Logger.info_ "closed head" >>= fun () -> begin + tlog_coll # set_previous_checksum hcso; match hio with - | None -> Lwt.return () + | None -> Lwt.return None | Some head_i -> begin Logger.info_f_ "head_i = %s" (Sn.string_of head_i) >>= fun () -> - tlog_coll # remove_below head_i + tlog_coll # remove_below head_i >>= fun () -> + Lwt.return (Some (Sn.succ head_i)) end end + let stop_fuse stop = if !stop then @@ -105,7 +109,8 @@ let stop_fuse stop = else Lwt.return () -let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id mr_name ((module S : Store.STORE with type t = s),store,tlog_coll) +let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id mr_name + ((module S : Store.STORE with type t = s), store, (tlog_coll:Tlogcollection.tlog_collection)) = let current_i = tlog_coll # get_last_i () in Logger.info_f_ "catchup_tlog %s" (Sn.string_of current_i) >>= fun () -> @@ -114,9 +119,32 @@ let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id mr_name ((mo let mr_addresses = Node_cfg.client_addresses mr_cfg and mr_name = Node_cfg.node_name mr_cfg in Logger.info_f_ "getting last_entries from %s" mr_name >>= fun () -> - let head_saved_cb hfn = - Logger.info_f_ "head_saved_cb %s" hfn >>= fun () -> - head_saved_epilogue hfn tlog_coll >>= fun () -> + + let r_validate_i = ref (Some current_i) in + let validate_i () = + let _validate_i = !r_validate_i in + begin + r_validate_i := None; + _validate_i + end + in + + let f_entry (i, value) = + let validate = + match validate_i () with + | None -> false + | Some _ -> true + in + tlog_coll # log_value_explicit i value ~validate false None >>= fun () -> + stop_fuse stop + in + + let f_head ic = + tlog_coll # save_head ic >>= fun () -> + let hfn = tlog_coll # get_head_name () in + Logger.info_f_ "head_saved %s" hfn >>= fun () -> + head_saved_epilogue hfn tlog_coll >>= fun io -> + let () = r_validate_i := io in let when_closed () = Logger.debug_ "when_closed" >>= fun () -> let target_name = S.get_location store in @@ -126,20 +154,20 @@ let catchup_tlog (type s) ~tls_ctx ~stop other_configs ~cluster_id mr_name ((mo Lwt.return () in + let f_file name length ic = + let validate_i = validate_i () in + tlog_coll # save_tlog_file ~validate_i name length ic + in + let copy_tlog connection = make_remote_nodestream cluster_id connection >>= fun (client:nodestream) -> - let f (i,value) = - tlog_coll # log_value_explicit i value false None >>= fun _ -> - stop_fuse stop - in - - client # iterate current_i f tlog_coll ~head_saved_cb + client # iterate current_i ~f_entry ~f_head ~f_file in Lwt.catch (fun () -> - _with_client_connection ~tls_ctx mr_addresses copy_tlog >>= fun () -> - Logger.info_f_ "catchup_tlog completed" + _with_client_connection ~tls_ctx mr_addresses copy_tlog >>= fun () -> + Logger.info_f_ "catchup_tlog completed" ) (fun exn -> Logger.warning_ ~exn "catchup_tlog failed") >>= fun () -> @@ -163,22 +191,29 @@ let make_f ~stop (type s) (module S : Store.STORE with type t = s) me log_i acc in match !acc with | None -> - let () = acc := Some(i,value) in + let () = acc := Some (None, i, value) in Logger.debug_f_ "value %s has no previous" (Sn.string_of i) >>= fun () -> Lwt.return () - | Some (pi,pv) -> + | Some (pivo, pi, pv) -> if pi < i then begin + let () = + match pivo with + | None -> () (* the first value is already validated in catchup_tlog *) + | Some piv -> + if not (Value.is_valid_next piv pv) + then raise (Value.ValueCheckSumError (pi, pv)) + in log_i pi >>= fun () -> prepare_and_insert_value (module S) me store pi pv >>= fun () -> - let () = acc := Some(i,value) in + let () = acc := Some (Some pv, i, value) in Lwt.return () end else begin Logger.debug_f_ "%s => skip" (Sn.string_of pi) >>= fun () -> - let () = acc := Some(i,value) in + let () = acc := Some (pivo, i, value) in Lwt.return () end @@ -186,8 +221,15 @@ let make_f ~stop (type s) (module S : Store.STORE with type t = s) me log_i acc let epilogue (type s) (module S : Store.STORE with type t = s) me acc store = match !acc with | None -> Lwt.return () - | Some(i,value) -> + | Some (pivo, i, value) -> begin + let () = + match pivo with + | None -> () + | Some piv -> + if not (Value.is_valid_next piv value) + then raise (Value.ValueCheckSumError (i, value)) + in Logger.debug_f_ "%s => store" (Sn.string_of i) >>= fun () -> prepare_and_insert_value (module S) me store i value end @@ -295,6 +337,8 @@ let verify_n_catchup_store (type s) ~stop me ?(apply_last_tlog_value=false) ((mo let si_o = S.consensus_i store in Logger.info_f_ "verify_n_catchup_store; too_far_i=%s current_i=%s si_o:%s" (Sn.string_of too_far_i) (Sn.string_of current_i) (io_s si_o) >>= fun () -> + S.validate store tlog_coll >>= fun () -> + begin match too_far_i, si_o with | i, None when i <= 0L -> Lwt.return () | i, Some j when i = j -> Lwt.return () @@ -310,84 +354,14 @@ let verify_n_catchup_store (type s) ~stop me ?(apply_last_tlog_value=false) ((mo Logger.fatal_ msg >>= fun () -> let maybe a = function | None -> a | Some b -> b in Lwt.fail (StoreAheadOfTlogs(maybe (-1L) si_o, too_far_i)) + end let last_entries (type s) (module S : Store.STORE with type t = s) store tlog_collection (start_i:Sn.t) (oc:Lwt_io.output_channel) = - (* This one is kept (for how long?) - for x-version clusters during upgrades - *) - Logger.warning_f_ "DEPRECATED : last_entries " >>= fun () -> Logger.debug_f_ "last_entries %s" (Sn.string_of start_i) >>= fun () -> let consensus_i = S.consensus_i store in - begin - match consensus_i with - | None -> Lwt.return () - | Some ci -> - begin - tlog_collection # get_infimum_i () >>= fun inf_i -> - let too_far_i = Sn.succ ci in - Logger.debug_f_ - "inf_i:%s too_far_i:%s" (Sn.string_of inf_i) - (Sn.string_of too_far_i) - >>= fun () -> - begin - if start_i < inf_i - then - begin - Llio.output_int oc 2 >>= fun () -> - tlog_collection # dump_head oc - end - else - Lwt.return start_i - end - >>= fun start_i2-> - - - let step = Sn.of_int (!Tlogcommon.tlogEntriesPerFile) in - let rec loop_parts (start_i2:Sn.t) = - if Sn.rem start_i2 step = Sn.start && - Sn.add start_i2 step < too_far_i - then - begin - Logger.debug_f_ "start_i2=%Li < %Li" start_i2 too_far_i - >>= fun () -> - Llio.output_int oc 3 >>= fun () -> - tlog_collection # dump_tlog_file start_i2 oc - >>= fun start_i2' -> - loop_parts start_i2' - end - else - Lwt.return start_i2 - in - loop_parts start_i2 - >>= fun start_i3 -> - Llio.output_int oc 1 >>= fun () -> - let f entry = - let i = Entry.i_of entry - and v = Entry.v_of entry - in - Tlogcommon.write_entry oc i v - in - Lwt.catch - (fun () -> tlog_collection # iterate start_i3 too_far_i f) - (function - | Tlogcommon.TLogUnexpectedEndOfFile _ -> Lwt.return () - | ex -> Lwt.fail ex) >>= fun () -> - Sn.output_sn oc (-1L) - end - end - >>= fun () -> - Logger.info_f_ "done with_last_entries" - - -let last_entries2 - (type s) (module S : Store.STORE with type t = s) - store tlog_collection (start_i:Sn.t) (oc:Lwt_io.output_channel) - = - Logger.debug_f_ "last_entries2 %s" (Sn.string_of start_i) >>= fun () -> - let consensus_i = S.consensus_i store in begin match consensus_i with diff --git a/src/node/catchup_test.ml b/src/node/catchup_test.ml index b39035a0..3ab8345c 100644 --- a/src/node/catchup_test.ml +++ b/src/node/catchup_test.ml @@ -38,8 +38,9 @@ let _fill tlog_coll n = let k = Printf.sprintf "key%i" i and v = Printf.sprintf "value%i" i in let u = Update.Set (k,v) in - let value = Value.create_client_value [u] sync in - tlog_coll # log_value (Sn.of_int i) value >>= fun () -> + let sni = Sn.of_int i in + let value = Value.create_client_value tlog_coll sni [u] sync in + tlog_coll # log_value sni value >>= fun () -> _loop (i+1) end in @@ -58,11 +59,11 @@ let _fill2 tlog_coll n = in let u = Update.Set(k,v) in let u2 = Update.Set(k2,v2) in - let value = Value.create_client_value [u] sync in - let value2 = Value.create_client_value [u2] sync in let sni = Sn.of_int i in - tlog_coll # log_value sni value >>= fun () -> - tlog_coll # log_value sni value2 >>= fun () -> + let value = Value.create_client_value tlog_coll sni [u] sync in + let value2 = Value.create_client_value tlog_coll sni [u2] sync in + tlog_coll # log_value sni value >>= fun () -> + tlog_coll # log_value sni value2 >>= fun () -> _loop (i+1) end in @@ -84,9 +85,9 @@ let _fill3 tlog_coll n = let u = Update.Set(k,v) in let u2 = Update.Set(k2,v2) in let u3 = Update.Sequence [Update.Set(k3,v3); Update.Assert_exists("nonExistingKey")] in - let value = Value.create_client_value [u; u3] sync in - let value2 = Value.create_client_value [u2; u3] sync in let sni = Sn.of_int i in + let value = Value.create_client_value tlog_coll sni [u; u3] sync in + let value2 = Value.create_client_value tlog_coll sni [u2; u3] sync in tlog_coll # log_value sni value >>= fun () -> tlog_coll # log_value sni value2 >>= fun () -> _loop (i+1) diff --git a/src/node/collapser_test.ml b/src/node/collapser_test.ml index 0a356d38..f6b43fe0 100644 --- a/src/node/collapser_test.ml +++ b/src/node/collapser_test.ml @@ -25,6 +25,12 @@ let compressor = Compression.Snappy module S = (val (Store.make_store_module (module Batched_store.Local_store))) + +let test_dn = "/tmp/collapser" +let _tlf_dir = "/tmp/collapser_tlf" +let _head_dir = "/tmp/collapser_head" + + let _should_fail x error_msg success_msg = Lwt.catch (fun () -> @@ -47,8 +53,8 @@ let _make_values tlc n = let k = Printf.sprintf "sqrt(%i)" a in let v = Printf.sprintf "%f" (sqrt (float a)) in let update = Update.Set(k, v) in - let value = Value.create_client_value [update] sync in let sni = Sn.of_int i in + let value = Value.create_client_value tlc sni [update] sync in tlc # log_value sni value >>= fun _wr_result -> loop (i+1) in @@ -63,7 +69,7 @@ let test_collapse_until (dn, tlf_dir, head_dir) = tlc # close () >>= fun () -> Lwt_unix.sleep 5.0 >>= fun () -> (* give it time to generate the .tlc *) (* now collapse first file into a tc *) - let storename = "head.db" in + let storename = Filename.concat test_dn "head.db" in File_system.unlink storename >>= fun () -> let store_methods = (Batched_store.Local_store.copy_store2, storename, 0.0) in @@ -83,10 +89,6 @@ let test_collapse_until (dn, tlf_dir, head_dir) = Lwt.return () -let test_dn = "/tmp/collapser" -let _tlf_dir = "/tmp/collapser_tlf" -let _head_dir = "/tmp/collapser_head" - let test_collapse_many (dn, tlf_dir, head_dir) = let () = Tlogcommon.tlogEntriesPerFile := 100 in Logger.debug_f_ "test_collapse_many_regime dn=%s, tlf_dir=%s, head_dir=%s" dn tlf_dir head_dir >>= fun () -> diff --git a/src/node/node_main.ml b/src/node/node_main.ml index 7e564080..be9019a5 100644 --- a/src/node/node_main.ml +++ b/src/node/node_main.ml @@ -301,7 +301,7 @@ module X = struct let vni' = v',n,i in begin match v' with - | Value.Vm (m, _) -> + | (_, Value.Vm (m, _)) -> let now = Int64.of_float (Unix.gettimeofday ()) in let m_old_master = S.who_master store in let new_master = @@ -318,7 +318,7 @@ module X = struct end else Lwt.return () - | Value.Vc _ -> Lwt.return () + | (_, Value.Vc _) -> Lwt.return () end >>= fun () -> S.on_consensus store vni' >>= fun r -> let t1 = Unix.gettimeofday () in @@ -336,11 +336,11 @@ module X = struct tlog_coll # log_value_explicit i v sync marker >>= fun () -> begin match v with - | Value.Vc (us,_) -> + | (_, Value.Vc (us,_)) -> let size = List.length us in let () = Statistics.new_harvest statistics size in Lwt.return () - | Value.Vm _ -> Lwt.return () + | (_, Value.Vm _) -> Lwt.return () end >>= fun () -> let t1 = Unix.gettimeofday() in let d = t1 -. t0 in @@ -903,6 +903,18 @@ let _main_2 (type s) let rc = 50 in Logger.fatal_f_ "[rc=%i] Somebody has been messing with the available tlogs" rc >>= fun () -> Lwt.return rc + | Value.ValueCheckSumError (i, value) -> + let rc = 51 in + Logger.fatal_f_ "[rc=%i] Value has a checksum error, tlog diverged from other nodes: sn=%s, value=%s" + rc (Sn.string_of i) (Value.value2s value) >>= fun () -> + Lwt.return rc + | Store.StoreChecksumError (i, scs, tcs) -> + let rc = 52 in + Logger.fatal_f_ + "[rc=%i] Store and tlog have different checksums: i=%s, store_cs=%s, tlog_cs=%s" + rc (Sn.string_of i) (Log_extra.option2s Checksum.Crc32.string_of scs) + (Log_extra.option2s Checksum.Crc32.string_of tcs) >>= fun () -> + Lwt.return rc | exn -> begin Logger.fatal_ ~exn "going down" >>= fun () -> diff --git a/src/node/simple_store.ml b/src/node/simple_store.ml index 5e52ed12..a30ed66d 100644 --- a/src/node/simple_store.ml +++ b/src/node/simple_store.ml @@ -45,6 +45,7 @@ let __j_key = "*j" let __interval_key = "*interval" let __routing_key = "*routing" let __master_key = "*master" +let __checksum_key = "*checksum" (* let __lease_key = "*lease" *) (* let __lease_key2 = "*lease2" *) let __prefix = "@" diff --git a/src/node/store.ml b/src/node/store.ml index 7f276700..b87e3d47 100644 --- a/src/node/store.ml +++ b/src/node/store.ml @@ -31,6 +31,7 @@ type update_result = | Update_fail of Arakoon_exc.rc * string exception CorruptStore +exception StoreChecksumError of (Sn.t * Checksum.Crc32.t option * Checksum.Crc32.t option) type key_or_transaction = | Key of transaction_lock @@ -42,6 +43,8 @@ sig type t val make_store : lcnum:int -> ncnum:int -> ?read_only:bool -> string -> t Lwt.t val consensus_i : t -> Sn.t option + val get_checksum : t -> Checksum.Crc32.t option + val validate : t -> Tlogcollection.tlog_collection -> unit Lwt.t val flush : t -> unit Lwt.t val close : ?flush : bool -> ?sync:bool -> t -> unit Lwt.t val get_location : t -> string @@ -64,8 +67,6 @@ sig val get_succ_store_i : t -> int64 val get_catchup_start_i : t -> int64 - val incr_i : t -> unit Lwt.t - val set_master : t -> transaction -> string -> float -> unit Lwt.t val set_master_no_inc : t -> string -> float -> unit Lwt.t val clear_self_master : t -> string -> unit @@ -297,6 +298,33 @@ struct let consensus_i store = store.store_i + let get_checksum store = + try + let cs_string = S.get store.s __checksum_key in + let cs = Checksum.Crc32.checksum_from (Llio.make_buffer cs_string 0) in + Some cs + with Not_found -> + None + + let validate store tlog_coll = + let io = _consensus_i store.s in + match io with + | None -> Lwt.return () + | Some i -> + let entry = ref None in + let check e = Lwt.return (entry := Some e) in + tlog_coll # iterate i (Sn.succ i) check >>= fun () -> + let store_cs = get_checksum store in + let tlog_cs = + match !entry with + | None -> None + | Some e -> Value.checksum_of (Tlogcommon.Entry.v_of e) in + if store_cs <> tlog_cs + then + Lwt.fail (StoreChecksumError (i, store_cs, tlog_cs)) + else + Lwt.return () + let _get_j store = try let jstring = S.get store.s __j_key in @@ -306,41 +334,33 @@ struct let _set_j store tx j = S.set store.s tx __j_key (string_of_int j) - let _new_i = function - | None -> Sn.start - | Some i -> Sn.succ i - - let _incr_i store tx = - let old_i = _consensus_i store.s in - let new_i = _new_i old_i in - let new_is = + let _set_i_checksum store tx (i, cso) = + let is = let buf = Buffer.create 10 in - let () = Sn.sn_to buf new_i in + let () = Sn.sn_to buf i in Buffer.contents buf in - let () = S.set store.s tx __i_key new_is in + let csso = + match cso with + | None -> None + | Some cs -> + let buf = Buffer.create 4 in + let () = Checksum.Crc32.checksum_to buf cs in + Some (Buffer.contents buf) + in + let () = S.set store.s tx __i_key is in + let () = S.put store.s tx __checksum_key csso in let () = _set_j store tx 0 in - store.store_i <- Some new_i; - Logger.debug_f_ "Store.incr_i old_i:%s -> new_i:%s" - (Log_extra.option2s Sn.string_of old_i) (Sn.string_of new_i) - - let incr_i store = - if quiesced store - then - begin - let new_i = _new_i store.store_i in - store.store_i <- Some new_i; - Lwt.return () - end - else - S.with_transaction store.s (fun tx -> _incr_i store tx) + store.store_i <- Some i; + Logger.debug_f_ "Store.set_i_checksum i:%s, checksum:%s" + (Sn.string_of i) (Log_extra.option2s Checksum.Crc32.string_of cso) let _set_i store i = if quiesced store then store.store_i <- Some i else - failwith "_set_i is only meant to be used on a quiesced store to cheat with tlog replay" + failwith "_set_i is only meant to be used on a quiesced store" let _with_transaction_lock store f = Lwt_mutex.with_lock store._tx_lock_mutex (fun () -> @@ -863,7 +883,8 @@ struct let f u = _insert_update store u kt in Lwt_list.map_s f us - let _insert_value store (value:Value.t) kt = + let _insert_value store i value kt = + let cs = Value.checksum_of value in let updates = Value.updates_from_value value in let j = _get_j store in let rec skip n l = @@ -881,7 +902,7 @@ struct end >>= fun () -> _insert_updates store updates' kt >>= fun (urs:update_result list) -> - _with_transaction store kt (fun tx -> _incr_i store tx) >>= fun () -> + _with_transaction store kt (fun tx -> _set_i_checksum store tx (i, cs)) >>= fun () -> let prepend_oks n l = let rec inner l = function | 0 -> l @@ -908,12 +929,12 @@ struct if quiesced store then begin - incr_i store >>= fun () -> + _set_i store i; Lwt.return [Ok None] end else begin - _insert_value store value kt + _insert_value store i value kt end in inner t) @@ -946,16 +967,19 @@ struct begin begin match v with - | Value.Vm (m, ls) -> set_master_no_inc store m ls - | Value.Vc _ -> Lwt.return () + | (_, Value.Vm (m, ls)) -> set_master_no_inc store m ls + | (_, Value.Vc _) -> Lwt.return () end >>= fun () -> - incr_i store >>= fun () -> + _set_i store i; Lwt.return [Ok None] end else - _with_transaction_lock store (fun key -> _insert_value store v (Key key))) + _with_transaction_lock store (fun key -> _insert_value store i v (Key key))) - let get_succ_store_i store = _new_i (consensus_i store) + let get_succ_store_i store = + match consensus_i store with + | None -> Sn.start + | Some i -> Sn.succ i let get_catchup_start_i = get_succ_store_i diff --git a/src/node/store_test.ml b/src/node/store_test.ml index 403f0e4f..4413b3da 100644 --- a/src/node/store_test.ml +++ b/src/node/store_test.ml @@ -78,14 +78,14 @@ let assert_not_exists k (store:S.t) = let failing_value = let k2 = "key2" and v2 = "value2" in - Value.create_client_value [Update.Sequence [Update.Set(k2, v2); + Value.create_client_value_nocheck [Update.Sequence [Update.Set(k2, v2); Update.Assert_exists "notExists"]] false let value_asserts = let k1 = "key1" and v1 = "value1" in - [(Value.create_client_value [Update.Set(k1,v1)] false, + [(Value.create_client_value_nocheck [Update.Set(k1,v1)] false, [assert_k_v k1 v1]); (failing_value, [assert_not_exists "key2"]);] @@ -114,7 +114,7 @@ let test_safe_insert_value_with_partial_value_update () = let u1 = Update.TestAndSet(k, Some "value1", Some "illegal") and u2 = Update.TestAndSet(k, None, Some "value1") and u3 = Update.Set("key2", "bla") in - let paxos_value = Value.create_client_value [u1;u2;u3] false in + let paxos_value = Value.create_client_value_nocheck [u1;u2;u3] false in S._with_transaction_lock store (fun k -> S._insert_update store u1 (Store.Key k)) >>= fun _ -> S._with_transaction_lock store (fun k -> S._insert_update store u2 (Store.Key k)) >>= fun _ -> diff --git a/src/node/sync_backend.ml b/src/node/sync_backend.ml index aa198649..591486f6 100644 --- a/src/node/sync_backend.ml +++ b/src/node/sync_backend.ml @@ -195,12 +195,6 @@ struct Catchup.last_entries (module S) store tlog_collection start_i oc ) - method last_entries2 (start_i:Sn.t) (oc:Lwt_io.output_channel) = - self # with_blocked_collapser start_i - (fun () -> - Catchup.last_entries2 (module S) store tlog_collection start_i oc - ) - method range_entries ~consistency (first:string option) finc (last:string option) linc max = let start = Unix.gettimeofday() in diff --git a/src/paxos/master.ml b/src/paxos/master.ml index e58af51d..658901cf 100644 --- a/src/paxos/master.ml +++ b/src/paxos/master.ml @@ -40,11 +40,11 @@ let master_consensus (type s) constants {mo;v;n;i; lew} () = in let inject_e = EGen (fun () -> match v with - | Value.Vm _ -> + | (_, Value.Vm _) -> let event = Multi_paxos.FromClient [(Update.Nop, fun _ -> Lwt.return ())] in Lwt.ignore_result (constants.inject_event event); Lwt.return () - | Value.Vc _ -> + | (_, Value.Vc _) -> begin let inject_lease_expired ls = let event = Multi_paxos.LeaseExpired (ls) in @@ -116,7 +116,7 @@ let stable_master (type s) constants ((n,new_i, lease_expire_waiters) as current end else (* if is_empty lease_expire_waiters *) let log_e = ELog (fun () -> "stable_master: half-lease_expired: update lease." ) in - let v = Value.create_master_value me in + let v = Value.create_master_value constants.tlog_coll new_i me in let ms = {mo = None; v;n;i = new_i;lew = []} in Fsm.return ~sides:[log_e] (Master_dictate ms) in @@ -137,7 +137,7 @@ let stable_master (type s) constants ((n,new_i, lease_expire_waiters) as current begin let updates, finished_funs = List.split ufs in let synced = List.fold_left (fun acc u -> acc || Update.is_synced u) false updates in - let v = Value.create_client_value updates synced in + let v = Value.create_client_value constants.tlog_coll new_i updates synced in let ms = {mo = Some finished_funs;v;n;i = new_i; lew = lease_expire_waiters} in diff --git a/src/paxos/multi_paxos_fsm.ml b/src/paxos/multi_paxos_fsm.ml index a384cc73..f48915e2 100644 --- a/src/paxos/multi_paxos_fsm.ml +++ b/src/paxos/multi_paxos_fsm.ml @@ -71,7 +71,7 @@ let promises_check_done constants state () = let bv,bf = begin match v_s with - | [] -> Value.create_master_value me, 0 + | [] -> Value.create_master_value constants.tlog_coll i me, 0 | hd::_ -> hd end in let nnodes = List.length constants.others + 1 in diff --git a/src/paxos/value.ml b/src/paxos/value.ml index 1182f701..bf2a49eb 100644 --- a/src/paxos/value.ml +++ b/src/paxos/value.ml @@ -18,90 +18,163 @@ limitations under the License. open Update -type t = +type content = | Vc of (Update.t list * bool) (* is_synced *) | Vm of (string * float) -let create_client_value (us:Update.t list) (synced:bool) = Vc (us, synced) -let create_master_value - ?(lease_start = Unix.gettimeofday ()) - m = - Vm (m,lease_start) +type t = Checksum.Crc32.t option * content + +exception ValueCheckSumError of Sn.t * t + +let content_to buf = function + | Vc (us,synced) -> begin + Llio.char_to buf 'c'; + Llio.bool_to buf synced; + Llio.list_to buf Update.to_buffer us + end + | Vm (m,l) -> begin + Llio.char_to buf 'm'; + Llio.string_to buf m; + Llio.int64_to buf (Int64.of_float l) + end + +let value_to buf (cso, c) = + match cso with + | None -> begin + Llio.int_to buf 0xff; + content_to buf c + end + | Some cs -> begin + Llio.int_to buf 0x100; + Checksum.Crc32.checksum_to buf cs; + content_to buf c + end + +let content_from buf = + let c = Llio.char_from buf in + match c with + | 'c' -> + let synced = Llio.bool_from buf in + let us = Llio.list_from buf Update.from_buffer in + Vc (us, synced) + | 'm' -> + let m = Llio.string_from buf in + let l = Llio.int64_from buf in + Vm (m, Int64.to_float l) + | _ -> failwith "demarshalling error" + +let value_from buf = + let pos = Llio.buffer_pos buf in + match Llio.int_from buf with + | 0x100 -> + let cs = Checksum.Crc32.checksum_from buf in + (Some cs, content_from buf) + | 0xff -> (None, content_from buf) + | _ -> + (* this is for backward compatibility: + formerly, we logged updates iso values *) + let () = Llio.buffer_set_pos buf pos in + let u = Update.from_buffer buf in + let synced = Update.is_synced u in + (None, Vc ([u], synced)) + +let value2s ?(values=false) (cso, c) = + let css = match cso with + | None -> "_" + | Some cs -> Checksum.Crc32.string_of cs + in + match c with + | Vc (us,synced) -> + let uss = Log_extra.list2s (fun u -> Update.update2s u ~values) us in + Printf.sprintf "(%s, Vc (%s,%b)" css uss synced + | Vm (m,l) -> Printf.sprintf "(%s, Vm (%s,%f))" css m l + +let _string_of_content c = + let buf = Buffer.create 64 in + let () = content_to buf c in + Buffer.contents buf + +let checksum tlog_coll i c = + let s = _string_of_content c in + let cso = tlog_coll # get_previous_checksum i in + Checksum.Crc32.update cso s + +let create_client_value_nocheck us synced = (None, Vc (us, synced)) -let is_master_set = function - | Vc _ -> false - | Vm _ -> true +let create_master_value_nocheck + ?(lease_start = Unix.gettimeofday ()) + m = + (None, Vm (m, lease_start)) +let create_first_value c = + let s = _string_of_content c in + let cs = Checksum.Crc32.calculate s in + (Some cs, c) + +let create_value tlog_coll i c = + let cs = checksum tlog_coll i c in + (Some cs, c) + +let create_first_client_value us synced = + let c = Vc (us, synced) in + create_first_value c + +let create_client_value tlog_coll i us synced = + let c = Vc (us, synced) in + create_value tlog_coll i c + +let create_first_master_value + ?(lease_start = Unix.gettimeofday ()) + m = + let c = Vm (m, lease_start) in + create_first_value c + +let create_master_value + tlog_coll i + ?(lease_start = Unix.gettimeofday ()) + m = + let c = Vm (m, lease_start) in + create_value tlog_coll i c + +let is_valid tlog_coll i (cso, c) = + match cso with + | None -> true + | Some cs -> cs = checksum tlog_coll i c + +let is_valid_next (pcso, _) (cso, c) = + match cso with + | None -> true + | Some cs -> + let s = _string_of_content c in + let cs' = Checksum.Crc32.update pcso s in + cs = cs' + +let is_master_set = function + | (_, Vc _) -> false + | (_, Vm _) -> true let is_other_master_set me = function - | Vm (m, _) -> m <> me - | Vc _ -> false + | (_, Vm (m, _)) -> m <> me + | (_, Vc _) -> false let is_synced = function - | Vc (_,s) -> s - | Vm _ -> false + | (_, Vc (_,s)) -> s + | (_, Vm _) -> false let clear_self_master_set me = function - | Vm (m,_) when m = me -> Vm(m, 0.0) - | Vc _ - | Vm _ as v -> v + | (cso, Vm (m, _)) when m = me -> (cso, Vm (m, 0.)) + | (_, Vc _) + | (_, Vm _) as v -> v let fill_other_master_set me = function - | Vm (m,_) when m <> me -> - let now = Unix.gettimeofday () in - Vm(m,now) - | Vc _ - | Vm _ as v -> v + | (cso, Vm (m, _)) when m <> me -> + let lease_start = Unix.gettimeofday () in + (cso, Vm (m, lease_start)) + | (_, Vc _) + | (_, Vm _) as v -> v let updates_from_value = function - | Vc (us,_) -> us - | Vm (m,l) -> [Update.MasterSet(m,l)] - -let value_to buf v= - let () = Llio.int_to buf 0xff in - match v with - | Vc (us,synced) -> - Llio.char_to buf 'c'; - Llio.bool_to buf synced; - Llio.list_to buf Update.to_buffer us - | Vm (m,l) -> - begin - Llio.char_to buf 'm'; - Llio.string_to buf m; - Llio.int64_to buf (Int64.of_float l) - end - -let value_from b = - let pos = Llio.buffer_pos b in - let i0 = Llio.int_from b in - if i0 = 0xff - then - let c = Llio.char_from b in - match c with - | 'c' -> - let synced = Llio.bool_from b in - let us = Llio.list_from b Update.from_buffer in - let r = Vc(us,synced) in - r - | 'm' -> let m = Llio.string_from b in - let l = Llio.int64_from b in - Vm (m,Int64.to_float l) - | _ -> failwith "demarshalling error" - else - begin - (* this is for backward compatibility: - formerly, we logged updates iso values *) - let () = Llio.buffer_set_pos b pos in - let u = Update.from_buffer b in - let synced = Update.is_synced u in - let r = Vc ([u], synced) in - r - end - + | (_, Vc (us,_)) -> us + | (_, Vm (m,l)) -> [Update.MasterSet(m,l)] - -let value2s ?(values=false) = function - | Vc (us,synced) -> - let uss = Log_extra.list2s (fun u -> Update.update2s u ~values) us in - Printf.sprintf "(Vc (%s,%b)" uss synced - | Vm (m,l) -> Printf.sprintf "(Vm (%s,%f))" m l +let checksum_of (cso, _) = cso diff --git a/src/system/startup.ml b/src/system/startup.ml index 770e7be5..eb0e0610 100644 --- a/src/system/startup.ml +++ b/src/system/startup.ml @@ -79,8 +79,9 @@ let _make_tlog_coll ~compressor tlcs values tlc_name tlf_dir head_dir Mem_tlogcollection.make_mem_tlog_collection tlc_name tlf_dir head_dir ~fsync node_id ~fsync_tlog_dir >>= fun tlc -> let rec loop i = function | [] -> Lwt.return () - | v :: vs -> + | (_, c) :: vs -> begin + let v = Value.create_value tlc i c in tlc # log_value i v >>= fun () -> loop (Sn.succ i) vs end @@ -92,7 +93,7 @@ let _make_tlog_coll ~compressor tlcs values tlc_name tlf_dir head_dir let stop = ref (ref false) let node_ts = ref [] -let _make_run ~stores ~tlcs ~now ~values ~get_cfgs name () = +let _make_run_rc ~stores ~tlcs ~now ~values ~get_cfgs name () = let module S = struct include LS @@ -112,11 +113,14 @@ let _make_run ~stores ~tlcs ~now ~values ~get_cfgs name () = ~name ~daemonize:false ~catchup_only:false - ~stop:!stop - >>= fun _ -> Lwt.return () in - node_ts := t :: !node_ts; + ~stop:!stop in + node_ts := (Lwt.map ignore t) :: !node_ts; t +let _make_run ~stores ~tlcs ~now ~values ~get_cfgs name () = + _make_run_rc ~stores ~tlcs ~now ~values ~get_cfgs name () >>= fun _ -> + Lwt.return () + let _dump_tlc ~tlcs node = let tlc0 = Hashtbl.find tlcs node in let printer entry = @@ -131,9 +135,9 @@ let _dump_tlc ~tlcs node = let post_failure () = let lease_period = 2 in - let node0 = "was_master" in - let node1 = "was_slave1" in - let node2 = "was_slave2" in + let node0 = "post_failure_was_master" in + let node1 = "post_failure_was_slave1" in + let node2 = "post_failure_was_slave2" in let node0_cfg = _make_cfg node0 0 lease_period in let node1_cfg = _make_cfg node1 1 lease_period in let node2_cfg = _make_cfg node2 2 lease_period in @@ -157,8 +161,8 @@ let post_failure () = } in let get_cfgs () = cluster_cfg in - let v0 = Value.create_master_value ~lease_start:0. node0 in - let v1 = Value.create_client_value [Update.Set("x","y")] false in + let v0 = Value.create_master_value_nocheck ~lease_start:0. node0 in + let v1 = Value.create_client_value_nocheck [Update.Set("x","y")] false in let tlcs = Hashtbl.create 5 in let stores = Hashtbl.create 5 in let now = Unix.gettimeofday () in @@ -191,9 +195,9 @@ let post_failure () = let restart_slaves () = let lease_period = 2 in - let node0 = "slave0" in - let node1 = "slave1" in - let node2 = "was_master" in + let node0 = "restart_slaves_slave0" in + let node1 = "restart_slaves_slave1" in + let node2 = "restart_slaves_was_master" in let node0_cfg = _make_cfg node0 0 lease_period in let node1_cfg = _make_cfg node1 1 lease_period in let node2_cfg = _make_cfg node2 2 lease_period in @@ -217,8 +221,8 @@ let restart_slaves () = } in let get_cfgs () = cluster_cfg in - let v0 = Value.create_master_value ~lease_start:0. node2 in - let v1 = Value.create_client_value [Update.Set("xxx","xxx")] false in + let v0 = Value.create_master_value_nocheck ~lease_start:0. node2 in + let v1 = Value.create_client_value_nocheck [Update.Set("xxx","xxx")] false in let tlcs = Hashtbl.create 5 in let stores = Hashtbl.create 5 in let now = Unix.gettimeofday () in @@ -249,9 +253,9 @@ let restart_slaves () = let ahead_master_loses_role () = let lease_period = 2 in - let node0 = "slave0" in - let node1 = "slave1" in - let node2 = "was_master" in + let node0 = "ahead_master_loses_role_slave0" in + let node1 = "ahead_master_loses_role_slave1" in + let node2 = "ahead_master_loses_role_was_master" in let node0_cfg = _make_cfg node0 0 lease_period in let node1_cfg = _make_cfg node1 1 lease_period in let node2_cfg = _make_cfg node2 2 lease_period in @@ -275,9 +279,9 @@ let ahead_master_loses_role () = } in let get_cfgs () = cluster_cfg in - let v0 = Value.create_master_value ~lease_start:0. node0 in - let v1 = Value.create_client_value [Update.Set("xxx","xxx")] false in - let v2 = Value.create_client_value [Update.Set("invalidkey", "shouldnotbepresent")] false in + let v0 = Value.create_master_value_nocheck ~lease_start:0. node0 in + let v1 = Value.create_client_value_nocheck [Update.Set("xxx","xxx")] false in + let v2 = Value.create_client_value_nocheck [Update.Set("invalidkey", "shouldnotbepresent")] false in let tlcs = Hashtbl.create 5 in let stores = Hashtbl.create 5 in let now = Unix.gettimeofday () in @@ -309,20 +313,12 @@ let ahead_master_loses_role () = Lwt_list.iter_s check_store [node0;node1;node2] -let setup () = Lwt.return () -let teardown () = - !stop := true; - stop := ref false; - Lwt.join !node_ts >>= fun () -> - node_ts := []; - Logger.debug_ "teardown" - let interrupted_election () = let lease_period = 4 in let cluster_id = "ricky" in - let wannabe_master = "wannabe_master" in - let node2 = "node2" in - let node3 = "node3" in + let wannabe_master = "interrupted_election_wannabe_master" in + let node2 = "interrupted_election_node2" in + let node3 = "interrupted_election_node3" in let wannabe_master_cfg = _make_cfg wannabe_master 0 lease_period in let node2_cfg = _make_cfg node2 1 lease_period in let node3_cfg = _make_cfg node3 2 lease_period in @@ -346,7 +342,7 @@ let interrupted_election () = } in let get_cfgs () = cluster_cfg in - let v0 = Value.create_master_value ~lease_start:0. wannabe_master in + let v0 = Value.create_master_value_nocheck ~lease_start:0. wannabe_master in let tlcs = Hashtbl.create 5 in let stores = Hashtbl.create 5 in let now = Unix.gettimeofday () in @@ -428,6 +424,76 @@ let interrupted_election () = Lwt_list.iter_s (_dump_tlc ~tlcs) [node2; node3] +let power_failure () = + let lease_period = 2 in + let node0 = "power_failure_slave0" in + let node1 = "power_failure_slave1" in + let node2 = "power_failure_was_master" in + let node0_cfg = _make_cfg node0 0 lease_period in + let node1_cfg = _make_cfg node1 1 lease_period in + let node2_cfg = _make_cfg node2 2 lease_period in + let cluster_cfg = + {cfgs = [node0_cfg;node1_cfg;node2_cfg]; + log_cfgs = [_make_log_cfg ()]; + batched_transaction_cfgs = [_make_batched_transaction_cfg ()]; + _master = Elected; + quorum_function = Quorum.quorum_function; + _lease_period = 2; + cluster_id = "ricky"; + plugins = []; + nursery_cfg = None; + overwrite_tlog_entries = None; + max_value_size = Node_cfg.default_max_value_size; + max_buffer_size = Node_cfg.default_max_buffer_size; + client_buffer_capacity = Node_cfg.default_client_buffer_capacity; + lcnum = 8192; + ncnum = 4096; + tls = None; + } + in + let get_cfgs () = cluster_cfg in + let v0 = Value.create_master_value_nocheck ~lease_start:0. node2 in + let v1 = Value.create_client_value_nocheck [Update.Set("a","a")] false in + let v2 = Value.create_client_value_nocheck [Update.Set("b","b")] false in + let v3 = Value.create_client_value_nocheck [Update.Set("c","c")] false in + let tlcs = Hashtbl.create 5 in + let stores = Hashtbl.create 5 in + let now = Unix.gettimeofday () in + + let run_node0 = _make_run ~stores ~tlcs ~now ~get_cfgs ~values:[v0;v1] node0 in + let run_node1 = _make_run ~stores ~tlcs ~now ~get_cfgs ~values:[v0;v1] node1 in + let run_previous_master () = + Lwt_unix.sleep 2.0 >>= fun () -> + _make_run_rc ~stores ~tlcs ~now ~get_cfgs ~values:[v0;v1;v2;v3] node2 () >>= function + | 51 -> Lwt.return () + | rc -> + let msg = Printf.sprintf "it threw the wrong exception: %i" rc in + Lwt.return (OUnit.assert_bool msg false) + in + let eventually_stop () = + Lwt_unix.sleep 5.0 >>= fun () -> + Lwt.return (OUnit.assert_bool "node power_failure_was_master should have failed" false) + in + + Logger.debug_ "start of scenario" >>= fun () -> + Lwt.pick [run_node0 (); + run_node1 (); + begin Lwt_unix.sleep 2.0 >>= fun () -> run_previous_master () end; + eventually_stop (); + ] + >>= fun () -> + Logger.debug_ "end of scenario" >>= fun () -> + Lwt_list.iter_s (_dump_tlc ~tlcs) [node0; node1; node2] + + +let setup () = Lwt.return () +let teardown () = + !stop := true; + stop := ref false; + Lwt.join !node_ts >>= fun () -> + node_ts := []; + Logger.debug_ "teardown" + let w f = Extra.lwt_bracket setup f teardown let suite = "startup" >:::[ @@ -435,4 +501,5 @@ let suite = "startup" >:::[ "restart_slaves" >:: w restart_slaves; "ahead_master_loses_role" >:: w ahead_master_loses_role; "interrupted_election" >:: w interrupted_election; + "power_failure" >:: w power_failure; ] diff --git a/src/tlog/compression_test.ml b/src/tlog/compression_test.ml index 1a68d123..942d6bb1 100644 --- a/src/tlog/compression_test.ml +++ b/src/tlog/compression_test.ml @@ -38,7 +38,7 @@ let test_compress_file which () = begin let v = Printf.sprintf "value%Li" i in let updates = [Update.Set ("x", v)] in - let value = Value.create_client_value updates false in + let value = Value.create_client_value_nocheck updates false in writer # log_value i value >>= fun _ -> loop (Int64.succ i) end diff --git a/src/tlog/mem_tlogcollection.ml b/src/tlog/mem_tlogcollection.ml index 6a006e78..82d7dc74 100644 --- a/src/tlog/mem_tlogcollection.ml +++ b/src/tlog/mem_tlogcollection.ml @@ -17,11 +17,13 @@ limitations under the License. open Tlogcollection open Tlogcommon + class mem_tlog_collection _name = object (self: #tlog_collection) val mutable data = [] val mutable last_entry = (None: Entry.t option) + val mutable previous_checksum = None method validate_last_tlog () = Lwt.return (TlogValidComplete, last_entry, None) @@ -52,6 +54,20 @@ class mem_tlog_collection _name = | None -> None | Some e -> Some (Entry.v_of e, Entry.i_of e) + method get_previous_checksum i = + match last_entry with + | None -> previous_checksum + | Some pe -> + let pi = Entry.i_of pe in + if pi = i + then previous_checksum + else + if pi = Sn.pred i + then Value.checksum_of (Entry.v_of pe) + else None + + method set_previous_checksum cso = + previous_checksum <- cso method iterate from_i too_far_i f = let data' = @@ -66,15 +82,27 @@ class mem_tlog_collection _name = method dump_tlog_file _start_i _oc = failwith "dump_tlog_file not supported" - method save_tlog_file _name _length _ic = failwith "save_tlog_file not supported" + method save_tlog_file ?validate_i:_ _name _length _ic = + failwith "save_tlog_file not supported" method which_tlog_file _start_i = failwith "which_tlog_file not supported" - method log_value_explicit i (v:Value.t) _sync marker = - let entry = Entry.make i v 0L marker in - let () = data <- entry::data in - let () = last_entry <- (Some entry) in - Lwt.return () + method log_value_explicit i (v:Value.t) ?(validate = true) _sync marker = + if validate && not (Value.is_valid self i v) + then Lwt.fail (Value.ValueCheckSumError (i, v)) + else + let entry = Entry.make i v 0L marker in + let () = data <- entry::data in + let () = + match last_entry with + | None -> () + | Some pe -> + let pi = Entry.i_of pe in + if pi < i + then previous_checksum <- Value.checksum_of (Entry.v_of pe) + in + let () = last_entry <- (Some entry) in + Lwt.return () method log_value i v = self #log_value_explicit i v false None diff --git a/src/tlog/tlc2.ml b/src/tlog/tlc2.ml index 9e82a511..e477dabc 100644 --- a/src/tlog/tlc2.ml +++ b/src/tlog/tlc2.ml @@ -211,7 +211,7 @@ let fold_read tlog_dir tlf_dir file_name | exn -> Lwt.fail exn ) -type validation_result = (Entry.t option * Index.index) +type validation_result = (Entry.t option * Checksum.Crc32.t option * Index.index) let _make_close_marker node_id = "closed:" ^ node_id let _make_open_marker node_id = "opened:" ^ node_id @@ -220,21 +220,33 @@ let _validate_one tlog_name node_id ~check_marker : validation_result Lwt.t = Logger.debug_f_ "Tlc2._validate_one %s" tlog_name >>= fun () -> let e2s e = let i = Entry.i_of e in Printf.sprintf "(%s,_)" (Sn.string_of i) in let prev_entry = ref None in + let prev_checksum = ref None in let new_index = Index.make tlog_name in Lwt.catch (fun () -> let first = Sn.of_int 0 in let folder, _, index = folder_for tlog_name None in - let do_it ic = folder ic ~index Sn.start None ~first None - (fun _a0 entry -> + let do_it ic = folder ic ~index Sn.start None ~first (None, None) + (fun _ entry -> let () = Index.note entry new_index in - let r = Some entry in - let () = prev_entry := r in - Lwt.return r) + let pcs = match !prev_entry with + | None -> None + | Some e -> + if Entry.i_of e < Entry.i_of entry + then + let cs = Value.checksum_of (Entry.v_of e) in + let () = prev_checksum := cs in + cs + else + !prev_checksum + in + let eo = Some entry in + let () = prev_entry := eo in + Lwt.return (eo, pcs)) in Lwt_io.with_file tlog_name ~mode:Lwt_io.input do_it - >>= fun eo -> + >>= fun (eo, pcs) -> begin if not check_marker then Lwt.return eo @@ -253,16 +265,16 @@ let _validate_one tlog_name node_id ~check_marker : validation_result Lwt.t = >>= fun eo' -> Logger.debug_f_ "XX:a=%s" (Log_extra.option2s e2s eo') >>= fun () -> Logger.debug_f_ "After validation index=%s" (Index.to_string new_index) >>= fun () -> - Lwt.return (eo', new_index) + Lwt.return (eo', pcs, new_index) ) (function - | Unix.Unix_error(Unix.ENOENT,_,_) -> Lwt.return (None, new_index) + | Unix.Unix_error(Unix.ENOENT,_,_) -> Lwt.return (None, None, new_index) | exn -> Lwt.fail exn ) let _validate_list tlog_names node_id ~check_marker= - Lwt_list.fold_left_s (fun _ tn -> _validate_one tn node_id ~check_marker) (None,None) tlog_names >>= fun (eo,index) -> + Lwt_list.fold_left_s (fun _ tn -> _validate_one tn node_id ~check_marker) (None, None, None) tlog_names >>= fun (eo, _, index) -> Logger.info_f_ "_validate_list %s => %s" (String.concat ";" tlog_names) (Index.to_string index) >>= fun () -> Lwt.return (TlogValidIncomplete, eo, index) @@ -397,7 +409,7 @@ let iterate_tlog_dir tlog_dir tlf_dir ~index start_i too_far_i f = class tlc2 ?(compressor=Compression.Snappy) (tlog_dir:string) (tlf_dir:string) (head_dir:string) (new_c:int) - (last:Entry.t option) (index:Index.index) + (last:Entry.t option) (last_checksum:Checksum.Crc32.t option) (index:Index.index) (node_id:string) ~(fsync:bool) ~(fsync_tlog_dir:bool) = let inner = @@ -414,6 +426,7 @@ class tlc2 val mutable _inner = inner (* ~ pos in file *) val mutable _outer = new_c (* ~ pos in dir *) val mutable _previous_entry = last + val mutable _previous_checksum = last_checksum val mutable _compression_q = Lwt_buffer.create_fixed_capacity 5 val mutable _compression_thread = None val mutable _compressing = false @@ -505,31 +518,41 @@ class tlc2 Lwt.ignore_result t - method log_value_explicit i value sync marker = - Lwt_mutex.with_lock _write_lock - (fun () -> - begin - self # _prelude i >>= fun file -> - let p = F.file_pos file in - let oc = F.oc_of file in - Tlogcommon.write_entry oc i value >>= fun () -> - Lwt_io.flush oc >>= fun () -> + method log_value_explicit i value ?(validate = true) sync marker = + if validate && not (Value.is_valid self i value) + then Lwt.fail (Value.ValueCheckSumError (i, value)) + else + Lwt_mutex.with_lock _write_lock + (fun () -> begin - if sync || fsync - then F.fsync file - else Lwt.return () - end - >>= fun () -> - let () = match _previous_entry with - | None -> _inner <- _inner +1 - | Some pe-> - let pi = Entry.i_of pe in if pi < i then _inner <- _inner +1 - in - let entry = Entry.make i value p marker in - _previous_entry <- Some entry; - Index.note entry _index; - Lwt.return () - end) + self # _prelude i >>= fun file -> + let p = F.file_pos file in + let oc = F.oc_of file in + Tlogcommon.write_entry oc i value >>= fun () -> + Lwt_io.flush oc >>= fun () -> + begin + if sync || fsync + then F.fsync file + else Lwt.return () + end + >>= fun () -> + let () = match _previous_entry with + | None -> _inner <- _inner +1 + | Some pe-> + let pi = Entry.i_of pe in + if pi < i + then + begin + _inner <- _inner + 1; + let pcs = Value.checksum_of (Entry.v_of pe) in + _previous_checksum <- pcs + end + in + let entry = Entry.make i value p marker in + _previous_entry <- Some entry; + Index.note entry _index; + Lwt.return () + end) method log_value i value = self # log_value_explicit i value fsync None @@ -668,7 +691,7 @@ class tlc2 method get_last_i () = match _previous_entry with | None -> Sn.start - | Some pe -> let pi = Entry.i_of pe in pi + | Some pe -> Entry.i_of pe method get_last_value i = match _previous_entry with @@ -691,6 +714,26 @@ class tlc2 | None -> None | Some pe -> Some (Entry.v_of pe, Entry.i_of pe) + method get_previous_checksum i = + match _previous_entry with + | None -> _previous_checksum + | Some pe -> + let pi = Entry.i_of pe in + if pi = i + then _previous_checksum + else + if pi = Sn.pred i + then Value.checksum_of (Entry.v_of pe) + else + let msg = Printf.sprintf "get_previous_checksum %s > %s can't look back so far" + (Sn.string_of pi) (Sn.string_of i) + in + failwith msg + + method set_previous_checksum cso = + _previous_entry <- None; + _previous_checksum <- cso + method close ?(wait_for_compression=false) () = Lwt_mutex.lock _write_lock >>= fun () -> Logger.debug_ "tlc2::close()" >>= fun () -> @@ -794,14 +837,26 @@ class tlc2 let next_i = Sn.add start_i (Sn.of_int !tlogEntriesPerFile) in Lwt.return next_i - method save_tlog_file name length ic = + method save_tlog_file ?(validate_i = None) name length ic = (* what with rotation (jump to new tlog), open streams, ...*) let canon = get_full_path tlog_dir tlf_dir name in let tmp = canon ^ ".tmp" in Logger.debug_f_ "save_tlog_file: %s" tmp >>= fun () -> Lwt_io.with_file ~mode:Lwt_io.output tmp (fun oc -> Llio.copy_stream ~length ~ic ~oc) >>= fun () -> - File_system.rename tmp canon - + match validate_i with + | None -> File_system.rename tmp canon + | Some i -> + let folder, _, index = folder_for tmp None in + let too_far_i = Some (Sn.succ i) in + let f _ entry = Lwt.return (Some (Entry.v_of entry)) in + let ic_f ic = folder ic ~index i too_far_i ~first:i None f in + Lwt_io.with_file ~mode:Lwt_io.input tmp ic_f >>= fun vo -> + match vo with + | None -> failwith "tlog file is empty" + | Some value -> + if not (Value.is_valid self i value) + then Lwt.fail (Value.ValueCheckSumError (i, value)) + else File_system.rename tmp canon method remove_oldest_tlogs count = get_tlog_names tlog_dir tlf_dir >>= fun existing -> @@ -856,7 +911,7 @@ let maybe_correct new_c last index = let make_tlc2 ~compressor tlog_dir tlf_dir head_dir ~fsync node_id ~fsync_tlog_dir = Logger.debug_f_ "make_tlc2 %S" tlog_dir >>= fun () -> get_last_tlog tlog_dir tlf_dir >>= fun (new_c, fn) -> - _validate_one fn node_id ~check_marker:true >>= fun (last, index) -> + _validate_one fn node_id ~check_marker:true >>= fun (last, previous_checksum, index) -> maybe_correct new_c last index >>= fun (new_c,last,new_index) -> Logger.debug_f_ "make_tlc2 after maybe_correct %s" (Index.to_string new_index) >>= fun () -> let msg = @@ -865,7 +920,7 @@ let make_tlc2 ~compressor tlog_dir tlf_dir head_dir ~fsync node_id ~fsync_tlog_d | Some e -> let i = Entry.i_of e in "Some" ^ (Sn.string_of i) in Logger.debug_f_ "post_validation: last_i=%s" msg >>= fun () -> - let col = new tlc2 tlog_dir tlf_dir head_dir new_c last new_index ~compressor node_id ~fsync ~fsync_tlog_dir in + let col = new tlc2 tlog_dir tlf_dir head_dir new_c last previous_checksum new_index ~compressor node_id ~fsync ~fsync_tlog_dir in (* rewrite last entry with ANOTHER marker so we can see we got here *) begin match last with diff --git a/src/tlog/tlc2_test.ml b/src/tlog/tlc2_test.ml index c75a436d..0b322e77 100644 --- a/src/tlog/tlc2_test.ml +++ b/src/tlog/tlc2_test.ml @@ -33,12 +33,12 @@ let prepare_tlog_scenarios (dn,factory) = let old_tlog_entries_value = !Tlogcommon.tlogEntriesPerFile in Tlogcommon.tlogEntriesPerFile := 5 ; factory dn "node_name" >>= fun (tlog_coll:tlog_collection) -> - let value = Value.create_master_value ~lease_start:0. "me" in - tlog_coll # log_value 0L value >>= fun () -> - tlog_coll # log_value 1L value >>= fun () -> - tlog_coll # log_value 2L value >>= fun () -> - tlog_coll # log_value 3L value >>= fun () -> - tlog_coll # log_value 4L value >>= fun () -> + let value i = Value.create_master_value tlog_coll i ~lease_start:0. "me" in + tlog_coll # log_value 0L (value 0L) >>= fun () -> + tlog_coll # log_value 1L (value 1L) >>= fun () -> + tlog_coll # log_value 2L (value 2L) >>= fun () -> + tlog_coll # log_value 3L (value 3L) >>= fun () -> + tlog_coll # log_value 4L (value 4L) >>= fun () -> tlog_coll # close () >>= fun _ -> Lwt.return old_tlog_entries_value @@ -47,7 +47,7 @@ let test_interrupted_rollover (dn, tlx_dir, factory) = (*let fn = Tlc2.get_full_path dn tlx_dir "001.tlog" in Unix.unlink fn; *) factory dn "node_name" >>= fun tlog_coll -> - let value = Value.create_master_value ~lease_start:0. "me" in + let value = Value.create_master_value tlog_coll 5L ~lease_start:0. "me" in tlog_coll # log_value 5L value >>= fun () -> tlog_coll # close () >>= fun _ -> Tlc2.get_tlog_names dn tlx_dir >>= fun tlog_names -> @@ -77,12 +77,12 @@ let test_validate_at_rollover_boundary (dn, tlx_dir, factory) = end; OUnit.assert_equal ~msg lasti 4L; factory dn "node_name" >>= fun (tlog_coll:tlog_collection) -> - let value = Value.create_master_value ~lease_start:0. "me" in - tlog_coll # log_value 5L value >>= fun _ -> - tlog_coll # log_value 6L value >>= fun _ -> - tlog_coll # log_value 7L value >>= fun _ -> - tlog_coll # log_value 8L value >>= fun _ -> - tlog_coll # log_value 9L value >>= fun _ -> + let value i = Value.create_master_value tlog_coll i ~lease_start:0. "me" in + tlog_coll # log_value 5L (value 5L) >>= fun _ -> + tlog_coll # log_value 6L (value 6L) >>= fun _ -> + tlog_coll # log_value 7L (value 7L) >>= fun _ -> + tlog_coll # log_value 8L (value 8L) >>= fun _ -> + tlog_coll # log_value 9L (value 9L) >>= fun _ -> Tlc2.get_tlog_names dn tlx_dir >>= fun tlog_names -> let n = List.length tlog_names in Tlogcommon.tlogEntriesPerFile := old_tlog_entries_value; @@ -93,7 +93,7 @@ let test_iterate4 (dn, tlx_dir, factory) = Logger.debug_ "test_iterate4" >>= fun () -> let () = Tlogcommon.tlogEntriesPerFile := 100 in factory dn "node_name" >>= fun (tlc:tlog_collection) -> - let value = Value.create_client_value [Update.Set("test_iterate4","xxx")] false in + let value = Value.create_client_value_nocheck [Update.Set("test_iterate4","xxx")] false in Tlogcollection_test._log_repeat tlc value 120 >>= fun () -> Lwt_unix.sleep 3.0 >>= fun () -> (* TODO: compression should have callback *) let extension = Tlc2.extension Compression.Snappy in @@ -118,8 +118,9 @@ let test_iterate5 (dn, _tlx_dir, factory) = begin let sync = false in let is = string_of_int i in - let value = Value.create_client_value [Update.Set("test_iterate_" ^ is ,is)] sync in - tlc # log_value (Sn.of_int i) value >>= fun _ -> + let sni = Sn.of_int i in + let value = Value.create_client_value tlc sni [Update.Set("test_iterate_" ^ is ,is)] sync in + tlc # log_value sni value >>= fun _ -> begin if i mod 3 = 2 then @@ -156,7 +157,7 @@ let test_iterate6 (dn, _tlx_dir, factory) = begin let is = string_of_int i in let sni = Sn.of_int i in - let value = Value.create_client_value [Update.Set("test_iterate_" ^ is ,is)] sync in + let value = Value.create_client_value tlc sni [Update.Set("test_iterate_" ^ is ,is)] sync in begin if i != 19 then @@ -164,7 +165,7 @@ let test_iterate6 (dn, _tlx_dir, factory) = else begin tlc # log_value sni value >>= fun _ -> - let value2 = Value.create_client_value [Update.Set("something_else","gotcha")] sync in + let value2 = Value.create_client_value tlc sni [Update.Set("something_else","gotcha")] sync in tlc # log_value sni value2 end end >>= fun _ -> @@ -202,12 +203,12 @@ let test_compression_bug (dn, tlx_dir, factory) = if i = n then Lwt.return () else let key = Printf.sprintf "test_compression_bug_%i" i in - let value = Value.create_client_value [Update.Set(key, v)] sync in let sni = Sn.of_int i in + let value = Value.create_client_value tlc sni [Update.Set(key, v)] sync in tlc # log_value sni value >>= fun () -> loop (i+1) in - tlc # log_value 0L (Value.create_client_value [Update.Set("xxx","XXX")] false) >>= fun () -> + tlc # log_value 0L (Value.create_client_value tlc 0L [Update.Set("xxx","XXX")] false) >>= fun () -> loop 1 >>= fun () -> tlc # close ~wait_for_compression:true () >>= fun () -> File_system.stat (tlx_dir ^ "/000.tlx") >>= fun stat -> @@ -237,12 +238,12 @@ let test_compression_previous (dn, tlx_dir, factory) = if i = n then Lwt.return () else let key = Printf.sprintf "test_compression_bug_%i" i in - let value = Value.create_client_value [Update.Set(key, v)] sync in let sni = Sn.of_int i in + let value = Value.create_client_value tlc sni [Update.Set(key, v)] sync in tlc # log_value sni value >>= fun () -> loop (i+1) in - tlc # log_value 0L (Value.create_client_value [Update.Set("xxx","XXX")] false) >>= fun () -> + tlc # log_value 0L (Value.create_client_value tlc 0L [Update.Set("xxx","XXX")] false) >>= fun () -> loop 1 >>= fun () -> tlc # close ~wait_for_compression:true () >>= fun () -> @@ -296,6 +297,7 @@ let suite = "tlc2" >::: ("test_iterate6", test_iterate6); ("validate", Tlogcollection_test.test_validate_normal); ("validate_corrupt", Tlogcollection_test.test_validate_corrupt_1); + ("test_checksum", Tlogcollection_test.test_checksum); ("test_rollover_1002", Tlogcollection_test.test_rollover_1002); ("test_rollover_boundary", test_validate_at_rollover_boundary); ("test_interrupted_rollover", test_interrupted_rollover); diff --git a/src/tlog/tlog_main.ml b/src/tlog/tlog_main.ml index cf897c59..8e9d42e2 100644 --- a/src/tlog/tlog_main.ml +++ b/src/tlog/tlog_main.ml @@ -113,7 +113,7 @@ let make_tlog tlog_name (i:int) = let sni = Sn.of_int i in let t = let f oc = Tlogcommon.write_entry oc sni - (Value.create_client_value [Update.Update.Nop] false) + (Value.create_first_client_value [Update.Update.Nop] false) in Lwt_io.with_file ~mode:Lwt_io.output tlog_name f in diff --git a/src/tlog/tlogcollection.ml b/src/tlog/tlogcollection.ml index e28d1c22..0d1cfe41 100644 --- a/src/tlog/tlogcollection.ml +++ b/src/tlog/tlogcollection.ml @@ -21,16 +21,18 @@ class type tlog_collection = object method validate_last_tlog: unit -> (tlogValidity * Entry.t option * Index.index) Lwt.t method iterate: Sn.t -> Sn.t -> (Entry.t -> unit Lwt.t) -> unit Lwt.t method log_value : Sn.t -> Value.t -> unit Lwt.t - method log_value_explicit : Sn.t -> Value.t -> bool -> string option -> unit Lwt.t + method log_value_explicit : Sn.t -> Value.t -> ?validate:bool -> bool -> string option -> unit Lwt.t method get_last_i: unit -> Sn.t method get_last_value: Sn.t -> Value.t option (* Lwt.t *) method get_last: unit -> (Value.t * Sn.t) option + method get_previous_checksum: Sn.t -> Checksum.Crc32.t option + method set_previous_checksum: Checksum.Crc32.t option -> unit method close : ?wait_for_compression : bool -> unit -> unit Lwt.t method get_infimum_i : unit -> Sn.t Lwt.t method dump_head : Lwt_io.output_channel -> Sn.t Lwt.t method save_head : Lwt_io.input_channel -> unit Lwt.t method dump_tlog_file : Sn.t -> Lwt_io.output_channel -> Sn.t Lwt.t - method save_tlog_file : string -> int64 -> Lwt_io.input_channel -> unit Lwt.t + method save_tlog_file : ?validate_i:Sn.t option -> string -> int64 -> Lwt_io.input_channel -> unit Lwt.t method get_head_name : unit -> string method get_tlog_from_i : Sn.t -> Sn.t method get_tlog_count: unit -> int Lwt.t diff --git a/src/tlog/tlogcollection_test.ml b/src/tlog/tlogcollection_test.ml index e29b11fc..b54cd78a 100644 --- a/src/tlog/tlogcollection_test.ml +++ b/src/tlog/tlogcollection_test.ml @@ -59,13 +59,14 @@ let setup factory test_name () = let teardown (dn, tlf_dir, _factory) = Logger.info_f_ "teardown %s,%s" dn tlf_dir -let _make_set_v k v= Value.create_client_value [Update.Set (k,v)] false +let _make_set_v k v= Value.create_client_value_nocheck [Update.Set (k,v)] false -let _log_repeat tlc (value:Value.t) n = +let _log_repeat tlc ((_, c):Value.t) n = let rec loop i = if i = (Sn.of_int n) then Lwt.return () else begin + let value = Value.create_value tlc i c in tlc # log_value i value >>= fun _wr_result -> loop (Sn.succ i) end @@ -100,7 +101,7 @@ let test_rollover_1002 (dn, tlf_dir, factory) = let test_get_value_bug (dn, _tlf_dir, factory) = Logger.info_ "test_get_value_bug" >>= fun () -> factory dn "node_name" >>= fun (c0:tlog_collection) -> - let v0 = Value.create_master_value ~lease_start:0. "XXXX" in + let v0 = Value.create_master_value c0 0L ~lease_start:0. "XXXX" in c0 # log_value 0L v0 >>= fun _wr_result -> c0 # close () >>= fun () -> factory dn "node_name" >>= fun c1 -> @@ -245,8 +246,27 @@ let test_validate_corrupt_1 (dn, tlf_dir, factory) = OUnit.assert_bool msg false; Lwt.return () ) - >>= fun () -> - Lwt.return () + +let test_checksum (dn, tlf_dir, factory) = + Logger.info_f_ "test_checksum %s, %s" dn tlf_dir >>= fun () -> + factory dn "node_name" >>= fun (tlc:tlog_collection) -> + let value = _make_set_v "XXX" "X" in + _log_repeat tlc value 10 >>= fun () -> + let value = Value.create_client_value tlc 9L [Update.Set ("XXX", "XX")] false in + Lwt.catch + (fun () -> + tlc # log_value 10L value >>= fun () -> + OUnit.assert_bool "the checksum should be wrong" false; + Lwt.return () + ) + (function + | Value.ValueCheckSumError _ -> Lwt.return () + | exn -> + let () = ignore exn in + let msg = Printf.sprintf "it threw the wrong exception %s" "?" in + OUnit.assert_bool msg false; + Lwt.return () + ) let wrap factory test (name:string) = lwt_bracket (setup factory name) test teardown @@ -256,6 +276,7 @@ let wrap_memory name = wrap create_test_tlc name let suite_mem = "mem_tlogcollection" >::: [ "rollover" >:: wrap_memory test_rollover "rollover"; + "checksum" >:: wrap_memory test_checksum "checksum"; (* "get_value_bug" >:: wrap_memory test_get_value_bug; (* assumption that different tlog_collections with the same name have the same state *) *) diff --git a/src/tools/arakoon_crc32c.c b/src/tools/arakoon_crc32c.c index 42c32546..6ea0d3ae 100644 --- a/src/tools/arakoon_crc32c.c +++ b/src/tools/arakoon_crc32c.c @@ -67,6 +67,11 @@ CAMLprim value update_crc32c(value crc32c, value buffer, value offset, value len const int32_t ioffset = Int_val(offset); const uint8_t * buffer2 = (const uint8_t *) String_val(buffer) + ioffset; int32_t crc32c2 = Int32_val(crc32c); - const uint32_t res = bsd_update_crc32c(crc32c2, buffer2, ilength); + uint32_t res = 0; + if (has_sse_4_2) { + res = crc32c_sse4_2(crc32c2, buffer2, ilength); + } else { + res = bsd_update_crc32c(crc32c2, buffer2, ilength); + } CAMLreturn(caml_copy_int32(res)); } diff --git a/src/tools/bsd_crc32c.c b/src/tools/bsd_crc32c.c index 6992584a..69d906e7 100644 --- a/src/tools/bsd_crc32c.c +++ b/src/tools/bsd_crc32c.c @@ -786,9 +786,9 @@ static uint32_t _crc32c_sse4_2_byte(uint32_t crc, return crc; } -static uint32_t crc32c_sse4_2(uint32_t crc, +uint32_t crc32c_sse4_2(uint32_t crc, unsigned char const *p, - size_t len){ + uint32_t len){ unsigned int q = len >> 3; unsigned int rem = len & 0x7; unsigned long *ptmp = (unsigned long *)p; diff --git a/src/tools/bsd_crc32c.h b/src/tools/bsd_crc32c.h index bdd26381..f2b63ca4 100644 --- a/src/tools/bsd_crc32c.h +++ b/src/tools/bsd_crc32c.h @@ -11,8 +11,8 @@ extern "C" { uint32_t bsd_calculate_crc32c(const unsigned char *buffer, unsigned int length); uint32_t bsd_update_crc32c(uint32_t crc32c, const unsigned char *buffer, unsigned int length); -uint32_t sse4_2_crc32c(const unsigned char *buffer, - uint32_t length); +uint32_t sse4_2_crc32c(const unsigned char *buffer, uint32_t length); +uint32_t crc32c_sse4_2(uint32_t crc, unsigned char const *p, uint32_t len); #if defined(__cplusplus) } diff --git a/src/tools/checksum.ml b/src/tools/checksum.ml new file mode 100644 index 00000000..2fa6d367 --- /dev/null +++ b/src/tools/checksum.ml @@ -0,0 +1,49 @@ +(* +Copyright (2010-2014) INCUBAID BVBA + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*) + +module type ChecksumType = sig + type t + + val string_of : t -> string + val checksum_to : Buffer.t -> t -> unit + val checksum_from : Llio.buffer -> t + + val calculate : string -> t + val update : t -> string -> t +end + +module Make (Cs : ChecksumType) = struct + include Cs + + let update cs s = + match cs with + | None -> calculate s + | Some cs -> Cs.update cs s +end + +module Crc32Digest : ChecksumType = struct + type t = int32 + + let string_of = Printf.sprintf "%lx" + let checksum_to = Llio.int32_to + let checksum_from = Llio.int32_from + + let calculate s = Crc32c.calculate_crc32c s 0 (String.length s) + let update cs s = Crc32c.update_crc32c cs s 0 (String.length s) +end + + +module Crc32 = Make (Crc32Digest)