Pipeline with std::thread vectors and queue

Question

here is my code, it works but after few iterations it slows down and stop without any error, I'm looking for a more efficient solution.

 #include <iostream>
 #include <vector>
 #include <algorithm>

 #include "SafeQueue.h"

 using namespace std;

 const int DATA_MAG = 256;

 struct Frame
 {
    int num;

    char type;

    bool encoded;

    vector<vector<int>> grid;
};

void do_join(thread& t)
{
    t.join();
}

void join_all(vector<thread>& v)
{
    for_each(v.begin(),v.end(),do_join);
}

void create_input (Queue<Frame>& stream, int num_frames, int height, int width)
{
    for (int i = 0; i < num_frames; i++)
    {
        vector<vector<int>>tmp_grid(height, vector<int>(width, 0));

        Frame frame;

        for (int j = 0; j < height; j++)
        {
            for (int k = 0; k < width; k++)
            {
                tmp_grid[j][k] = rand()%DATA_MAG;
            }
        }

        frame.grid = tmp_grid;
        frame.num = i;

        stream.push(frame);
    }
}


void decide_type(int preset, Queue<Frame>& stream, Queue<Frame>& typed, vector<char>& param, int num_frames)
{
    cout<<"hello from decide"<<" "<<endl;

    for(int i = 0; i < num_frames; i++)
    {
        Frame tmp = stream.pop();

        int j = rand() % 10;

        if(j < preset)
        {
            tmp.type = 'I';
        }

        else
        {
            tmp.type = 'B';
        }

        param[tmp.num] = tmp.type;

        typed.push(tmp);
    }
}

void decode_flow(int preset, Queue<Frame>& typed, vector<Frame>& encoded,
                    vector<char>& parameters, int num_frames, int height, int width)
{
    cout<<"hello from decode"<<" "<<endl;

    for(int i = 0; i < num_frames; i++)
    {
        Frame f = typed.pop();

        if (f.type == 'I')
        {
            cout<<"hi from I"<<" "<<endl;
            for (int j = 0; j < height; j++)
            {
                for (int k = 0; k < width; k++)
                {
                    f.grid[j][k] = f.grid[j][k] * 2;
                }
            }
        }

        else cout<<"hi from B"<<" "<<endl;

        encoded.push_back(f);
    }
}




int main()
{
    srand(time(NULL));

    int num_threadsXstage = 2;

    int width = 500;
    int height = 500;

    int num_frames = 100;

    int frames_thread = num_frames/num_threadsXstage;

    int preset = 3;

    vector<Frame> final;

    //Vectors of threads
    vector<thread> typer;
    vector<thread> encoder;

    //Vector of parameters
    vector<char> parameters(num_frames);

    //Working queues
    Queue<Frame> created;
    Queue<Frame> typed;

    //Final vector
    vector<Frame> encoded(num_frames);

    //Movie creation

    create_input(created, num_frames, height, width);



for (int i = 0; i < num_threadsXstage; i++)
    {
        //stage 1
        typer.push_back(thread(bind(&decide_type, preset, ref(created),
                                    ref(typed), ref(parameters), frames_thread)));

        //stage 2
        encoder.push_back(thread(bind(&decode_flow, preset, ref(typed), ref(encoded),
                                      ref(parameters), frames_thread, height, width)));
    }


    // JOIN

    join_all(typer);

    join_all(encoder);


    for (int i = 0; i < num_frames; i++)
    {
        Frame k = typed.pop();

        cout<<k.type<<" ";
    }

    cout<<endl<<endl;

    for (int i = 0; i < num_frames; i++)
    {
        cout<<parameters[i]<<" ";
    }
}

And this is the code of my thread safe queue:

#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>

using namespace std;

template <typename T>

class Queue
{
private:
    queue<T> queue_;
    mutex mutex_;
    condition_variable cond_;

public:

    T pop()
    {
        unique_lock<std::mutex> mlock(mutex_);
        while (queue_.empty())
        {
            cond_.wait(mlock);
        }

        auto val = queue_.front();
        queue_.pop();
        return val;
    }

    void pop(T& item)
    {
        unique_lock<std::mutex> mlock(mutex_);

        while (queue_.empty())
        {
            cond_.wait(mlock);
        }
        item = queue_.front();
        queue_.pop();
    }

    void push(const T& item)
    {
        unique_lock<std::mutex> mlock(mutex_);
        queue_.push(item);
        mlock.unlock();
        cond_.notify_one();
    }
    Queue()=default;
    Queue(const Queue&) = delete;            // disable copying
    Queue& operator=(const Queue&) = delete; // disable assignment

};

Answer 1

I see a number of things that may help you improve your code.

Don't abuse using namespace std

Putting using namespace std at the top of every program is a bad habit that you'd do well to avoid. It's absolutely a mistake to put it into a header file.

Use whitespace to improve readability

Some blank lines are useful for visually separating functions, for example, but I find that this code has so many blank lines that it actually impedes understanding rather than enhancing it. Oddly, by contrast, individual lines could benefit from more whitespace. Instead of this:

for_each(v.begin(),v.end(),do_join);

I'd write this:

for_each(v.begin(), v.end(), do_join);

Fix your formatting

There are inconsistent spaces at the beginning of lines, inconsistent indentation and inconsistent use of whitespace as mentioned above. Being consistent helps others read and understand your code.

Use locks efficiently

There is an easy simplification that can be made to both versions of pop. We have this code:

while (queue_.empty())
{
    cond_.wait(mlock);
}

Better would be to use the form of wait that's specifically designed for this kind of situation:

cond_.wait(mlock, [this]{return !queue_.empty();});

Eliminate spurious std::bind calls

In both cases in this code, std::bind can simply be eliminated because std::thread's constructor can just as easily take all arguments directly.

Avoid data races

In decide_type and decide_flow, std::cout is used without locks. This is a problem since both threads are attempting to use the same resource, potentially at the same time. You probably aren't really concerned with interleaved output here, but I thought I'd point it out so that others reading this will be aware.

Also the use of encoded.push_back() within decode_flow is definitely a problem without a mutex. I'd rewrite it like this:

{
    std::lock_guard<std::mutex> lock(enc_mutex);
    encoded.push_back(f);
}

Naturally, there will need to be a declaration of enc_mutex. I put it at file scope like this:

static std::mutex enc_mutex;

Be careful with empty queues

In this implementation an attempt to pop from an empty queue will cause an infinite wait unless some other thread pushes something into the queue. At the end of main, I suspect you intended to print out encoded rather than typed. If so replace this:

for (int i = 0; i < num_frames; i++) {
    Frame k = typed.back();
    std::cout << k.type << " ";
}

with this:

for (const auto &k : encoded) {
    std::cout << k.type << " ";
}