806 improve consensus throughput latest

Consensust.h

@@ -23,7 +23,7 @@
 
 #pragma once
 
-#define DEFAULT_RUNNING_TIME_S 30
+#define DEFAULT_RUNNING_TIME_S 180
 #define STUCK_TEST_TIME 5
 
 class Consensust {

chains/Schain.cpp

@@ -84,6 +84,7 @@
 #include "messages/NetworkMessageEnvelope.h"
 #include "monitoring/MonitoringAgent.h"
 #include "monitoring/StuckDetectionAgent.h"
+#include "monitoring/OptimizerAgent.h"
 #include "network/ClientSocket.h"
 #include "network/IO.h"
 #include "network/Sockets.h"
@@ -241,6 +242,7 @@ Schain::Schain( weak_ptr< Node > _node, schain_index _schainIndex, const schain_
       node( _node ),
       schainIndex( _schainIndex ) {
     lastCommittedBlockTimeStamp = TimeStamp( 0, 0 );
+    setTimeStampValuesFromConfig();
 
     // construct monitoring, timeout and stuck detection agents early
     monitoringAgent = make_shared< MonitoringAgent >( *this );
@@ -291,10 +293,7 @@ Schain::Schain( weak_ptr< Node > _node, schain_index _schainIndex, const schain_
         getNode()->registerAgent( this );
 
 
-        if ( getNode()->getPatchTimestamps().count( "verifyDaSigsPatchTimestamp" ) > 0 ) {
-            this->verifyDaSigsPatchTimestampS =
-                getNode()->getPatchTimestamps().at( "verifyDaSigsPatchTimestamp" );
-        }
+
 
     } catch ( ExitRequestedException& ) {
         throw;
@@ -309,6 +308,7 @@ void Schain::constructChildAgents() {
     MONITOR( __CLASS_NAME__, __FUNCTION__ )
 
     try {
+        optimizerAgent = make_shared< OptimizerAgent >( *this );
         oracleResultAssemblyAgent = make_shared< OracleResultAssemblyAgent >( *this );
         pricingAgent = make_shared< PricingAgent >( *this );
         catchupClientAgent = make_shared< CatchupClientAgent >( *this );
@@ -404,6 +404,13 @@ void Schain::lockWithDeadLockCheck( const char* _functionName ) {
             result = ( ( uint64_t ) getLastCommittedBlockID() ) - committedIDOld;
             if ( !getNode()->isSyncOnlyNode() ) {
                 proposeNextBlock( true );
+            } else {
+                // on sync nodes we get candidate block and throw it away immediately
+                // this is to clean skaled queues
+                if (extFace) {    // if extFace is null we are in consensus tests and there is no skaled
+                    u256 stateRoot = 0;
+                    extFace->pendingTransactions( getNode()->getMaxTransactionsPerBlock(), stateRoot );
+                }
             }
         }
 
@@ -541,6 +548,13 @@ void Schain::proposeNextBlock( bool _isCalledAfterCatchup ) {
 
         proposedBlockArrived( myProposal );
 
+        if (getOptimizerAgent()->skipSendingProposalToTheNetwork(_proposedBlockID)) {
+            // a node skips sending and saving its proposal during
+            // optimized block consensus, if the node was not a winner
+            // last time
+            return; // dont propose
+        }
+
         LOG( debug, "PROPOSING BLOCK NUMBER:" << to_string( _proposedBlockID ) );
 
         auto db = getNode()->getProposalHashDB();
@@ -884,7 +898,10 @@ void Schain::daProofArrived( const ptr< DAProof >& _daProof ) {
         if ( _daProof->getBlockId() <= getLastCommittedBlockID() )
             return;
 
-        auto pv = getNode()->getDaProofDB()->addDAProof( _daProof );
+        // this will add the DAProof to DB. If there are enough DAProofs in DB
+        // to start binary consensus, this will return binary proposal vector of 1s and 0s
+        auto pv =
+            addDAProofToDBAndCalculateProposalVectorIfItsTimeToStartBinaryConsensus( _daProof );
 
 
         if ( pv != nullptr ) {
@@ -1462,3 +1479,51 @@ uint64_t Schain::getVerifyDaSigsPatchTimestampS() const {
 
 
 mutex Schain::vdsMutex;
+
+// this function is called on arrival of each DA proof
+// if it is time to make binary proposals it will return a vector of 0s and 1s
+// for normal consensus it will happen when 2t+1 DA proofs  arrive (which is 11)
+// for optimized consensus it will happen when a DA proof from the previous winner arrives
+ptr<BooleanProposalVector>
+Schain::addDAProofToDBAndCalculateProposalVectorIfItsTimeToStartBinaryConsensus(
+    const ptr<DAProof> &_daProof) {
+
+    ptr<BooleanProposalVector> pv;
+
+    if (getOptimizerAgent()->doOptimizedConsensus(_daProof->getBlockId(), getLastCommittedBlockTimeStamp().getS())) {
+        // when we do optimized block consensus only the previous winner
+        // proposes and provides da proof
+        // proposals from other nodes, if sent made by mistake, are ignored
+        auto lastWinner = getOptimizerAgent()->getPreviousWinner( _daProof->getBlockId() );
+        if (_daProof->getProposerIndex() == lastWinner) {
+            getNode()->getDaProofDB()->addDAProof(_daProof);
+            pv = make_shared<BooleanProposalVector>(getNodeCount(), lastWinner);
+        }
+    } else {
+        // do things regular way
+        // the binary proposal vector is formed and the consensus is started when
+        // 2/3 of nodes  (11) submit a da proof
+        pv = getNode()->getDaProofDB()->addDAProof(_daProof);
+    }
+    return pv;
+}
+
+// returns true if fastConsensusPatch ie enabled
+bool Schain::fastConsensusPatchEnabled(uint64_t _blockTimeStampSec ) {
+    return fastConsensusPatchTimestampS != 0 && _blockTimeStampSec >= fastConsensusPatchTimestampS;
+}
+
+// macro to set patchstamp variable from connfig
+#define SET_TIMESTAMP_FROM_CONFIG(__TIMESTAMP_NAME__) \
+    { \
+        auto& timestamps = getNode()->getPatchTimestamps(); \
+        if (timestamps.count(#__TIMESTAMP_NAME__) > 0) { \
+            __TIMESTAMP_NAME__ = timestamps.at(#__TIMESTAMP_NAME__); \
+        } \
+    }
+
+// set all timestamp values from config
+void Schain::setTimeStampValuesFromConfig() {
+    SET_TIMESTAMP_FROM_CONFIG(verifyDaSigsPatchTimestampS)
+    SET_TIMESTAMP_FROM_CONFIG(fastConsensusPatchTimestampS)
+}

chains/Schain.h

@@ -61,7 +61,7 @@ class CatchupServerAgent;
 class MonitoringAgent;
 class TimeoutAgent;
 class StuckDetectionAgent;
-
+class OptimizerAgent;
 
 class BlockProposalServerAgent;
 
@@ -136,6 +136,7 @@ class Schain : public Agent {
 
     ptr< OracleResultAssemblyAgent > oracleResultAssemblyAgent;
 
+    ptr<OptimizerAgent> optimizerAgent;
 
     ptr< IO > io;
 
@@ -176,6 +177,7 @@ class Schain : public Agent {
     ptr< NodeInfo > thisNodeInfo = nullptr;
 
     uint64_t verifyDaSigsPatchTimestampS = 0;
+    uint64_t fastConsensusPatchTimestampS = 0;
 
     // If a BlockError analyzer is added to the queue
     // its analyze(CommittedBlock _block) function will be run on commit
@@ -390,4 +392,12 @@ class Schain : public Agent {
 
     const ptr<CatchupClientAgent> &getCatchupClientAgent() const;
 
+    ptr< OptimizerAgent > getOptimizerAgent() const;
+
+    bool fastConsensusPatchEnabled( uint64_t _blockTimeStampSec );
+
+    void setTimeStampValuesFromConfig();
+
+    ptr<BooleanProposalVector>
+    addDAProofToDBAndCalculateProposalVectorIfItsTimeToStartBinaryConsensus(const ptr<DAProof> &_daProof);
 };

chains/SchainGettersSetters.cpp

@@ -268,6 +268,13 @@ ptr< CryptoManager > Schain::getCryptoManager() const {
     return cryptoManager;
 }
 
+
+ptr< OptimizerAgent > Schain::getOptimizerAgent() const {
+    CHECK_STATE( optimizerAgent );
+    return optimizerAgent;
+}
+
+
 void Schain::createBlockConsensusInstance() {
     blockConsensusInstance = make_shared< BlockConsensusAgent >( *this );
 }
@@ -320,8 +327,6 @@ void Schain::updateLastCommittedBlockInfo( uint64_t _lastCommittedBlockID,
     tpsAverage = ( blockSizeAverage * 1000 ) / blockTimeAverageMs;
     getRandomForBlockId( ( uint64_t ) lastCommittedBlockID );
 
-    if ( getNode()->isSyncOnlyNode() )
-        return;
 }
 
 

datastructures/BooleanProposalVector.cpp

@@ -63,6 +63,27 @@ BooleanProposalVector::BooleanProposalVector( node_count _nodeCount, const strin
 }
 
 
+
+// This is boolean proposal constructor in case of optimized proposal. In this
+// case we proposer 1 for the previos winner and 0 for everyone else
+BooleanProposalVector::BooleanProposalVector( node_count _nodeCount, schain_index _previousWinner )
+        : nodeCount( _nodeCount ) {
+
+    CHECK_ARGUMENT( _previousWinner > 0 );
+    CHECK_ARGUMENT( _previousWinner <= (uint64_t )_nodeCount );
+    proposals.push_back( false );
+
+    for ( uint64_t i = 1; i <= _nodeCount; i++ ) {
+        if ( i == _previousWinner ) {
+            proposals.push_back( true );
+            trueCount++;
+        } else {
+            proposals.push_back( false );
+        }
+    }
+}
+
+
 bool BooleanProposalVector::getProposalValue( schain_index _index ) {
     CHECK_STATE( proposals.size() == nodeCount + 1 );
     CHECK_STATE( _index <= ( uint64_t ) nodeCount );

datastructures/BooleanProposalVector.h

@@ -37,6 +37,8 @@ class BooleanProposalVector : public DataStructure {
 
     BooleanProposalVector( node_count _nodeCount, const string& _vectorStr );
 
+    BooleanProposalVector( node_count _nodeCount, schain_index _singleWinner );
+
     bool getProposalValue( schain_index _index );
 
     [[nodiscard]] uint64_t getTrueCount() const;

db/BlockProposalDB.cpp

@@ -36,6 +36,7 @@
 #include "exceptions/FatalError.h"
 #include "leveldb/db.h"
 #include "monitoring/LivelinessMonitor.h"
+#include "monitoring/OptimizerAgent.h"
 #include "node/Node.h"
 #include "pendingqueue/PendingTransactionsAgent.h"
 #include "thirdparty/json.hpp"
@@ -79,8 +80,17 @@ void BlockProposalDB::addBlockProposal( const ptr< BlockProposal > _proposal ) {
 
     addProposalToCacheIfDoesNotExist( _proposal );
 
+
     // save own proposal to levelDB
     if ( _proposal->getProposerIndex() == getSchain()->getSchainIndex() ) {
+
+        // for optimized consensus only previous winner proposed.
+        // non-winners skip sending proposal and do not need to save it to the db
+        // since saving proposal to the db is done to be able to resend it in case of a
+        // crash
+        if (getSchain()->getOptimizerAgent()->skipSendingProposalToTheNetwork(_proposal->getBlockID()))
+                return;
+
         serializeProposalAndSaveItToLevelDB( _proposal );
     }
 }