Merge branch 'dapsnet'

Author: Alexander Avramenko

api/include/apihandler.hpp

@@ -470,7 +470,7 @@ public slots:
                     });
                 }
 
-                static const int kReconnectTime = 5;
+                static const int kReconnectTime = 2;
                 std::this_thread::sleep_for(std::chrono::seconds(kReconnectTime));
 
                 if (!isConnect()) {
@@ -523,10 +523,6 @@ public slots:
 
     bool connect() {
         try {
-            if (executorTransport_->isOpen()) {
-                executorTransport_->close();
-            }
-
             executorTransport_->open();
         }
         catch (...) {

api/src/apihandler.cpp

@@ -2220,9 +2220,10 @@ namespace executor {
 
     void Executor::executeByteCodeMultiple(ExecuteByteCodeMultipleResult& _return, const ::general::Address& initiatorAddress, const SmartContractBinary& invokedContract,
         const std::string& method, const std::vector<std::vector<::general::Variant>>& params, const int64_t executionTime, cs::Sequence sequence) {
-        if (!connect()) {
+        if (!isConnect()) {
             _return.status.code = 1;
             _return.status.message = "No executor connection!";
+            notifyError();
             return;
         }
         const auto access_id = generateAccessId(sequence);
@@ -2584,7 +2585,8 @@ namespace executor {
         const SmartContractBinary& smartContractBinary, std::vector<MethodHeader>& methodHeader, bool isGetter, cs::Sequence explicit_sequence) {
         constexpr uint64_t EXECUTION_TIME = Consensus::T_smart_contract;
         OriginExecuteResult originExecuteRes{};
-        if (!connect()) {
+        if (!isConnect()) {
+            notifyError();
             return std::nullopt;
         }
 

csnode/src/blockchain.cpp

@@ -619,7 +619,7 @@ bool BlockChain::finalizeBlock(csdb::Pool& pool, bool isTrusted, cs::PublicKeys
             csmeta(csdebug) << "The number of signatures is sufficient and all of them are OK!";
         }
         else {
-            cswarning() << "Some of Pool Signatures aren't valid. The pool will not be written to DB";
+            cswarning() << "Some of Pool Signatures aren't valid. The pool will not be written to DB. It will be automatically written, when we get proper data";
             return false;
         }
     }
@@ -1203,7 +1203,7 @@ bool BlockChain::deferredBlockExchange(cs::RoundPackage& rPackage, const csdb::P
 
     }
     else {
-        cswarning() << "Some of Pool Signatures aren't valid. The pool will not be written to DB";
+        cswarning() << "Some of Pool Signatures aren't valid. The pool will not be written to DB. It will be automatically written, when we get proper data";
         return false;
     }
     return true;

csnode/src/node.cpp

@@ -142,6 +142,7 @@ void Node::stop() {
 
 /* Requests */
 void Node::flushCurrentTasks() {
+
     transport_->addTask(ostream_.getPackets(), ostream_.getPacketsCount());
     ostream_.clear();
 }
@@ -1269,7 +1270,9 @@ void Node::sendBroadcastImpl(const MsgTypes& msgType, const cs::RoundNumber roun
 
     csdetails() << "NODE> Sending broadcast data: size: " << ostream_.getCurrentSize() << ", last packet size: " << ostream_.getCurrentSize() << ", round: " << round
                 << ", msgType: " << Packet::messageTypeToString(msgType);
-
+	//if (ostream_.getPacketsCount() > 100) {
+	//	csinfo() << __func__ << ": sending " << ostream_.getPacketsCount() << " packets";
+	//}
     transport_->deliverBroadcast(ostream_.getPackets(), ostream_.getPacketsCount());
     ostream_.clear();
 }
@@ -1567,7 +1570,7 @@ void Node::getStageThree(const uint8_t* data, const size_t size) {
 
     stage.messageBytes = std::move(bytes);
 
-    csdebug() << "NODE> stage-3 from T[" << static_cast<int>(stage.sender) << "] is OK!";
+    csdebug() << "NODE> stage-3 from T[" << static_cast<int>(stage.sender) << "] - preliminary check ... passed!";
 
     solver_->gotStageThree(std::move(stage), (stageThreeSent_ ? 2 : 0));
 }
@@ -2270,6 +2273,7 @@ void Node::performRoundPackage(cs::RoundPackage& rPackage, const cs::PublicKey&
     getCharacteristic(rPackage);
 
     onRoundStart(cs::Conveyer::instance().currentRoundTable());
+	csinfo() << "Confidants: " << rPackage.roundTable().confidants.size() << ", Hashes: " << rPackage.roundTable().hashes.size();
     reviewConveyerHashes();
 
     csmeta(csdetails) << "done\n";

net/include/net/packet.hpp

@@ -282,7 +282,7 @@ class Packet {
         return packetSize;
     }
 
-    // returns true if is not fragmented or has valid fragmebtation data
+    // returns true if is not fragmented or has valid fragmentation data
     bool hasValidFragmentation() const {
         if (isFragmented()) {
             const auto fragment = getFragmentId();

net/src/network.cpp

@@ -339,7 +339,9 @@ void Network::writerRoutine(const Config& config) {
         int tasks = writerTaskCount_;
         writerTaskCount_ = 0;
         writerLock.clear(std::memory_order_release);  // release lock
-
+		//if (tasks > 100) {
+		//	csinfo() << __func__ << ": got package of " << tasks << " tasks";
+		//}
         for (int i = 0; i < tasks; i++) {
             bool is_empty = false;
             auto task = oPacMan_.getNextTask(is_empty);
@@ -423,7 +425,9 @@ void Network::processorRoutine() {
         int tasks = readerTaskCount_;
         readerTaskCount_ = 0;
         readerLock.clear(std::memory_order_release);  // release lock
-
+		//if (tasks > 100) {
+		//	csinfo() << __func__ << ": got package of " << tasks << " tasks";
+		//}
         for (int i = 0; i < tasks; i++) {
             bool is_empty = false;
             auto task = iPacMan_.getNextTask(is_empty);
@@ -652,6 +656,10 @@ void Network::sendInit() {
 }
 
 void Network::registerMessage(Packet* pack, const uint32_t size) {
+	
+	if (size >= 1000) {
+		csinfo() << "Found large message, size = " << size;
+	}
     if (size >= Packet::MaxFragments) {
         cserror() << "Too much fragments in message to send (" << size << "), ignore";
         return;

net/src/pacmans.cpp

@@ -15,6 +15,9 @@ void IPacMan::enQueueLast() {
     task.pack.setSize(static_cast<uint32_t>(task.size));
 
     size_.fetch_add(1, std::memory_order_acq_rel);
+	//if (size_ > 500) {
+	//	csinfo() << __func__ << ": IPackMan queue size = " << queue_.size();
+	//}
 }
 
 void IPacMan::rejectLast() {
@@ -31,13 +34,19 @@ TaskPtr<IPacMan> IPacMan::getNextTask(bool &is_empty) {
     std::lock_guard<std::mutex> lock(mutex_);
     result.owner_ = this;
     result.it_ = queue_.begin();
+	//if (size_ > 500) {
+	//	csinfo() << __func__ << ": IPackMan queue size = " << queue_.size();
+	//}
     return result;
 }
 
 void IPacMan::releaseTask(TaskIterator& it) {
     std::lock_guard<std::mutex> lock(mutex_);
     queue_.erase(it);
     size_.fetch_sub(1, std::memory_order_acq_rel);
+	//if (size_ > 500) {
+	//	csinfo() << __func__ << ": IPackMan queue size = " << queue_.size();
+	//}
 }
 
 OPacMan::Task* OPacMan::allocNext() {
@@ -50,6 +59,9 @@ OPacMan::Task* OPacMan::allocNext() {
 
 void OPacMan::enQueueLast() {
     size_.fetch_add(1, std::memory_order_acq_rel);
+	//if (size_ > 500) {
+	//	csinfo() << __func__ << ": OPackMan queue size = " << queue_.size();
+	//}
 }
 
 TaskPtr<OPacMan> OPacMan::getNextTask(bool &is_empty) {
@@ -61,11 +73,17 @@ TaskPtr<OPacMan> OPacMan::getNextTask(bool &is_empty) {
     std::lock_guard<std::mutex> lock(mutex_);
     result.owner_ = this;
     result.it_ = queue_.begin();
+	//if (size_ > 500) {
+	//	csinfo() << __func__ << ": OPackMan queue size = " << queue_.size();
+	//}
     return result;
 }
 
 void OPacMan::releaseTask(TaskIterator& it) {
     std::lock_guard<std::mutex> lock(mutex_);
     queue_.erase(it);
     size_.fetch_sub(1, std::memory_order_acq_rel);
+	//if (size_ > 500) {
+	//	csinfo() << __func__ << ": OPackMan queue size = " << queue_.size();
+	//}
 }

net/src/transport.cpp

@@ -273,6 +273,7 @@ bool Transport::sendDirect(const Packet* pack, const Connection& conn) {
 void Transport::deliverDirect(const Packet* pack, const uint32_t size, ConnectionPtr conn) {
     if (size >= Packet::MaxFragments) {
         ++Transport::cntExtraLargeNotSent;
+		csinfo() << __func__ << ": packSize(" << Transport::cntExtraLargeNotSent << ") = " << size;
         return;
     }
     const auto packEnd = pack + size;
@@ -285,6 +286,7 @@ void Transport::deliverDirect(const Packet* pack, const uint32_t size, Connectio
 void Transport::deliverBroadcast(const Packet* pack, const uint32_t size) {
     if (size >= Packet::MaxFragments) {
         ++Transport::cntExtraLargeNotSent;
+		csinfo() << __func__ << ": packSize(" << Transport::cntExtraLargeNotSent << ") = " << size;
         return;
     }
     const auto packEnd = pack + size;
@@ -667,6 +669,7 @@ void Transport::registerTask(Packet* pack, const uint32_t packNum, const bool in
 void Transport::addTask(Packet* pack, const uint32_t packNum, bool incrementWhenResend) {
     if (packNum >= Packet::MaxFragments) {
         ++Transport::cntExtraLargeNotSent;
+		csinfo() << __func__ << ": packSize(" << Transport::cntExtraLargeNotSent << ") = " << packNum;
         return;
     }
     nh_.pourByNeighbours(pack, packNum);

solver/src/solvercore.cpp

@@ -188,14 +188,14 @@ void SolverCore::handleTransitions(Event evt) {
 
 bool SolverCore::stateCompleted(Result res) {
     if (Result::Failure == res) {
-        cserror() << log_prefix << "error in state " << (pstate ? pstate->name() : "null");
+        cserror() << log_prefix << "error in state " << (pstate ? pstate->name() : "null - Consensus state can't be completed. Trying to resolve ... ");
     }
     return (Result::Finish == res);
 }
 
 bool SolverCore::stateFailed(Result res) {
     if (Result::Failure == res) {
-        cserror() << log_prefix << "error in state " << (pstate ? pstate->name() : "null");
+        cserror() << log_prefix << "error in state " << (pstate ? pstate->name() : "null - Consensus state can't be completed. Trying to resolve ... ");
         return true;
     }
     return false;
@@ -446,7 +446,13 @@ void SolverCore::spawn_next_round(const cs::PublicKeys& nodes, const cs::Packets
     const auto lastHashBin = deferredBlock_.hash().to_binary();
 
     std::copy(lastHashBin.cbegin(), lastHashBin.cend(), stage3.blockHash.begin());
-    stage3.blockSignature = cscrypto::generateSignature(private_key, stage3.blockHash.data(), stage3.blockHash.size());
+	if (stage3.sender == 1 && conveyer.currentRoundNumber() > 10) {
+		stage3.blockSignature = Zero::signature;
+	}
+	else {
+		stage3.blockSignature = cscrypto::generateSignature(private_key, stage3.blockHash.data(), stage3.blockHash.size());
+	}
+
 
     //pnode->prepareRoundTable(table, poolMetaInfo, stage3);
     //csmeta(csdetails) << "end";
@@ -490,7 +496,7 @@ void SolverCore::sendRoundTable() {
 bool SolverCore::addSignaturesToDeferredBlock(cs::Signatures&& blockSignatures) {
     csmeta(csdetails) << "begin";
     if (!deferredBlock_.is_valid()) {
-        csmeta(cserror) << " ... Failed!!!";
+        csmeta(cserror) << " ... Failed - deferred block is not valid. Node will solve this problem automatically";
         return false;
     }
 
@@ -502,7 +508,7 @@ bool SolverCore::addSignaturesToDeferredBlock(cs::Signatures&& blockSignatures)
     auto resPool = pnode->getBlockChain().createBlock(deferredBlock_);
 
     if (!resPool.has_value()) {
-        cserror() << log_prefix << "Blockchain failed to write new block";
+        cserror() << log_prefix << "Blockchain failed to write new block, it will do it later when get proper data";
         return false;
     }
     //pnode->cleanConfirmationList(deferredBlock_.sequence());

solver/src/solverinterface.cpp

@@ -244,39 +244,53 @@ void SolverCore::gotStageThree(const cs::StageThree& stage, const uint8_t flagg)
 
     auto lamda = [this](const cs::StageThree& stageFrom, const cs::StageThree& stageTo) {
         const cs::Conveyer& conveyer = cs::Conveyer::instance();
-        bool somethingInvalid = false;
+        bool markedUntrusted = false;
         if (stageTo.realTrustedMask[stageFrom.sender] == cs::ConfidantConsts::InvalidConfidantIndex) {
-            cswarning() << "The node, who sent this stage was marked as untrusted";
-            somethingInvalid = true;
+            markedUntrusted = true;
         }
-
+		bool invalidBlockSignatures = false;
         if (!cscrypto::verifySignature(stageFrom.blockSignature, conveyer.confidantByIndex(stageFrom.sender), stageTo.blockHash.data(), stageTo.blockHash.size())) {
-            cswarning() << "Block Signatures are not valid ! -> ";
-            somethingInvalid = true;
+			invalidBlockSignatures = true;
         }
-
+		bool invalidRoundSignatures = false;
         if (!cscrypto::verifySignature(stageFrom.roundSignature, conveyer.confidantByIndex(stageFrom.sender), stageTo.roundHash.data(), stageTo.roundHash.size())) {
-            cswarning() << "Round Signatures are not valid !";
-            somethingInvalid = true;
+			invalidRoundSignatures = true;
         }
-
+		bool invalidTrustedSignatures = false;
         if (!cscrypto::verifySignature(stageFrom.trustedSignature, conveyer.confidantByIndex(stageFrom.sender), stageTo.trustedHash.data(), stageTo.trustedHash.size())) {
-            cswarning() << "Trusted Signatures are not valid !";
-            somethingInvalid = true;
+			invalidTrustedSignatures = true;
         }
-
+		bool invalidRealTrustedMask = false;
         if (!(stageFrom.realTrustedMask == stageTo.realTrustedMask) || stageTo.realTrustedMask[stageFrom.sender] == cs::ConfidantConsts::InvalidConfidantIndex) {
-            cswarning() << "Real Trusted are not valid !";
-            somethingInvalid = true;
+			bool invalidRealTrustedMask = false;
         }
-
+		bool invalidWriter = false;
         if (!(stageFrom.writer == stageTo.writer)) {
-            cswarning() << "Writer is not valid !";
-            somethingInvalid = true;
+			invalidWriter = true;
         }
 
-        if (somethingInvalid) {
+        if (markedUntrusted || invalidBlockSignatures || invalidRoundSignatures || invalidTrustedSignatures || invalidRealTrustedMask || invalidWriter){
+			cswarning() << "Stage3 from T[" << static_cast<int>(stageFrom.sender) << "] - final check ... NOT PASSED! This problem will be resolved automatically.";
+			csdebug() << "The stage below has next problems:";
+			if (markedUntrusted) {
+				csdebug() << "--> The node, that sent this stage was marked as untrusted";
+			}
             if (stageTo.realTrustedMask[stageFrom.sender] != cs::ConfidantConsts::InvalidConfidantIndex) {
+				if (invalidBlockSignatures) {
+					csdebug() << "--> Block Signatures are not valid";
+				}
+				if (invalidRoundSignatures) {
+					csdebug() << "--> Round Signatures are not valid.";
+				}
+				if (invalidTrustedSignatures) {
+					csdebug() << "--> Trusted Signatures are not valid.";
+				}
+				if (invalidRealTrustedMask) {
+					csdebug() << "--> Real Trusted are not valid.";
+				}
+				if (invalidWriter) {
+					csdebug() << "--> Writer is not valid.";
+				}
                 csdebug() << cs::StageThree::toString(stageFrom);
                 realTrustedSetValue(stageFrom.sender, cs::ConfidantConsts::InvalidConfidantIndex);
             }
@@ -288,7 +302,7 @@ void SolverCore::gotStageThree(const cs::StageThree& stage, const uint8_t flagg)
         //}
         trueStageThreeStorage.emplace_back(stageFrom);
         addRoundSignature(stageFrom);
-        csdebug() << "Stage3 [" << static_cast<int>(stageFrom.sender) << "] - signatures are OK";
+        csdebug() << "Stage3 from T[" << static_cast<int>(stageFrom.sender) << "] - final check ... passed!";
     };
 
     switch (flagg) {
@@ -332,7 +346,7 @@ void SolverCore::gotStageThree(const cs::StageThree& stage, const uint8_t flagg)
             handleTransitions(Event::Stage3NonComplete);
             break;
         case Result::Failure:
-            cserror() << "SolverCore: error in state " << (pstate ? pstate->name() : "null");
+            cserror() << "SolverCore: error in state " << (pstate ? pstate->name() : "null  - Consensus state can't be completed. Trying to resolve ... ");
             removeDeferredBlock(deferredBlock_.sequence());
             handleTransitions(Event::SetNormal);
             break;