怎麼寫執行緒安全的複製(移動)建構子及複製(移動)賦值運算子

會呼叫建構子就是還沒有東西需要被保護。

呼叫解構子時需要保護成員變數基本上就是設計錯誤,任何時候都不該在變數還被使用時銷毀它,這跟 thread-safe 本身沒有關係。

因此以上兩點不在本文範圍。

現假設有自訂類別如下,將其改為 thread-safe 要怎麼做呢?

class Foo {
public:
    // copy constructor
    Foo(const Foo& rhs) : id_(rhs.id_), ticks_(rhs.ticks_) {}

    // move constructor
    Foo(Foo&& rhs) : id_(rhs.id_), ticks_(std::move(rhs.ticks_)) {}

    // copy assignment
    Foo& operator=(const Foo& rhs) {
        if (this != &rhs) {
            id_ = rhs.id_;
            ticks_ = rhs.ticks_;
        }

        return *this;
    }

    // move assignment
    Foo& operator=(Foo&& rhs) {
        if (this != &rhs) {
            id_ = rhs.id_;
            ticks_ = std::move(rhs.ticks_);
        }

        return *this;
    }

public:
    void SetID(int id) {
        id_ = id;
    }

private:
    int id_;
    std::vector ticks_;
};

首先 #include <mutex>,並新增成員變數 mutable std::mutex m_mutex。

copy constructor 不能使用 initializer list 的方式初始化了,因為沒有辦法在 initializer list 鎖定 rhs。

Foo(const Foo& rhs) : id_(rhs.id_), ticks_(rhs.ticks_) {}

考慮以下程式碼:

Foo foo1;
foo1.SetID(1); // use thread 1 write foo1
Foo foo2(foo1); // use thread 2 read foo1

讀取 rhs 時,可能會有其他 thread 正在修改 rhs,因此複製建構子需要改成先將 rhs 的 mutex 鎖住,再讀取 rhs 並建構自己。

Foo(const Foo& rhs) {
    std::lock_guard<std::mutex> lock(rhs.m_mutex);
    id_ = rhs.id_;
    ticks_ = rhs.ticks_;
}

move constructor 也是一樣。

Foo(Foo&& rhs) {
    std::lock_guard<std::mutex> lock(rhs.m_mutex);
    id_ = rhs.id_;
    ticks_ = std::move(rhs.ticks_);
}

接下來討論 copy assignment 及 move assignment。

Foo& operator=(const Foo& rhs) {
    std::lock_guard<std::mutex> lockThis(m_mutex);
    std::lock_guard<std::mutex> lockRhs(rhs.m_mutex);
    if (this != &rhs) {
        id_ = rhs.id_;
        ticks_ = rhs.ticks_;
    }

    return *this;
}

一開始直覺可能會這樣寫,但這會有 deak lock,考慮以下程式碼:

Foo foo;
foo = foo;

自己賦值給自己,根據程式,先鎖定 this,再鎖定 rhs,但其實都是 foo,也就是同一條 thread 鎖定自己兩次會 dead lock,我們可以移動 lock 的位置,但請不要改成std::recursive_mutex,這會隱藏 lock 的錯誤使用:

Foo& operator=(const Foo& rhs) {
    if (this != &rhs) {
        std::lock_guard<std::mutex> lockThis(m_mutex); // step 1
        std::lock_guard<std::mutex> lockRhs(rhs.m_mutex); // step 2
        id_ = rhs.id_;
        ticks_ = rhs.ticks_;
    }

    return *this;
}

很好,lock 之前會先檢查是不是自己賦值給自己,但這還有其他問題,考慮以下程式碼:

Foo foo1, foo2;
foo1 = foo2; // thread 1
foo2 = foo1; // thread 2

現假設 thread 1 及 thread 2 的執行順序如下

1. thread1 執行 step1,鎖住 this 成功,也就是鎖住了自己 (foo1)。
2. thread2 執行 step1,鎖住 this 成功,也就是鎖住了自己 (foo2)。
3. thread1 執行 step2,鎖住失敗,因為 rhs 也就是 foo2 已被 thread 2 鎖住。
4. thread2 執行 step2,鎖住失敗,因為 rhs 也就是 foo1 已被 thread 1 鎖住。

至此 thread 1 及 thread 2 都在等待對方釋放鎖,dead lock。

怎麼修改呢?見下列程式粗體部分:

Foo& operator=(const Foo& rhs) {
    if (this != &rhs) {
        std::lock(m_mutex, rhs.m_mutex);
        std::lock_guard<std::mutex> lockThis(m_mutex, std::adopt_lock);
        std::lock_guard<std::mutex> lockRhs(rhs.m_mutex, std::adopt_lock);
        id_ = rhs.id_;
        ticks_ = rhs.ticks_;
    }

    return *this;
}

std::lock 可以將所有的 mutex 一起鎖定,如果某一個 mutex 鎖不成功就先 unlock,至於 std::adopt_lock 則是告訴 std::lock_guard 當執行到你這行 code 時,std::lock 已經鎖住了mutex,你就不用再重複鎖定免得 dead lock,但是你還是要負責在離開 block 時幫我自動解鎖。
 

如果你的編譯器支援 C++17 的話,有另一種替代寫法如下:

Foo& operator=(const Foo& rhs) {
    if (this != &rhs) {
        std::scoped_lock lockAll(m_mutex, rhs.m_mutex);
        id_ = rhs.id_;
        ticks_ = rhs.ticks_;
    }

    return *this;
}

move assignment 也是一樣的作法,不再贅述。

最後,如果你的 class 是使用讀寫鎖實作的話,也就是用 std::shared_mutex 配合 std::unique_lock 及 std::shared_lock,則 copy constructor 改成讀鎖:

Foo(const Foo& rhs) {
     std::shared_lock<std::shared_mutex> lock(rhs.m_mutex);
     id_ = rhs.id_;
     ticks_ = rhs.ticks_;
}

copy assignment 改成寫鎖加讀鎖:

Foo& operator=(const Foo& rhs) {
    if (this != &rhs) {
        std::unique_lock<std::shared_mutex> lockThis(m_mutex, std::defer_lock);
        std::shared_lock<std::shared_mutex> lockRhs(rhs.m_mutex, std::defer_lock);
        std::lock(lockThis, lockRhs);
        id_ = rhs.id_;
        ticks_ = rhs.ticks_;
    }

    return *this;
}

注意這時 lockThis 及 lockRhs 改成 std::defer_lock,也就是不獲得 mutex 的所有權,只負責表明是讀寫鎖,且負責離開作用域的釋放鎖,讓 std::lock 來執行真正的鎖定。

為什麼使用讀寫鎖要後呼叫 std::lock,而使用 std::lock_guard 時先呼叫 std::lock 呢?

其實只是因為使用讀寫鎖時,std::lock 要知道哪個是讀鎖,哪個是寫鎖而已,畢竟 std::lock 要知道它內部一個一個呼叫的是 std::shared_mutex 的 lock 還是 lock_shared,因此std::unique_lock 及 std::shared_lock 要寫在前,先表明型別。

move 版本也是一樣,只是非內建型別要用 std::move 包起來而已。

發佈留言