Boost Asio多线程性能异常及C++11线程池wait方法问题排查

1. Boost Asio io_service Performance Anomaly #

Let's break down why your Boost Asio-based implementation didn't scale well beyond 2 threads:

• Task Granularity & Lock Contention: If you were dispatching tiny tasks (like individual pixels or small pixel blocks) to the io_service, the overhead of locking the task queue and thread scheduling would eat up most of your parallel gains. With 2 threads, lock contention is minimal, so you see a nice speedup. But as you add more threads, every thread spends more time waiting to acquire the queue lock instead of processing pixels—this is why 3-7 threads perform almost like single-threaded. At 8 threads, you might be hitting your CPU's core count (assuming an 8-core CPU), so scheduling overhead is reduced a bit, leading to a small improvement.
• Cache Thrashing: A 20000×20000 bool array (especially if using std::vector<bool> which is bit-packed) means adjacent pixels sit in the same cache line. When multiple threads access overlapping cache lines, the CPU has to invalidate and reload cache lines constantly, killing performance. 2 threads might split the image into two large, non-overlapping blocks that fit nicely into cache, but more threads mean smaller blocks with more cache line overlaps or frequent invalidations.
• io_service Work Handling: Double-check you're using io_service::work to keep threads alive when there are no tasks. Without it, threads might exit prematurely, leading to underutilization—though this is less likely given your solid 2-thread result.

Fixes for Boost Asio: #

• Coarser Task Granularity: Split the image into large, contiguous blocks (e.g., 20000 / N rows per thread, where N is your thread count) instead of tiny tasks. This cuts down queue lock contention and improves cache locality.
• Avoid Bit-Packed Arrays: Replace std::vector<bool> with std::vector<uint8_t>—bit packing makes memory access patterns far less efficient for parallel processing, since accessing a single bit requires manipulating an entire byte.
• Pin Threads to Cores: Use pthread_setaffinity_np (Linux) or SetThreadAffinityMask (Windows) to bind each io_service thread to a specific CPU core. This reduces cross-core scheduling overhead and improves cache consistency.

2. Custom C++11 ThreadPool Wait() Issue & Code Review #

Since your custom thread pool scales well but has a flaky wait() method, here are the most common bugs to look for, plus a checklist for code review:

Common Bugs in wait() Implementations: #

1. Incorrect Atomic Counter Handling:

   • If you're using a non-atomic counter to track pending tasks, race conditions can mess up the count (e.g., two threads decrementing it at the same time, leading to an undercount). Always use std::atomic<int> or std::atomic_size_t for task counting.
   • Make sure you increment the counter before enqueuing the task. If you increment after, wait() might see a counter of 0 before all tasks are added to the queue.

2. Condition Variable Misuse:

   • Never use if to check the wait condition—always use while. Spurious wakeups are allowed by the C++ standard, so your wait loop should recheck the condition every time it's woken up.
   • Example of a correct wait loop:

     std::unique_lock<std::mutex> lock(m_mutex);
     m_cv.wait(lock, [this](){ return m_pending_tasks == 0; });
     

   • Ensure every task completion properly notifies the condition variable. After decrementing the pending task counter, call m_cv.notify_all() (safer than notify_one() for multiple waiting threads).

3. Race Conditions Between Task Submission & Wait:

   • If you allow adding tasks while wait() is running, make sure the counter and queue are protected by the same mutex. A common mistake is not locking the mutex when incrementing the counter, leading to wait() seeing an incorrect pending task count.

Code Review Checklist for Your ThreadPool: #

• Atomicity: Are all shared variables (task counters, queue state) accessed with proper synchronization (atomic types or mutex locks)?
• Condition Variable Logic: Does wait() use a while loop to check the pending task count, and is the condition variable notified every time a task finishes?
• Task Enqueue Order: Is the task counter incremented before the task is added to the queue? This ensures wait() sees the correct number of pending tasks.
• Thread Safety: Are all public methods (enqueue, wait, shutdown) properly synchronized to prevent race conditions between task submission and waiting?

Example of a Correct wait() Implementation Snippet: #

class ThreadPool {
private:
    std::vector<std::thread> m_threads;
    std::queue<std::function<void()>> m_tasks;
    std::mutex m_mutex;
    std::condition_variable m_cv;
    std::atomic_size_t m_pending_tasks = 0;
    bool m_stop = false;

public:
    // Constructor: Spawns worker threads
    explicit ThreadPool(size_t num_threads) {
        for (size_t i = 0; i < num_threads; ++i) {
            m_threads.emplace_back(&ThreadPool::worker_thread, this);
        }
    }

    // Enqueue a task
    void enqueue(std::function<void()> func) {
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            if (m_stop) throw std::runtime_error("Enqueue on stopped ThreadPool");
            m_tasks.push(std::move(func));
            m_pending_tasks++;
        }
        m_cv.notify_one();
    }

    // Wait for all pending tasks to complete
    void wait() {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_cv.wait(lock, [this]() {
            return m_pending_tasks == 0 && !m_stop;
        });
    }

    // Shutdown the pool (call before destruction)
    void shutdown() {
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_stop = true;
        }
        m_cv.notify_all();
        for (auto& thread : m_threads) {
            if (thread.joinable()) thread.join();
        }
    }

    ~ThreadPool() {
        if (!m_stop) shutdown();
    }

private:
    void worker_thread() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(m_mutex);
                m_cv.wait(lock, [this]() {
                    return m_stop || !m_tasks.empty();
                });
                if (m_stop && m_tasks.empty()) return;
                task = std::move(m_tasks.front());
                m_tasks.pop();
            }
            task();
            m_pending_tasks--;
            m_cv.notify_all();
        }
    }
};

内容的提问来源于stack exchange，提问作者Istu7