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