Module rr_recon

replica reconcilication module.

Copyright © 2011-2016 Zuse Institute Berlin

Version: $Id$

Behaviours: gen_component.

Authors: Maik Lange (malange@informatik.hu-berlin.de).

Description

replica reconcilication module

Data Types

client_version()

client_version() = non_neg_integer()

export for testing

db_chunk_kv()

db_chunk_kv() = [{rt_chord:key(), client_version()}]

exit_reason()

exit_reason() = empty_interval
              | recon_node_crash
              | sync_finished
              | sync_finished_remote

kvi_tree()

kvi_tree() = mymaps:mymap()

KeyShort::non_neg_integer() => {VersionShort::non_neg_integer(), Idx::non_neg_integer()}

merkle_cmp_request()

merkle_cmp_request() = 
    {Hash :: merkle_tree:mt_node_key() | none,
     IsLeaf :: boolean()}

merkle_sync()

merkle_sync() = 
    {[merkle_sync_send()],
     [merkle_sync_rcv()],
     SynRcvLeafCount :: non_neg_integer(),
     merkle_sync_direct()}

merkle_sync_direct()

merkle_sync_direct() = 
    {MyKItems :: [rt_chord:key()],
     MyLeafCount :: non_neg_integer(),
     OtherNodeCount :: non_neg_integer()}

merkle_sync_rcv()

merkle_sync_rcv() = 
    {MyMaxItemsCount :: non_neg_integer(),
     MyKVItems :: merkle_tree:mt_bucket()}

merkle_sync_send()

merkle_sync_send() = 
    {OtherMaxItemsCount :: non_neg_integer(),
     MyKVItems :: merkle_tree:mt_bucket()}

message()

message() = request()
          | {reconcile_req,
             DiffBFBin :: bitstring(),
             OtherBFCount :: non_neg_integer(),
             OtherDiffCount :: non_neg_integer(),
             SenderPid :: comm:mypid()}
          | {resolve_req, BinReqIdxPos :: bitstring()}
          | {resolve_req,
             DBChunk :: {bitstring(), bitstring()},
             Payload :: any(),
             SigSize :: signature_size(),
             SenderPid :: comm:mypid()}
          | {resolve_req, shutdown, Payload :: any()}
          | {101,
             SenderPid :: comm:mypid(),
             ToCheck :: bitstring(),
             MaxItemsCount :: non_neg_integer()}
          | {101,
             ToCheck :: bitstring(),
             MaxItemsCount :: non_neg_integer()}
          | {102,
             FlagsBin :: bitstring(),
             MaxItemsCount :: non_neg_integer()}
          | {resolve_req,
             HashesK :: bitstring(),
             HashesV :: bitstring()}
          | {resolve_req,
             BinKeyList :: bitstring(),
             SyncSendP1E_real :: float()}
          | {create_struct2,
             SenderI :: intervals:interval(),
             {get_state_response, MyI :: intervals:interval()}}
          | {process_db,
             {get_chunk_response,
              {intervals:interval(), db_chunk_kv()}}}
          | {shutdown, exit_reason()}
          | {fd_notify,
             fd:event(),
             DeadPid :: comm:mypid(),
             Reason :: fd:reason()}
          | {'DOWN',
             MonitorRef :: reference(),
             process,
             Owner :: pid(),
             Info :: any()}

method()

method() = trivial | shash | bloom | merkle_tree | art

| iblt.

parameters()

parameters() = #trivial_recon_struct{interval =
                                         intervals:interval(),
                                     reconPid = comm:mypid()
                                               | undefined,
                                     exp_delta = number(),
                                     db_chunk = {bitstring(),
                                                  bitstring()}
                                               | {bitstring(),
                                                  bitstring(),
                                                  db_chunk_kv()},
                                     sig_size =
                                         signature_size()}
             | #shash_recon_struct{interval =
                                       intervals:interval(),
                                   reconPid = comm:mypid()
                                             | undefined,
                                   exp_delta = number(),
                                   db_chunk = bitstring()
                                             | {bitstring(),
                                                db_chunk_kv()},
                                   sig_size = signature_size(),
                                   p1e = float()}
             | #bloom_recon_struct{interval =
                                       intervals:interval(),
                                   reconPid = comm:mypid()
                                             | undefined,
                                   exp_delta = number(),
                                   bf = bitstring()
                                       | bloom:bloom_filter(),
                                   item_count =
                                       non_neg_integer(),
                                   hf_count = pos_integer(),
                                   p1e = float()}
             | #merkle_params{interval = intervals:interval(),
                              reconPid = comm:mypid()
                                        | undefined,
                              exp_delta = number(),
                              branch_factor = pos_integer(),
                              num_trees = pos_integer(),
                              bucket_size = pos_integer(),
                              p1e = float(),
                              ni_item_count =
                                  non_neg_integer()}
             | #art_recon_struct{art = art:art(),
                                 reconPid = comm:mypid()
                                           | undefined,
                                 branch_factor = pos_integer(),
                                 bucket_size = pos_integer()}

recon_dest()

recon_dest() = rt_chord:key() | random

request()

request() = {start, method(), DestKey :: recon_dest()}
          | {create_struct,
             method(),
             SenderI :: intervals:interval(),
             SenderMaxItems :: non_neg_integer()}
          | {start_recon,
             bloom,
             #bloom_recon_struct{interval =
                                     intervals:interval(),
                                 reconPid = comm:mypid()
                                           | undefined,
                                 exp_delta = number(),
                                 bf = bitstring()
                                     | bloom:bloom_filter(),
                                 item_count =
                                     non_neg_integer(),
                                 hf_count = pos_integer(),
                                 p1e = float()}}
          | {start_recon,
             merkle_tree,
             #merkle_params{interval = intervals:interval(),
                            reconPid = comm:mypid()
                                      | undefined,
                            exp_delta = number(),
                            branch_factor = pos_integer(),
                            num_trees = pos_integer(),
                            bucket_size = pos_integer(),
                            p1e = float(),
                            ni_item_count = non_neg_integer()}}
          | {start_recon,
             art,
             #art_recon_struct{art = art:art(),
                               reconPid = comm:mypid()
                                         | undefined,
                               branch_factor = pos_integer(),
                               bucket_size = pos_integer()}}

signature_size()

signature_size() = 0..160

use an upper bound of 160 (SHA-1) to limit automatic testing

stage()

stage() = req_shared_interval
        | build_struct
        | reconciliation
        | resolve

state()

state() = 
    #rr_recon_state{ownerPid = pid(),
                    dest_rr_pid = comm:mypid(),
                    dest_recon_pid = comm:mypid() | undefined,
                    method = method() | undefined,
                    sync_interval@I = intervals:interval(),
                    max_items@I = non_neg_integer() | undefined,
                    params = parameters() | {},
                    struct = sync_struct() | {},
                    stage = stage(),
                    initiator = boolean(),
                    merkle_sync = merkle_sync(),
                    misc = [{atom(), term()}],
                    kv_list = db_chunk_kv() | [db_chunk_kv()],
                    k_list = [rt_chord:key()],
                    stats = rr_recon_stats:stats(),
                    to_resolve =
                        {ToSend :: rr_resolve:kvv_list(),
                         ToReqIdx :: [non_neg_integer()]}}

sync_struct()

sync_struct() = #trivial_recon_struct{interval =
                                          intervals:interval(),
                                      reconPid = comm:mypid()
                                                | undefined,
                                      exp_delta = number(),
                                      db_chunk = {bitstring(),
                                                   bitstring()}
                                                | {bitstring(),
                                                   bitstring(),
                                                   db_chunk_kv()},
                                      sig_size =
                                          signature_size()}
              | #shash_recon_struct{interval =
                                        intervals:interval(),
                                    reconPid = comm:mypid()
                                              | undefined,
                                    exp_delta = number(),
                                    db_chunk = bitstring()
                                              | {bitstring(),
                                                 db_chunk_kv()},
                                    sig_size =
                                        signature_size(),
                                    p1e = float()}
              | #bloom_recon_struct{interval =
                                        intervals:interval(),
                                    reconPid = comm:mypid()
                                              | undefined,
                                    exp_delta = number(),
                                    bf = bitstring()
                                        | bloom:bloom_filter(),
                                    item_count =
                                        non_neg_integer(),
                                    hf_count = pos_integer(),
                                    p1e = float()}
              | merkle_tree:merkle_tree()
              | [merkle_tree:mt_node()]
              | #art_recon_struct{art = art:art(),
                                  reconPid = comm:mypid()
                                            | undefined,
                                  branch_factor =
                                      pos_integer(),
                                  bucket_size = pos_integer()}

Function Index

bitstring_to_k_list_k/3Converts the bitstring from pos_to_bitstring/4 into keys at the appropriate positions in KList.
bitstring_to_k_list_kv/3Converts the bitstring from pos_to_bitstring/4 into keys at the appropriate positions in KVList.
calc_n_subparts_p1e/2Splits P1E into N equal independent sub-processes and returns the P1E to use for each of these sub-processes: p_sub = 1 - (1 - p1e)^(1/n).
calc_n_subparts_p1e/3Splits P1E into N further (equal) independent sub-processes and returns the P1E to use for the next of these sub-processes with the previous sub-processes having a (combined) failure probability of PrevP1.
calc_signature_size_nm_pair/6Calculates the minimum number of bits needed to have a hash collision probability of P1E, given we compare N hashes with M other hashes pairwise with each other (assuming that ExpDelta percent of them are different).
check_config/0Checks whether config parameters exist and are valid.
find_sync_interval/2Gets a randomly selected sync interval as an intersection of the two given intervals as a sub interval of A inside a single quadrant.
get_chunk_filter/1
get_chunk_kv/1
init/1init module.
key_dist/2
map_interval/2Maps interval B into interval A.
map_key_to_interval/2Maps any key (K) into a given interval (I).
map_key_to_quadrant/2Maps an abitrary key to its associated key in replication quadrant Q.
map_rkeys_to_quadrant/2Returns a key in the given replication quadrant Q from a list of replica keys.
merkle_compress_hashlist/4Transforms a list of merkle keys, i.e.
merkle_decompress_hashlist/3Transforms the compact binary representation of merkle hash lists from merkle_compress_hashlist/2 back into the original form.
on/2
pos_to_bitstring/4Converts a list of positions to a bitstring where the x'th bit is set if the x'th position is in the list.
quadrant_intervals/0Gets the quadrant intervals.
quadrant_subints_/3Gets all sub intervals of the given interval which lay only in a single quadrant.
start/2
start_gen_component/5
tester_create_kvi_tree/1
tester_is_kvi_tree/1

Function Details

start_gen_component/5

start_gen_component(Module :: module(),
                    Handler :: gen_component:handler(),
                    Args :: term(),
                    Options :: [gen_component:option()],
                    Self :: pid()) ->
                       no_return() | ok

on/2

on(Msg :: message(), State :: state()) -> state() | kill

calc_signature_size_nm_pair/6

calc_signature_size_nm_pair(N :: non_neg_integer(),
                            M :: non_neg_integer(),
                            HashType :: key | key_version,
                            ExpDelta :: number(),
                            P1E :: float(),
                            MaxSize :: signature_size()) ->
                               SigSize :: signature_size()

Calculates the minimum number of bits needed to have a hash collision probability of P1E, given we compare N hashes with M other hashes pairwise with each other (assuming that ExpDelta percent of them are different). NOTE: if multiple items are affected by the collision, simply divide P1E by the number of affected items!

pos_to_bitstring/4

pos_to_bitstring(Pos :: [non_neg_integer()],
                 AccBin :: [bitstring()],
                 BitsSet :: non_neg_integer(),
                 FinalSize :: non_neg_integer()) ->
                    Result :: [bitstring()]

Converts a list of positions to a bitstring where the x'th bit is set if the x'th position is in the list. The final bitstring may be created with erlang:list_to_bitstring(lists:reverse(Result)). A total of FinalSize bits will be used. PreCond: sorted list Pos, 0 <= every pos < FinalSize

bitstring_to_k_list_k/3

bitstring_to_k_list_k(PosBitString :: bitstring(),
                      KList :: [rt_chord:key()],
                      Acc :: [rt_chord:key()]) ->
                         Result :: [rt_chord:key()]

Converts the bitstring from pos_to_bitstring/4 into keys at the appropriate positions in KList. Result is reversly sorted. NOTE: This is essentially the same as bitstring_to_k_list_kv/3 but we need the separation because of the opaque RT keys.

bitstring_to_k_list_kv/3

bitstring_to_k_list_kv(PosBitString :: bitstring(),
                       KVList :: db_chunk_kv(),
                       Acc :: [rt_chord:key()]) ->
                          Result :: [rt_chord:key()]

Converts the bitstring from pos_to_bitstring/4 into keys at the appropriate positions in KVList. Result is reversly sorted.

merkle_compress_hashlist/4

merkle_compress_hashlist(Nodes :: [merkle_tree:mt_node()],
                         Bin,
                         SigSizeI :: signature_size(),
                         SigSizeL :: signature_size()) ->
                            Bin

Transforms a list of merkle keys, i.e. hashes, into a compact binary representation for transfer.

merkle_decompress_hashlist/3

merkle_decompress_hashlist(Bin :: bitstring(),
                           SigSizeI :: signature_size(),
                           SigSizeL :: signature_size()) ->
                              Hashes :: [merkle_cmp_request()]

Transforms the compact binary representation of merkle hash lists from merkle_compress_hashlist/2 back into the original form.

calc_n_subparts_p1e/2

calc_n_subparts_p1e(N :: number(), P1E :: float()) ->
                       P1E_sub :: float()

Splits P1E into N equal independent sub-processes and returns the P1E to use for each of these sub-processes: p_sub = 1 - (1 - p1e)^(1/n). This is based on p0e(total) = (1 - p1e(total)) = p0e(each)^n = (1 - p1e(each))^n.

calc_n_subparts_p1e/3

calc_n_subparts_p1e(N :: number(),
                    P1E :: float(),
                    PrevP1 :: float()) ->
                       P1E_sub :: float()

Splits P1E into N further (equal) independent sub-processes and returns the P1E to use for the next of these sub-processes with the previous sub-processes having a (combined) failure probability of PrevP1. This is based on p0e(total) = (1 - p1e(total)) = p0e(each)^n = (1 - p1e(each))^n.

get_chunk_filter/1

get_chunk_filter(DBEntry :: db_entry:entry()) -> boolean()

get_chunk_kv/1

get_chunk_kv(DBEntry :: db_entry:entry()) ->
                {rt_chord:key(), client_version() | -1}

map_key_to_interval/2

map_key_to_interval(Key :: rt_chord:key(),
                    I :: intervals:interval()) ->
                       rt_chord:key() | none

Maps any key (K) into a given interval (I). If K is already in I, K is returned. If K has more than one associated key in I, the closest one is returned. If all associated keys of K are not in I, none is returned.

key_dist/2

key_dist(Key1 :: rt_chord:key(), Key2 :: rt_chord:key()) ->
            number()

map_key_to_quadrant/2

map_key_to_quadrant(Key :: rt_chord:key(), Q :: rt_beh:segment()) ->
                       rt_chord:key()

Maps an abitrary key to its associated key in replication quadrant Q.

map_rkeys_to_quadrant/2

map_rkeys_to_quadrant(RKeys :: [rt_chord:key(), ...],
                      Q :: rt_beh:segment()) ->
                         rt_chord:key()

Returns a key in the given replication quadrant Q from a list of replica keys.

quadrant_intervals/0

quadrant_intervals() -> [intervals:non_empty_interval(), ...]

Gets the quadrant intervals.

quadrant_subints_/3

quadrant_subints_(A :: intervals:interval(),
                  Quadrants :: [intervals:interval()],
                  AccIn :: [intervals:interval()]) ->
                     AccOut :: [intervals:interval()]

Gets all sub intervals of the given interval which lay only in a single quadrant.

find_sync_interval/2

find_sync_interval(A :: intervals:continuous_interval(),
                   B :: intervals:continuous_interval()) ->
                      intervals:interval()

Gets a randomly selected sync interval as an intersection of the two given intervals as a sub interval of A inside a single quadrant. Result may be empty, otherwise it is also continuous!

map_interval/2

map_interval(A :: intervals:continuous_interval(),
             B :: intervals:continuous_interval()) ->
                intervals:interval()

Maps interval B into interval A. PreCond: the second (continuous) interval must be in a single quadrant! The result is thus also only in a single quadrant. Result may be empty, otherwise it is also continuous!

init/1

init(State :: state()) -> state()

init module

start/2

start(SessionId :: rrepair:session_id(),
      SenderRRPid :: comm:mypid()) ->
         {ok, pid()}

check_config/0

check_config() -> boolean()

Checks whether config parameters exist and are valid.

tester_create_kvi_tree/1

tester_create_kvi_tree(KVList ::
                           [{KeyShort :: non_neg_integer(),
                             {VersionShort :: non_neg_integer(),
                              Idx :: non_neg_integer()}}]) ->
                          kvi_tree()

tester_is_kvi_tree/1

tester_is_kvi_tree(Map :: any()) -> boolean()


Generated by EDoc, Aug 2 2016, 13:42:09.