Function ThreadRun

serving/processor/storage/feature_store_mgr.cc:25–95

namespace {
void ThreadRun(AsyncFeatureStoreMgr* mgr, int idx,
               bool is_update_thread) {
  std::mutex* mu = nullptr;
  std::condition_variable* cv = nullptr;
  sparse_task_queue* queue = nullptr;

  if (is_update_thread) {
    mu = mgr->GetUpdateMutex(idx);
    cv = mgr->GetUpdateCV(idx);
    queue = mgr->GetUpdateSparseTaskQueue(idx);
  } else {
    mu = mgr->GetMutex(idx);
    cv = mgr->GetCV(idx);
    queue = mgr->GetSparseTaskQueue(idx);
  }

  const int try_count = 64;
  int curr_try_count = 0;
  SparseTask* task = nullptr;
  bool succeeded = false;

  while ((succeeded = queue->try_dequeue(task)) ||
         !mgr->ShouldStop()) {
    if (!succeeded) {
      ++curr_try_count;
      if (curr_try_count <= try_count) {
        continue;
      }
      curr_try_count = 0;

      if (is_update_thread) {
        // Ready going to sleep
        *(mgr->GetUpdateSleepingFlag(idx)) = true;
        *(mgr->GetUpdateReadyFlag(idx)) = false;
      } else {
        // Ready going to sleep
        *(mgr->GetSleepingFlag(idx)) = true;
        *(mgr->GetReadyFlag(idx)) = false;
      }

      {
        // try to wait signal when have no elements in the queue
        std::unique_lock<std::mutex> lock(*mu);
        cv->wait(lock, [is_update_thread, mgr, idx] {
          return (is_update_thread ?
                      *(mgr->GetUpdateReadyFlag(idx)) :
                      *(mgr->GetReadyFlag(idx))) ||
                 mgr->ShouldStop();
        });
        lock.unlock();
      }

      if (is_update_thread) {
        *(mgr->GetUpdateSleepingFlag(idx)) = false;
      } else {
        *(mgr->GetSleepingFlag(idx)) = false;
      }