如何避免日志线程空转？多线程日志优化方案咨询

问题概述 #

程序将超长文件列表分配给4个工作线程，每个线程把日志写入独立的logPool（vector<logRec>）。日志线程负责按顺序输出日志池内容，目前采用持续自旋检测更新，但工作线程初始化/处理大文件时，日志线程会持续空转浪费CPU资源。现有顺序版本代码如下：

struct logRec{ enum class type : unsigned { info = 0, error, end } typ;  /* end : marker for pool finalized, no-mo-msg */
               string msg; };
auto static const INFO = logRec::type::info, ERROR = logRec::type::error, END = logRec::type::end;
typedef vector<struct logRec> logPool;
vector<logPool> logPools;

//Print out log pools in order
void logger(){
  
  constexpr auto prLog = [](const size_t iPool, const size_t k) -> bool /*end reached ?*/ {
    switch(const auto& lrec = logPools[iPool][k]; lrec.typ){
      case INFO  : cout <<       lrec.msg      <<"\n"  <<std::flush; return false;
      case ERROR : cerr <<"\n"<< lrec.msg      <<"\n\n"<<std::flush; return false;
      [[unlikely]] case END : return true;
    }
    return true;
  };
  
  size_t last = 0, k = 0; bool end = false;
  for(auto i=0; i<logPools.size(); i++){ // sequential mode
    last = k = 0; end = false;
    while(!end){  // keep scanning pool #i until an end marker is reached
      for(k = last; k < logPools[i].size(); k++) if((end = prLog(i,k))) break;
      last = k;
    }
  }
}

当前日志线程与工作线程无同步，工作线程无需感知日志线程存在，但空转问题亟待解决。考虑过按日志条数（如1000条）或时间间隔（如300ms）触发日志线程的方案，寻求更优解。

优化方案 #

1. 条件变量+阈值触发（低资源消耗） #

给每个logPool配套原子计数器和条件变量，工作线程积累到指定条数或写入结束标记时唤醒日志线程，彻底避免空转。

修改要点：

• 扩展日志池结构，新增同步对象：

  struct PoolWithSync {
      logPool pool;
      std::atomic<size_t> new_logs = 0;
      std::condition_variable cv;
      std::mutex mtx;
      bool finished = false;
  };
  std::vector<PoolWithSync> logPools;
  

• 工作线程写入日志时更新计数器，达到阈值（如1000）或写入END时唤醒：

  // 工作线程写入日志示例
  void worker(size_t pool_idx, const string& msg) {
      std::lock_guard<std::mutex> lock(logPools[pool_idx].mtx);
      logPools[pool_idx].pool.push_back({INFO, msg});
      if (++logPools[pool_idx].new_logs >= 1000) {
          logPools[pool_idx].new_logs = 0;
          logPools[pool_idx].cv.notify_one();
      }
  }
  // 写入END标记时必须唤醒
  void finish_pool(size_t pool_idx) {
      std::lock_guard<std::mutex> lock(logPools[pool_idx].mtx);
      logPools[pool_idx].pool.push_back({END, ""});
      logPools[pool_idx].finished = true;
      logPools[pool_idx].cv.notify_one();
  }
  

• 日志线程逻辑改为等待通知后处理：

  void logger() {
      for (size_t i = 0; i < logPools.size(); ++i) {
          size_t last = 0;
          bool end = false;
          auto& pool_sync = logPools[i];
          while (!end) {
              std::unique_lock<std::mutex> lock(pool_sync.mtx);
              // 等待新日志或结束标记
              pool_sync.cv.wait(lock, [&](){
                  return pool_sync.new_logs > 0 || pool_sync.finished;
              });
              // 处理新增日志
              for (; last < pool_sync.pool.size(); ++last) {
                  if ((end = prLog(i, last))) break;
              }
              pool_sync.new_logs = 0;
          }
      }
  }
  

2. 定时休眠+自旋（最小代码改动） #

如果不想修改工作线程逻辑，可在日志线程空转时短暂休眠，平衡CPU消耗与日志输出延迟：

void logger(){
  
  constexpr auto prLog = [](const size_t iPool, const size_t k) -> bool /*end reached ?*/ {
    switch(const auto& lrec = logPools[iPool][k]; lrec.typ){
      case INFO  : cout <<       lrec.msg      <<"\n"  <<std::flush; return false;
      case ERROR : cerr <<"\n"<< lrec.msg      <<"\n\n"<<std::flush; return false;
      [[unlikely]] case END : return true;
    }
    return true;
  };
  
  size_t last = 0, k = 0; bool end = false;
  for(auto i=0; i<logPools.size(); i++){ // sequential mode
    last = k = 0; end = false;
    while(!end){  // keep scanning pool #i until an end marker is reached
      bool has_new = false;
      for(k = last; k < logPools[i].size(); k++) {
        if((end = prLog(i,k))) break;
        has_new = true;
      }
      last = k;
      // 无新日志且未结束时，休眠300ms
      if(!has_new && !end) {
          std::this_thread::sleep_for(std::chrono::milliseconds(300));
      }
    }
  }
}

3. 切换到非顺序遍历模式（零同步成本） #

你提到非顺序版本遍历所有日志池几乎不会空转，可调整日志线程逻辑为轮询所有池的新增日志，避免单池等待时的空转：

void logger(){
  
  constexpr auto prLog = [](const size_t iPool, const size_t k) -> bool /*end reached ?*/ {
    switch(const auto& lrec = logPools[iPool][k]; lrec.typ){
      case INFO  : cout <<       lrec.msg      <<"\n"  <<std::flush; return false;
      case ERROR : cerr <<"\n"<< lrec.msg      <<"\n\n"<<std::flush; return false;
      [[unlikely]] case END : return true;
    }
    return true;
  };
  
  std::vector<size_t> last_pos(logPools.size(), 0);
  std::vector<bool> pool_finished(logPools.size(), false);
  size_t finished_count = 0;

  while(finished_count < logPools.size()){
    for(size_t i=0; i<logPools.size(); i++){
      if(pool_finished[i]) continue;
      bool end = false;
      for(size_t k = last_pos[i]; k < logPools[i].size(); k++){
        if((end = prLog(i,k))) break;
        last_pos[i] = k+1;
      }
      if(end){
        pool_finished[i] = true;
        finished_count++;
      }
    }
    // 可选：如果一轮遍历无任何新日志，短暂休眠
    bool any_new = false;
    for(size_t i=0; i<logPools.size(); i++){
      if(!pool_finished[i] && last_pos[i] < logPools[i].size()){
        any_new = true;
        break;
      }
    }
    if(!any_new){
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
  }
}

方案选择 #

• 追求最低CPU消耗：选方案1，通过同步机制精准唤醒，完全避免空转。
• 希望最小化代码改动：选方案2，仅修改日志线程逻辑，无需调整工作线程。
• 业务允许非顺序输出：选方案3，零同步成本，空转概率极低。

内容的提问来源于stack exchange，提问作者Wasfi JAOUAD