# Pipeline with std::thread vectors and queue

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;
};

{
t.join();
}

{
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 width = 500;
int height = 500;

int num_frames = 100;

int preset = 3;

vector<Frame> final;

//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

//stage 2
}

// 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 <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

};


## 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);


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 << " ";
}

• Thanks for your useful advices, yes I intended the encoded vector, but remains one problem, it gives me now seg fault on the cout of the encoded vector, I think that maybe the concurrent push_back is not thread safe? – CIVI89 May 9 '16 at 14:43
• Definitely not thread safe! I've updated my answer to address this. – Edward May 9 '16 at 14:53
• Perfect! You think that in general this is a good way to implement a sort of multi-threaded pipeline? This should be a model that represent encoding algorithm like x264 (A LOT simplified) that instead use thread pools for each stage. But I've tried and synchronizing different thread pools is a real nightmare, so I've decided to use queues. – CIVI89 May 9 '16 at 15:00
• Yes, this is a frequently used idiom for synchronizing multiple threads and in fact, this is very much like one of the samples in the book C++ Concurrency in Action by Anthony Williams which is a book I'd highly recommend. – Edward May 9 '16 at 15:12
• By the way, if you found the answer useful, you can upvote it as well as selecting it. – Edward May 9 '16 at 15:13