【发布时间】：2019-05-27 00:54:03
【问题描述】：
我正在实现一个线程安全的“延迟同步”集作为由 shared_ptr 连接的节点的链接列表。该算法来自“多处理器编程的艺术”。我正在添加一个需要与现有函数线性化的is_empty() 函数：contains(), add(), remove()。在下面的代码中，您可以看到remove 是一个两步过程。首先它通过设置marked = nullptr“惰性”标记节点，然后物理移动链表next指针。
修改类以支持 is_empty()
template <class T>
class LazySet : public Set<T> {
public:
LazySet ();
bool contains (const T&) const;
bool is_empty () const;
bool add (const T&);
bool remove (const T&);
private:
bool validate(const std::shared_ptr<Node>&, const std::shared_ptr<Node>&);
class Node;
std::shared_ptr<Node> head;
std::shared_ptr<bool> counter; //note: type is unimportant, will never change true/fase
};
template <class T>
class LazySet<T>::Node {
public:
Node ();
Node (const T&);
T key;
std::shared_ptr<bool> marked; //assume initialized to = LazySet.counter
// nullptr means it's marked; otherwise unmarked
std::shared_ptr<Node> next;
std::mutex mtx;
};
支持is_empty的相关修改方法
template <class T>
bool LazySet<T>::remove(const T& k) {
std::shared_ptr<Node> pred;
std::shared_ptr<Node> curr;
while (true) {
pred = head;
curr = atomic_load(&(head->next));
//Find window where key should be in sorted list
while ((curr) && (curr->key < k)) {
pred = atomic_load(&curr);
curr = atomic_load(&(curr->next));
}
//Aquire locks on the window, left to right locking prevents deadlock
(pred->mtx).lock();
if (curr) { //only lock if not nullptr
(curr->mtx).lock();
}
//Ensure window didn't change before locking, and then remove
if (validate(pred, curr)) {
if (!curr) { //key doesn't exist, do nothing
//## unimportant ##
} else { //key exists, remove it
atomic_store(&(curr->marked), nullptr); //logical "lazy" remove
atomic_store(&(pred->next), curr->next) //physically remove
(curr->mtx).unlock();
(pred->mtx).unlock();
return true;
}
} else {
//## unlock and loop again ##
}
}
}
template <class T>
bool LazySet<T>::contains(const T& k) const {
std::shared_ptr<Node> curr;
curr = atomic_load(&(head->next));
//Find window where key should be in sorted list
while ((curr) && (curr->key < k)) {
curr = atomic_load(&(curr->next));
}
//Check if key exists in window
if (curr) {
if (curr->key == k) { //key exists, unless marked
return (atomic_load(&(curr->marked)) != nullptr);
} else { //doesn't exist
return false;
}
} else { //doesn't exist
return false;
}
}
Node.marked 最初是一个普通的布尔值，LazySet.counter 不存在。使它们成为 shared_ptrs 的选择是能够原子地修改节点数量的计数器和节点上的惰性删除标记。在remove() 中同时修改两者是is_empty() 与contains() 线性化的必要条件。 （如果没有双倍宽的 CAS 或其他东西，它就不能是单独的 bool 标记和 int 计数器。）我希望使用 shared_ptr 的use_count() 函数来实现计数器，但在多线程上下文中，由于relaxed_memory_order，它只是一个近似值。
我知道独立围栏通常是不好的做法，而且我对使用它们不太熟悉。 但是，如果我像下面那样实现is_empty，围栏是否会确保它不再是近似值，而是可靠计数器的精确值？
template <class T>
bool LazySet<T>::is_empty() const {
// ## SOME FULL MEMORY BARRIER
if (counter.use_count() == 1) {
// ## SOME FULL MEMORY BARRIER
return true
}
// ## SOME FULL MEMORY BARRIER
return false
}
我只问因为LWG Issue 2776 说：
如果不增加更多的围栏，我们就无法使
use_count()可靠。
【问题讨论】：
-
对于我们这些没有this book 的人，你能扩展一下“懒惰同步”的含义吗？正如您可能知道的那样，几乎不可能让集合上的计数器方法在返回时既是线程安全的又是有意义的。您可以容忍哪些边缘和竞争条件？
-
总的来说，在使用
shared_ptr<bool>作为一个花哨的atomic<int>和每个链表节点出于某种原因需要自己的mutex之间，我会说你的实现这个算法是错误的。 -
@selbie Lazy-synchronized 是两步删除过程：首先将其标记为已删除，然后物理删除它。这允许
contains无需等待。锁定的节点仍然可以遍历，只是不被修改，并且 contains 不必等待删除。 -
is_empty 对多处理器环境中的分布式集合来说根本不是一个有意义的问题。再多的 C++ 魔法也无法改变这一点。
-
这就是这个问题的动机。
use_count()认为什么是正确的？我知道use_count()可以说 1 而另一个指针已被重置或重新分配，但仍有待处理的操作。因为 refcount 的顺序是宽松的，所以“正确”的含义没有顺序一致的计数器（特别是 C++ 使用的 SC w/o Race）那么严格。一般来说，SC可以和串行化结合起来形成线性化。那么你能否强制执行更严格的内存顺序以让整个系统“正确”？
标签： c++ multithreading shared-ptr memory-barriers reference-counting