互斥锁mutex的选择

互斥锁mutex有2种方案：
1）C++11以后，使用std::mutex，当然，特殊应用场景下，也有另外三种：std::recursive_mutex（递归mutex类），std::timed_mutex（定时mutex类），recurisive_timed_mutex（定时递归mutex类）；

2）Linux平台下，使用NPTL提供的pthread_mutex。

muduo采用第2种方案，自定义MutexLock对pthread_mutex进行了轻度包装。

MutexLock类

MutexLock类，除了基础的lock/unlock，支持Condition（条件变量）外，还支持查询锁的持有线程tid（holder），断言当前线程是否持有锁等。

// CAPABILITY, RELEASE, ASSERT_CAPABILITY等宏定义适用于clang/SWIG编译器下的线程安全
// 其他编译器如GCC可以安全擦除, 这里保留, 可以作为标记提醒程序员

class CAPABILITY("mutex") MutexLock : public noncopyable
{
public:
    MutexLock()
    : holder_(0)
    {
        MCHECK(pthread_mutex_init(&mutex_, NULL)); // 宏函数MCHECK, 用于检查库函数/系统调用的返回值
    }
    ~MutexLock()
    {
        assert(holder_ == 0);
        MCHECK(pthread_mutex_destroy(&mutex_));
    }
    // must be called when locked, i.e. for assertion
    bool isLockedByThisThread() const // 断言 当前调用线程持有锁, 失败返回false, 不会导致程序终止
    {
        return holder_ == CurrentThread::tid();
    }
    void assertLocked() const ASSERT_CAPABILITY(this) // 断言 调用线程持有锁, 会导致程序终止, 用于debug
    {
        assert(isLockedByThisThread());
    }

    // internal usage

    /**
     * lock mutex_ and check lock status.
     * assign current thread's tid as the lock's holder
     */
    void lock() ACQUIRE()
    {
        MCHECK(pthread_mutex_lock(&mutex_));
        assignHolder();
    }
    /**
     * unassign the lock's holder, then unlock mutex_.
     */
    void unlock() RELEASE()
    {
        unassignHolder();
        MCHECK(pthread_mutex_unlock(&mutex_));
    }
    /* non-const, because return a pointer to class's (private) data */
    pthread_mutex_t* getPthreadMutex()
    {
        return &mutex_;
    }

private:
    friend class Condition;

    /* give up the MutexLock */
    class UnassignGuard : public noncopyable // 用于放弃锁, 如条件变量中需要放弃锁时可用到
    {
    public:
        explicit UnassignGuard(MutexLock& owner)
        : owner_(owner)
        {
            owner_.unassignHolder(); // clear lock owner_ holder
        }
        ~UnassignGuard()
        {
            owner_.assignHolder();  // set current thread as the lock holder
        }

    private:
        MutexLock& owner_; // 互斥锁对象引用
    };
    /**
     * set current thread as the lock holder
     * by assigning current thread's tid to holder
     */
    void assignHolder()
    {
        holder_ = CurrentThread::tid(); // 设置锁的持有线程tid为当前线程tid
    }
    /**
     * clear the lock holder
     */
    void unassignHolder() // 清除锁的持有线程tid
    {
        holder_ = 0;
    }

    pthread_mutex_t mutex_;
    pid_t holder_;         // 持有锁的线程tid, 初值0表示无效线程
};

注意：MutexLock是引用传递，要操作原来的MutexLock对象，需要传递引用。

MutexLockGuard类

类同std::lock_guard，MutexLockGuard通过RAII方式管理MutexLock资源：构造MutexLockGuard对象时，取得mutex_锁；释放对象时，释放mutex_锁。

// RAII方式管理MutexLock锁资源
class SCOPED_CAPABILITY MutexLockGuard : public noncopyable
{
public:
    explicit MutexLockGuard(MutexLock& mutex) ACQUIRE(mutex)
    : mutex_(mutex)
    {
        mutex_.lock();
    }
    ~MutexLockGuard() RELEASE()
    {
        mutex_.unlock();
    }

private:
    MutexLock& mutex_;
};

测试

思路：利用MutexLockGuard在local 作用域内，获取指定MutexLock互斥锁，对全局变量进行递增操作。多个线程同时进行这一操作，最终通过判断全局变量是否为预期值，来判断MutexLock和MutexLockGuard是否正常工作。

MutexLock g_mutex;
vector<int> g_vec;
const int kCount = 10*1000*1000;

int g_count = 0;
int g_threadfunc_count = 0;

int foo() __attribute__ ((__noinline__)); // 阻止foo内联

void threadFunc()
{
    for (int i = 0; i < kCount; ++i) {
        MutexLockGuard lock(g_mutex);
        g_vec.push_back(i);
        g_threadfunc_count++;
    }
}

int foo()
{
    MutexLockGuard lock(g_mutex);
    if (!g_mutex.isLockedByThisThread())
    {
        printf("FAIL\n");
        return -1;
    }

    ++g_count;
    return 0;
}

int main()
{
    printf("sizeof pthread_mutex_t: %zd\n", sizeof(pthread_mutex_t));
    printf("sizeof Mutex: %zd\n", sizeof(MutexLock));
    printf("sizeof pthread_cond_t: %zd\n", sizeof(pthread_cond_t));
    printf("sizeof Condition: %zd\n", sizeof(Condition));
    MCHECK(foo());
    if (g_count != 1)
    {
        printf("MCHECK calls twice.\n");
        abort();
    }

    const int kMaxThreads = 8;
    g_vec.reserve(kMaxThreads * kCount);

    Timestamp start(Timestamp::now());
    for (int i = 0; i < kCount; ++i) {
        g_vec.push_back(i);
    }
    printf("single thread without lock %f seconds\n", timeDifference(Timestamp::now(), start));

    start = Timestamp::now();
    threadFunc();
    printf("signle thread with lock %f seconds\n", timeDifference(Timestamp::now(), start));

    // 多线程核心测试部分
    // multi-thread invoke threadFunc try to get the same mutex at the same time
    for (int nthreads = 0; nthreads < kMaxThreads; ++nthreads)
    {
        std::vector<std::unique_ptr<Thread>> threads;
        g_vec.clear();
        g_threadfunc_count = 0;
        start = Timestamp::now();
        for (int i = 0; i < nthreads; ++i) {
            threads.emplace_back(new Thread(&threadFunc));
            threads.back()->start();
        }

        for (int i = 0; i < nthreads; ++i) {
            threads[i]->join();
        }
        printf("%d thread(s) with lock %f seconds\n", nthreads, timeDifference(Timestamp::now(), start));
        printf("g_threadfunc_count = %d\n", g_threadfunc_count);
    }
    return 0;
}

运行结果：

sizeof pthread_mutex_t: 40
sizeof Mutex: 48
sizeof pthread_cond_t: 48
sizeof Condition: 56
single thread without lock 0.166781 seconds
signle thread with lock 0.455777 seconds
0 thread(s) with lock 0.000000 seconds
g_threadfunc_count = 0
1 thread(s) with lock 0.425418 seconds
g_threadfunc_count = 10000000
2 thread(s) with lock 4.080142 seconds
g_threadfunc_count = 20000000
3 thread(s) with lock 4.136063 seconds
g_threadfunc_count = 30000000
4 thread(s) with lock 5.242511 seconds
g_threadfunc_count = 40000000
5 thread(s) with lock 6.609551 seconds
g_threadfunc_count = 50000000
6 thread(s) with lock 7.620478 seconds
g_threadfunc_count = 60000000
7 thread(s) with lock 8.231012 seconds
g_threadfunc_count = 70000000