Conveyor Thread Wrapper for Modern C++

class ConveyorThreadWrapperDemo : public ConveyorThreadWrapper<int> {
    enum class command {
        quit = 'q', // entire thread
        abort = 'a', // current task
        sleep = 's',
        wakeUp = 'w',
    };
    static const char* help() { return "a: task abort, q: quit, s: sleep, w: wake up, key: "; }
    static bool commandIs(char c, command cmd) { return (int)cmd == (int)c; }
public:
    using natural = unsigned long long int;
    static void Run(natural delayMs) {
        natural max = 100;
        IntegerConveyor cv;
        cv.CurrentTaskAborted = [](int) -> void {
            std::cout << ". Enter any character: ";
        }; //cv.CurrentTaskAborted
        using cType = decltype(cv);
        auto sleep = [delayMs] {
            std::this_thread::sleep_for(std::chrono::milliseconds(delayMs));
        }; //sleep
        auto Hailstone = [] (natural a) -> natural {
            if (a % 2 == 0)
                return a / 2;
            else
                return 3 * a + 1;
        }; //Hailstone (iteration)
        auto HailstoneStoppingTime = [Hailstone] (natural n) -> natural {
            natural stoppingTime = 0;
            while (n != 1) {
                n = Hailstone(n);
                ++stoppingTime;
            } //loop
            return stoppingTime;
        }; //HailstoneStoppingTime
        auto lambdaHailstone = [HailstoneStoppingTime, sleep, max] (cType::SyncDelegate& sync, int value) {
            for (natural count = 1; count < max; ++count) {
                natural stoppingTime = HailstoneStoppingTime(count + value);
                sync(false);
                std::cout
                    << count << ": " << help()
                    << value << " => Hailstone " << stoppingTime << std::endl;
                sleep();
            } //loop
        }; //lambdaHailstone
        natural mod = 1;
        mod = mod << 32; // NOT 1 < 32 !!!
        natural multiplier = 1664525; // Numerical Recipes
        natural increment = 1013904223; // Numerical Recipes
        auto Lgc = [mod, multiplier, increment](natural n) -> natural {
            return (n * multiplier + increment) % mod;
        }; //Lgc
        auto lambdaLgc = [Lgc, sleep, max](cType::SyncDelegate& sync, int value) {
            natural n = value;
            for (natural count = 0; count < max; count++) {
                n = Lgc(n);
                sync(false);
                std::cout
                    << help()
                    << value << " => LGC " << (n) << std::endl;
                sleep();
            } //loop
        }; //lambdaLgc
        SA::delegate<void(cType::SyncDelegate&, int)> delHailstone = lambdaHailstone;
        SA::delegate<void(cType::SyncDelegate&, int)> delLgc = lambdaLgc;
        cv.Start();
        char cmd;
        while (true) {
            std::cin >> cmd;
            if (commandIs(cmd, command::abort))
                cv.Abort();
            else if (commandIs(cmd, command::quit)) {
                cv.Abort(cType::AbortDepth::entierThread);
                break;
            } else if (commandIs(cmd, command::sleep))
                cv.PutToSleep();
            else if (commandIs(cmd, command::wakeUp))
                cv.WakeUp();
            else {
                int cmdValue = (int)cmd;
                if (cmdValue % 2 == 0)
                    cv.Enqueue(delHailstone, cmdValue);
                else
                    cv.Enqueue(delLgc, cmdValue);
            } //if
        } //loop
        cv.Join();
    } //Run
}; //class ConveyorThreadWrapperDemo

This demo shows the use of lambda expressions which are extremely convenient for the case. Alternatively, the mechanism of delegates also allows for the use of static functions and constant or non-constant instance functions, as it is explained in the article on delegates.

By the way, note a pretty obvious but important lambda expression technique: passing some objects through closure and not through the parameter list. For example, the lambda expression sleep uses the value delayMs from the parameter of the method Run as auto sleep = [delayMs].

For this application, this is essential. It could be passed in a parameter list, but let’s see how sleep is used. Ultimately, all those lambda expressions are called by the lambda expressions lambdaHailstone and lambdaLgc and also are passed through their closures. These two ultimate lambda expressions are never called in this context. Instead, they are assigned to the delegate instances. Their parameter lists should fit the profile of the delegate (shown here) and cannot accept other actual parameters.

Pay attention that multiple characters can be entered sequentially while some task is being executed. It demonstrates the cases when new tasks are queued while one of the tasks is being executed. If, say, N tasks were queued during execution of another tasks, it will take up to N+1 current task aborts to bring the thread to a wait state. When the presently executing task is aborted, the next task is taken for execution from the queue.

If shallow abort is not performed for a while, the current task eventual completes execution. It puts the thread to sleep. The thread can be woken up by adding another task to the queue. I intentionally specified small number of iterations, to demonstrate this condition. There is no a need to add another virtual function or a delegate to hook up this event by the user code, because the appropriate action, if needed, can be simple written as a part of the task delegate code.

In this code sample, I tried to shorten the text at the expense of purely mathematical algorithms and leave the “system” aspects. At the same time, the fun part is about those algorithms.

Demonstrated Algorithms

Let’s make a short fun break not really related to the topic of this article. In the demo application, to demonstrate passing different delegates to the queue, I’ve implemented the choice between the two (the chosen algorithm depends on the parity of the code point of the user-entered character): calculation of the total stopping time of a sequence of Hailstone numbers and linear congruential generator.

First thing is a very intricate mathematical problem, most likely, not solved at the moment of writing; the problem is considered as an “extraordinarily difficult problem, out of reach of the present day mathematics”. The problem is called Collatz conjecture, or 3n+1 conjecture. It says that repeated sequence of numbers defined by the recursive process n->n/2 for even n and n->3n+1 for odd n will eventually reach the value of 1 for any initial n. So far, we don’t know numbers which don’t behave this way, but it’s not proven that such numbers don’t exit. In my demo, I calculate the Hailstone number sequences themselves, but display only the numbers of iterations required to achieve 1, which is called total stopping time for a given initial number n. For initial n, I take the sum of the code point of the user-entered character and the number of iteration of an outer cycle. This total stopping time value and its dependency on the initial number is one of the real mathematical wonders — do read the article; it is very interesting.

The second algorithm is a very simple algorithm for generation of a deterministic sequence of pseudo-randomized numbers: linear congruential generator. The algorithm requires three carefully chosen numeric parameters, which I take according to Numerical Recipes.

The implementations of these algorithms are way too trivial to show them here; everyone is welcome to look at them at in the downloadable source code.

Giving the Wrapper a Task

This is how a task can be queued:

template<typename PARAM>
class ConveyorThreadWrapper : public ThreadWrapper {
public:
    
    // ...

    void Enqueue(TaskDelegate& task, PARAM argument) {
        std::lock_guard<std::mutex> lock(queueMutex);
        queue.push(new QueueItem(task, argument));
        queueCondition.notify_one();
    } //Enqueue

    // ...

}; //class ConveyorThreadWrapper

template<typename PARAM>
class ConveyorThreadWrapper : public ThreadWrapper {
// ...
private:

    // ...

    struct QueueItem {
        QueueItem() = default;
        QueueItem(TaskDelegate& del, PARAM param) {
            delegate = &del;
            parameter = param;
        } //QueueItem
        TaskDelegate* delegate;
        PARAM parameter;
    }; //struct QueueItem

    std::queue<QueueItem*> queue; // protected by queueState and queueMutex
    std::mutex queueMutex;
    std::condition_variable queueCondition;

    void getTask(QueueItem& itemCopy) {
        std::unique_lock<std::mutex> ul(queueMutex);
        queueCondition.wait(ul, [=] {
            return queue.size() > 0;
        });
        auto front = queue.front();
        itemCopy.delegate = front->delegate;
        itemCopy.parameter = front->parameter;
        queue.pop();
        delete front;
    } //getTask
}; //class ConveyorThreadWrapper

class ThreadWrapper {
// ...
protected:

    // ...

    static void InitializeSyncPointDelegate(SA::delegate<void(bool)>& del, ThreadWrapper* instance) {
        del = SA::delegate<void(bool)>::create<ThreadWrapper, &ThreadWrapper::SyncPoint>(instance);
    } //InitializeSyncPointDelegate

    // ...

}; //class ThreadWrapper

class ThreadWrapper {
// ...
protected:

    void Body() override final {
        while (true) {
            QueueItem task;
            getTask(task);
            try {
                (*task.delegate)(syncDelegate, task.parameter);
            } catch (ShallowThreadAbortException&) {
                SetAbort(false);
                OnCurrentTaskAborted(task.parameter);
            } //exception
        } //loop
    } //Body

    // ...

}; //class ThreadWrapper

class ThreadWrapper {
public:

    // ...

    enum class AbortDepth { currentTask, entierThread, };
    void Abort(AbortDepth depth = AbortDepth::currentTask) {
        SetAbort(true, depth == AbortDepth::currentTask);
    } //Abort

    // ...

}; //class ThreadWrapper

using SyncDelegate = SA::delegate<void(bool)>;
using TaskDelegate = SA::delegate<void(SyncDelegate&, PARAM)>;