mardi 24 août 2021

Issue with condition variables in C++

We have implemented TaskRunner whose functions will be called by different threads to start, stop and post tasks. TaskRunner will internally create a thread and if the queue is not empty, it will pop the task from queue and executes it. Start() will check if the thread is running. If not creates a new thread. Stop() will join the thread. The code is as below.

bool TaskRunnerImpl::PostTask(Task* task) {
  tasks_queue_.push_back(task);
  return true;
}

void TaskRunnerImpl::Start() {
  std::lock_guard<std::mutex> lock(is_running_mutex_);
  if(is_running_) {
    return;
  }
  is_running_ = true;

  runner_thread_ = std::thread(&TaskRunnerImpl::Run, this);
}

void TaskRunnerImpl::Run() {
  while(is_running_) {
    if(tasks_queue_.empty()) {
      continue;
    }
    Task* task_to_run = tasks_queue_.front();
    task_to_run->Run();
    tasks_queue_.pop_front();

    delete task_to_run;
  }
}

void TaskRunnerImpl::Stop() {
  std::lock_guard<std::mutex> lock(is_running_mutex_);
  is_running_ = false;
  if(runner_thread_.joinable()) {
    runner_thread_.join();
  }
}

This code is working fine. Continuously tasks are getting pushed and thread is executing those tasks. We want to use conditional variables now otherwise the thread will be continuously checking whether the task queue is empty or not. We implemented as below.

  • Thread function (Run()) will wait on condition variable.
  • PostTask() will signal if some one posts a task.
  • Stop() will signal if some one calls stop.

Implemented code as below.

bool TaskRunnerImpl::PostTask(Task* task, uint64_t delay_milliseconds) {
    std::lock_guard<std::mutex> taskGuard(m_task_mutex);
    tasks_queue_.push_back(task);
    m_task_cond_var.notify_one();
    INFO("{} : {} : {}", __FUNCTION__, delay_milliseconds, tasks_queue_.size());
    return true;
}

void TaskRunnerImpl::Start() {
    INFO("{}", __FUNCTION__);
    std::lock_guard<std::mutex> taskGuard(m_task_mutex);

    if(!is_running_) {
        is_running_ = true;
        runner_thread_ = std::thread(&TaskRunnerImpl::Run, this);
    }
}

void TaskRunnerImpl::Run() {
    while(true) {
        INFO("{} : {}", __FUNCTION__, 1);

        {
            std::unique_lock<std::mutex> mlock(m_task_mutex);
            INFO("{} : Locked Mutex", __FUNCTION__);
            m_task_cond_var.wait(mlock, [this]() {
                INFO("{} : Checking Condition", __FUNCTION__);
                return !(is_running_ && tasks_queue_.empty());
            });


                INFO("{} : Came out of wait", __FUNCTION__);
            if(!is_running_) {
                return;
            }

           INFO("{} : Escaped if cond", __FUNCTION__);
            if(!tasks_queue_.empty()) {
                INFO("{} : {} : {}", __FUNCTION__, 2, tasks_queue_.size());    // NO LOGS AFTER THIS GETTING PRINTED
                Task* task_to_run = tasks_queue_.front();
                task_to_run->Run();
                INFO("{} : Deleting Task", __FUNCTION__);
                tasks_queue_.pop_front();
                INFO("{} : After Deletion : {}", __FUNCTION__, tasks_queue_.size());
                delete task_to_run;
            }
        INFO("{} : Out of scope", __FUNCTION__);
        }

        INFO("{} : End of iteration", __FUNCTION__);
    }

    INFO("{} : returning", __FUNCTION__);
}

void TaskRunnerImpl::Stop() {
    {
        std::lock_guard<std::mutex> taskGuard(m_task_mutex);
        is_running_ = false;
        INFO("{} : Signalling STOP", __FUNCTION__);
        m_task_cond_var.notify_one();
    }

    INFO("{} : {}", __FUNCTION__, 1);

    if(runner_thread_.joinable()) {
        runner_thread_.join();
    }
}

Not sure what is wrong with the code. I am getting following output.

TaskRunnerImpl.cpp:34:INFO: Start
TaskRunnerImpl.cpp:45:INFO: Run : 1
TaskRunnerImpl.cpp:49:INFO: Run : Locked Mutex
TaskRunnerImpl.cpp:51:INFO: operator() : Checking Condition
TaskRunnerImpl.cpp:29:INFO: PostTask : 0 : 1
TaskRunnerImpl.cpp:29:INFO: PostTask : 0 : 2
TaskRunnerImpl.cpp:51:INFO: operator() : Checking Condition
TaskRunnerImpl.cpp:56:INFO: Run : Came out of wait
TaskRunnerImpl.cpp:61:INFO: Run : Escaped if cond
TaskRunnerImpl.cpp:63:INFO: Run : 2 : 2

That means the log is getting printed before executing the task and after that no logs. Usually PostTask() will be called continuously to post the tasks to queue. But with the new code no logs after task run. So I am assuming the thread function is holding the mutex and PostTask() is not able to push the tasks to queue. But unable to understand why there are no logs after executing the task. If I revert back to original code, the code is working as expected. Can anyone please let me know if there is any issue with the code.

Aucun commentaire:

Enregistrer un commentaire