16 AsyncJobQueueOnThread

The AsyncJobQueueOnThread<Job> creates a single thread responsible for asynchronously processing jobs in its queue of objects of type Job.

template<typename Job>
class AsyncJobQueueOnThread
{
public:
    void Start();
    void Stop();
    void Push(const Job& job);
};

Objects of type Job must be callable objects (i.e. implement operator()), and support default construction and copy construction.

E.g.

struct MyJob
{
    MyJob(Task* task = nullptr) : task(task) {}

    void operator()
    {
        task->Run();
    }

    Task* task;
};

Correct usage of the AsyncJobQueueOnThread involves

1. A single call to Start()
2. Any number of calls to Push()
3. A single call to Stop(). This must come after all calls to Push().

Implementation

The implementation has a ManualResetEvent member which is signalled when the queue transitions to non-empty and non-signalled when the queue transitions to empty.

The ManualResetEvent must be signalled inside the lock on the mutex

Consider that the event is signalled outside the mutex lock in the Push method:

void Push(Job* job)
{
    bool wasEmpty;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        wasEmpty = queue_.empty();
        queue_.push_back(job);
    }
    if (wasEmpty)
    {
        evtQueueNonEmpty_.Signal();
    }
}

Job* Pop()
{
    Job* job;
    bool becomeEmpty;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty())
        {
            job = nullptr;
            becomeEmpty = true;
        }
        else
        {
            job = queue_.back();
            queue_.pop_back();
            becomeEmpty = queue_.empty();
        }
    }
    if (becomeEmpty)
    {
        evtQueueNonEmpty_.Reset();
    }
    return job;
}

Consider this sequence of events:

1. Push inserts job into empty queue
2. Pop removes job from queue making it empty again
3. Pop resets event
4. Push signals event

--> event is signalled incorrectly!

So instead must signal/reset the event within the lock on the mutex:

void Push(Job* job)
{
    std::lock_guard<std::mutex> lock(mutex_);
    if (queue_.empty())
    {
        evtQueueNonEmpty_.Signal();
    }
    queue_.push_back(job);
}

Job* Pop()
{
    std::lock_guard<std::mutex> lock(mutex_);
    if (queue_.empty())
    {
        return nullptr;
    }
    else
    {
        Job* job = queue_.back();
        queue_.pop_back();
        if (queue_.empty())
        {
            evtQueueNonEmpty_.Reset();
        }
        return job;
    }
}

Shutting down thread correctly

The worker thread should only exit once Stop() has been called and the queue is empty.

Consider this implementation:

Stop()
{
    requestShutDown_ = true;
    evtQueueNonEmpty_.Signal();
    StopThread();
}    

WorkerThread()
{
    while(1)
    {
        evtQueueNonEmpty_.Wait();
        while(Job* job = Pop())
        {
            // Process the job
            job->Execute();
        }
        if (requestShutDown_) break;
    }
}

Consider this sequence of events:

1. WorkerThread breaks out of Pop() loop because queue is empty
2. Client pushes a job
3. Client calls Stop() --> sets requestShutDown_ = true
4. WorkerThread exits.

This is wrong because the worker thread has exited without processing all the jobs.

IDEA: Consider that we push a special value of a Job* that represents termination. Since this is pushed last, we know that all previous jobs were handled.

With this design we don't even need to signal the event in order to stop.

AsyncJobQueueOnThread.h

// AsyncJobQueueOnThread.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2013

#pragma once
#ifndef Ceda_cxThread_AsyncJobQueueOnThread_H
#define Ceda_cxThread_AsyncJobQueueOnThread_H

#ifdef CEDA_DEPRECATED

#include "ThreadName.h"
#include "ManualResetEvent.h"
#include "Ceda/cxUtils/CedaAssert.h"
#include "Ceda/cxUtils/xdeque.h"
#include <thread>
#include <mutex>

namespace ceda
{
template<typename Job>
class AsyncJobQueueOnThread
{
public:
    AsyncJobQueueOnThread() :
        finished_(false)
    {
        // Note that initially evtQueueNonEmpty_ is non signalled because queue is empty
    }
    void Start()
    {
        finished_ = false;
        thread_ = std::thread(
            [this]
            {
                SetThreadName("AsyncJobQueueOnThread");
    
                /*
                For performance the queue is processed in batches - by swapping out the entire queue each
                time.  This avoids the CPU expense of popping individual elements, and reduces contention
                for access to the queue between producers and consumer.  Typically the batches will be 
                quite small.  They are self adjusting to the extent that very small batches increases the
                per-job overheads resulting in larger batches to be used.
    
                In order to avoid continual reallocation of the page in the queue we reuse a local variable
                'q' and call clear2() which clears the queue without deleting the first page.
                */
                QUEUE q;
                while(1)
                {
                    // The worker thread is put to sleep using a manual reset event when there are no jobs 
                    // queued.  The event is signalled once the queue becomes non-empty.
                    evtQueueNonEmpty_.Wait();
        
                    while(1)
                    {
                        bool finished = PopAll(q);
                        for (auto& job : q)
                        {
                            job();
                        }
                        if (finished)
                        {
                            return;
                        }
                        if (q.empty()) break;

                        // Clear q without deleting the first page.
                        q.clear2();
                    }
                }
            });

    }
    void Stop()
    {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            finished_ = true;
            evtQueueNonEmpty_.Signal();
        }
        thread_.join();                  // Blocks until worker thread exits
    }
    void Push(const Job& job)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty())
        {
            // Queue transitioning from empty to non-empty so signal the event
            evtQueueNonEmpty_.Signal();
        }
        queue_.push_back(job);
    }

private:
    typedef xdeque<Job> QUEUE;

    bool PopAll(QUEUE& q)
    {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.swap(q);
        evtQueueNonEmpty_.Reset();
        return finished_;
    }

private:
    std::thread thread_;

    bool finished_;

    // The queued async load jobs.
    QUEUE queue_;

    // Manual-reset event which is in the signalled state if and only if the queue is non-empty
    ManualResetEvent evtQueueNonEmpty_;
    
    // Protect access to queue_ and finished_
    mutable std::mutex mutex_;
};
} // namespace ceda

#endif // CEDA_DEPRECATED

#endif // include guard

Proposal

The convention for the standard C++ library containers that have a capacity, is for the capacity to remain unchanged when the clear method is called. We should follow this same convention for an xdeque, rather than have a second method named clear2.

A new version of AsyncJobQueueOnThread is needed that uses a single boost asio io_context from an IoContextPool. There is no need for a dedicated thread. There is no need to use a ManualResetEvent to put a thread into an efficient wait state while there is no more work to do.

We should not require the tasks to be copyable. Instead we should only require them to be movable.

We should avoid any and all assumptions about what threads may call Start, Push and Stop or when they are called or in what order they are called - they may even be called concurrently by different threads. The only requirement is that a task cannot call Stop, or else it will dead-lock itself, similar to a AsyncPeriodicTimer.

The implementation should tend to post tasks to a single io_context which is assumed to be run by a single thread. Therefore all the asynchronous callbacks are implicitly serialised. Therefore no mutex is needed.