Avoid deadlocks because of reusing same pool for creating fibers

Since submit_eth_call_to_pool() is called from rust thread, blocking on
fiber promises is not appropriate and can lead to performance issues.
Move recovert_authorities() logic inside lambda executed in fiber pool.

Since eth call txn has single authority, parallelization logic inside
recover_authorities() only intoduces overhead without benefits. Replace
parallelization with single call to recover_authority().

Modify recover_senders_and_authorities() to recover senders and
authorities directly using recover_sender() and recover_authority()
instead of using the pool-based monad::recover_senders() and
monad::recover_authorities().

This avoids potential deadlock if all fibers in the pool are occupied,
since the function is called from within a fiber pool lambda context."

Refactor fiber pools to enable thread sharing between fiber groups.

Introduce FiberThreadPool and FiberGroup abstractions that allow multiple
fiber groups to share OS threads while maintaining separate task queues and
fiber limits.

Architecture:
  PriorityPool = FiberThreadPool + FiberGroup (isolated, dedicated threads)
  Shared setup = FiberThreadPool + multiple FiberGroups (shared threads)

Thread 0 runs a bootstrap fiber that handles fiber creation requests to
ensure fibers are created on threads with proper scheduler configuration.

Use trace_tx_exec_poo to parallelize trace transaction execution to
avoid deadlock because of using same pool.

Author: maxkozlovsky

category/core/CMakeLists.txt

@@ -126,6 +126,10 @@ add_library(
   "event/test_event_ctypes_metadata.c"
   # fiber
   "fiber/config.hpp"
+  "fiber/fiber_group.cpp"
+  "fiber/fiber_group.hpp"
+  "fiber/fiber_thread_pool.cpp"
+  "fiber/fiber_thread_pool.hpp"
   "fiber/priority_algorithm.cpp"
   "fiber/priority_algorithm.hpp"
   "fiber/priority_pool.cpp"

category/core/fiber/fiber_group.cpp

@@ -0,0 +1,85 @@
+// Copyright (C) 2025 Category Labs, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#include <category/core/fiber/fiber_group.hpp>
+
+#include <category/core/assert.h>
+#include <category/core/fiber/config.hpp>
+#include <category/core/fiber/fiber_thread_pool.hpp>
+#include <category/core/fiber/priority_properties.hpp>
+#include <category/core/fiber/priority_task.hpp>
+
+#include <boost/fiber/channel_op_status.hpp>
+#include <boost/fiber/fiber.hpp>
+#include <boost/fiber/operations.hpp>
+#include <boost/fiber/protected_fixedsize_stack.hpp>
+
+#include <cstddef>
+#include <future>
+#include <memory>
+#include <utility>
+
+MONAD_FIBER_NAMESPACE_BEGIN
+
+FiberGroup::FiberGroup(FiberThreadPool &pool, unsigned const n_fibers)
+    : pool_{pool}
+{
+    MONAD_ASSERT(n_fibers);
+
+    pool_.register_group();
+    fibers_.reserve(n_fibers);
+
+    // Create fibers via bootstrap channel so they're created on a thread with
+    // the proper scheduler configured.
+    pool_.submit_bootstrap_task([this, n_fibers] {
+        for (unsigned i = 0; i < n_fibers; ++i) {
+            auto *const properties = new PriorityProperties{nullptr};
+            boost::fibers::fiber fiber{
+                static_cast<boost::fibers::fiber_properties *>(properties),
+                std::allocator_arg,
+                boost::fibers::protected_fixedsize_stack{
+                    static_cast<size_t>(8 * 1024 * 1024)},
+                [this, properties] {
+                    PriorityTask task;
+                    while (channel_.pop(task) ==
+                           boost::fibers::channel_op_status::success) {
+                        properties->set_priority(task.priority);
+                        boost::this_fiber::yield();
+                        task.task();
+                        properties->set_priority(0);
+                    }
+                }};
+            fibers_.push_back(std::move(fiber));
+        }
+        start_.set_value();
+    });
+
+    start_.get_future().wait();
+}
+
+FiberGroup::~FiberGroup()
+{
+    channel_.close();
+
+    while (fibers_.size()) {
+        auto &fiber = fibers_.back();
+        fiber.join();
+        fibers_.pop_back();
+    }
+
+    pool_.unregister_group();
+}
+
+MONAD_FIBER_NAMESPACE_END

category/core/fiber/fiber_group.hpp

@@ -0,0 +1,77 @@
+// Copyright (C) 2025 Category Labs, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#pragma once
+
+#include <category/core/fiber/config.hpp>
+#include <category/core/fiber/priority_task.hpp>
+
+#include <boost/fiber/buffered_channel.hpp>
+#include <boost/fiber/fiber.hpp>
+
+#include <functional>
+#include <future>
+#include <vector>
+
+MONAD_FIBER_NAMESPACE_BEGIN
+
+class FiberThreadPool;
+
+/// FiberGroup represents a group of fibers that execute on a shared
+/// FiberThreadPool. Multiple FiberGroup instances can coexist on the same
+/// thread pool, allowing different components to have their own task queues
+/// and fiber limits while sharing the underlying OS threads.
+///
+/// Each FiberGroup has:
+/// - Its own buffered_channel for task submission
+/// - Its own set of fibers that process tasks from the channel
+/// - Reference to a shared FiberThreadPool for thread resources
+///
+/// The fibers in this group can migrate to any thread in the pool via
+/// work-stealing through the shared PriorityQueue.
+///
+/// FiberGroup must be destroyed before the FiberThreadPool it references.
+class FiberGroup final
+{
+    FiberThreadPool &pool_;
+
+    boost::fibers::buffered_channel<PriorityTask> channel_{1024};
+
+    std::vector<boost::fibers::fiber> fibers_{};
+
+    std::promise<void> start_{};
+
+    friend class FiberThreadPool;
+
+    FiberGroup(FiberThreadPool &pool, unsigned n_fibers);
+
+public:
+    FiberGroup(FiberGroup const &) = delete;
+    FiberGroup &operator=(FiberGroup const &) = delete;
+
+    ~FiberGroup();
+
+    void submit(uint64_t const priority, std::function<void()> task)
+    {
+        channel_.push({priority, std::move(task)});
+    }
+
+    unsigned num_fibers() const
+    {
+        return static_cast<unsigned>(fibers_.size());
+    }
+};
+
+MONAD_FIBER_NAMESPACE_END

category/core/fiber/fiber_thread_pool.cpp

@@ -0,0 +1,130 @@
+// Copyright (C) 2025 Category Labs, Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+#include <category/core/fiber/fiber_thread_pool.hpp>
+
+#include <category/core/assert.h>
+#include <category/core/fiber/config.hpp>
+#include <category/core/fiber/fiber_group.hpp>
+#include <category/core/fiber/priority_algorithm.hpp>
+#include <category/core/fiber/priority_properties.hpp>
+
+#include <boost/fiber/channel_op_status.hpp>
+#include <boost/fiber/fiber.hpp>
+#include <boost/fiber/operations.hpp>
+#include <boost/fiber/protected_fixedsize_stack.hpp>
+
+#include <atomic>
+#include <cstddef>
+#include <cstdio>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include <pthread.h>
+
+MONAD_FIBER_NAMESPACE_BEGIN
+
+FiberThreadPool::FiberThreadPool(
+    unsigned const n_threads, bool const prevent_spin)
+    : prevent_spin_{prevent_spin}
+{
+    MONAD_ASSERT(n_threads);
+
+    threads_.reserve(n_threads);
+
+    // Create worker threads (1 through n_threads-1) that wait for work via
+    // the shared queue with priority-based work-stealing scheduler.
+    for (unsigned i = n_threads - 1; i > 0; --i) {
+        auto thread = std::thread([this, i] {
+            char name[16];
+            std::snprintf(name, 16, "ftpool %u", i);
+            pthread_setname_np(pthread_self(), name);
+
+            boost::fibers::use_scheduling_algorithm<PriorityAlgorithm>(
+                queue_, prevent_spin_);
+
+            std::unique_lock<boost::fibers::mutex> lock{mutex_};
+            cv_.wait(lock, [this] { return done_; });
+        });
+        threads_.push_back(std::move(thread));
+    }
+
+    // Thread 0 runs a bootstrap fiber that handles fiber creation requests
+    // from FiberGroup constructors.
+    auto thread = std::thread([this] {
+        pthread_setname_np(pthread_self(), "ftpool 0");
+
+        boost::fibers::use_scheduling_algorithm<PriorityAlgorithm>(
+            queue_, prevent_spin_);
+
+        auto *const properties = new PriorityProperties{nullptr};
+        boost::fibers::fiber bootstrap_fiber{
+            static_cast<boost::fibers::fiber_properties *>(properties),
+            std::allocator_arg,
+            boost::fibers::protected_fixedsize_stack{
+                static_cast<size_t>(8 * 1024 * 1024)},
+            [this] {
+                std::function<void()> task;
+                while (bootstrap_channel_.pop(task) ==
+                       boost::fibers::channel_op_status::success) {
+                    task();
+                }
+            }};
+
+        bootstrap_fiber.detach();
+
+        std::unique_lock<boost::fibers::mutex> lock{mutex_};
+        cv_.wait(lock, [this] { return done_; });
+    });
+    threads_.push_back(std::move(thread));
+}
+
+FiberThreadPool::~FiberThreadPool()
+{
+    MONAD_ASSERT(
+        active_groups_.load(std::memory_order_relaxed) == 0 &&
+        "All FiberGroup instances must be destroyed before FiberThreadPool");
+
+    bootstrap_channel_.close();
+
+    {
+        std::unique_lock<boost::fibers::mutex> const lock{mutex_};
+        done_ = true;
+    }
+
+    cv_.notify_all();
+
+    while (threads_.size()) {
+        auto &thread = threads_.back();
+        thread.join();
+        threads_.pop_back();
+    }
+}
+
+std::unique_ptr<FiberGroup>
+FiberThreadPool::create_fiber_group(unsigned const n_fibers)
+{
+    MONAD_ASSERT(n_fibers);
+    MONAD_ASSERT(
+        !done_ && "Cannot create FiberGroup after FiberThreadPool shutdown");
+
+    return std::unique_ptr<FiberGroup>(new FiberGroup{*this, n_fibers});
+}
+
+MONAD_FIBER_NAMESPACE_END