Bytecoin v.2.0.6 release

ReleaseNotes.txt

@@ -1,3 +1,9 @@
+Release notes 2.0.6
+
+- Wallet synchronization enhancements
+- Correct application closing
+- Daemon stability enhancements
+
 Release notes 2.0.5
 
 - Fixed Bytecoin RPC Wallet hang issue

src/CryptoNoteConfig.h

@@ -185,7 +185,8 @@ const CheckpointData CHECKPOINTS[] = {
   {1268000, "d49fcaec1d53095e2c244913f123bfd4b26eabb6d75aca7b77a00de8aa8ac680"},
   {1272000, "2fb2c50328c8345d2f0a16b3ec4ea680a8a93730358494265ada9edbb9bfa1a6"},
   {1273000, "496a9238c654d79c48d269224aa75d61f51831bae6dc744f5e709bec11c7c9f2"},
-  {1278000, "de0225cd279ca27cc8d4f8da1b5b92ba0112e48b3777b8c50301846ccfc9146b"}
+  {1278000, "de0225cd279ca27cc8d4f8da1b5b92ba0112e48b3777b8c50301846ccfc9146b"},
+  {1283000, "826043db95e9801f038f254d223ce0d0912da269dcce1461b5f0f05ddfae9e1c"}
 };
 } // CryptoNote
 

src/Platform/Linux/System/Dispatcher.cpp

@@ -89,6 +89,7 @@ Dispatcher::Dispatcher() {
           mainContext.group = &contextGroup;
           mainContext.groupPrev = nullptr;
           mainContext.groupNext = nullptr;
+          mainContext.inExecutionQueue = false;
           contextGroup.firstContext = nullptr;
           contextGroup.lastContext = nullptr;
           contextGroup.firstWaiter = nullptr;
@@ -169,6 +170,10 @@ void Dispatcher::dispatch() {
     if (firstResumingContext != nullptr) {
       context = firstResumingContext;
       firstResumingContext = context->next;
+
+      assert(context->inExecutionQueue);
+      context->inExecutionQueue = false;
+
       break;
     }
 
@@ -251,7 +256,13 @@ bool Dispatcher::interrupted() {
 
 void Dispatcher::pushContext(NativeContext* context) {
   assert(context != nullptr);
+
+  if (context->inExecutionQueue)
+    return;
+
   context->next = nullptr;
+  context->inExecutionQueue = true;
+
   if(firstResumingContext != nullptr) {
     assert(lastResumingContext != nullptr);
     lastResumingContext->next = context;
@@ -414,6 +425,7 @@ void Dispatcher::contextProcedure(void* ucontext) {
   context.ucontext = ucontext;
   context.interrupted = false;
   context.next = nullptr;
+  context.inExecutionQueue = false;
   firstReusableContext = &context;
   ucontext_t* oldContext = static_cast<ucontext_t*>(context.ucontext);
   if (swapcontext(oldContext, static_cast<ucontext_t*>(currentContext->ucontext)) == -1) {

src/Platform/Linux/System/Dispatcher.h

@@ -30,6 +30,7 @@ struct NativeContext {
   void* ucontext;
   void* stackPtr;
   bool interrupted;
+  bool inExecutionQueue;
   NativeContext* next;
   NativeContextGroup* group;
   NativeContext* groupPrev;

src/Platform/OSX/System/Dispatcher.cpp

@@ -85,6 +85,7 @@ Dispatcher::Dispatcher() : lastCreatedTimer(0) {
           mainContext.group = &contextGroup;
           mainContext.groupPrev = nullptr;
           mainContext.groupNext = nullptr;
+          mainContext.inExecutionQueue = false;
           contextGroup.firstContext = nullptr;
           contextGroup.lastContext = nullptr;
           contextGroup.firstWaiter = nullptr;
@@ -145,6 +146,10 @@ void Dispatcher::dispatch() {
     if (firstResumingContext != nullptr) {
       context = firstResumingContext;
       firstResumingContext = context->next;
+
+      assert(context->inExecutionQueue);
+      context->inExecutionQueue = false;
+
       break;
     }
 
@@ -237,7 +242,12 @@ bool Dispatcher::interrupted() {
 
 void Dispatcher::pushContext(NativeContext* context) {
   assert(context!=nullptr);
+
+  if (context->inExecutionQueue)
+    return;
+
   context->next = nullptr;
+  context->inExecutionQueue = true;
   if (firstResumingContext != nullptr) {
     assert(lastResumingContext != nullptr);
     lastResumingContext->next = context;
@@ -388,6 +398,7 @@ void Dispatcher::contextProcedure(void* ucontext) {
   context.uctx = ucontext;
   context.interrupted = false;
   context.next = nullptr;
+  context.inExecutionQueue = false;
   firstReusableContext = &context;
   uctx* oldContext = static_cast<uctx*>(context.uctx);
   if (swapcontext(oldContext, static_cast<uctx*>(currentContext->uctx)) == -1) {

src/Platform/OSX/System/Dispatcher.h

@@ -31,6 +31,7 @@ struct NativeContext {
   void* uctx;
   void* stackPtr;
   bool interrupted;
+  bool inExecutionQueue;
   NativeContext* next;
   NativeContextGroup* group;
   NativeContext* groupPrev;

src/Platform/Windows/System/Dispatcher.cpp

@@ -65,6 +65,7 @@ Dispatcher::Dispatcher() {
         mainContext.group = &contextGroup;
         mainContext.groupPrev = nullptr;
         mainContext.groupNext = nullptr;
+        mainContext.inExecutionQueue = false;
         contextGroup.firstContext = nullptr;
         contextGroup.lastContext = nullptr;
         contextGroup.firstWaiter = nullptr;
@@ -131,6 +132,10 @@ void Dispatcher::dispatch() {
     if (firstResumingContext != nullptr) {
       context = firstResumingContext;
       firstResumingContext = context->next;
+
+      assert(context->inExecutionQueue);
+      context->inExecutionQueue = false;
+
       break;
     }
 
@@ -149,6 +154,10 @@ void Dispatcher::dispatch() {
     if (firstResumingContext != nullptr) {
       context = firstResumingContext;
       firstResumingContext = context->next;
+
+      assert(context->inExecutionQueue);
+      context->inExecutionQueue = false;
+
       break;
     }
 
@@ -224,7 +233,13 @@ bool Dispatcher::interrupted() {
 void Dispatcher::pushContext(NativeContext* context) {
   assert(GetCurrentThreadId() == threadId);
   assert(context != nullptr);
+
+  if (context->inExecutionQueue) {
+    return;
+  }
+
   context->next = nullptr;
+  context->inExecutionQueue = true;
   if (firstResumingContext != nullptr) {
     assert(lastResumingContext->next == nullptr);
     lastResumingContext->next = context;
@@ -359,6 +374,11 @@ void Dispatcher::pushReusableContext(NativeContext& context) {
 
 void Dispatcher::interruptTimer(uint64_t time, NativeContext* context) {
   assert(GetCurrentThreadId() == threadId);
+
+  if (context->inExecutionQueue) {
+    return;
+  }
+
   auto range = timers.equal_range(time);
   for (auto it = range.first; ; ++it) {
     assert(it != range.second);
@@ -376,6 +396,7 @@ void Dispatcher::contextProcedure() {
   NativeContext context;
   context.interrupted = false;
   context.next = nullptr;
+  context.inExecutionQueue = false;
   firstReusableContext = &context;
   SwitchToFiber(currentContext->fiber);
   for (;;) {

src/Platform/Windows/System/Dispatcher.h

@@ -29,6 +29,7 @@ struct NativeContextGroup;
 struct NativeContext {
   void* fiber;
   bool interrupted;
+  bool inExecutionQueue;
   NativeContext* next;
   NativeContextGroup* group;
   NativeContext* groupPrev;

src/Platform/Windows/System/TcpConnection.cpp

@@ -140,6 +140,10 @@ size_t TcpConnection::read(uint8_t* data, size_t size) {
     throw InterruptedException();
   }
 
+  if (context.interrupted) {
+    throw InterruptedException();
+  }
+
   assert(transferred <= size);
   assert(flags == 0);
   return transferred;
@@ -209,6 +213,10 @@ size_t TcpConnection::write(const uint8_t* data, size_t size) {
     throw InterruptedException();
   }
 
+  if (context.interrupted) {
+    throw InterruptedException();
+  }
+
   assert(transferred == size);
   assert(flags == 0);
   return transferred;

src/Platform/Windows/System/TcpConnector.cpp

@@ -157,6 +157,14 @@ TcpConnection TcpConnector::connect(const Ipv4Address& address, uint16_t port) {
                   }
                 }
               } else {
+                if (context2.interrupted) {
+                  if (closesocket(connection) != 0) {
+                    throw std::runtime_error("TcpConnector::connect, closesocket failed, " + errorMessage(WSAGetLastError()));
+                  } else {
+                    throw InterruptedException();
+                  }
+                }
+
                 assert(transferred == 0);
                 assert(flags == 0);
                 DWORD value = 1;

src/Platform/Windows/System/TcpListener.cpp

@@ -182,6 +182,15 @@ TcpConnection TcpListener::accept() {
             }
           }
         } else {
+          if (context2.interrupted) {
+            if (closesocket(connection) != 0) {
+              throw std::runtime_error("TcpConnector::connect, closesocket failed, " + errorMessage(WSAGetLastError()));
+            }
+            else {
+              throw InterruptedException();
+            }
+          }
+
           assert(transferred == 0);
           assert(flags == 0);
           if (setsockopt(connection, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, reinterpret_cast<char*>(&listener), sizeof listener) != 0) {

src/System/ContextGroup.cpp

@@ -83,6 +83,10 @@ void ContextGroup::wait() {
   if (contextGroup.firstContext != nullptr) {
     NativeContext* context = dispatcher->getCurrentContext();
     context->next = nullptr;
+
+    assert(!context->inExecutionQueue);
+    context->inExecutionQueue = true;
+
     if (contextGroup.firstWaiter != nullptr) {
       assert(contextGroup.lastWaiter->next == nullptr);
       contextGroup.lastWaiter->next = context;

src/Transfers/TransfersConsumer.cpp

@@ -36,6 +36,19 @@ namespace {
 
 using namespace CryptoNote;
 
+class MarkTransactionConfirmedException : public std::exception {
+public:
+    MarkTransactionConfirmedException(const Crypto::Hash& txHash) {
+    }
+
+    const Hash& getTxHash() const {
+        return m_txHash;
+    }
+
+private:
+    Crypto::Hash m_txHash;
+};
+
 void checkOutputKey(
   const KeyDerivation& derivation,
   const PublicKey& key,
@@ -218,17 +231,20 @@ uint32_t TransfersConsumer::onNewBlocks(const CompleteBlock* blocks, uint32_t st
   BlockingQueue<Tx> inputQueue(workers * 2);
 
   std::atomic<bool> stopProcessing(false);
+  std::atomic<size_t> emptyBlockCount(0);
 
   auto pushingThread = std::async(std::launch::async, [&] {
     for( uint32_t i = 0; i < count && !stopProcessing; ++i) {
       const auto& block = blocks[i].block;
 
       if (!block.is_initialized()) {
+        ++emptyBlockCount;
         continue;
       }
 
       // filter by syncStartTimestamp
       if (m_syncStart.timestamp && block->timestamp < m_syncStart.timestamp) {
+        ++emptyBlockCount;
         continue;
       }
 
@@ -308,7 +324,7 @@ uint32_t TransfersConsumer::onNewBlocks(const CompleteBlock* blocks, uint32_t st
     return std::tie(a.blockInfo.height, a.blockInfo.transactionIndex) < std::tie(b.blockInfo.height, b.blockInfo.transactionIndex);
   });
 
-  uint32_t processedBlockCount = 0;
+  uint32_t processedBlockCount = emptyBlockCount;
   try {
     for (const auto& tx : preprocessedTransactions) {
       processTransaction(tx.blockInfo, *tx.tx, tx);
@@ -324,6 +340,14 @@ uint32_t TransfersConsumer::onNewBlocks(const CompleteBlock* blocks, uint32_t st
         });
       }
     }
+  } catch (const MarkTransactionConfirmedException& e) {
+    m_logger(ERROR, BRIGHT_RED) << "Failed to process block transactions: failed to confirm transaction " << e.getTxHash() <<
+      ", remove this transaction from all containers and transaction pool";
+    forEachSubscription([&e](TransfersSubscription& sub) {
+      sub.deleteUnconfirmedTransaction(e.getTxHash());
+    });
+
+    m_poolTxs.erase(e.getTxHash());
   } catch (std::exception& e) {
     m_logger(ERROR, BRIGHT_RED) << "Failed to process block transactions, exception: " << e.what();
   } catch (...) {
@@ -539,9 +563,14 @@ void TransfersConsumer::processOutputs(const TransactionBlockInfo& blockInfo, Tr
 
   if (contains) {
     if (subscribtionTxInfo.blockHeight == WALLET_UNCONFIRMED_TRANSACTION_HEIGHT && blockInfo.height != WALLET_UNCONFIRMED_TRANSACTION_HEIGHT) {
-      // pool->blockchain
-      sub.markTransactionConfirmed(blockInfo, tx.getTransactionHash(), globalIdxs);
-      updated = true;
+      try {
+        // pool->blockchain
+        sub.markTransactionConfirmed(blockInfo, tx.getTransactionHash(), globalIdxs);
+        updated = true;
+      } catch (...) {
+          m_logger(ERROR, BRIGHT_RED) << "markTransactionConfirmed failed, throw MarkTransactionConfirmedException";
+          throw MarkTransactionConfirmedException(tx.getTransactionHash());
+      }
     } else {
       assert(subscribtionTxInfo.blockHeight == blockInfo.height);
     }