标准C++中的共享递归互斥锁

Question

shared_mutex计划用于C++ 17 的课程.并且shared_timed_mutex已经在C++ 14.(谁知道为什么他们按顺序排列,但无论如何.)然后有一个recursive_mutex和一个recursive_timed_mutex自C++ 11.我需要的是一个shared_recursive_mutex.我是否错过了标准中的某些内容,还是需要再等三年才能获得标准化版本？

如果目前没有这样的设施,那么仅使用标准C++的这种功能的简单(第一优先级)和有效(第二优先级)实现是什么？

Answer 1

互斥体的递归属性与术语所有者一起操作,在shared_mutex的情况下没有明确定义:几个线程可能同时.lock_shared()调用.

假设所有者是一个调用的线程.lock()(不是.lock_shared()!),递归共享互斥锁的实现可以简单地从shared_mutex以下派生:

class shared_recursive_mutex: public shared_mutex
{
public:
    void lock(void) {
        std::thread::id this_id = std::this_thread::get_id();
        if(owner == this_id) {
            // recursive locking
            count++;
        }
        else {
            // normal locking
            shared_mutex::lock();
            owner = this_id;
            count = 1;
        }
    }
    void unlock(void) {
        if(count > 1) {
            // recursive unlocking
            count--;
        }
        else {
            // normal unlocking
            owner = std::thread::id();
            count = 0;
            shared_mutex::unlock();
        }
    }

private:
    std::atomic<std::thread::id> owner;
    int count;
};

字段.owner需要声明为原子,因为在.lock()方法中,它会在没有保护的情况下进行检查以防止并发访问.

如果要递归调用.lock_shared()方法,则需要维护所有者的映射,并且应该使用一些额外的互斥锁来保护对该映射的访问.

允许具有活动的线程.lock()调用.lock_shared()使实现更复杂.

最后,允许线程提前从锁定.lock_shared()到.lock()是没有,没有,因为它当两个线程尝试执行推进导致可能的死锁.

同样,递归 共享互斥锁的语义将非常脆弱,因此最好不要使用它.

Answer 2

如果您使用的是 Linux / POSIX 平台，那么您很幸运，因为 C++ 互斥锁是根据 POSIX 互斥锁建模的。POSIX 提供了更多功能，包括递归、进程共享等。将 POSIX 原语包装到 C++ 类中是直接的。

POSIX 线程文档的良好切入点。

Answer 3

这是围绕类型 T 的快速线程安全包装：

template<class T, class Lock>
struct lock_guarded {
  Lock l;
  T* t;
  T* operator->()&&{ return t; }
  template<class Arg>
  auto operator[](Arg&&arg)&&
  -> decltype(std::declval<T&>()[std::declval<Arg>()])
  {
    return (*t)[std::forward<Arg>(arg)];
  }
  T& operator*()&&{ return *t; }
};
constexpr struct emplace_t {} emplace {};
template<class T>
struct mutex_guarded {
  lock_guarded<T, std::unique_lock<std::mutex>>
  get_locked() {
    return {{m},&t};
  }
  lock_guarded<T const, std::unique_lock<std::mutex>>
  get_locked() const {
    return {{m},&t};
  }
  lock_guarded<T, std::unique_lock<std::mutex>>
  operator->() {
    return get_locked();
  }
  lock_guarded<T const, std::unique_lock<std::mutex>>
  operator->() const {
    return get_locked();
  }
  template<class F>
  std::result_of_t<F(T&)>
  operator->*(F&& f) {
    return std::forward<F>(f)(*get_locked());
  }
  template<class F>
  std::result_of_t<F(T const&)>
  operator->*(F&& f) const {
    return std::forward<F>(f)(*get_locked());
  }
  template<class...Args>
  mutex_guarded(emplace_t, Args&&...args):
    t(std::forward<Args>(args)...)
  {}
  mutex_guarded(mutex_guarded&& o):
    t( std::move(*o.get_locked()) )
  {}
  mutex_guarded(mutex_guarded const& o):
    t( *o.get_locked() )
  {}
  mutex_guarded() = default;
  ~mutex_guarded() = default;
  mutex_guarded& operator=(mutex_guarded&& o)
  {
    T tmp = std::move(o.get_locked());
    *get_locked() = std::move(tmp);
    return *this;
  }
  mutex_guarded& operator=(mutex_guarded const& o):
  {
    T tmp = o.get_locked();
    *get_locked() = std::move(tmp);
    return *this;
  }

private:
  std::mutex m;
  T t;
};

您可以使用：

mutex_guarded<std::vector<int>> guarded;
auto s0 = guarded->size();
auto s1 = guarded->*[](auto&&e){return e.size();};

两者的作用大致相同，并且只有当互斥锁被锁定时才能访问受保护的对象。

从 @tsyvarev 的答案中窃取（进行一些细微的更改）我们得到：

class shared_recursive_mutex
{
  std::shared_mutex m
public:
  void lock(void) {
    std::thread::id this_id = std::this_thread::get_id();
    if(owner == this_id) {
      // recursive locking
      ++count;
    } else {
      // normal locking
      m.lock();
      owner = this_id;
      count = 1;
    }
  }
  void unlock(void) {
    if(count > 1) {
      // recursive unlocking
      count--;
    } else {
      // normal unlocking
      owner = std::thread::id();
      count = 0;
      m.unlock();
    }
  }
  void lock_shared() {
    std::thread::id this_id = std::this_thread::get_id();
    if (shared_counts->count(this_id)) {
      ++(shared_count.get_locked()[this_id]);
    } else {
      m.lock_shared();
      shared_count.get_locked()[this_id] = 1;
    }
  }
  void unlock_shared() {
    std::thread::id this_id = std::this_thread::get_id();
    auto it = shared_count->find(this_id);
    if (it->second > 1) {
      --(it->second);
    } else {
      shared_count->erase(it);
      m.unlock_shared();
    }
  }
private:
  std::atomic<std::thread::id> owner;
  std::atomic<std::size_t> count;
  mutex_guarded<std::map<std::thread::id, std::size_t>> shared_counts;
};

try_lock并try_lock_shared留下作为练习。

锁定和解锁共享锁定互斥锁两次（这是安全的，因为分支实际上是关于“该线程是否控制互斥锁”，并且另一个线程无法将该答案从“否”更改为“是”，反之亦然） 。您可以使用一把锁来->*代替->，这会使其更快（以逻辑上的一些复杂性为代价）。

上面不支持先有排它锁，然后再有共享锁。这很棘手。它不支持拥有共享锁，然后升级到唯一锁，因为当两个线程尝试这样做时，这基本上不可能阻止它发生死锁。

最后一个问题可能就是为什么递归共享互斥体是一个坏主意的原因。