feek_ptr到singleton而不是线程安全

Question

我正在编写将共享_ptr返回单例的函数。我希望当所有参考文献消失时，单身对象都会被摧毁。我的解决方案以使用静态feek_ptr和mutex ，但是我对其线程安全的测试并不始终如一。

这是一个说明问题的完整示例。

#include <gtest/gtest.h>
#include <atomic>
#include <mutex>
#include <memory>
#include <vector>
#include <thread>

using namespace std;

// Number of instances of this class are tracked in a static counter.
// Used to verify the get_singleton logic (below) is correct.
class CountedObject {
private:
    static atomic<int> instance_counter;

public:
    CountedObject() {
        int prev_counter = instance_counter.fetch_add(1);
        if (prev_counter != 0)
            // Somehow, 2 objects exist at the same time. Why?
            throw runtime_error("Constructed " + to_string(prev_counter + 1) +
                                " counted objects");
    }
    ~CountedObject() {
        instance_counter.fetch_sub(1);
    }
    static int count() {
        return instance_counter.load();
    }
};

atomic<int> CountedObject::instance_counter{0};

// Returns reference to a singleton that gets destroyed when all references
// are destroyed.
template <typename T>
std::shared_ptr<T> get_singleton() {
    static mutex mtx;
    static weak_ptr<T> weak;
    scoped_lock lk(mtx);
    shared_ptr<T> shared = weak.lock();
    if (!shared) {
        shared.reset(new T, [](T* ptr){ 
            scoped_lock lk(mtx);
            delete ptr;
        });
        weak = shared;
    }
    return shared;
}

// This test passes consistently.
TEST(GetSingletonTest, SingleThreaded) {
    ASSERT_EQ(CountedObject::count(), 0);

    auto ref1 = get_singleton<CountedObject>();
    auto ref2 = get_singleton<CountedObject>();
    ASSERT_EQ(CountedObject::count(), 1);

    ref1.reset();
    ASSERT_EQ(CountedObject::count(), 1);
    ref2.reset();
    ASSERT_EQ(CountedObject::count(), 0);
}

// This test does NOT pass consistently.
TEST(GetSingletonTest, MultiThreaded) {
    const int THREAD_COUNT = 2;
    const int ITERS = 1000;
    vector<thread> threads;
    for (int i = 0; i < THREAD_COUNT; ++i)
        threads.emplace_back([ITERS]{
            // Repeatedly obtain and release references to the singleton.
            // The invariant must hold that at most one instance ever exists
            // at a time.
            for (int j = 0; j < ITERS; ++j) {
                auto local_ref = get_singleton<CountedObject>();
                local_ref.reset();
            }
        });
    for (auto& t : threads)
        t.join();
}

在我的系统（ARM64 Linux，g ++ 7.5.0）上，多线程测试通常会失败：

[==========] Running 2 tests from 1 test case.
[----------] Global test environment set-up.
[----------] 2 tests from GetSingletonTest
[ RUN      ] GetSingletonTest.SingleThreaded
[       OK ] GetSingletonTest.SingleThreaded (0 ms)
[ RUN      ] GetSingletonTest.MultiThreaded
terminate called after throwing an instance of 'std::runtime_error'
  what():  Constructed 2 counted objects
Aborted (core dumped)

我省略了cout来自代码的消息，但分别添加了消息以调试什么是发生：

[thread id,  message]
...
547921465808 Acquired lock
547921465808 Weak ptr expired, constructing
547921465808 Releasing lock
547929858512 Acquired lock
547929858512 Weak ptr expired, constructing
terminate called after throwing an instance of 'std::runtime_error'
  what():  Constructed 2 counted objects
Aborted (core dumped)

看来线程1确定单例已过期并重新构造它。然后螺纹2醒来， 也确定了单例已经过期，即使线程1刚刚填充了它 - 好像线程2正在使用弱的“过时”版本指针。

如何将分配从线程1立即可见到线程2的弱？这可以通过C ++ 20的atomic＆lt; feal_ptr＆lt; t＆gt;＆gt;来实现吗？我的项目仅限于C ++ 17，因此不幸的是，这对我来说不是一个选择。

感谢您的帮助！

Answer 1

锁定下的分配将可见到任何其他线程检查feek_ptr锁定下的分配。

存在多个实例的原因是驱动器的操作顺序不能保证否则。 ！它等同于“有确定性，有一个对象的实例”的倒数。可能会有一个实例，因为在eleter有机会采取行动之前，参考计数已减少。

想象一下此序列：

• 线程1：〜共享_ptr被调用。
• 线程1：强参考计数减少。
• 线程1：比较等于零，并且eleter开始。
• 线程2：get_singleton被调用。
• 线程2：获取锁。
• 线程2：！
• 线程1：...与此同时，我们正在等待获取锁...
• 线程2：新实例是构造和分配的，锁定了。
• 线程1：现在，eleter可以获取锁定并删除原始实例。

编辑：

为此，您需要一些指示没有实例。只有在实例被销毁之后才能重置。其次，您需要决定如果遇到存在实例但正在破坏实例的情况下该怎么办：

template <typename T>
std::shared_ptr<T> get_singleton() {
    static mutex mtx;
    static weak_ptr<T> weak;
    static bool instance;

    while (true) {
        scoped_lock lk(mtx);

        shared_ptr<T> shared = weak.lock();
        if (shared)
            return shared;

        if (instance)
            // allow deleters to make progress and try again
            continue;

        instance = true;
        shared.reset(new T, [](T* ptr){ 
            scoped_lock lk(mtx);
            delete ptr;
            instance = false;
        });
        weak = shared;
        return shared;
    }
}

要说以不同的方式，状态机器中的某些状态：

1. 不存在T t
2. 正在构造T，但是可以't仍在访问
3. t，并且可以访问
4. t正在被销毁，并且无法再访问

bool，让我们知道我们处于州＃4。一种方法是循环，直到我们不再处于州＃4为止。我们放下锁，使删除器可以取得进展。当我们重新锁定锁定时，另一个线程可能已经涌入并创建了实例，因此我们需要从顶部开始并重新检查所有内容。

Answer 2

非常感谢Jeff的解释和回答。我正在接受他的答案，但我想发布一个替代实现，在我的（公认的）测试中，表现稍好一些。您的里程可能会有所不同。

template <typename T>
std::shared_ptr<T> get_singleton() {
    static mutex mtx;
    static weak_ptr<T> weak;
    static atomic<bool> deleted{true};
    scoped_lock lk(mtx);
    shared_ptr<T> shared = weak.lock();
    if (!shared) {
        // The refcount is 0, but is the object actually deleted yet?
        // Spin until it is.
        bool expected;
        do {
            expected = true;
        } while (!deleted.compare_exchange_weak(expected, false));

        // The previous object is definitely deleted. Ready to make a new one.
        shared.reset(new T, [](T* ptr){ 
            delete ptr;
            deleted.store(true);
        });
        weak = shared;
    }
    return shared;
}

这可以从eleter中删除锁定的采集，并用原子标志代替，该原子标志在对象实际删除时变为真。