CODEWITHSUNDEEP
c++multithreading
user2840470
user2840470

Reputation: 969

Waking up a thread waiting on a condition in infinite loop

I have a pretty basic producer / consumer implementation. The producer is the "main" thread, and the consumer is executed on a separate thread. However the consumer needs to be explicitly started, using a Start() function. This sets the "processing" flag to true (used in the infinite while loop).

Once in the while loop, the consumer then uses a condition variable to see if there is data in the queue to process. If yes, it does its work, goes back to the top of the infinite loop, then the condition variable, and so on.

The problem I am having is the consumer is waiting for data in the queue, and I want to stop processing. How can I wake up the consumer? I have provided some example code below, removing some major components, just showing the high level design (everything is not actually public).

// Consumer object
class Consumer {
public: 
  std::mutex mtx_;
  bool processing_ = false;
  std::thread processing_thread_;
  std::queue<int> data_;
  std::condition_variable cv_;
  ~Consumer() {
    // Make sure the processing thread is stopped
    {
      std::lock_guard<std::mutex> lock(mtx_);
      processing_ = false;
    }

    if (processing_thread_.joinable()) {

      processing_thread_.join();
    }
  }
  void Start() {
    std::lock_guard<std::mutex> lock(mtx_);

    processing_ = true;
    processing_thread_ = std::thread(
      &Consumer::Run,
       this);
  }
  void Stop() {
    std::lock_guard<std::mutex> lock(mtx_);

    processing_ = false;
  } 
  void AddData(int d) {
    std::lock_guard<std::mutex> lock(mtx_);
    data_.push(d);
    cv_.notify_one();
  }
  bool IsDataAvailable() const {
    return (!data.empty());
  }
  void Run() {

    // The infinite loop
    while (processing_) {
      // This is where I get stuck waiting even tho processing has been 
      // changed to false by the main thread
      std::unique_lock<std::mutex> lock(mtx_);
      cv_.wait(lock, std::bind(
        &Consumer::IsDataAvailable, this));
      // do some processing
    }
  }

}; // end of consumer

// Somewhere in main trying to stop the processing thread cause I am
// done processing OR my_consumer goes out of scope and tries to join
// ... 
  my_consumer.Stop();
}
// my_consumer goes out of scope here calling destructor.

Upvotes: 1

Views: 1536

Answers (1)

Maxim Egorushkin
Maxim Egorushkin

Reputation: 136276

A couple of changes is required for the consumer to wait for change in processing_:

~Consumer() {
  if (processing_thread_.joinable()) {
    Stop();
    processing_thread_.join();
  }
}
// ...
void Stop() {
  std::lock_guard<std::mutex> lock(mtx_);
  processing_ = false;
  cv_.notify_one();
} 
// ...

void Run() {
  for(;;) {
    std::unique_lock<std::mutex> lock(mtx_);
    // Wait till something is put into the queue or stop requested.
    cv_.wait(lock, [this]() { return !processing_ || !data_.empty(); });
    if(!data_.empty())
      // Process queue elements.
    else if(!processing_)
      return; // Only exit when the queue is empty.
  }
}

Upvotes: 1

Related Questions