Relaying stdin data from one thread to another

Question

I'd like to know if this piece of code is thread-safe. I'm learning something about threads, queues and synchronization, and I'd like to make sure this is correct, before moving forward to next step.

The code is very simple. It just runs two threads (producer and consumer). Producer reads data from stdin (until "0" is entered) and pushes into a queue. Consumer, just prints queue contents each 10 secs.

The way I tried to make it thread-safe:

• using an atomic boolean variable, running, in the consumer loop
• push to and pop from queue using mutex variable, m_queueMutex.

I cannot use c++11 standard. That's why I used boost library.

The code compiles and works as expected using GCC. To compile:
g++ -g -o queues queues.cpp -lboost_thread -lboost_system -lboost_chrono

Any other advice will be very welcome.

Code:

#include <iostream>
#include <queue>
#include <string>
#include <boost/thread.hpp>
#include <boost/thread/scoped_thread.hpp>
#include <boost/chrono.hpp>
#include <boost/atomic.hpp>

std::queue<std::string> myqueue;
boost::atomic<bool> running; 
boost::mutex m_queueMutex;

void push(std::string& val) { 

    boost::mutex::scoped_lock lock(m_queueMutex);
    myqueue.push(val); 
}

std::string pop() { 
    std::string val = ""; 

    boost::mutex::scoped_lock lock(m_queueMutex);

    if (!myqueue.empty()){  
        val = myqueue.front(); 
        myqueue.pop();
    }

    return val; 
}


void wait(int seconds) {
  boost::this_thread::sleep_for(boost::chrono::seconds(seconds));
}

void consume () { 
    while (running || !myqueue.empty()) { 
        if (!myqueue.empty()) { 
            std::string dt = pop();

            wait(10); 
            std::cout << "[CONS] -- Got: " << dt << std::endl; 
        }
        wait (1); 
    }

    std::cout << "[CONS] -- Quiting.. " << std::endl; 
}

void produce() {
    std::string data; 

    do { 
        std::cout << "# "; 
        std::cin >> data; 
        if (data != "0") {
            push(data); 
            std::cout << "[PROD] -- Add: " << data << std::endl; 
        }
    } while (data != "0"); 

    std::cout << "[PROD] -- Quiting.. " << std::endl; 

    running = false; 
}

int main() {
    std::cout << "Started... " << std::endl;

    running = true; 

    boost::thread producer = boost::thread(produce);
    boost::thread consumer = boost::thread(consume); 
    //boost::scoped_thread<> producer((boost::thread(produce)));
    //boost::scoped_thread<> consumer((boost::thread(consume)));

    producer.join(); 
    consumer.join(); 

    std::cout << "Ended... " << std::endl;
}

Answer 1

Right away I can see a potential problem. It appears that it is possible for the consumer thread to access "myqueue" outside of any lock.

void consume () { 
while (running || !myqueue.empty()) { 

This will probably still function just fine, but it is only as thread-safe as the queue's implementation for 'empty'. Unless the queue itself can guarantee that such an operation is safe, then the code is not thread safe. It is up to the specifics of the queues implementation, and the optimizations done by the compiler / processor, weather or not this would work as expected.

One other minor thing I notice is that the queue is being checked for empty twice. Once in the consumer before the global function pop() is called, and again in the pop function itself. This is a bit confusing to read, and should not be necessary. It may be better to lock the queue first, and then process items later, like so.

void consume()
{
bool hasWork = true;
while (running && hasWork)
{
    std::string item = "";
    mutex.lock();
    if (!myqueue.empty())
    {
        item = myqueue.pop();
    }
    else
    {
        hasWork = false;
    }
    mutex.unlock();

    // Do something with "item"
}
}

The idea here is that you are locking the queue only when you are performing operations on it, and leaving it unlocked while actually doing work on whatever was pulled form the queue. This is especially important when the individual work items may take significant time.

Other than that, it looks like you've got the right idea about needing to protect access to the queue's push and pop methods, it's just that you've got to protect all the other methods that touch 'myqueue' also.

Again, this would probably still work just fine on most implementations, but it's better to know you're safe.

Answer 2

My main problem with this is that myqueue is exposed and easily accessible. This makes it very easy for somebody else to write some other code that uses myqueue without knowing that is potentially being used by two other threads.

You should hide the actual queue inside a class a control all access to the queue object to make sure that it is accessed only in a thread safe manner via member methods.

I would use conditional variables to make sure you don't lock up a processor basically doing a busy wait.

Also C++ has its own thread system (since C++11 now 5 years ago); there is no need to use the boost version anymore this has been subsumed into the standard.