Merge pull request #3716

a2b557f 6795bd0 209ec96 ed2c81e a830db2 57ea902 31a895e ba8331c f7f1917 41be339 f025ae9 ef2cb63 dcfd299 5d3e702 2704624 2771a18 0e4c7d0 (moneromooo-monero)

src/blockchain_utilities/blockchain_import.cpp

@@ -164,7 +164,7 @@ int pop_blocks(cryptonote::core& core, int num_blocks)
   return num_blocks;
 }
 
-int check_flush(cryptonote::core &core, std::list<block_complete_entry> &blocks, bool force)
+int check_flush(cryptonote::core &core, std::vector<block_complete_entry> &blocks, bool force)
 {
   if (blocks.empty())
     return 0;
@@ -176,7 +176,7 @@ int check_flush(cryptonote::core &core, std::list<block_complete_entry> &blocks,
   if (!force && new_height % HASH_OF_HASHES_STEP)
     return 0;
 
-  std::list<crypto::hash> hashes;
+  std::vector<crypto::hash> hashes;
   for (const auto &b: blocks)
   {
     cryptonote::block block;
@@ -312,7 +312,7 @@ int import_from_file(cryptonote::core& core, const std::string& import_file_path
   MINFO("Reading blockchain from bootstrap file...");
   std::cout << ENDL;
 
-  std::list<block_complete_entry> blocks;
+  std::vector<block_complete_entry> blocks;
 
   // Skip to start_height before we start adding.
   {
@@ -437,7 +437,7 @@ int import_from_file(cryptonote::core& core, const std::string& import_file_path
         {
           cryptonote::blobdata block;
           cryptonote::block_to_blob(bp.block, block);
-          std::list<cryptonote::blobdata> txs;
+          std::vector<cryptonote::blobdata> txs;
           for (const auto &tx: bp.txs)
           {
             txs.push_back(cryptonote::blobdata());

src/common/threadpool.cpp

@@ -36,16 +36,17 @@
 #include "common/util.h"
 
 static __thread int depth = 0;
+static __thread bool is_leaf = false;
 
 namespace tools
 {
 threadpool::threadpool(unsigned int max_threads) : running(true), active(0) {
   boost::thread::attributes attrs;
   attrs.set_stack_size(THREAD_STACK_SIZE);
   max = max_threads ? max_threads : tools::get_max_concurrency();
-  unsigned int i = max;
+  size_t i = max ? max - 1 : 0;
   while(i--) {
-    threads.push_back(boost::thread(attrs, boost::bind(&threadpool::run, this)));
+    threads.push_back(boost::thread(attrs, boost::bind(&threadpool::run, this, false)));
   }
 }
 
@@ -60,20 +61,25 @@ threadpool::~threadpool() {
   }
 }
 
-void threadpool::submit(waiter *obj, std::function<void()> f) {
-  entry e = {obj, f};
+void threadpool::submit(waiter *obj, std::function<void()> f, bool leaf) {
+  CHECK_AND_ASSERT_THROW_MES(!is_leaf, "A leaf routine is using a thread pool");
   boost::unique_lock<boost::mutex> lock(mutex);
-  if ((active == max && !queue.empty()) || depth > 0) {
+  if (!leaf && ((active == max && !queue.empty()) || depth > 0)) {
     // if all available threads are already running
     // and there's work waiting, just run in current thread
     lock.unlock();
     ++depth;
+    is_leaf = leaf;
     f();
     --depth;
+    is_leaf = false;
   } else {
     if (obj)
       obj->inc();
-    queue.push_back(e);
+    if (leaf)
+      queue.push_front({obj, f, leaf});
+    else
+      queue.push_back({obj, f, leaf});
     has_work.notify_one();
   }
 }
@@ -91,17 +97,20 @@ threadpool::waiter::~waiter()
   }
   try
   {
-    wait();
+    wait(NULL);
   }
   catch (const std::exception &e)
   {
     /* ignored */
   }
 }
 
-void threadpool::waiter::wait() {
+void threadpool::waiter::wait(threadpool *tpool) {
+  if (tpool)
+    tpool->run(true);
   boost::unique_lock<boost::mutex> lock(mt);
-  while(num) cv.wait(lock);
+  while(num)
+    cv.wait(lock);
 }
 
 void threadpool::waiter::inc() {
@@ -113,24 +122,30 @@ void threadpool::waiter::dec() {
   const boost::unique_lock<boost::mutex> lock(mt);
   num--;
   if (!num)
-    cv.notify_one();
+    cv.notify_all();
 }
 
-void threadpool::run() {
+void threadpool::run(bool flush) {
   boost::unique_lock<boost::mutex> lock(mutex);
   while (running) {
     entry e;
     while(queue.empty() && running)
+    {
+      if (flush)
+        return;
       has_work.wait(lock);
+    }
     if (!running) break;
 
     active++;
     e = queue.front();
     queue.pop_front();
     lock.unlock();
     ++depth;
+    is_leaf = e.leaf;
     e.f();
     --depth;
+    is_leaf = false;
 
     if (e.wo)
       e.wo->dec();

src/common/threadpool.h

@@ -59,15 +59,15 @@ class threadpool
     public:
     void inc();
     void dec();
-    void wait();  //! Wait for a set of tasks to finish.
+    void wait(threadpool *tpool);  //! Wait for a set of tasks to finish.
     waiter() : num(0){}
     ~waiter();
   };
 
   // Submit a task to the pool. The waiter pointer may be
   // NULL if the caller doesn't care to wait for the
   // task to finish.
-  void submit(waiter *waiter, std::function<void()> f);
+  void submit(waiter *waiter, std::function<void()> f, bool leaf = false);
 
   unsigned int get_max_concurrency() const;
 
@@ -78,6 +78,7 @@ class threadpool
     typedef struct entry {
       waiter *wo;
       std::function<void()> f;
+      bool leaf;
     } entry;
     std::deque<entry> queue;
     boost::condition_variable has_work;
@@ -86,7 +87,7 @@ class threadpool
     unsigned int active;
     unsigned int max;
     bool running;
-    void run();
+    void run(bool flush = false);
 };
 
 }

src/cryptonote_basic/connection_context.h

@@ -52,7 +52,7 @@ namespace cryptonote
     };
 
     state m_state;
-    std::list<crypto::hash> m_needed_objects;
+    std::vector<crypto::hash> m_needed_objects;
     std::unordered_set<crypto::hash> m_requested_objects;
     uint64_t m_remote_blockchain_height;
     uint64_t m_last_response_height;