Cleanup threadpool

Author: avevad

common/threadpool.hpp

@@ -11,17 +11,8 @@
 
 namespace ton {
 
-class ThreadPool {
-public:
-  ThreadPool(size_t num_threads = std::thread::hardware_concurrency()) = delete;
-  ~ThreadPool() = delete;
-
-  template <typename Iter, typename OIter>
-  static void invoke_task_group(Iter tasks_beg, Iter tasks_end, OIter res_beg, size_t num_threads = 0);
-};
-
-template <typename Iter, typename  OIter>
-void ThreadPool::invoke_task_group(Iter tasks_beg, Iter tasks_end, OIter res_beg, size_t num_threads) {
+template <typename Iter, typename OIter>
+void invoke_task_group(Iter tasks_beg, Iter tasks_end, OIter res_beg, size_t num_threads = 0) {
   if (num_threads == 0) {
     num_threads = std::thread::hardware_concurrency();
   }
@@ -34,14 +25,14 @@ void ThreadPool::invoke_task_group(Iter tasks_beg, Iter tasks_end, OIter res_beg
   }
   num_threads = std::min(num_threads, n);
   std::atomic_size_t cur_pos{0};
-  std::mutex panic_mutex;
+  std::mutex error_mutex;
   std::optional<std::exception_ptr> error;
 
   {
     std::vector<std::thread> workers;
     workers.reserve(num_threads);
     for (size_t id = 0; id < num_threads && cur_pos.load() < n; id++) {
-      workers.emplace_back([&cur_pos, n, in, out, &error, &panic_mutex]() -> void {
+      workers.emplace_back([&cur_pos, n, in, out, &error, &error_mutex]() -> void {
         while (true) {
           size_t pos = cur_pos.fetch_add(1);
           if (pos >= n) {
@@ -50,14 +41,14 @@ void ThreadPool::invoke_task_group(Iter tasks_beg, Iter tasks_end, OIter res_beg
           try {
             out[pos] = in[pos]();
           } catch (...) {
-            std::lock_guard panic_lock(panic_mutex);
+            std::lock_guard error_lock(error_mutex);
             error = std::current_exception();
           }
         }
       });
     }
 
-    for (auto &worker : workers) {
+    for (auto& worker : workers) {
       worker.join();
     }
   }
@@ -71,4 +62,4 @@ void ThreadPool::invoke_task_group(Iter tasks_beg, Iter tasks_end, OIter res_beg
   }
 }
 
-}
+}  // namespace ton

validator/impl/validate-query.cpp

@@ -6168,29 +6168,30 @@ bool ValidateQuery::check_transactions() {
   LOG(INFO) << "checking all transactions";
   std::deque<StdSmcAddress> account_addresses;
   std::deque<CheckAccountTxsCtx> account_contexts;
-  std::vector<std::function<unsigned char /*bool*/ ()>> account_tasks;
-
-  account_blocks_dict_->check_for_each_extra([this, &account_addresses, &account_contexts, &account_tasks] (Ref<vm::CellSlice> value, Ref<vm::CellSlice> extra, td::ConstBitPtr key, int key_len) {
-    CHECK(key_len == 256);
-    StdSmcAddress address = key;
-    account_addresses.push_back(address);
-
-    account_contexts.emplace_back();
-    CheckAccountTxsCtx& ctx = account_contexts.back();
-    if (account_expected_defer_all_messages_.count(address)) {
-      ctx.defer_all_messages = true;
-    }
+  std::vector<std::function<bool()>> account_tasks;
+
+  account_blocks_dict_->check_for_each_extra(
+      [this, &account_addresses, &account_contexts, &account_tasks](Ref<vm::CellSlice> value, Ref<vm::CellSlice> extra,
+                                                                    td::ConstBitPtr key, int key_len) {
+        CHECK(key_len == 256);
+        StdSmcAddress address = key;
+        account_addresses.push_back(address);
+
+        account_contexts.emplace_back();
+        CheckAccountTxsCtx& ctx = account_contexts.back();
+        if (account_expected_defer_all_messages_.count(address)) {
+          ctx.defer_all_messages = true;
+        }
 
-    account_tasks.emplace_back([this, address, &ctx, acc_tr = std::move(value)] {
-      unsigned char result = check_account_transactions_ts(address, acc_tr, ctx);
-      return result;
-    });
-    return true;
-  });
+        account_tasks.emplace_back([this, address, &ctx, acc_tr = std::move(value)] {
+          return check_account_transactions_ts(address, acc_tr, ctx);
+        });
+        return true;
+      });
 
   try {
-    std::vector<unsigned char /*bool*/> account_results(account_tasks.size(), false);
-    ThreadPool::invoke_task_group(account_tasks.begin(), account_tasks.end(), account_results.begin());
+    std::vector<int> account_results(account_tasks.size(), false);
+    invoke_task_group(account_tasks.begin(), account_tasks.end(), account_results.begin());
     for (auto& ok : account_results) {
       if (!ok) {
         return false;
@@ -6203,7 +6204,7 @@ bool ValidateQuery::check_transactions() {
   }
 
   for (size_t pos = 0; pos < account_addresses.size(); pos++) {
-    auto &ctx = account_contexts[pos];
+    auto& ctx = account_contexts[pos];
 
     for (auto& e : ctx.msg_proc_lt) {
       msg_proc_lt_.emplace_back(std::move(e));