Linux(muduo网络库):17---高效的多线程日志之（多线程异步日志、其他方案）

• 本文内容衔接于前一篇文章（日志概述、功能需求、性能需求）：https://blog.csdn.net/qq_41453285/article/details/105091022

一、多线程中的日志安全

• 多线程程序对日志库提出了新的需求：线程安全。即多个线程可以并发写日志，两个线程的日志消息不会出现交织
• 线程安全不难办到， 简单的办法是：
  • 用一个全局mutex保护IO
  • 或者每个线程单独写一个日志文件（Google C++日志库的默认多线程实现即如此）
• 但这两种做法的高效性就堪忧了。前者会造成全部线程抢一个锁，后者有可能让业务线程阻塞在写磁盘操作上

二、异步日志

异步日志概述

• 我认为一个多线程程序的每个进程最好只写一个日志文件，这样分析日志更容易，不必在多个文件中跳来跳去。再说多线程写多个文件也不一定能提速，见此处的分析
• 解决办法不难想到，用一个背景线程负责收集日志消息，并写入日志文件，其他业务线程只管往这个“日志线程”发送日志消息，这称为“异步日志”

应避免使用磁盘IO，而使用异步日志

• 在多线程服务程序中，异步日志（叫“非阻塞日志”似乎更准确）是必需的
• 原因如下：
  • 如果在网络IO线程或业务线程中直接往磁盘写数据的话， 写操作偶尔可能阻塞长达数秒之久（原因很复杂，可能是磁盘或磁盘控制器复位）
  • 这可能导致请求方超时，或者耽误发送心跳消息，在分布式系统中更可能造成多米诺骨牌效应，例如误报死锁引发自动failover等
• 因此，在正常的实时业务处理流程中应该彻底避免磁盘IO，这在使用one loop per thread模型的非阻塞服务端程序中尤为重要，因为线程是复用的，阻塞线程意味着影响多个客户连接

muduo异步日志库的实现（双缓冲技术）

• 我们需要一个“队列”来将日志前端的数据传送到后端（日志线程），但这个“队列”不必是现成的BlockingQueue<std::string>，因为不用每次产生一条日志消息都通知（notify()）接收方
• muduo日志库采用的是双缓冲（double buffering）技术（http://en.wikipedia.org/wiki/Multiple_buffering）
• 基本思路是：
  • 准备两块buffer：A和B，前端负责往buffer A填数据（日志消息），后端负责将buffer B的数据写入文件
  • 当buffer A写满之后，交换A和B。让前端则往buffer B填入新的日志消息，后端将buffer A的数据写入文件。如此往复...
• 用两个buffer的好处是：
  • 在新建日志消息的时候不必等待磁盘文件操作，也避免每条新日志消息都触发（唤醒）后端日志线程
  • 换言之，前端不是将一条条日志消息分别传送给后端，而是将多条日志消息拼成一个大的buffer传送给后端，相当于批处理，减少了线程唤醒的频度，降低开销
  • 另外，为了及时将日志消息写入文件，即便buffer A未满，日志库也会每3秒执行一次上述交换写入操作
• muduo异步日志的性能开销大约是：前端每写一条日志消息耗时1.0μs ～1.6μs

三、关键代码

• 实际实现采用了四个缓冲区，这样可以进一步减少或避免日志前端的等待
• 下面的代码摘录于muduo/base/AsyncLogginf.h

数据结构如下

   

    

     

      

     
     

      

       typedef muduo::detail::FixedBuffer<muduo::detail::kLargeBuffer> Buffer;
      
     
    

     

      

     
     

      

       typedef 
       std::
       vector<
       std::
       unique_ptr<Buffer>> BufferVector;
      
     
    

     

      

     
     

      

       typedef BufferVector::value_type             BufferPtr;
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       muduo::MutexLock mutex_;
      
     
    

     

      

     
     

      

       muduo::Condition cond_;
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       BufferPtr        currentBuffer_; 
       //当前缓冲
      
     
    

     

      

     
     

      

       BufferPtr        nextBuffer_;    
       //预备缓冲
      
     
    

     

      

     
     

      

       BufferVector     buffers_;       
       //待写入文件的已填满的缓冲
      
     
   

• Buffer类型是FixedBuffer class template的一份具体实现 （instantiation），其大小为4MB，可以存至少1000条日志消息
• BufferVector::value_type的类型为C++11中的std::unique_ptr，具备移动语义（move semantics），而且能自动管理对象生命期
• mutex_：用于保护后面的四个数据成员
• buffers_：存放的是供后端写入的buffer

发送方（前端）代码

• 这个函数时我们在前一篇文章（https://blog.csdn.net/qq_41453285/article/details/105091022）中介绍的回调函数output()的实现

   

    

     

      

     
     

      

       void AsyncLogging::append(
       const 
       char* logline, 
       int len)
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           muduo::
       MutexLockGuard lock(mutex_);
      
     
    

     

      

     
     

      
    
       //most common case: buffer is not full, copy data here
      
     
    

     

      

     
     

      
    
       if (currentBuffer_->avail() > len)                          
       //1
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      

               currentBuffer_->append(logline, len);                   
       //2
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      
    
       //buffer is full, push it, and find next spare buffer
      
     
    

     

      

     
     

      
    
       else
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      

               buffers_.push_back(
       std::move(currentBuffer_));          
       //3
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      
        
       if (nextBuffer_) 
       //is there is one already, use it //4
      
     
    

     

      

     
     

      

               {
      
     
    

     

      

     
     

      

                   currentBuffer_ = 
       std::move(nextBuffer_); 
       //移动，而非复制
      
     
    

     

      

     
     

      

               }
      
     
    

     

      

     
     

      
        
       else 
       //allocate a new one //5
      
     
    

     

      

     
     

      

               {
      
     
    

     

      

     
     

      

                   currentBuffer_.reset(
       new Buffer); 
       // Rarely happens
      
     
    

     

      

     
     

      

               }
      
     
    

     

      

     
     

      

               currentBuffer_->append(logline, len);                   
       //6
      
     
    

     

      

     
     

      

               cond_.notify();                                         
       //7
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      

       }
      
     
   

• 前端在生成一条日志消息的时候会调用AsyncLogging::append()
• 函数解释：
  • 如果当前缓冲currentBuffer_剩余的空间足够大 （代码段1处），则会直接把日志消息拷贝（追加）到当前缓冲中（代码段2处）， 这是最常见的情况。这里拷贝一条日志消息并不会带来多大开销。前后端代码的其余部分都没有拷贝，而是简单的指针交换
  • 否则，说明当前缓冲已经写满，就把它送入/移入buffers_（代码段3处），并试图把预备好的另一块缓冲nextBuffer_移用 （move）为当前缓冲（代码段4处），然后追加日志消息并通知/唤醒后端开始写入日志数据（代码段6、7处）
  • 以上两种情况在临界区之内都没有耗时的操作，运行时间为常数
  • 如果前端写入速度太快，一下子把两块缓冲都用完了，那么只好分配一块新的buffer，作为当前缓冲（代码段5处），这是极少发生的情况

发送方（后端）代码

• 下面只给出了threadFunc()函数最关键的临界区代码，其他代码参阅源文件

   

    

     

      

     
     

      

       void AsyncLogging::threadFunc()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      
    
       //...
      
     
    

     

      

     
     

      
    
       BufferPtr newBuffer1(new Buffer);  
       //8
      
     
    

     

      

     
     

      
    
       BufferPtr newBuffer2(new Buffer);  
       //9
      
     
    

     

      

     
     

      
    
       //...
      
     
    

     

      

     
     

      

           BufferVector buffersToWrite;    
      
     
    

     

      

     
     

      

           buffersToWrite.reserve(
       16);
      
     
    

     

      

     
     

      
    
       while (running_)
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      
        
       //...
      
     
    

     

      

     
     

      
        
       //swap out what need to be written, keep CS short
      
     
    

     

      

     
     

      

               {
      
     
    

     

      

     
     

      

                   muduo::
       MutexLockGuard lock(mutex_);
      
     
    

     

      

     
     

      
            
       if (buffers_.empty())  
       // unusual usage!
      
     
    

     

      

     
     

      

                   {
      
     
    

     

      

     
     

      

                       cond_.waitForSeconds(flushInterval_);      
       //10
      
     
    

     

      

     
     

      

                   }
      
     
    

     

      

     
     

      

                   buffers_.push_back(
       std::move(currentBuffer_));
       //移动，而非复制 11
      
     
    

     

      

     
     

      

                   currentBuffer_ = 
       std::move(newBuffer1);       
       //移动，而非复制 12
      
     
    

     

      

     
     

      

                   buffersToWrite.swap(buffers_);                
       //内部指针交换，而非复制 13
      
     
    

     

      

     
     

      
            
       if (!nextBuffer_)                             
       //14
      
     
    

     

      

     
     

      

                   {
      
     
    

     

      

     
     

      

                       nextBuffer_ = 
       std::move(newBuffer2);      
       //移动，而非复制
      
     
    

     

      

     
     

      

                   }
      
     
    

     

      

     
     

      

               }
      
     
    

     

      

     
     

      
        
       //...
      
     
    

     

      

     
     

      
        
       //output buffersToWrite to file 15
      
     
    

     

      

     
     

      
        
       //re-fill newBuffer1 and newBuffer2 16
      
     
    

     

      

     
     

      

           }
      
     
    

     

      

     
     

      
    
       //flush output...
      
     
    

     

      

     
     

      

       }
      
     
   

• 函数解释：
  • 首先准备好两块空闲的buffer，以备在临界区内交换（代码段8、9处）
  • 在临界区内，等待条件触发（代码段10处）。这里的条件有两个：
    • 其一是超时，其二是前端写满了一个或多个buffer
    • 注意这里是非常规的condition variable用法，它没有使用while循环，而且等待时间有 上限。
  • 当“条件”满足时，先将当前缓冲（currentBuffer_）移入 buffers_（代码段11处），并立刻将空闲的newBuffer1移为当前缓冲（代码段12处）。注意这整段代码位于临界区之内，因此不会有任何race condition
  • 接下来将buffers_与buffersToWrite交换（代码段13处），后面的代码可以在临界区之外安全地访问buffersToWrite，将其中的日志数据写入文件（代码段15处）
  • 临界区里最后干的一件事情是用newBuffer2替换nextBuffer_（代码段14处），这样前端始终有一个预备buffer可供调配。nextBuffer_可以减少前端临界区分配内存的概率，缩短前端临界区长度
  • 注意到后端临界区内也没有耗时的操作，运行时间为常数
  • 代码段16处会将buffersToWrite内的buffer重新填充newBuffer1和 newBuffer2，这样下一次执行的时候还有两个空闲buffer可用于替换前端的当前缓冲和预备缓冲。最后，这四个缓冲在程序启动的时候会全部填充为0，这样可以避免程序热身时page fault引发性能不稳定

四、运行图示

• 以下再用图表展示前端和后端的具体交互情况。一开始先分配好四 个缓冲区A、B、C、D，前端和后端各持有其中两个。前端和后端各有 一个缓冲区数组，初始时都是空的。

第一种情况

• 第一种情况是前端写日志的频度不高，后端3秒超时后将“当前缓冲 currentBuffer_”写入文件。如下图所示（图中变量名为简写，下同）：

• 执行过程如下：
  • 在第2.9秒的时候，currentBuffer_使用了80％
  • 在第3秒的时候后端线程醒过来，先把currentBuffer_送入buffers_（代码段11处），再把newBuffer1移用为currentBuffer_（代码段12处）
  • 随后第3+秒，交换buffers_和buffersToWrite（代码段13处），离开临界区，后端开始将buffer A写入文件
  • 写完（write done）之后再把newBuffer1重新填上，等待下一次cond_.waitForSeconds()返回

• 后面在画图时将有所简化，不再画出buffers_和buffersToWrite交换 的步骤。

第二种情况

• 第二种情况，在3秒超时之前已经写满了当前缓冲，于是唤醒后端线程开始写入文件，如下图所示：

• 执行过程如下：
  • 在第1.5秒的时候，currentBuffer_使用了80％
  • 第1.8秒， currentBuffer_写满，于是将当前缓冲送入buffers_（代码段3处），并将 nextBuffer_移用为当前缓冲（代码段4处），然后唤醒后端线程开始写 入
  • 当后端线程唤醒之后（第1.8+秒），先将currentBuffer_送入 buffers_（代码段11处），再把newBuffer1移用为currentBuffer_（代码段12处），然后交换buffers_和buffersToWrite（代码段13处），最后用newBuffer2替换 nextBuffer_（代码段14处），即保证前端有两个空缓冲可用
  • 离开临界区之后，将buffersToWrite中的缓冲区A和B写入文件，写完之后重新填充newBuffer1和newBuffer2，完成一次循环

• 上面这两种情况都是最常见的，再来看一看前端需要分配新buffer 的两种情况。

第三种情况

• 第三种情况，前端在短时间内密集写入日志消息，用完了两个缓冲，并重新分配了一块新的缓冲，如下图所示：

• 执行过程如下：
  • 在第1.8秒的时候，缓冲A已经写满，缓冲B也接近写满，并且已经 notify()了后端线程，但是出于种种原因，后端线程并没有立刻开始工 作
  • 到了第1.9秒，缓冲B也已经写满，前端线程新分配了缓冲E
  • 到了第1.8+秒，后端线程终于获得控制权，将C、D两块缓冲交给前端，并开始将A、B、E依次写入文件。一段时间之后，完成写入操作，用A、 B重新填充那两块空闲缓冲
  • 注意这里有意用A和B来填充 newBuffer1/2，而释放了缓冲E，这是因为使用A和B不会造成page fault
• 思考题：阅读代码并回答，缓冲E是何时在哪个线程释放的？

第四种情况

• 第四种情况，文件写入速度较慢，导致前端耗尽了两个缓冲，并分配了新缓冲，如下图所示：

• 执行过程如下：
  • 前1.8+秒的场景和前面“第二种情况”相同，前端写满了一个缓冲， 唤醒后端线程开始写入文件
  • 之后，后端花了较长时间（大半秒）才将数据写完。这期间前端又用完了两个缓冲，并分配了一个新的缓冲，这期间前端的notify()已经丢失
  • 当后端写完（write done）后，发现buffers_不为空（代码段10处的if语句），立刻进入下一循环。即替换前端的两个缓冲， 并开始一次写入C、D、E
  • 假定前端在此期间产生的日志较少，请读者补全后续的情况

五、改进措施

• 前面我们一共准备了四块缓冲，应该足以应付日常的需求。如果需要进一步增加buffer数目，可以改用下面的数据结构：

  

   

    

     

    
    

     

      BufferPtr        currentBuffer_; 
      //当前缓冲
     
    
   

    

     

    
    

     

      BufferVector     emptyBuffers_;  
      //空闲缓冲
     
    
   

    

     

    
    

     

      BufferVector     fullBuffers_;   
      //已写满的缓冲
     
    
  

使用方法

• 步骤如下：
  • 初始化时在emptyBuffers_中放入足够多空闲buffer，这样前端几乎不会遇到需要在临界区内新分配buffer的情况，这是一种空间换时间的做法
  • 为了避免短时突发写大量日志造成新分配的buffer占用过多内存，后端代码应该保证emptyBuffers_和fullBuffers_的长度之和不超过某个定值
  • buffer在前端和后端之间流动，形成一个循环，如下图所示

• 这种方式muduo没有实现，读者可以自己实现

六、如果日志消息堆积怎么办？

• 万一前端陷入死循环，拼命发送日志消息，超过后端的处理（输出）能力，会导致什么后果？
  • 对于同步日志来说：这不是问题，因为阻塞IO自然就限制了前端的写入速度，起到了节流阀（throttling）的作用
  • 但是对于异步日志来说：这就是典型的生产速度高于消费速度问题，会造成数据在内存中堆积，严重时引发性能问题（可用内存不足） 或程序崩溃（分配内存失败）
• muduo日志库处理日志堆积的方法很简单：
  • 直接丢掉多余的日志buffer，以腾出内存，见muduo/base/AsyncLogging.cc第92~101行代码
  • 这样可以防止日志库本身引起程序故障，是一种自我保护措施
  • 将来或许可以加上网络报警功能，通知人工介入，以尽快修复故障

七、其他方案

直接传递指针

• 当然在前端和后端之间高效传递日志消息的办法不止这一种：
  • 比方说使用常规的muduo::BlockingQueue<std::string>或 muduo::BoundedBlockingQueue<std::string>在前后端之间传递日志消息，其中每个std::string是一条消息
  • 这种做法每条日志消息都要分配内存，特别是在前端线程分配的内存要由后端线程释放，因此对malloc的实现要求较高，需要针对多线程特别优化
  • 另外，如果用这种方案，那么需要修改LogStream的Buffer，使之直接将日志写到std::string中，可节省一次内存拷贝
• 相比前面展示的直接拷贝日志消息的做法，这个传递指针的方案似乎会更高效，但是据我测试（代码参阅recipes/logging/AsyncLogging*），直接拷贝日志数据的做法比传递指针快3倍（在每条日志消息不大于4kB的时候），估计是内存分配的开销所致
• 因此muduo日志库只提供了上面介绍的这一种异步日志机制。这再次说明“性能”不能凭感觉说了算，一定要有典型场景的测试数据作为支撑

全局锁

• muduo现在的异步日志实现用了一个全局锁。尽管临界区很小，但是如果线程数目较多，锁争用（lock contention）也可能影响性能
• 一种解决办法是像Java的ConcurrentHashMap那样用多个桶子（bucket），前端写日志的时候再按线程id哈希到不同的bucket中，以减少contention。 这种方案的后端实现较为复杂，有兴趣的读者可以试一试

• 为了简化实现，目前muduo日志库只允许指定日志文件的名字，不允许指定其路径。日志库会把日志文件写到当前路径，因此可以在启动脚本（shell脚本）里改变当前路径，以达到相同的目的
• Linux默认会把core dump写到当前目录，而且文件名是固定的core。为了不让新的core dump文件冲掉旧的，我们可以通过sysctl设置kernel.core_pattern参数（也可以修改/rpoc/sys/kernel/core_pattern），让每次core dump都产生不同的文件。例如设为%e.%t.%p.%u.core，其中各个参数的意义见man 5 core。另外也可以使用Apport来收集有用的诊断信 息，见http://wiki.ubuntu.com/Apport