基于boost::interprocess的生产者消费者模型：消费者CPU占满问题排查

场景概述 #

两个二进制程序本应轮流递增共享内存中的数值，但实际运行时消费者进程CPU占用100%，完全阻塞生产者进程；仅在消费者中添加短暂延迟（如usleep(1)），就能实现预期的交替运行效果。

相关代码 #

共享结构体定义（my_struct.h） #

#pragma once

#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/containers/vector.hpp>
#include <boost/interprocess/allocators/allocator.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/interprocess/sync/interprocess_condition.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

namespace bip = boost::interprocess;

namespace my_namespace {

static const char *name = "MySharedMemory";

struct MyStruct {
  bip::interprocess_mutex mutex;
  bip::interprocess_condition cond;
  unsigned long counter;

  MyStruct(): mutex(), cond(), counter(0) {
  }
};

}  // namespace my_namespace

生产者进程代码 #

#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>

#include <iostream>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/allocators/allocator.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/thread/locks.hpp>

#include "my_struct.h"

bool exit_flag = false;

void my_handler(int) {
  exit_flag = true;
}

namespace bip = boost::interprocess;

int main() {
  struct sigaction sigIntHandler;
  sigIntHandler.sa_handler = my_handler;
  sigemptyset(&sigIntHandler.sa_mask);
  sigIntHandler.sa_flags = 0;
  sigaction(SIGINT, &sigIntHandler, NULL);

  bip::shared_memory_object::remove(my_namespace::name);
  auto memory = bip::managed_shared_memory(bip::create_only, my_namespace::name, 65536);
  auto *data = memory.construct<my_namespace::MyStruct>(my_namespace::name)();

  long unsigned iterations = 0;
  while (!exit_flag) {
    boost::interprocess::scoped_lock lock(data->mutex);
    data->counter++;
    std::cout << "iteration:" << iterations << "Counter: " << data->counter << std::endl;
    ++iterations;
    auto start = boost::posix_time::microsec_clock::universal_time();
    auto wait_time = start + boost::posix_time::milliseconds(1000);
    auto ret = data->cond.timed_wait(lock, wait_time);
    if (!ret) {
      std::cout << "Timeout" << std::endl;
    }
  }
  return 0;
}

消费者进程代码 #

#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <sched.h>
#include <chrono>

#include <iostream>
#include <thread>
#include <mutex>

#include "my_struct.h"

bool exit_flag = false;

void my_handler(int) {
  exit_flag = true;
}

namespace bip = boost::interprocess;

int fib(int x) {
  if ((x == 1) || (x == 0)) {
    return (x);
  } else {
    return (fib(x - 1) + fib(x - 2));
  }
}

int main() {
  struct sigaction sigIntHandler;
  sigIntHandler.sa_handler = my_handler;
  sigemptyset(&sigIntHandler.sa_mask);
  sigIntHandler.sa_flags = 0;
  sigaction(SIGINT, &sigIntHandler, nullptr);

  auto memory = bip::managed_shared_memory(bip::open_only, my_namespace::name);
  auto *data = memory.find<my_namespace::MyStruct>(my_namespace::name).first;

  long unsigned iterations = 0;
  while (!exit_flag) {
    {
      boost::interprocess::scoped_lock lock(data->mutex);
      std::this_thread::sleep_for(std::chrono::milliseconds(200));
      data->counter += 1;
      std::cout << "iteration:" << iterations << "Counter: " << data->counter << std::endl;
      ++iterations;
      std::cout << "notify_one" << std::endl;
      data->cond.notify_one();
    }
//    usleep(1);  // If I add this it works
  }
  return 0;
}

问题原因分析 #

1. 消费者的*忙等（Busy Waiting）*行为 #

消费者主循环在释放互斥锁后，没有任何阻塞或延迟逻辑，会立刻发起下一次锁竞争。现代CPU调度器为了减少上下文切换开销，会优先调度正在运行的进程，导致消费者持续抢占CPU和互斥锁，生产者完全没有获取锁的机会。

2. 条件变量的使用逻辑缺陷 #

生产者调用cond.timed_wait()时会释放锁并进入等待状态，等待消费者的notify_one()唤醒。但消费者在调用notify_one()后立刻重新抢锁，生产者刚被唤醒就会在锁竞争中失败，继续被阻塞。添加usleep(1)后，消费者主动放弃CPU时间片，调度器才有机会切换到生产者进程，让生产者获取锁完成自身逻辑。

3. 互斥锁的非公平性 #

默认的interprocess_mutex是抢占式的，没有公平调度机制。当消费者释放锁后立刻再次请求锁，调度器会倾向于让同一个进程继续持有锁，进一步加剧了生产者的“饿死”问题，导致完全无法执行。

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