递归共享互斥

Question

我一直在寻找一个recursive shared mutex的实现，以便在一个大型多线程应用程序中处理一个非常特殊的数据树。

因为boost和stdlib没有这种特殊类型，所以我自己写了一个。但是，我不确定我是不是错过了什么.

您可以在github上通过测试找到完整的实现。用GCC 5.4进行测试。

特性

• 具有锁和lock_shared的独占所有权的递归。
• 使用lock_shared实现可共享所有权的递归。
• 试图获得独占所有权的线程比试图获得可共享所有权的新线程具有更高的优先级。
• 最大等待写入者，所有权级别:uint32 32：：max

recursive_shared_mutex.hpp

/*
 * Copyright (c) 2017 Toni Neubert, all rights reserved.
 */
#pragma once

#include <atomic>
#include <mutex>
#include <thread>
#include <unordered_map>

class RecursiveSharedMutex {
public:
    /**
     * @brief Constructs the mutex.
     */
    RecursiveSharedMutex();

    /**
     * @brief Locks the mutex for exclusive write access for this thread.
     *              Blocks execution as long as write access is not available:
     *              * other thread has write access
     *              * other threads try to get write access
     *              * other threads have read access
     *
     *              A thread may call lock repeatedly.
     *              Ownership will only be released after the thread makes a matching number of calls to unlock.
     */
    void lock();

    /**
     * @brief Locks the mutex for sharable read access.
     *              Blocks execution as long as read access is not available:
     *              * other thread has write access
     *              * other threads try to get write access
     *
     *              A thread may call lock repeatedly.
     *              Ownership will only be released after the thread makes a matching number of calls to unlock_shared.
     */
    void lock_shared();

    /**
     * @brief Unlocks the mutex for this thread if its level of ownership is 1. Otherwise reduces the level of ownership
     *              by 1.
     */
    void unlock();

    /**
     * @brief Unlocks the mutex for this thread if its level of ownership is 1. Otherwise reduces the level of ownership
     *              by 1.
     */
    void unlock_shared();

private:
    /// Protects data access of mutex.
    std::mutex _mtx;
    /// Number of threads waiting for exclusive write access.
    std::atomic<uint32_t> _waitingWriters;
    /// Thread id of writer.
    std::atomic<std::thread::id> _writerThreadId;
    /// Level of ownership of writer thread.
    uint32_t _writersOwnership;
    /// Level of ownership of reader threads.
    std::unordered_map<std::thread::id, uint32_t> _readersOwnership;
};

recursive_shared_mutex.cpp

/*
 * Copyright (c) 2017 Toni Neubert, all rights reserved.
 */
#include "recursive_shared_mutex.hpp"
#include <cassert>

RecursiveSharedMutex::RecursiveSharedMutex() :
    _waitingWriters(0),
    _writersOwnership(1) {
}

void RecursiveSharedMutex::lock() {
    // Case 1:
    // * Thread has no ownership.
    // * Zero readers, no writer.
    // -> The thread gets exclusive ownership as writer.

    // Case 2:
    // * Thread has no ownership.
    // * Many readers, no writer.
    // -> Gets exclusive ownership as writer and waits until last reader is unlocked.

    // Case 3:
    // * Thread has no ownership.
    // * Zero readers, one writer.
    // -> Gets exclusive ownership as writer after other writer thread is unlocked.

    // Case 4:
    // * Thread has no ownership.
    // * Zero readers, one writer.
    // * Many threads try to get exclusive ownership.
    // -> Various attempts until exclusive ownership as writer has been acquired. The acquisition order is arbitrarily.

    // Case 5:
    // * Thread has exclusive ownership.
    // * Zero readers, one writer.
    // -> Increases threads level of ownership.

    // Case 6:
    // * Thread has sharable ownership.
    // -> Deadlock.

    auto threadId = std::this_thread::get_id();
    {
        // Increase level of ownership if thread has already exclusive ownership.
        std::lock_guard<std::mutex> lock(_mtx);
        if (_writerThreadId == threadId) {
            ++_writersOwnership;
            return;
        }
    }

    // Notify the new waiting writer.
    assert(_waitingWriters != 0x7fffffff);
    _waitingWriters += 1;

    for (;;) {
        // Attempt to get exclusive ownership.
        std::thread::id emptyThreadId;
        if (_writerThreadId.compare_exchange_weak(emptyThreadId, threadId)) {
            for (;;) {
                // Wait until no readers exist.
                std::lock_guard<std::mutex> lock(_mtx);
                if (_readersOwnership.size() == 0) {
                    // Notify a waiting writer is gone.
                    --_waitingWriters;
                    return;
                }
            }
        }
    }
}

void RecursiveSharedMutex::lock_shared() {
    // Case 1:
    // * Thread has/has no ownership.
    // * Zero/Many readers, no writer.
    // -> The thread gets shared ownership as reader.

    // Case 2:
    // * Thread has no ownership.
    // * Zero readers, one writer.
    // -> Waits until writer thread unlocked. The thread gets shared ownership as reader.

    // Case 3:
    // * Thread has sharable ownership.
    // * Many readers, no writer.
    // -> Increases threads level of ownership.

    // Case 4:
    // * Thread has exclusive ownership.
    // * Zero readers, one writer.
    // -> Increases threads level of ownership.

    // Case 5:
    // * Thread has no ownership.
    // * Zero/Many readers, one/no writer.
    // * Many threads try to get exclusive ownership.
    // -> Waits until all exclusive ownership requests are handled. The thread gets shared ownership as reader.

    auto threadId = std::this_thread::get_id();
    {
        // Increase level of ownership if thread has already exclusive ownership.
        std::lock_guard<std::mutex> lock(_mtx);

        // As writer.
        if (_writerThreadId == threadId) {
            ++_writersOwnership;
            return;
        }

        // As reader.
        if (_readersOwnership.count(threadId) != 0) {
            ++_readersOwnership[threadId];
            return;
        }
    }

    for (;;) {
        std::lock_guard<std::mutex> lock(_mtx);

        // Wait until no writer is waiting or writing.
        if (_waitingWriters.load() != 0 || _writerThreadId != std::thread::id()) {
            continue;
        }

        // Add new reader ownership.
        _readersOwnership.insert(std::make_pair(threadId, 1));
        return;
    }
}

void RecursiveSharedMutex::unlock() {
    // Case 1:
    // * Thread has exclusive ownership.
    // -> If threads level of ownership is 0, releases exclusive ownership otherwise decrements threads level of
    //      ownership.

    // Case 2:
    // * Thread has no/has sharable ownership.
    // -> In debug mode: Assert will terminate program.
    // -> In release mode: Undefined behaviour! Case 1 will occur.

    assert(std::this_thread::get_id() == _writerThreadId);

    std::lock_guard<std::mutex> lock(_mtx);
    {
        // Decrease writer threads level of ownership if not 1.
        if (_writersOwnership != 1) {
            --_writersOwnership;
            return;
        }
    }

    // Reset threads ownership.
    _writerThreadId = std::thread::id();
}

void RecursiveSharedMutex::unlock_shared() {
    // Case 1:
    // * Thread has sharable ownership.
    // -> If reader threads level of ownership is 0, releases sharable ownership otherwise decrements reader threads
    //    level of ownership.

    // Case 2:
    // * Thread has exclusive ownership.
    // -> Decrements threads level of ownership.

    // Case 3:
    // * Thread has no ownership.
    // -> In debug mode: Assert will terminate program.
    // -> In release mode: Undefined behaviour!

    // Reduce readers recursive depth.
    // Remove reader from map if depth == 0.
    auto threadId = std::this_thread::get_id();

    std::lock_guard<std::mutex> lock(_mtx);

    // Decrease writer threads level of ownership if not 1.
    if (_writerThreadId == threadId) {
        --_writersOwnership;
        return;
    }

    assert(_readersOwnership.count(threadId) == 1);

    // Decrease threads level of ownership if not 1.
    if (_readersOwnership[threadId] != 1) {
        --_readersOwnership[threadId];
        return;
    }

    // Remove readers ownership.
    _readersOwnership.erase(threadId);
}

Answer 1

有一个想法是这样的:如果在紧的for-循环中的互斥线外有一个线程屈服，那不是更好吗？或者我们扩大到这个程度？

我也一直在使用这个try_lock加法；

bool RecursiveSharedMutex::try_lock() {
    auto threadId = std::this_thread::get_id();
    std::thread::id emptyThreadId;
    std::lock_guard<std::mutex> lock(_mtx);

    if (_writerThreadId == threadId) {
        ++_writersOwnership;
        return true;
    }

    if (_readersOwnership.size() == 0 && _writerThreadId.compare_exchange_weak(emptyThreadId, threadId))
        return true;

    return false;
}