具有两个独占锁组的共享锁

Question

我有两种方法“log”和“measure”，它们永远不应该同时执行。 所以我尝试使用“std::mutex”来执行此操作，如下所示：

void log(std::string message)
{
    mtx.lock();
    someLogFunctionality();
    mtx.unlock();
}

void measure()
{        
    mtx.lock();
    someMeasureFunctionality();
    mtx.unlock();
}

现在事实证明，也可以在不锁定的情况下并行多次调用“log”，同样适用于“measure”。 （原因：someLogFunctionality() 和 someMeasureFunctionality() 相互干扰，但同一个方法可能会被并行调用多次）

当时我查看了“std::shared_mutex”，但对我来说有两个问题：

1.) 使用 shared_mutex 我可以仅将 lock_shared 用于其中一种方法（日志或测量），但另一种方法必须使用排他锁（并且不能再次并行执行多次）

void log(std::string message)
{
    mtx.lock_shared();
    someLogFunctionality();
    mtx.unlock_shared();
}

void measure()
{        
    mtx.lock(); // This should also be shared but among another "group"
    someMeasureFunctionality();
    mtx.unlock();
}

2.) 我不能使用 C++17（我正在使用的环境中的约束）

您对我如何实现这一点有什么建议吗？

Answer 1

根据 alexb 的回复，我编写了以下当前适用于我的互斥锁类（目前仅在一个简单的多线程示例应用程序中试用过）

请注意，它不受“饥饿”保护。简而言之：如果 lockLogging 被频繁调用（反之亦然），则不能确保 lockMeasure 会获得锁。

class MyMutex
{
private:
    std::atomic<int> log_executors;
    std::atomic<int> measure_executors;

    std::mutex mtx;
    std::condition_variable condition;

public:
    MyMutex() : log_executors(0), measure_executors(0) {}
    
    ~MyMutex() {}

    void lockMeasure()
    {   
        std::unique_lock<std::mutex> lock(mtx);

        while(log_executors) {
            condition.wait(lock); 
        }
        measure_executors++; 
    }
    
    void unlockMeasure()
    {   
        std::unique_lock<std::mutex> lock(mtx);

        measure_executors--; 
        if (!measure_executors)
        {
          condition.notify_all();
        }
    }
    
    void lockLogging()
    {         
        std::unique_lock<std::mutex> lock(mtx);

        while(measure_executors) {
          condition.wait(lock); 
        }
        log_executors++;
    }

    void unlockLogging()
    {         
        std::unique_lock<std::mutex> lock(mtx);

        log_executors--; 
        if (!log_executors)
        {
          condition.notify_all(); 
        }
    }

    static MyMutex& getInstance()
    {
        static MyMutex _instance;
        return _instance;
    }    
};

用法：

void measure()
{
    MyMutex::getInstance().lockMeasure();

    someMeasureFunctionality();

    MyMutex::getInstance().unlockMeasure();
}

void log()
{
    MyMutex::getInstance().lockLogging();

    someLogFunctionality();

    MyMutex::getInstance().unlockLogging();
}

Answer 2

您需要一些比 shared_mutex 更复杂的屏障逻辑（顺便说一句，shared_mutex 不是多平台编译的最佳选择）。例如，您可以使用互斥锁、条件变量和 2 个变量进行屏障同步。它不占用 CPU，您不能使用 sleeps 进行检查。

#include <mutex>
#include <condition_variable>
#include <atomic>

std::atomic<int> log_executors = 0;
std::atomic<int> measure_executors = 0;

std::mutex mutex;
std::condition_variable condition;

void log(std::string message) {
  {
    std::unique_lock<std::mutex> lock(mutex);

    log_executors++;  // Register current executor and prevent from entering new measure executors

    // Wait until all measure executors will go away
    while(measure_executors) {
      condition.wait(lock);  // wait condition variable signal. Mutex will be unlocked during wait
    }
  }

  // here lock is freed
  someLogFunctionality(); // execute logic


  {
    std::unique_lock<std::mutex> lock(mutex);
    log_executors--;  // unregister current execution
    condition.notify_all();  // send signal and unlock all waiters
  }
}

void measure()
{        
  {
    std::unique_lock<std::mutex> lock(mutex);

    measure_executors++;  // Register current executor and prevent from entering new log executors
    while(log_executors) {
      condition.wait(lock);  // wait until all measure executors will gone
    }
  }

  someMeasureFunctionality();

  {
    std::unique_lock<std::mutex> lock(mutex);
    measure_executors--;  // unregister current execution
    condition.notify_all(); // send signal and unlock all waiters
  }
}

Answer 3

您可以拥有一个主锁来授予对信号量变量的访问权限：

void log(std::string message)
{
    acquire(LOG);
    someLogFunctionality();
    release(LOG);
}

void measure()
{        
    acquire(MEASURE);
    someMeasureFunctionality();
    release(MEASURE);
}

void acquire(int what) {
    for (;;) {
        mtx.lock();
        if (owner == NONE) {
            owner = what;
        }
        if (owner == what) {
            // A LOG was asked while LOG is running
            users[owner]++;
            mtx.unlock();
            return;
        }
        mtx.unlock();
        // Some sleep would be good
        usleep(5000);
    }
}

void release(int what) {
    mtx.lock();
    if (owner != what) {
        // This is an error. How could this happen?
    }
    if (users[what] <= 0) {
        // This is an error. How could this happen?
    }
    users[what]--;
    if (0 == users[what]) {
        owner = NONE;
    }
    mtx.unlock();
}

在这种情况下，例如：

owner is NONE
LOG1 acquires LOG. It can do so because owner is NONE
MEASURE1 acquires LOG. It starts spinning in place because owner != MEASURE
MEASURE2 acquires LOG. It starts spinning in place because owner != MEASURE
LOG2 acquires LOG. It can do so because owner is LOG, users[LOG]=2
LOG2 releases LOG. users[LOG]=1
LOG1 releases LOG. users[LOG]=0, so owner becomes NONE
MEASURE2 by pure chance acquires mtx before MEASURE1, finds owner=NONE and goes
MEASURE1 finds owner=MEASURE and sets users[MEASURE]=2

在上面，请注意对 measure() 的第二次调用实际上执行得更早一些。这应该没问题。但是，如果您想保持调用“序列化”，即使它们是并行发生的，您将需要为每个所有者提供一个堆栈和更复杂的代码。