Message passing between threads using a command file

Question

This project asked for 4 threads that has a command file with instructions such as SEND, Receive and quit. When the file says "2 send" the thread that in the second place in the array should wake up and receive its message. I need to know how to make a thread read it's message if the command file has a message for it?

Answer 1

The biggest issue I see for your design is the fact that each thread reads its line randomly independent from any other thread. After this it would have to check wether the current line is actually meant for it i.e. starting with the appropriate number. What happens if not ? Too complicated.

I would split this issue up into one reader thread and a set of worker threads. The first reads lines from a file and dispatches it to the workers by pushing it into the current workers queue. All synchronized with a per worker mutex and conditional variable The following is implemented in C++11 but should as well be doable in pthread_* style.

#include <thread> 
#include <iostream> 
#include <queue> 
#include <mutex> 
#include <fstream> 
#include <list> 
#include <sstream> 
#include <condition_variable> 

class worker  { 
public: 
    void operator()(int n) { 
        while(true) { 
            std::unique_lock<std::mutex> l(_m); 
            _c.wait(l); 
            if(!_q.empty()) { 
                { 
                    std::unique_lock<std::mutex> l(_mm); 
                    std::cerr << "#" << n << " " << _q.back() <<std::endl; 
                } 
                _q.pop(); 
            } 
        } 
    } 
private: 
    std::mutex              _m; 
    std::condition_variable _c; 
    std::queue<std::string> _q; 
    //  Only needed to synchronize I/O 
    static std::mutex       _mm; 
    // Reader may write into our queue 
    friend class reader; 
}; 

std::mutex       worker::_mm; 

class reader  { 
public: 
    reader(worker & w0,worker & w1,worker & w2,worker & w3) { 
        _v.push_back(&w0); 
        _v.push_back(&w1); 
        _v.push_back(&w2); 
        _v.push_back(&w3); 
    } 
    void operator()() { 
        std::ifstream fi("commands.txt"); 
        std::string s; 

        while(std::getline(fi,s)) { 
            std::stringstream ss(s); 
            int n; 
            if((ss >> n >> std::ws) && n>=0 && n<_v.size()) { 
                std::string s0; 
                if(std::getline(ss,s0)) { 
                    std::unique_lock<std::mutex> l(_v[n]->_m); 
                    _v[n]->_q.push(s0); 
                    _v[n]->_c.notify_one(); 
                } 
            } 
        } 

        std::cerr << "done" << std::endl; 
    } 
private: 
    std::vector<worker *> _v; 

}; 

int main(int c,char **argv) { 

    worker w0; 
    worker w1; 
    worker w2; 
    worker w3; 

    std::thread tw0([&w0]() { w0(0); }); 
    std::thread tw1([&w1]() { w1(1); }); 
    std::thread tw2([&w2]() { w2(2); }); 
    std::thread tw3([&w3]() { w3(3); }); 

    reader r(w0,w1,w2,w3); 

    std::thread tr([&r]() { r(); }); 

    tr.join(); 
    tw0.join(); 
    tw1.join(); 
    tw2.join(); 
    tw3.join(); 
} 

The example code only reads from "commands.txt" until EOF. I assume you'd like to read continuously like the "tail -f" command. That's however not doable with std::istream.

The code of course is clumsy but I guess it gives you an idea. One should for example add a blocking mechanism if the workers are way too slow processing their stuff and the queues may eat up all the precious RAM.