Computing Issues: Januar 2016

Three Threads Stream

About

Many applications are performing an editing of their input. Especially in image processing. They all need to buffer their input, if they want to handle the processing. Especially, if you want to process the reading, modify and writing of data not in one thread. So, you need buffers between the threads.
Here it describes the problem and gives a solution in form of an implementation.
Afterwards this document discuss the timing problematic with too slow or too fast threads.
Below, you can add some comments about your opinion. Don't hesitate to ask some questions.

Problem

As in "about" described, we want to process data for reading, modifying and writing. Of course, we can use one single data address to process the reading, modifying and writing in one single thread. But now we want to do this with three different threads. One for reading, one for modifying and one for writing.
So you can say: lets use one data address for all three threads. To avoid a race condition (simultaneous writing and reading), we use a mutex, which we can lock.

The issue is, that when we use a mutex, it could be, that only the input thread is accessing the data, while the modify thread is every time again waiting on the lock. This happens, because when the modify thread is checking, if the mutex is free again, the input thread has lock the mutex again. In that way, the input thread is much more faster than the modifying thread. And in this way, we omit a lot of data to modify and write.

The modifying thread is called working thread further on. And the writing thread is called output thread in the implementation.

So it is not suitable to use a mutex here. Because, we have too fast access on the mutex. More detailed, the threads run in different speed, so it can be, that one thread is running twice as fast as the other. Or one thread is running at whole, while the other thread do nothing.
So at least we need two conditions: not_full and not_empty to indicate the different states of the cyclic buffer, if only one thread is running. This buffer and conditions are invented by batptiste-wicht[1].

The implementation

The buffer "BoundedBuffer"[1] is slightly modified. It is now a template based class.
We need two buffers. One to transfer the input data to working thread, called inputToWorkingBuffer, and one to transfer modified data to output thread called workingToOutputBuffer. There are two operations on that buffers. One for fetching the last data and one for depositing new data.
There are streams handling the fetching and depositing of the buffers. See the operators << and >> therefore. The code is also improved by r-value references. The r-value references permit the data to be moved into the buffer and moved from the buffer. So the copy process is reduced as much as possible. In the end, there are only one copy from input stream to buffer and one copy to output stream.

The BoundedBuffer

#pragma once
/**
 * Under the license of LGPL V3.0
 * http://www.gnu.org/licenses/lgpl-3.0.de.html
 * Copyright © 2015 Free Software Foundation, Inc. <http://fsf.org/>
 * Everyone is permitted to copy and distribute verbatim copies or changing
 * of this document is allowed.
 * Changed documents must contain this license notice.
 * You can use another license for your own changes.
 */

#include <mutex>
#include <condition_variable>
#include <iostream>
#include <atomic>

/**
 * @brief The BoundedBuffer class
 * A cyclic buffer, which has the read index everytime behind the write index.
 * The buffer size is given in the constructor.
 * @see http://baptiste-wicht.com/posts/2012/04/c11-concurrency-tutorial-advanced-locking-and-condition-variables.html
 */
template<typename T>
class BoundedBuffer {
public:
 typedef T value_type;
 value_type* buffer;
 size_t capacity;

 size_t front;
 size_t rear;
 std::atomic_size_t count;

 std::mutex lock;

 std::condition_variable not_full;
 std::condition_variable not_empty;

    BoundedBuffer(size_t capacity) : capacity(capacity), front(0), rear(0) {
        count.store(0);
  buffer = new value_type[capacity];
        // XXX here you can allocate your frames once for the whole capacity.
        // Or you did not need to allocate, instead just move a pointer.
 }

 ~BoundedBuffer(){
  delete[] buffer;
        // XXX here you would deallocate all the frames.
 }

    void deposit(value_type &&data){
  std::unique_lock<std::mutex> l(lock);

        not_full.wait(l, [this](){return count.load() != capacity; });

  buffer[rear] = data;
  rear = (rear + 1) % capacity;
  ++count;

  not_empty.notify_one();
 }

    value_type && fetch(){
  std::unique_lock<std::mutex> l(lock);

        not_empty.wait(l, [this](){return count.load() != 0; });

  value_type result = buffer[front];
  front = (front + 1) % capacity;
  --count;

  not_full.notify_one();

        return std::move(result);
 }

    BoundedBuffer<t> & operator<<(typename BoundedBuffer<t>::value_type &&rhs)
    {
        deposit(std::move(rhs));
        // Return-Value-Optimization (RVO) will simplify the return value.
        return *this;
    }

    BoundedBuffer<t> & operator>>(typename BoundedBuffer<t>::value_type &rhs) // const - not possible, because front. Solution: front as mutable.
    {
        rhs = std::move(fetch());
        return *this;
    }

    template <t>
    friend std::ostream & operator<<(std::ostream &lhs, BoundedBuffer<t> &rhs);

    template <t>
    friend std::istream & operator>>(std::istream &lhs, BoundedBuffer<t> &rhs);

};

template<typename T>
std::ostream & operator<<(std::ostream &lhs, BoundedBuffer<t> &rhs)
{
    lhs << rhs.fetch();
    return lhs;
}

template<typename T>
std::istream & operator>>(std::istream &lhs, BoundedBuffer<t> &rhs)
{
    T value;
    lhs >> value;
    rhs.deposit(std::move(value));
    return lhs;
}

The main

/**
 * Under the license of LGPL V3.0
 * http://www.gnu.org/licenses/lgpl-3.0.de.html
 * Copyright © 2015 Free Software Foundation, Inc. <http://fsf.org/>
 * Everyone is permitted to copy and distribute verbatim copies or changing
 * of this document is allowed.
 * Changed documents must contain this license notice.
 * You can use another license for your own changes.
 */
#include <iostream>
#include <chrono>
#include <thread>
#include <atomic>
#include <iostream>
#include <future>       // for std::async

#include "BoundedBuffer.h"

//  Every-where you find XXX you have to change something

/**
 * some specifications
 * Every-where you find XXX you have to change something
 */
namespace ttt {
    // XXX check the parameters:
    // the data type of the buffers. E.g. pointer on image frame. Here an integer.
    typedef int dataType;
    // capacity of all two buffers.
    static const int capacity = 100;
    // here only for simulation of input elements. This is the count.
    static const int inputMax = 10000;
    // delay in the three threads.
    static const std::chrono::milliseconds delay {8};
    // used to indicate the end of inputThread.
    static std::atomic_bool isFinished;
}


using namespace std;

// input thread
template<typename T>
void threadInputFnc(BoundedBuffer<t> &buffer, std::chrono::milliseconds delayMs)
{
    for (T i=0; i < ttt::inputMax; ++i)
    {
        // XXX here you have to grab the frame and move it to the buffer.
        buffer << std::move(i);
        std::this_thread::sleep_for(delayMs);
    }
    ttt::isFinished.store(true);
}

// working thread
template<typename T>
void threadWorkingFnc(BoundedBuffer<t> &outBuffer, BoundedBuffer<t> &inBuffer, std::chrono::milliseconds delayMs)
{
    T x;
    while (!ttt::isFinished.load())
    {
        inBuffer >> x;
        std::this_thread::sleep_for(delayMs);
        // XXX here you make your calculations on the frame. Here just multiple of 2.
        outBuffer << x*2;
    }
}

// output thread
template<typename T>
void threadOutputFnc(BoundedBuffer<t> &buffer, std::chrono::milliseconds delayMs)
{
    while (!ttt::isFinished.load())
    {
        // XXX the output can be designed in other ways.
        // you can clean the frame pointer here. Also release the memory, if you allocated above.
        std::cout << '\r' << buffer;
        std::cout.flush();
        std::this_thread::sleep_for(delayMs);
    }
}


int main()
{
    cout << "starting ..." << endl;

    BoundedBuffer<ttt::dataType> inputToWrokingBuffer(ttt::capacity);
    BoundedBuffer<ttt::dataType> workingToOutputBuffer(ttt::capacity);

    auto inputThread = std::async(std::launch::async, threadInputFnc<ttt::dataType>
                                  , std::ref(inputToWrokingBuffer)      // ref, because & ref in thread fnc.
                                  , ttt::delay);

    auto workingThread = std::async(std::launch::async, threadWorkingFnc<ttt::dataType>
                                    , std::ref(workingToOutputBuffer)
                                    , std::ref(inputToWrokingBuffer)
                                    , ttt::delay);

    auto outputThread = std::async(std::launch::async, threadOutputFnc<ttt::dataType>
                                   , std::ref(workingToOutputBuffer)
                                   , ttt::delay);

    outputThread.get();
    workingThread.get();
    inputThread.get();

    cout << "Finished." << endl;

    return 0;
}

There are three threads in the main declared: inputThread, workingThread and outputThread. With get(), we wait of finishing the threads in reverse order of creation. std::ref(buffer) is needed on calling threads, because the buffer should be changed outside of the threads.

If all works correctly, we get the output of only even numbers from 0 to ttt::inputMax.

Timing Issues

There are 2^3 timing issues. This is because, every thread can be too slow or too fast. Those, timing issues are handled by some tricks. For example, if the outputThread is too fast, we can repeat the last frame.

All tricks are:

If working is too slow or input too fast, overwrite old data in inputToWorkingBuffer.
If input is too slow or working too fast, repeat return of last data in inputToWorkingBuffe.
If working is too slow or output too fast, repeat display of last data.
If output is too slow or working too fast, overwrite old data in workingToOutputBuffer.

fetch() must be changed to:

And we have to change deposit() slightly. All correct code will be in the gitHub[2].

The front size_t needs to be changed to atomic, because it is used in fetch() and deposit().

Have a lot of fun, Sebastian