muduo C++ 网络库——线程同步精要

深藏阁楼爱情的钟 2022-05-22 08:58 280阅读 0赞

并发编程有两种基本模型：**message passing 消息传递** 和 **shared memory 内存共享**

运行在多台机器上的多个进程的并行编程只有一种实用模型： **message passing**

线程同步的四项原则：按重要性排列

1.首要原则是最低限度地共享对象，减少需要同步的场合。一个对象能不暴露给别的进程就不要暴露。

2.其次是使用高级的并发编程构件，如TaskQueue,Producer-Consumer Queue, CountDownLatch等等

3.使用底层同步原语(primitives)时，只用非递归的互斥器和条件变量，慎用读写锁，不要用信号量。

4.除了使用atomic整数之外，不自己编写lock-free代码，也不要用“内核级”同步原语。

重点讲第3条：底层同步原语的使用

## 互斥器：使用得最多的同步原语 ##

概念补充：RAII——资源的有效期与持有资源的对象的生命期严格绑定，即由对象的构造函数完成资源的分配(获取),同时由析构函数完成资源的释放。只要对象能正确地析构，就不会出现资源泄漏问题。

互斥器保护了临界区，任何一个时刻最多只能有一个线程在此mutex划出的临界区内活动。单独使用mutex时，主要为了保护共享数据。原则：

\-用RAII手法封装mutex的创建，销毁，加锁，解锁四个操作。

\-只用非递归的mutex（即不可重入的mutex）

\-不手工调用lock()和unlock()函数，一切交给Guard对象的构造和析构函数负责。Guard对象的生命期正好等于临界区。

\-每次构造Guard对象的时候，思考已持有的锁，防止因加锁顺序不同而导致死锁。

### 1.只使用非递归的mutex ###

mutex分为**递归(recursive)**和**非递归(non-recursive)**两种。另一种叫法是：**可重入**与**非可重入**。

在同一线程里多次对non-recursive mutex加锁会立刻导致死锁，而recursive mutex不用考虑线程自己把自己锁死

但recursive mutex可能会隐藏代码里的一些问题，例如：

拿到一个锁就开始修改对象，但是没想到外层代码也拿到了锁，正在修改同一个对象。

MutexLock mutex;
    std::vector<Foo> foos;
    
    void post(const Foo& f)
    {
        MutexLockGuard lock(mutex);
        foos.push_back(f);
    }
    
    void traverse()
    {
        MutexLockGuard lock(mutex);
        for (std::vector<Foo>::const_iterator it = foos.begin(); it != foos.end(); ++it)
        {
            it->doit();
        }
    }

post加锁，然后修改foos对象；traverse()加锁，然后遍历foos向量。这都是正确的。

但是，如果Foo::doit()间接调用了post(),那么就会出现戏剧性的后果：  
1.mutex是非递归的，于是就死锁了。

2.mutex是递归的，由于push\_back()可能导致vector迭代器失效，程序偶尔会crash。

这种情况non-recursive就能暴露出程序的逻辑错误。死锁比较容易debug，把各个线程的调用栈打出来，很容易看出来是怎么死的。(gdb中使用thread apply all bt命令)

如果一个函数既可能在已加锁的情况下调用，又可能在未加锁的情况下调用，那么拆成两个函数：

1.跟原来的函数同名，函数加锁，转而调用第2个函数

2.给函数名加上后缀WithLockHold，不加锁，把原来的函数体搬过来。

void post(const Foo& f)
    {
        MutexLockGuard lock(mutex);
        postWithLockHold(f);
    }
    
    void postWithLockHold(const Foo& f)
    {
        foos.push_back(f);
    }

加锁情况下调用postWithLockHold,未加锁情况就调用post。

### 2.死锁    ###

考虑下面这个线程与自己死锁的例子：

#include "../Mutex.h"
    
    class Request
    {
     public:
      void process() // __attribute__ ((noinline))
      {
        muduo::MutexLockGuard lock(mutex_);
        print();
      }
    
      void print() const // __attribute__ ((noinline))
      {
        muduo::MutexLockGuard lock(mutex_);
      }
    
     private:
      mutable muduo::MutexLock mutex_;
    };
    
    int main()
    {
      Request req;
      req.process();
    }

line 8 和 line 14造成了死锁。可以按照之前的方法从Request::print()抽取出Request::printWithLockHold()。并让print和process都调用它即可。

两个线程死锁的例子：

Inventory（清单）class，记录当前的Request对象：成员函数都是线程安全的

class Inventory
    {
     public:
      void add(Request* req)
      {
        muduo::MutexLockGuard lock(mutex_);
        requests_.insert(req);
      }
    
      void remove(Request* req) __attribute__ ((noinline))
      {
        muduo::MutexLockGuard lock(mutex_);
        requests_.erase(req);
      }
    
      void printAll() const;
    
     private:
      mutable muduo::MutexLock mutex_;
      std::set<Request*> requests_;
    };
    
    Inventory g_inventory;

Request class 与 Inventory class 的交互逻辑很简单，在处理(process)请求的时候，往g\_inventory中添加自己。析构的时候从g\_inventory中移除自己。

class Request
    {
     public:
      void process()            // __attribute__ ((noinline))
      {
        muduo::MutexLockGuard lock(mutex_);
        g_inventory.add(this);
        // ...
      }
    
      ~Request() __attribute__ ((noinline))
      {
        muduo::MutexLockGuard lock(mutex_);
        sleep(1);
        g_inventory.remove(this);
      }
    
      void print() const __attribute__ ((noinline))
      {
        muduo::MutexLockGuard lock(mutex_);
        // ...
      }
    
     private:
      mutable muduo::MutexLock mutex_;
    };

Inventory class 还有一个功能是打印全部的Request对象：printAll()

void Inventory::printAll() const
    {
      muduo::MutexLockGuard lock(mutex_);
      sleep(1);
      for (std::set<Request*>::const_iterator it = requests_.begin(); it != requests_.end(); ++it)
      {
        (*it)->print();
      }
      printf("Inventory::printAll() unlocked\n");
    }

运行下面这个程序会产生死锁：

void threadFunc()
    {
      Request* req = new Request;
      req->process();
      delete req;       //~Request()
    }
    
    int main()
    {
      muduo::Thread thread(threadFunc);
      thread.start();
      usleep(500 * 1000);
      g_inventory.printAll();
      thread.join();
    }

main()线程先调用 **Inventory::printAll()** 再调用**Request::print()** ;

而threadFunc()线程，先调用 **Request::~Request()**, 再调用**Inventory::remove()**,这两个调用序列对两个mutex的加锁顺序正好相反，于是造成了经典的死锁。

解决方案很简单，要么把print()移出printAll()的临界区；要么把remove()移出~Request()的临界区，比如交换Request类中13行和15行的代码。

## 条件变量 ##

互斥器是加锁原语，用来排他性地访问共享数据，使用mutex时，我们希望立即拿到锁，然后尽快访问数据，用完之后尽快解锁，这样才能不影响并发性和性能。

先介绍moduo中condition的封装：

class Condition : boost::noncopyable
    {
        public:
            explicit Condition(MutexLock& mutex) : mutex_(mutex)
            {   pthread_cond_init(&pcond_,NULL);    }
            
            ~Condition() {  pthread_cond_destroy(&pcond_);  }
            void wait() {   pthread_cond_wait(&pcond_, mutex_.getPthreadMutex());   }
            void notify()   {   pthread_cond_signal(&pcond_ );  }
            void notifyAll()    {   pthread_cond_broadcast(&pcond_);    }
    
        private:
            MutexLock&  mutex_;
            pthread_cond_t  pcond_;
    };

如果需要等待某个条件成立，应该使用条件变量。条件变量的学名叫管程(monitor)。

对于wait端：

1.必须与mutex一起使用，该布尔表达式的读写需受此mutex保护

2.在mutex已上锁的时候才能调用wait()

3.把判断布尔条件和wait()放到while循环中

代码：

muduo::MutexLock mutex;
    muduo::Condition cond(mutex);
    std::deque<int> queue;
    
    int dequeue()
    {
        MutexLockGuard lock(mutex);
        while(queue.empty())                //必须用循环；必须在判断之后再wait()
        {
            cont.wait();        //这一步会原子地unlock mutex并进入等待。不会与enqueue死锁;wait()执行完毕后会自动重新加锁
        }
        assert(!queue.empty());             //判断条件是否变化
        int top = queue.front();
        queue.pop_front();
    }
    
    void enqueue(int x)
    {
        MutexLockGuard lock(mutex);
        queue.push_back(x);
        cond.notify();          //muduo:Condition采用了notify()和notifyAll()为函数名，避免重载signal这个术语(如上)
    }

上面代码中必须用while循环来等待条件变量，而不用if语句。原因是spurious wakeup。

条件变量是非常底层的同步原语，很少直接使用，一般用它来实现高层的同步措施，如CountDownLatch。

倒计时(CountDownLatch)是一种常用且易用的同步手段。主要有两种用途：

1.主线程发起多个子线程，等这些子线程各自都完成一定的任务之后，主线程才继续执行。通常用于主线程等待多个子线程完成初始化。

2.主线程发起多个子线程，子线程都等待主线程，主线程完成其他一些任务之后通知所有子线程开始执行。通常用于多个子线程等待主线程发出“起跑”命令。

CountDownLatch接口：

class CountDownLatch : boost::nonqcopyable
    {
        public:
            explicit CountDownLatch(int count);     //倒数几次
            void wait();                            //等待计数值变为0
            void countDown();                       //计数减1
            
        private:
            mutable Mutexlock mutex_;
            Condition condition_;
            int count_;
    };

CountDownLatch实现：

/* 构造函数：mutex_应先于condition_构造 */
    class CountDownLatch
    {
        public:
            CountDownLatch(int count): mutex_(),condition_(mutex_),count_(count) {}
        private:
            mutable MutexLock mutex_;       //顺序很重要，先mutex后condition
            Condition condition_;
            int count_:
    }
    
    void CountDownLatch::wait()
    {
        MutexLockGuard  lock(mutex_);
        while(count_ > 0)
            condition_.wait();
    }
    
    void CountDownLatch::countDown()
    {
        MutexLockGuard  lock(mutex_);
        --count_;
        if (count == 0)
            condition_.notyfiAll();
    }

互斥器和条件变量构成了多线程编程的全部必备同步原语，用它们即可完成任何多线程同步任务，二者不能相互替代。

sleep()不是同步原语

sleep只能出现在测试代码中。

生产代码中线程的等待可分为两种：一种是等待资源可用(要么等在select/poll/epoll\_wait上，要么等在条件变量)；一种是等着进入临界区(等在mutex上)以便读写共享数据。

不要使用下面这种业余做法：

while(true) {
        if (dataAvailable)
            sleep(some_time);
        else
            consumeData();
    }

等待某个事件的发生，正确的做法是用select()等价物或Condition,抑或高层同步工具；在用户态做轮询(polling)是低效的。