thrift之TTransport層的堵塞的套接字I/O傳輸類TSocket


本節將介紹第一個實現具體傳輸功能的類TSocket,這個類是基於TCP socket實現TTransport的接口。下面具體介紹這個類的相關函數功能實現。
  1.構造函數
  分析一個類的功能首先看它的定義和構造函數實現,先看看它的定義:

class TSocket : public TVirtualTransport<TSocket> { ......}

  由定義可以看書TSocket繼承至虛擬傳輸類,並且把自己當做模板參數傳遞過去,所以從虛擬傳輸類繼承下來的虛擬函數(如read_virt)調用非虛擬函數(如read)就是TSocket自己實現的。
  TSocket類的構造函數有4個,當然還有一個析構函數。四個構造函數就是根據不同的參數來構造,它們的聲明如下:

  TSocket();//所有參數都默認
  TSocket(std::string host, int port);//根據主機名和端口構造一個socket
  TSocket(std::string path);//構造unix域的一個socket
  TSocket(int socket);//構造一個原始的unix句柄socket

  四個構造函數分別用於不同的情況下來產生不同的TSocket對象,不過這些構造函數都只是簡單的初始化一些最基本的成員變量,而沒有真正的連接socket。它們初始化的變量基本如下:

  TSocket::TSocket() :
    host_(""),
    port_(0),
    path_(""),
    socket_(-1),
    connTimeout_(0),
    sendTimeout_(0),
    recvTimeout_(0),
    lingerOn_(1),
    lingerVal_(0),
    noDelay_(1),
    maxRecvRetries_(5) {
    recvTimeval_.tv_sec = (int)(recvTimeout_/1000);
    recvTimeval_.tv_usec = (int)((recvTimeout_%1000)*1000);
    cachedPeerAddr_.ipv4.sin_family = AF_UNSPEC;
  }

  大部分簡單的參數都采用初始化列表初始化了,需要簡單計算的就放在函數體內初始化,其他幾個都是這種情況。下面需要單獨介紹一下的是unix domain socket。
  socket API原本是為網絡通訊設計的,但后來在socket的框架上發展出一種IPC機制,就是UNIX Domain Socket。雖然網絡socket也可用於同一台主機的進程間通訊(通過loopback地址127.0.0.1),但是UNIX Domain Socket用於IPC更有效率:不需要經過網絡協議棧,不需要打包拆包、計算校驗和、維護序號和應答等,只是將應用層數據從一個進程拷貝到另一個進程。這是因為,IPC機制本質上是可靠的通訊,而網絡協議是為不可靠的通訊設計的。UNIX Domain Socket也提供面向流和面向數據包兩種API接口,類似於TCP和UDP,但是面向消息的UNIX Domain Socket也是可靠的,消息既不會丟失也不會順序錯亂。
  UNIX Domain Socket是全雙工的,API接口語義豐富,相比其它IPC機制有明顯的優越性,目前已成為使用最廣泛的IPC機制,比如X Window服務器和GUI程序之間就是通過UNIX Domain Socket通訊的。
  使用UNIX Domain Socket的過程和網絡socket十分相似,也要先調用socket()創建一個socket文件描述符,address family指定為AF_UNIX,type可以選擇SOCK_DGRAM或SOCK_STREAM,protocol參數仍然指定為0即可。
  UNIX Domain Socket與網絡socket編程最明顯的不同在於地址格式不同,用結構體sockaddr_un表示,網絡編程的socket地址是IP地址加端口號,而UNIX Domain Socket的地址是一個socket類型的文件在文件系統中的路徑,這個socket文件由bind()調用創建,如果調用bind()時該文件已存在,則bind()錯誤返回。
  打開連接函數open
  首先看這個函數的代碼實現,如下:

  void TSocket::open() {
    if (isOpen()) {//如果已經打開就直接返回
      return;
    }
    if (! path_.empty()) {//如果unix路徑不為空就打開unix domian socket
      unix_open();
    } else {
      local_open();//打開通用socket
    }
  }

  Open函數又根據路徑為不為空(不為空就是unix domain socket)調用相應的函數來繼續打開連接,首先看看打開unix domain socket,代碼如下:

  void TSocket::unix_open(){
    if (! path_.empty()) {//保證path_不為空
      // Unix Domain SOcket does not need addrinfo struct, so we pass NULL
      openConnection(NULL);//調用真正的打開連接函數
    }
  }

  由代碼可以看出,真正實現打開連接的函數是openConnection,這個函數根據傳遞的參數來決定是否是打開unix domain socket,實現代碼如下(這個函數代碼比較多,其中除了錯誤部分代碼省略):

  void TSocket::openConnection(struct addrinfo *res) {
    if (isOpen()) {
      return;//如果已經打開了直接返回
    }
    if (! path_.empty()) {//根據路徑是否為空創建不同的socket
      socket_ = socket(PF_UNIX, SOCK_STREAM, IPPROTO_IP);//創建unix domain socket
    } else {
      socket_ = socket(res->ai_family, res->ai_socktype, res->ai_protocol);//創建通用的網絡通信socket
    }
    if (sendTimeout_ > 0) {//如果發生超時設置大於0就調用設置發送超時函數設置發送超時
      setSendTimeout(sendTimeout_);
    }
    if (recvTimeout_ > 0) {//如果接收超時設置大於0就調用設置接收超時函數設置接收超時
      setRecvTimeout(recvTimeout_);
    }
    setLinger(lingerOn_, lingerVal_);//設置優雅斷開連接或關閉連接參數
    setNoDelay(noDelay_);//設置無延時
  #ifdef TCP_LOW_MIN_RTO
    if (getUseLowMinRto()) {//設置是否使用較低的最低TCP重傳超時 
      int one = 1;
      setsockopt(socket_, IPPROTO_TCP, TCP_LOW_MIN_RTO, &one, sizeof(one));
    }
  #endif
    //如果超時已經存在設置連接為非阻塞
    int flags = fcntl(socket_, F_GETFL, 0);//得到socket_的標識
    if (connTimeout_ > 0) {//超時已經存在
      if (-1 == fcntl(socket_, F_SETFL, flags | O_NONBLOCK)) {//設置為非阻塞
      }
    } else {
      if (-1 == fcntl(socket_, F_SETFL, flags & ~O_NONBLOCK)) {//設置為阻塞
      }
    }
    // 連接socket
    int ret;
    if (! path_.empty()) {//unix domain socket
  #ifndef _WIN32 //window不支持
      struct sockaddr_un address;
      socklen_t len;
      if (path_.length() > sizeof(address.sun_path)) {//path_長度不能超過最長限制
      }
      address.sun_family = AF_UNIX;
      snprintf(address.sun_path, sizeof(address.sun_path), "%s", path_.c_str());
      len = sizeof(address);
      ret = connect(socket_, (struct sockaddr *) &address, len);//連接unix domain socket
  #else
        //window不支持unix domain socket
  #endif
  
    } else {
      ret = connect(socket_, res->ai_addr, res->ai_addrlen);//連接通用的非unix domain socket
    }
    if (ret == 0) {//失敗了就會執行后面的代碼,用poll來監聽寫事件
      goto done;//成功了就直接跳轉到完成處
    }
    struct pollfd fds[1];//定於用於poll的描述符
    std::memset(fds, 0 , sizeof(fds));//初始化為0
    fds[0].fd = socket_;//描述符為socket
    fds[0].events = POLLOUT;//接收寫事件
    ret = poll(fds, 1, connTimeout_);//調用poll,有一個超時值
    if (ret > 0) {
      // 確保socket已經被連接並且沒有錯誤被設置
      int val;
      socklen_t lon;
      lon = sizeof(int);
      int ret2 = getsockopt(socket_, SOL_SOCKET, SO_ERROR, cast_sockopt(&val), &lon);//得到錯誤選項參數
      if (val == 0) {// socket沒有錯誤也直接到完成處了
        goto done;
      }
    } else if (ret == 0) {// socket 超時
      //相應處理代碼省略
    } else {
      // poll()出錯了,相應處理代碼省略
    }
  
   done:
    fcntl(socket_, F_SETFL, flags);//設置socket到原來的模式了(阻塞)
    if (path_.empty()) {//如果是unix domain socket就設置緩存地址
      setCachedAddress(res->ai_addr, res->ai_addrlen);
    }
  }

  上面這個函數代碼確實比較長,不過還好都是比較簡單的代碼實現,沒有什么很繞的代碼,整個流程也很清晰,在代碼中也有比較詳細的注釋了。下面繼續看通用socket打開函數local_open(它也真正的執行打開功能也是調用上面剛才介紹的那個函數,只是傳遞了具體的地址信息):

  void TSocket::local_open(){
  #ifdef _WIN32
      TWinsockSingleton::create();//兼容window平台
  #endif // _WIN32
    if (isOpen()) {//打開了就直接返回
      return;
    }
    if (port_ < 0 || port_ > 0xFFFF) {//驗證端口是否為有效值
      throw TTransportException(TTransportException::NOT_OPEN, "Specified port is invalid");
    }
    struct addrinfo hints, *res, *res0;
    res = NULL;
    res0 = NULL;
    int error;
    char port[sizeof("65535")];
    std::memset(&hints, 0, sizeof(hints));//內存設置為0
    hints.ai_family = PF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
    sprintf(port, "%d", port_);
    error = getaddrinfo(host_.c_str(), port, &hints, &res0);//根據主機名得到所有網卡地址信息
    // 循環遍歷所有的網卡地址信息,直到有一個成功打開
    for (res = res0; res; res = res->ai_next) {
      try {
        openConnection(res);//調用打開函數
        break;//成功就退出循環
      } catch (TTransportException& ttx) {
        if (res->ai_next) {//異常處理,是否還有下一個地址,有就繼續
          close();
        } else {
          close();
          freeaddrinfo(res0); // 清除地址信息內存和資源
          throw;//拋出異常
        }
      }
    }
    freeaddrinfo(res0);//釋放地址結構內存
  }

  整個local_open函數就是根據主機名得到所有的網卡信息,然后依次嘗試打開,直到打開一個為止就退出循環,如果所有都不成功就拋出一個異常信息。
  讀函數read
  在實現讀函數的時候需要注意區分返回錯誤為EAGAIN的情況,因為當超時和系統資源耗盡都會產生這個錯誤(沒有明顯的特征可以區分它們),所以Thrift在實現的時候設置一個最大的嘗試次數,如果超過這個了這個次數就認為是系統資源耗盡了。下面具體看看read函數的實現,代碼如下(省略一些參數檢查和錯誤處理的代碼):

  uint32_t TSocket::read(uint8_t* buf, uint32_t len) {
    int32_t retries = 0;//重試的次數
    uint32_t eagainThresholdMicros = 0;
    if (recvTimeout_) {//如果設置了接收超時時間,那么計算最大時間間隔來判斷是否系統資源耗盡
      eagainThresholdMicros = (recvTimeout_*1000)/ ((maxRecvRetries_>0) ? maxRecvRetries_ : 2);
    }
   try_again:
    struct timeval begin;
    if (recvTimeout_ > 0) {
      gettimeofday(&begin, NULL);//得到開始時間
    } else {
      begin.tv_sec = begin.tv_usec = 0;//默認為0,不需要時間來判斷是超時了
    }
    int got = recv(socket_, cast_sockopt(buf), len, 0);//從socket接收數據
    int errno_copy = errno; //保存錯誤代碼
    ++g_socket_syscalls;//系統調用次數統計加1
    if (got < 0) {//如果讀取錯誤
      if (errno_copy == EAGAIN) {//是否為EAGAIN
        if (recvTimeout_ == 0) {//如果沒有設置超時時間,那么就是資源耗盡錯誤了!拋出異常
          throw TTransportException(TTransportException::TIMED_OUT, "EAGAIN (unavailable resources)");
        }
        struct timeval end;
        gettimeofday(&end, NULL);//得到結束時間,會改變errno,所以前面需要保存就是這個原因
        uint32_t readElapsedMicros =  (((end.tv_sec - begin.tv_sec) * 1000 * 1000)//計算消耗的時間
                                       + (((uint64_t)(end.tv_usec - begin.tv_usec))));
        if (!eagainThresholdMicros || (readElapsedMicros < eagainThresholdMicros)) {
          if (retries++ < maxRecvRetries_) {//重試次數還小於最大重試次數
            usleep(50);//睡眠50毫秒
            goto try_again;//再次嘗試從socket讀取數據
          } else {//否則就認為是資源不足了
            throw TTransportException(TTransportException::TIMED_OUT, "EAGAIN (unavailable resources)");
          }
        } else {//推測為超時了
          throw TTransportException(TTransportException::TIMED_OUT, "EAGAIN (timed out)");
        }
      }
      if (errno_copy == EINTR && retries++ < maxRecvRetries_) {//如果是中斷並且重試次數沒有超過
        goto try_again;//那么重試
      }
      #if defined __FreeBSD__ || defined __MACH__
      if (errno_copy == ECONNRESET) {//FreeBSD和MACH特殊處理錯誤代碼
        return 0;
      }
      #endif
  #ifdef _WIN32
      if(errno_copy == WSAECONNRESET) {//win32平台處理錯誤代碼
        return 0; // EOF
      }
  #endif
    return got;
  }

  整個讀函數其實沒有什么特別的,主要的任務就是錯誤情況的處理,從這里可以看出其實實現一個功能是很容易的,但是要做到穩定和容錯性確實需要發很大功夫。
  寫函數write
  寫函數和讀函數實現差不多,主要的代碼還是在處理錯誤上面,還有一點不同的是寫函數寫的內容可能一次沒有發送完畢,所以是在一個while循環中一直發送直到指定的內容全部發送完畢。代碼實現如下:

  void TSocket::write(const uint8_t* buf, uint32_t len) {
    uint32_t sent = 0;//記錄已經發送了的字節數
    while (sent < len) {//是否已經發送了指定的字節長度
      uint32_t b = write_partial(buf + sent, len - sent);//調部分寫入函數
      if (b == 0) {//發送超時過期了
        throw TTransportException(TTransportException::TIMED_OUT, "send timeout expired");
      }
      sent += b;//已經發送的字節數
    }
  }
  上面的函數還沒有這種的調用send函數發送寫入的內容,而是調用部分寫入函數write_partial寫入,這個函數實現如下:
  uint32_t TSocket::write_partial(const uint8_t* buf, uint32_t len) {
    uint32_t sent = 0;
    int flags = 0;
  #ifdef MSG_NOSIGNAL 
    //使用這個代替SIGPIPE 錯誤,代替我們檢查返回EPIPE錯誤條件和關閉socket的情況
    flags |= MSG_NOSIGNAL;//設置這個標志位
  #endif 
    int b = send(socket_, const_cast_sockopt(buf + sent), len - sent, flags);//發送數據
    ++g_socket_syscalls;//系統調用計數加1
    if (b < 0) { //錯誤處理
      if (errno == EWOULDBLOCK || errno == EAGAIN) {
        return 0;//應該阻塞錯誤直接返回
      }
      int errno_copy = errno;//保存錯誤代碼
      if (errno_copy == EPIPE || errno_copy == ECONNRESET || errno_copy == ENOTCONN) {
        close();//連接錯誤關閉掉socket
      }
    }
    return b;//返回寫入的字節數
  }

  這個寫入的實現邏輯和過程也是非常簡單的,只是需要考慮到各種錯誤的情況並且相應的處理之。
  其他函數
  TSocket類還有一些其他函數,不過功能都比較簡單,比如設置一些超時和得到一些成員變量值的函數,哪些函數一般都是幾句代碼完成了。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM