From 23a3fa9b398300323e09a795aa3671d306796d9f Mon Sep 17 00:00:00 2001 From: Asanka Indrajith Date: Thu, 19 Dec 2019 01:29:46 -0500 Subject: [PATCH] Consensus with network clock (#71) --- examples/rndcontract/rnd_contract | Bin 0 -> 16656 bytes src/cons/cons.cpp | 62 ++++++++++++++++--- src/cons/state_handler.cpp | 11 ++-- test/vm-cluster/cluster.sh | 17 +++-- test/vm-cluster/consensus-test-continuous.sh | 2 +- test/vm-cluster/setup-vm.sh | 3 + 6 files changed, 74 insertions(+), 21 deletions(-) create mode 100755 examples/rndcontract/rnd_contract diff --git a/examples/rndcontract/rnd_contract b/examples/rndcontract/rnd_contract new file mode 100755 index 0000000000000000000000000000000000000000..97e396402703687bfd8832960f90998d589b5f3b GIT binary patch literal 16656 zcmeHOdvqMtdB3x($7&_5*4D!|hInjjupP9LjQqkjNPaO&WMjaPq=5{p-IcWUYFFKz zH8zPri6Y8Yc4{|-V>q-p4JCxbNl8y=eVjOzF#(h0G^S}&l8|!@HAI3?R1&~7!Rqh3 zbHCN>csJ=e?LR%o*VcUZ`+bl5-N)RWd*{wQTe`P;G)-{wifaXNt$rKnP=W(%R0gC& ztPxeXE)okx1^6lJ0Y8(~Lr{!f!?CUKK0uT;B?tN59!>WY>`yk}K=OCs8wh$pjq zn)j?()x4@blu3nF$a<51(!GBBEixo(R^>{{8^9Ar@-1Csuh0JUmQT#LF4?~PoqKNB zp8d&nq}zl)?l+myf*!)=;BVPw`z~Gk&KM65Acg69uzDdRxs|xiE`r}y0$*GL?<#@c zUjnBxiwtCGdBE>-cl0 zy8$e6gW&7&SFB#!fzJ|ig|6$!3BmVc1bMLNLMw&YwR4+^Vq$&qj1^1o+_ou^O2&4EdlKZXZy=Ra z?o3;?s3NtO2mieCN;$dVX{by*rm%*1yq50tZz(=| zNxX`Deg`}yo`1QYT4?|x8JFpjGCrk#k9@BFWB#QDwm!{)6c4A<4%|6EPB?JoTAEBb za2jKr&N^_KyEySUqu(T&n>g_}Bb??dPM=hMDgp3(NU}Q|IL(Wk`W-mW^&}f~;LhK} zJr118;56*OxlKuSp94q7 zc|h_W;%R734om)ABY5h%$wA4#Lp*iakne z9Xq@Cmp_EM8T)H5mt@21(3)<%7dF4GTdVN7Gr<;S3!-nXN}xz);x5<9}$Z# zsJ(sqbZ@9`!55|f+fPud&+xF1lj8oBJ<`px^RPFLpZ6NMDdYIWb%yqw@#1-FE*vgY z4r|C^DSyKoZiJf1wiqKD-fuwe7K*3F7~60T+(a#EJYnGST=_ESX%x;d1@@a0aEbbm z`t6gw(4d_0`CkuWDCCYkjHWVjPagn@8e?TkmcqoyZ8_UEJ|bmaYq>EtN@8Oyh*-am z!ceE(7$-*V`6E=q#48_8PfvayBgxo5N?m_+SUP#`AEoqbkSc+(--sgn4!j+ujx=)r z1SbzKR(>L*t!nvrqb4iw5PJCuoEkT5rTip)oj5fRx? zfi}foZpVL`qUHD(!g8{lki2iGila0(zejbTa4v;XZuHyaeatQd;FS_QG6No^`zdd; zltP-(e&KgA!k3^vu??aJ-#@@?qq@a++*B6vXkP3r63(F?aI)H;$hx&z(GecEPe|a;J8UN0%MXoiq-iVJ4*0 z$CcAJjfo#nS;vjYaU(wd`QhPVj7nXDWo7P}yo1~`$6xiq)5&G0a?gy7lC^Q@zWdAR zjvJAz#;CDY7IEj-jj{a{mofGgD$>~hG%1gbQY)d!7Rwqw3ufdPU4?ZAnn)TPpJ>XB zo+O(wSu^0$gLVp`LG&Yt#-u`S^hxC3Y(j@N%+_W9Joy1dL@**EvQFyX{qmdIff3o3 zz^O^s$(F9$_XP5e&UyLV^t2q###r`j)VR7`-U?Rv#7`i2;lzQ?PjueemHW5OTRJg! zEcsqQP7TdOZ|=-}*p+*)d)aGpjz0dOcj5-v7^81ldfPAfe(uhl>(0HmIrnzwbi>QW z$O+9@`^)SbG>3os_RiZnZ}0p}r+MPQ%>3~iOzqO2n&mb1tDDV)`(pZ%jJ{qzmxTvn zx|P!D(JwWS*sI5)aqC*YNcVhHL{iyAR8OWXogN8wQimN4YHhxE9W9G#^8T0qnx3W^ z>kR>0%39Y=Mg{aIpn0w!p;} zxYz<0Ti{{~Tx@}hE%5)X1$e)j_orzIMu{o!kxr>a7w?UpQJbOjZ1wy;_g0CQtD9qsl0X@ zR`$Fu<-O)Pwz?Qp0OP|d5RQ9F>G7U3m;VukvmTbEvVJH|LRdaZzC-!La+>ARa!TQlL6^)j`64NBLmY520B0A?4>*%_W`ll-H|1 zMLN@zU+Qlp9ib7^>VJwDO)_iz-y#c-WIFs4Bq@`O;eUvja>?xQFDGXelKG_nK4N?t zy>qz3znRqhS`wt+zn+*Htp^H&{$s=hC1d$lP&u=;K}h!a|AQPfXt{XZkMIoeJz zU+{l{)aGgyuo3?$QoB_9BAEO9t4Oj$`z){n{*%NsYt#&f{68ipB$>nhpAyp|naBOV zB&Jm|Px;RfvqEcz%@IHCa+h5#V;lEBK$4Z(p8XdA8Uuc_v~swXM`SP!g%qQbWtScQ*#)KH&N4Qgp55!wnyqSU|= ztrr15%!AGP?{ww&ym4^G&t^OXHtQmDHd7yly9KDemxk^3Oc z&ee-$7CBM9LYjPyva4m*1KsL%l%0JHL20#=b>FYr!`-;G>$~#`-h=yw>aUb673MuS zJk-#W?G;)BDS8i{zPs^T@T1MSO|q{JHCP#;$)@riJUr9@%>@Z58Xsz`^+5A-D%X4P zQOKe~TVivM4>jPGy3m%=pzTgy zwF}{=1;Ri*OxMrD@NBnz!05Vp1Oyd;a&2JNLR4pVNV{&H9rcy-y6UMa9djSpIFFpx zm*>OqU>TuU~qVN!tmB=KI`>^p5;xxQO~0K ztZ&;7Jd5ie@q`*YS2Xy#eIvfxef{73mG8@*tL6w#nfE(%@pR4t(JHJ8Uz2BP{a^U% zeUA|SGrGK@L~nz7yeVC=-phQtZv%$6rXiX*&hvnT(|sa_mgz~PBD-)5<_44MEbEx0 z13H9c!n#z)r(_I7P>YQ?@Pv_dZa&`CZ&iYvb<1+I8$0YX%}YK7+$F@sZu7n zUDZyFmrZ6fvFHjJhHMSlh=VCRFeuy2Lx-QtWNZ&L!C*MrW)qkZcePF0ZL-om?CmZr z2y!62$3&AmDo=gg(w`cLwWPzl*<@@(UkrmN9swIpNBTF|=OOjxKE1hDo*ru+OsA|^#7d=`ac&H~D}(64xw7n_ zDZ7B~JG(rxjVtlT(5cdjtgmfn_CJSDtI%Qs*=p8Srk3}428h7%FO2O+p zr_75bBCiVz@hZXV&qCavU#Ax0)q>aSg?J4f_??KI_ljDf7g=|CMIe8U+G*t#b%NLb zg?La*(Zhl(dGXx1zRm@L_2LBNo}zetem}u!<;7cI-bX0J8*#o_cS17n#fdgv_dA7N zI?<*V)1NtaS*VX+^CHPBcz2^v=oP#R;>LOR!;P!M*@ZCLhj$ZZ1oSgrCjL{N2$@0g zU;@U2{<_1}kJKH6A9mpj70&xU?2DG68qS#-s(q+DO0?M0inRX#aPrUlNTf??Fa9)~ za^>@p62#>Y{))w4s$O50erAi$1=+(4E#~KGDXza@oIEG}TsTgCtn}S+zAEtx$L+h) z&xPZv8Wot0b~vrZKTikqf$P#&;e{T~i=Fd*S?XUnK0glJc|MT6nZSb>&q2Mgl-?5h zjS|1`{=KV&J}qvGmHT)JoSr<3>Hi!!#p%B9epN#McO~#TG;Co_?Uwim@M7hz06xnz zNAUg-w;LVMFQ$JJ=_68}$LtLsA;Kl}vl0)8LDlauRlsNo{o{}aaewX5bq+<+)3UxZ z^X*5K^uZE-o&-*Q-2Hx}g#M2t9=Krb`(g?GGbM0Z`BQtk%cX^OVU6qx(BgHGs*97{ zBmH0aJ4IVF#rkCp{1iR^9!XmnE8E+Pb;8W~anl+wad&b#(B-d^&?~e-%)OnXzct3ac*x zTq4(31vAoi`7yx^FAw;;zX>Q5%a0ZcA#@yywvYYu2M2{3^6Q2|$n~v&{e^>DC7MZ@ z{WvLyZwu@%9cC2pfr5MxrAJwlp77OY4>KB$%1}#FN2)I`3U%aH6@`##_GB`uLiVQ; zg>w0+g#2)$5T{Qr3K7%97ZN;}sGEy^4^jAaMxh6Lu~vwQP-gD{9?C)SWr(ExoWl|+ zmL3!#oFj~dIyZJTTj4&%Vc#Isla1q3Dr{GUi0cpIJCsm#Zxa4&iWR1m?8XW*j@6`- zF(FIG5@9k>xxs`bLb7Q?$cFk-U@cljh3KAyQc4cXP^@3wh5b=DVa_(SZ%5mL^C%%a z5RV`x8r30mhX`RR7{KifEY_3lgJw9{hj~M>@nmm`^WmPJbZj?s_--u5d9;#qa{FK8 z>B$jOAfBo0#OoHO?k7mnb*9U~(5@)k^ZJIV?o>vef8Q8xL&j;(>ma6!Sy559{bu0r z;hhzi&+8?oOO+iji?BskkhX)vu3A3+apj0e%2H;Qb*#s97v%Jgk9l6t<*BlF$FE><7c$hQY|raHrc+AM z-F{sEFDm!p6z+PZm1yd9t%rkuj7>zx)=XE`Qzs%n$^Y~;t?nhc*QWe>r*Z&<003wac z>&%7inZ5y$(>}jm(QO+phdr(cons::ctx.curr_hash_state.c_str()) << ")"; + // node has finished a consensus round (all 4 stages). + LOG_INFO << "****Stage 3 consensus reached**** (lcl:" << ctx.lcl + << " state:" << *reinterpret_cast(cons::ctx.curr_hash_state.c_str()) << ")"; } } } - // We have finished a consensus stage. - + // Node has finished a consensus stage. // Transition to next stage. ctx.stage = (ctx.stage + 1) % 4; - // after a stage proposal we will just busy wait for proposals. - util::sleep(conf::cfg.roundtime / 4); + //Here nodes try to synchronise nodes stages using network clock. + uint64_t now = util::get_epoch_milliseconds(); + + // round start is the floor + uint64_t round_start = ((uint64_t)(now / conf::cfg.roundtime)) * conf::cfg.roundtime; + + uint64_t next_stage_start = 0; + + // Compute start time of next stage. + // Last stage (stage 3) waiting twice as any other stage's waiting time. + // This is for a node to catch up from lcl/state desync, + // becuase we are waiting (round time + time to next stage) to sync. + if (ctx.stage == 3) + next_stage_start = round_start + conf::cfg.roundtime; + else + next_stage_start = round_start + (int64_t)(ctx.stage * ((double)conf::cfg.roundtime / 5.0)); + + // Compute stage time wait. + // Node wait between stages to collect enough proposals from previous stages from other nodes. + int64_t to_wait = next_stage_start - now; + + LOG_DBG << "now = " << now << ", roundtime = " << conf::cfg.roundtime << ", round_start = " << round_start << ", next_stage_start = " << next_stage_start << ", to_wait = " << to_wait; + + // If a node doesn't have enough time (due to network delay) to recieve/send reliable stage proposals for next stage, + // it will continue particapating in this round, otherwise will join in next round(s). + if (to_wait < floor(conf::cfg.roundtime / 10)) //todo: self claculating/adjusting network delay + { + uint64_t next_round = round_start; + while (to_wait < floor(conf::cfg.roundtime / 10)) + { + next_round += conf::cfg.roundtime; + to_wait = next_round - now; + } + + LOG_INFO << "we missed a round, waiting " << to_wait << " and resetting to stage 0"; + ctx.stage = 0; + util::sleep(to_wait); + } + else + { + // after a stage proposal we will just busy wait for proposals. + util::sleep(to_wait); + } } /** @@ -411,7 +453,7 @@ p2p::proposal create_stage123_proposal(vote_counter &votes) // time is voted on a simple sorted (highest to lowest) and majority basis, since there will always be disagreement. int32_t highest_time_vote = 0; - for(auto itr = votes.time.rbegin(); itr != votes.time.rend(); ++itr) + for (auto itr = votes.time.rbegin(); itr != votes.time.rend(); ++itr) { const uint64_t time = itr->first; const int32_t numvotes = itr->second; @@ -872,4 +914,4 @@ void increment(std::map &counter, const T &candidate) counter.try_emplace(candidate, 1); } -} // namespace cons \ No newline at end of file +} // namespace cons diff --git a/src/cons/state_handler.cpp b/src/cons/state_handler.cpp index dca12bde..56442a22 100644 --- a/src/cons/state_handler.cpp +++ b/src/cons/state_handler.cpp @@ -12,11 +12,9 @@ namespace cons { // Max number of requests that can be awaiting response at any given time. -constexpr uint16_t MAX_AWAITING_REQUESTS = 4; +constexpr uint16_t MAX_AWAITING_REQUESTS = 1; // Syncing loop sleep delay. -constexpr uint16_t SYNC_LOOP_WAIT = 50; -// No of loop cycles to wait for a response before resubmitting request. -constexpr uint16_t MAX_AWAITING_CYCLES = 1000 / SYNC_LOOP_WAIT; +constexpr uint16_t SYNC_LOOP_WAIT = 100; // List of state responses flatbuffer messages to be processed. std::list candidate_state_responses; @@ -37,7 +35,7 @@ void request_state_from_peer(const std::string &path, const bool is_file, const p2p::peer_outbound_message msg(std::make_unique(1024)); fbschema::p2pmsg::create_msg_from_state_request(msg.builder(), sr, lcl); - p2p::send_message_to_random_peer(msg); + p2p::send_message_to_random_peer(msg); //todo: send to a node that hold the majority state to improve reliability of retrieving state. } int create_state_response(p2p::peer_outbound_message &msg, const p2p::state_request &sr) @@ -159,7 +157,8 @@ int run_state_sync_iterator() // Check for long-awaited responses and re-request them. for (auto &[hash, request] : submitted_requests) { - if (request.waiting_cycles < MAX_AWAITING_CYCLES) + // We wait for half of round time before each request is resubmitted. + if (request.waiting_cycles < (conf::cfg.roundtime / (SYNC_LOOP_WAIT * 2))) { // Increment counter. request.waiting_cycles++; diff --git a/test/vm-cluster/cluster.sh b/test/vm-cluster/cluster.sh index 4ddbbdb7..02b4d1e6 100755 --- a/test/vm-cluster/cluster.sh +++ b/test/vm-cluster/cluster.sh @@ -7,17 +7,26 @@ mode=$1 hpcore=$(realpath ../..) -if [ "$mode" = "new" ] || [ "$mode" = "run" ] || [ "$mode" = "update" ]; then +if [ "$mode" = "new" ] || [ "$mode" = "run" ] || [ "$mode" = "update" ] || [ "$mode" = "kill" ]; then echo "" else - echo "Invalid command. new | run | update expected." + echo "Invalid command. new | run | update | kill expected." exit 1 fi if [ $mode = "run" ]; then let nodeid=$2-1 vmip=${vmips[$nodeid]} - sshpass -p $vmpass ssh geveo@$vmip 'sudo ./hpcore run contract' + sshpass -p $vmpass ssh geveo@$vmip 'nohup sudo ./hpcore run contract' + sshpass -p $vmpass ssh geveo@$vmip 'tail -f nohup.out' + exit 0 +fi + +if [ $mode = "kill" ]; then + let nodeid=$2-1 + vmip=${vmips[$nodeid]} + sshpass -p $vmpass ssh geveo@$vmip 'sudo kill $(pidof hpcore)' + sshpass -p $vmpass ssh geveo@$vmip 'sudo kill $(pidof hpstatemon)' exit 0 fi @@ -80,7 +89,7 @@ do mypeers=$(joinarr peers $j) myunl=$(joinarr pubkeys $j) - node -p "JSON.stringify({...require('./cfg/node$n.json'),binary:'/usr/bin/node',binargs:'/home/geveo/contract.js',peers:${mypeers},unl:${myunl}}, null, 2)" > ./cfg/node$n.cfg + node -p "JSON.stringify({...require('./cfg/node$n.json'),binary:'/usr/bin/node',binargs:'/home/geveo/contract.js',peers:${mypeers},unl:${myunl},loggers:['console', 'file']}, null, 2)" > ./cfg/node$n.cfg # Copy local cfg file back to remote vm. vmip=${vmips[j]} diff --git a/test/vm-cluster/consensus-test-continuous.sh b/test/vm-cluster/consensus-test-continuous.sh index e0c04084..df7ea450 100644 --- a/test/vm-cluster/consensus-test-continuous.sh +++ b/test/vm-cluster/consensus-test-continuous.sh @@ -20,7 +20,7 @@ while true; do echo 'starting ...' STARTTIME=`date +%s` - nohup ~/hpcore run ~/contract > $PIPE 2>> $PIPE 3>MARKER & + nohup sudo ~/hpcore run ~/contract > $PIPE 2>> $PIPE 3>MARKER & PID=$! sleep 1 diff --git a/test/vm-cluster/setup-vm.sh b/test/vm-cluster/setup-vm.sh index 1f8a6663..6f2a9322 100755 --- a/test/vm-cluster/setup-vm.sh +++ b/test/vm-cluster/setup-vm.sh @@ -13,6 +13,7 @@ if [ $mode = "new" ]; then sshpass -p $vmpass scp $hpcore/build/hpcore \ $hpcore/build/hpstatemon \ $hpcore/examples/echocontract/contract.js \ + $hpcore/examples/rndcontract/rnd_contract \ /usr/local/lib/x86_64-linux-gnu/libfuse3.so.3 \ /usr/local/bin/fusermount3 \ ./consensus-test-continuous.sh \ @@ -25,5 +26,7 @@ else sshpass -p $vmpass scp $hpcore/build/hpcore \ $hpcore/build/hpstatemon \ $hpcore/examples/echocontract/contract.js \ + $hpcore/examples/rndcontract/rnd_contract \ + ./consensus-test-continuous.sh \ geveo@$vmip:~/ fi \ No newline at end of file