如何在 C++ 中实现自己的线程安全共享指针？

Question

我需要为嵌入式设备制作自己的简单线程安全共享指针类。 我按照 Jeff Alger 的书（面向真正程序员的 C++）中的描述制作了计数主指针和句柄。这是我的消息来源：

template <class T>
class counting_ptr {
public:
    counting_ptr() : m_pointee(new T), m_counter(0) {}
    counting_ptr(const counting_ptr<T>& sptr) :m_pointee(new T(*(sptr.m_pointee))), m_counter(0)      {}
    ~counting_ptr() {delete m_pointee;}
    counting_ptr<T>& operator=(const counting_ptr<T>& sptr)
    {
        if (this == &sptr) return *this;
        delete m_pointee;
        m_pointee = new T(*(sptr.m_pointee));
        return *this;
    }
    void grab() {m_counter++;}
    void release() 
    {
        if (m_counter > 0) m_counter--;
        if (m_counter <= 0)
            delete this;
    }

    T* operator->() const {return m_pointee;}

private:
    T* m_pointee;
    int m_counter;
};


template <class T>
class shared_ptr {
private:
    counting_ptr<T>* m_pointee;

public:
    shared_ptr() : m_pointee(new counting_ptr<T>()) { m_pointee->grab(); }
    shared_ptr(counting_ptr<T>* a_pointee) : m_pointee(a_ptr) { m_pointee->grab(); }
    shared_ptr(const shared_ptr<T>& a_src) : m_pointee(a_src.m_pointee) {m_pointee->grab(); }
    ~shared_ptr()  { m_pointee->release(); }

    shared_ptr<T>& operator=(const shared_ptr<T>& a_src)
    {
        if (this == &a_src) return *this;
        if (m_pointee == a_src.m_pointee) return *this;
        m_pointee->release();
        m_pointee = a_src.m_pointee;
        m_pointee->grab();
        return *this;
    }
    counting_ptr<T>* operator->() const {return m_pointee;}
};

如果在一个线程中使用，这会很好。假设我有两个线程：

//thread 1
shared_ptr<T> p = some_global_shared_ptr;
//thread 2
some_global_shared_ptr = another_shared_ptr;

这种情况下，如果其中一个线程在内存分配/解除分配和计数器更改之间被中断，我会得到未定义的行为。当然，我可以将 shared_ptr::release() 包含在关键部分中，这样可以安全地删除指针。但是我可以用复制构造函数做什么？构造函数可能会在 m_pointee 构造期间被另一个线程中断，该线程将删除此 m_pointee。 我认为使 shared_ptr 分配线程安全的唯一方法是将分配（或创建）包含在关键部分中。但这必须在“用户代码”中完成。换句话说，shared_ptr 类的用户必须注意安全。 是否有可能以某种方式改变这种实现以使 shared_ptr 类线程安全？

=== 编辑 ===

经过一些调查（感谢 Jonathan），我意识到我的 shared_ptr 有三个不安全的地方：

• 非原子计数器更改
• 非原子赋值运算符（复制过程中可以删除源对象）
• shared_ptr 复制构造函数（与之前的案例非常相似）

前两种情况可以通过添加关键部分轻松解决。但是我不知道如何将临界区添加到复制构造函数中？ a_src.m_pointee 的副本在构造函数中的任何其他代码执行之前创建，并且可以在调用 grab 之前删除。正如乔纳森在他的评论中所说，解决这个问题非常困难。 我做了这样的测试：

typedef shared_ptr<....> Ptr;
Ptr p1, p2;

//thread 1
while (true)
{
    Ptr p;
    p2 = p;
}

//thread 2
while (!stop)
{
    p1 = p2;
    Ptr P(p2);
}

当然，它崩溃了。但我尝试在 VS 2013 C++ 中使用 std::shared_ptr 。它有效！ 因此可以为shared_ptr 创建线程安全的复制构造函数。但是 stl 来源对我来说太难了，我不明白他们是如何做到的。请任何人解释一下它在 STL 中是如何工作的？

=== 编辑 2 ===

对不起，std::shared_ptr 的测试出错了。它并没有像 boost::shared_ptr 那样完全通过。有时复制构造函数无法进行复制，因为在复制过程中源被删除。在这种情况下，将创建空指针。

Answer 1

这很难解决，我会认真考虑您是否真的需要支持单个对象的并发读取和写入（boost::shared_ptr 和 std::shared_ptr 确实不支持，除非所有访问是通过 atomic_xxx() 函数完成的，这些函数为 shared_ptr 重载并且通常会获取锁。

首先，您需要将shared_ptr<T>::m_pointee 更改为atomic<counting_ptr<T>*>，以便您可以在其中自动存储新值。 counting_ptr<T>::m_counter 需要是 atomic<int>，这样引用计数的更新可以自动完成。

您的赋值运算符是一个大问题，您需要至少重新排序操作，以便首先增加引用计数，并避免time of check to time of use 错误，像这样（不是甚至编译，更不用说测试了）：

shared_ptr<T>& operator=(const shared_ptr<T>& a_src)
{
    counter_ptr<T>* new_ptr = a_src.m_pointee.load();
    new_ptr->grab();
    counter_ptr<T>* old_ptr = m_pointee.exchange(new_ptr);
    old_ptr->release();
    return *this;
}

这种形式对自赋值是安全的（如果两个对象共享相同的指针，它只会增加引用计数然后再次减少它）。在您尝试复制 a_src 时更改它仍然不安全。考虑最初a_src.m_pointee->m_counter == 1 的情况。当前线程可以调用load() 来获取另一个对象的指针，然后第二个线程可以在该指针上调用release()，这将delete 它，使grab() 调用未定义的行为，因为它访问的对象具有被销毁并且内存已经被释放。解决这个问题需要进行相当大的重新设计，并且可能需要一次对两个单词进行操作的原子操作。

做到这一点是可能的，但很难，您应该重新考虑是否有必要，或者使用它的代码是否可以避免在其他线程正在读取对象时修改对象，除非用户有锁定互斥锁或其他形式的手动同步。

Answer 2

经过一些调查，我可以得出结论，不可能在线程安全理解如下的情况下创建线程安全的 shared_ptr 类：

//thread 1
shared_ptr<T> p = some_global_shared_ptr;
//thread 2
some_global_shared_ptr = another_shared_ptr;

此示例不保证第一个线程中的p 将指向some_global_shared_ptr 的旧值或新值。通常，此示例会导致未定义的行为。使示例安全的唯一方法是将两个运算符包装到关键部分或相互排除中。

shared_ptr类的拷贝构造函数导致的主要问题。其他问题可以使用shared_ptr 方法中的临界区来解决。

Answer 3

只需从 CmyLock 继承您的类，您就可以使一切线程安全。 我在所有代码中使用这个类已经很多年了，通常与类 CmyThread 结合使用，它创建了一个具有非常安全互斥锁的线程。也许我的回答有点晚了，但上面的答案不是好的做法。

/** Constructor */
CmyLock::CmyLock()
{
    (void) pthread_mutexattr_init( &m_attr);
    pthread_mutexattr_settype( &m_attr, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init( &m_mutex, &m_attr);
}

/** Lock the thread for other threads. */
void CmyLock::lock()
{
    pthread_mutex_lock( &m_mutex);
}

/** Unlock the thread for other threads. */
void CmyLock::unlock()
{
    pthread_mutex_unlock( &m_mutex);
}

这里也是线程类。尝试请将 CmyLock 和 CmyThread 类复制到您的项目中并告诉它何时工作！虽然它是为 Linux 设计的，但 Windows 和 Mac 也应该能够运行它。

对于包含文件：

// @brief Class to create a single thread.
class CmyThread : public CmyLock
{
friend void *mythread_interrupt(void *ptr);

public:
    CmyThread();
    virtual ~CmyThread();
    virtual void startWorking() {}
    virtual void stopWorking() {}
    virtual void work();
    virtual void start();
    virtual void stop();
    bool isStopping() { return m_stopThread; }
    bool isRunning() { return m_running && !m_stopThread; }

private:
    virtual void run();

private:
    bool                m_running;      ///< Thread is now running.
    pthread_t           m_thread;       ///< Pointer to thread.
    bool                m_stopThread;   ///< Indicate to stop thread.
};

C++ 文件：

/** @brief Interrupt handler.
 *  @param ptr [in] SELF pointer for the instance.
 */
void *mythread_interrupt(void *ptr)
{
    CmyThread *irq =
            static_cast<CmyThread*> (ptr);
    if (irq != NULL)
    {
        irq->run();
    }
    return NULL;
}
/** Constructor new thread. */
CmyThread::CmyThread()
: m_running( false)
, m_thread( 0)
, m_stopThread( false)
{
}

/** Start thread. */
void CmyThread::start()
{
    m_running =true;
    m_stopThread =false;
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
    int stack_size =8192*1024;
    pthread_attr_setstacksize(&attr, stack_size);
    pthread_create(&m_thread, &attr, mythread_interrupt, (void*) this);
}

/** Thread function running. */
void CmyThread::run()
{
    startWorking();
    while (m_running && m_stopThread==false)
    {
        work();
    }
    m_running =false;
    stopWorking();
    pthread_exit(0);
}

/** Function to override for a thread. */
virtual void CmyThread::work()
{
    delay(5000);
}

例如，这里是存储和检索 1000 条数据的简单示例：

    class a : public CmyLock
    {
        set_safe(int *data)
        {
            lock();
            fileContent =std::make_shared<string>(data); 
            unlock();
        }

        get_safe(char *data)
        {
            lock();
            strcpy( data, fileContent->c_str());
            unlock();
        }
        std::shared_ptr<string> fileContent;
    };