1. 程式人生 > >Muduo網路庫原始碼分析(三)執行緒間使用eventfd通訊和EventLoop::runInLoop系列函式

Muduo網路庫原始碼分析(三)執行緒間使用eventfd通訊和EventLoop::runInLoop系列函式

先說第一點,執行緒(程序)間通訊有很多種方式(pipe,socketpair),為什麼這裡選擇eventfd?

eventfd 是一個比 pipe 更高效的執行緒間事件通知機制,一方面它比 pipe 少用一個 file descripor,節省了資源;另一方面,eventfd 的緩衝區管理也簡單得多,全部“buffer” 只有定長8 bytes,不像 pipe 那樣可能有不定長的真正 buffer。

最重要的一點:當我們想要編寫併發型伺服器的時候,eventfd 可以完美取代 pipe去通知(喚醒)其他的程序(執行緒)。比如經典的非同步IO reactor/selector 應用場景,去喚醒select的呼叫。可以和事件通知機制完美的的結合。


(一)eventfd

#include <sys/eventfd.h>
int eventfd(unsigned int initval, intflags);
簡單的應用示例:
#include <sys/eventfd.h>  
#include <unistd.h>  
#include <stdlib.h>  
#include <stdio.h>  
#include <stdint.h>             /* Definition of uint64_t */  
  
#define handle_error(msg) \  
   do { perror(msg); exit(EXIT_FAILURE); } while (0)  
  
int  
main(int argc, char *argv[])  
{  
   uint64_t u;  
      
   int efd = eventfd(10, 0);  
   if (efd == -1)  
       handle_error("eventfd");  
     
   int ret = fork();  
   if(ret == 0)  
   {  
       for (int j = 1; j < argc; j++) {  
           printf("Child writing %s to efd\n", argv[j]);  
           u = atoll(argv[j]);  
           ssize_t s = write(efd, &u, sizeof(uint64_t));  
           if (s != sizeof(uint64_t))  
               handle_error("write");  
       }  
       printf("Child completed write loop\n");  
  
       exit(EXIT_SUCCESS);  
   }  
   else  
   {  
       sleep(2);  
  
       ssize_t s = read(efd, &u, sizeof(uint64_t));  
       if (s != sizeof(uint64_t))  
           handle_error("read");  
       printf("Parent read %llu from efd\n",(unsigned long long)u);  
       exit(EXIT_SUCCESS);  
   }  
}  

(二)EventLoop::loop、runInLoop、queueInLoop、doPendingFunctors

先看一下這四個函式總體的流程圖:



依次解釋:

// 該函式可以跨執行緒呼叫
void EventLoop::quit()
{
    quit_ = true;
    if (!isInLoopThread())
    {
        wakeup();
    }
}

//使用eventfd喚醒
void EventLoop::wakeup()
{
  uint64_t one = 1;
  //ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);
  ssize_t n = ::write(wakeupFd_, &one, sizeof one);
  if (n != sizeof one)
  {
    LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
  }
}
如果不是當前IO執行緒呼叫quit,則需要喚醒(wakeup())當前IO執行緒,因為它可能還阻塞在poll的位置(EventLoop::loop()),這樣再次迴圈判斷 while (!quit_) 才能退出迴圈。
// 事件迴圈,該函式不能跨執行緒呼叫
// 只能在建立該物件的執行緒中呼叫
void EventLoop::loop()
{// 斷言當前處於建立該物件的執行緒中
  assertInLoopThread();
    while (!quit_)
    {
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);

        eventHandling_ = true;
        for (ChannelList::iterator it = activeChannels_.begin();
                it != activeChannels_.end(); ++it)
        {
            currentActiveChannel_ = *it;
            currentActiveChannel_->handleEvent(pollReturnTime_);
        }
        currentActiveChannel_ = NULL;
        eventHandling_ = false;
       <span style="color:#ff0000;"> doPendingFunctors();</span>
    }
}

// 為了使IO執行緒在空閒時也能處理一些計算任務
// 在I/O執行緒中執行某個回撥函式,該函式可以跨執行緒呼叫
void EventLoop::runInLoop(const Functor& cb)
{
  if (isInLoopThread())
  {
    // 如果是當前IO執行緒呼叫runInLoop,則同步呼叫cb
    cb();
  }
  else
  {
    // 如果是其它執行緒呼叫runInLoop,則非同步地將cb新增到佇列,讓IO執行緒處理
    queueInLoop(cb);
  }
}

void EventLoop::queueInLoop(const Functor& cb)
{
  {
  MutexLockGuard lock(mutex_);
  pendingFunctors_.push_back(cb);
  }

  // 呼叫queueInLoop的執行緒不是當前IO執行緒則需要喚醒當前IO執行緒,才能及時執行doPendingFunctors();

  // 或者呼叫queueInLoop的執行緒是當前IO執行緒(比如在doPendingFunctors()中執行functors[i]() 時又呼叫了queueInLoop())
  // 並且此時正在呼叫pending functor,需要喚醒當前IO執行緒
  // 因為在此時doPendingFunctors() 過程中又添加了任務,故迴圈回去poll的時候需要被喚醒返回,進而繼續執行doPendingFunctors()

  // 只有當前IO執行緒的事件回撥中呼叫queueInLoop才不需要喚醒
 //  即在handleEvent()中呼叫queueInLoop 不需要喚醒,因為接下來馬上就會執行doPendingFunctors();
  if (!isInLoopThread() || callingPendingFunctors_)
  {
    wakeup();
  }
}

// 該函式只會被當前IO執行緒呼叫
void EventLoop::doPendingFunctors()
{
  std::vector<Functor> functors;
  callingPendingFunctors_ = true;

  {
  MutexLockGuard lock(mutex_);
  functors.swap(pendingFunctors_);
  }

  for (size_t i = 0; i < functors.size(); ++i)
  {
    functors[i]();
  }
  callingPendingFunctors_ = false;
}
關於doPendingFunctors 的補充說明 1、不是簡單地在臨界區內依次呼叫Functor,而是把回撥列表swap到functors中,這樣一方面減小了臨界區的長度(意味著不會阻塞其它執行緒的queueInLoop()),另一方面,也避免了死鎖(因為Functor可能再次呼叫queueInLoop()) 2、由於doPendingFunctors()呼叫的Functor可能再次呼叫queueInLoop(cb),這時,queueInLoop()就必須wakeup(),否則新增的cb可能就不能及時呼叫了 3、muduo沒有反覆執行doPendingFunctors()直到pendingFunctors_為空而是每次poll 返回就執行一次,這是有意的,否則IO執行緒可能陷入死迴圈,無法處理IO事件。 總結一下就是: 假設我們有這樣的呼叫:loop->runInLoop(run),說明想讓IO執行緒執行一定的計算任務,此時若是在當前的IO執行緒,就馬上執行run();如果是其他執行緒呼叫的,那麼就執行queueInLoop(run),將run非同步新增到佇列,當loop內處理完事件後,就執行doPendingFunctors(),也就執行到了run();最後想要結束執行緒的話,執行quit。
參考: 《linux多執行緒服務端程式設計》 http://blog.csdn.net/yusiguyuan/article/details/40593721?utm_source=tuicool&utm_medium=referral