Module rr_recon

replica reconcilication module.

Copyright © 2011-2015 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

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()],
     merkle_sync_direct()}

merkle_sync_direct()

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

merkle_sync_rcv()

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

merkle_sync_send()

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

message()

message() = request()
          | {resolve_req, BinReqIdxPos :: bitstring()}
          | {resolve_req,
             DBChunk :: bitstring(),
             DiffIdx :: bitstring(),
             SigSize :: signature_size(),
             VSize :: signature_size(),
             SenderPid :: comm:mypid()}
          | {resolve_req,
             DBChunk :: bitstring(),
             SigSize :: signature_size(),
             VSize :: signature_size(),
             SenderPid :: comm:mypid()}
          | {resolve_req, shutdown}
          | {resolve_req, BinReqIdxPos :: bitstring()}
          | {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, Hashes :: bitstring()}
          | {resolve_req, idx, BinKeyList :: bitstring()}
          | {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,
                                     db_chunk = bitstring(),
                                     sig_size =
                                         signature_size(),
                                     ver_size =
                                         signature_size()}
             | #shash_recon_struct{interval =
                                       intervals:interval(),
                                   reconPid = comm:mypid()
                                             | undefined,
                                   db_chunk = bitstring(),
                                   sig_size = signature_size(),
                                   p1e_p2 = float()}
             | #bloom_recon_struct{interval =
                                       intervals:interval(),
                                   reconPid = comm:mypid()
                                             | undefined,
                                   bf = binary()
                                       | bloom:bloom_filter(),
                                   item_count =
                                       non_neg_integer(),
                                   hf_count = pos_integer(),
                                   p1e = float()}
             | #merkle_params{interval = intervals:interval(),
                              reconPid = comm:mypid()
                                        | undefined,
                              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(),
                                 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,
                                 bf = binary()
                                     | 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,
                            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(),
                               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(),
                    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,
                                      db_chunk = bitstring(),
                                      sig_size =
                                          signature_size(),
                                      ver_size =
                                          signature_size()}
              | #shash_recon_struct{interval =
                                        intervals:interval(),
                                    reconPid = comm:mypid()
                                              | undefined,
                                    db_chunk = bitstring(),
                                    sig_size =
                                        signature_size(),
                                    p1e_p2 = float()}
              | #bloom_recon_struct{interval =
                                        intervals:interval(),
                                    reconPid = comm:mypid()
                                              | undefined,
                                    bf = binary()
                                        | 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(),
                                  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 success probability of PrevP0 (a product of all (1-P1E_sub)).
calc_signature_size_nm_pair/4Calculates 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 the worst case, i.e.
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

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/4

calc_signature_size_nm_pair(N :: non_neg_integer(),
                            M :: non_neg_integer(),
                            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 the worst case, i.e. having M+N total hashes).

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(),
                    PrevP0 :: 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 success probability of PrevP0 (a product of all (1-P1E_sub)). 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.


Generated by EDoc, Feb 29 2016, 16:12:14.