muduo網絡庫學習筆記(五) 連接器Connector與監聽器Acceptor

目錄

• muduo網絡庫學習筆記(五) 連接器Connector與監聽器Acceptor
  • Connector
    • 系統函數connect
    • 處理非阻塞connect的步驟:
    • Connetor時序圖
  • Acceptor
    • 系統函數accept
    • Socket的封裝
    • Acceptor的封裝
    • Acceptor時序圖.
  • 簡單測試程序
    • Acceptor
    • Connctor
    • 運行日誌

muduo網絡庫學習筆記(五) 連接器Connector與監聽器Acceptor

標籤： muduo Connector Acceptor

---

本篇繼續爲前面封裝的EventLoop添加事件，到如今共給EventLoop添加了兩個fd,Timerfd，EventFd分別用於處理定時任務和通知事件.
今天添加的Acceptor會增長另外一個fd,此fd是是一個socket，用於監聽套接字鏈接.同時封裝非組賽網絡編程中的connect(2)的使用Connector.

Connector

在非阻塞網絡編程中，發起鏈接的基本方式是調用connect(2)，當socket變得可寫時代表鏈接創建完畢，其中要處理各類類型的錯誤，咱們把它封裝爲Connector class.
Connector 和 Acceptor 設計思路基本一致，只是Acceptor經過判斷套接字是否可讀來執行回調，而Connector是判斷套接字是否可寫來執行回調.
還有一點就是錯誤處理，socket可寫不必定就是鏈接創建好了 ， 當鏈接創建出錯時，套接口描述符變成既可讀又可寫,這時咱們能夠經過調用getsockopt來獲得套接口上待處理的錯誤(SO_ERROR).

其次非阻塞網絡編程中connect(2)的sockfd是一次性的，一旦出錯（好比對方拒絕鏈接），就沒法恢復，只能關閉重來。但Connector是能夠反覆使用的， 所以每次嘗試鏈接都要使用新的socket文件描述符和新的Channel對象。要注意的就是Channel的生命期管理了.

系統函數connect

#include <sys/types.h>          /* See NOTES */
   #include <sys/socket.h>
   
   int connect(int sockfd, const struct sockaddr *addr,
               socklen_t addrlen);

sockfd 試圖製做的一個鏈接到被綁定到addr指定地址的套接字。
addr和addrlen 服務端地址和長度.

retrun:
成功 返回0 ， 失敗 返回 -1.

處理非阻塞connect的步驟:

第一步:建立非阻塞socket,返回套接口描述符;
第二步:connect(2)開始創建鏈接;
第三步:判斷鏈接是否成功創建：

A:若是connect返回0,表示鏈接創建成功, 若是錯誤爲EINPROGRESS 表示鏈接正在進行，能夠等待select()變的可寫，經過getsockopt()來來獲得套接口上待處理的錯誤(SO_ERROR),鏈接是否創建成功.若是鏈接創建成功,這個錯誤值將是0,若是創建鏈接時遇到錯誤,則這個值是鏈接錯誤所對應的errno值(好比:ECONNREFUSED,ETIMEDOUT等).
B: EAGAIN、EADDRINUSE、EADDRNOTAVAIL、ECONNREFUSED、ENETUNREACH 像EAGAIN 這類代表本機臨時端口暫時用完的錯誤、能夠嘗試重連。
C: EACCES、EPERM、EAFNOSUPPORT、EALREADY、EBADF、EFAULT、ENOTSOCK 其餘真錯誤像無權限，協議錯誤，等直接關閉套接字.

Connector正是按這個步驟處理的鏈接.
暴露的接口只有start()和stop()
start()執行上述connect的步驟.
stop()關閉套接字，刪除註冊的通道，中止進行鏈接.

class Connector
{
public:
  typedef std::function<void (int sockfd)> NewConnectionCallback;

  Connector(EventLoop* loop, const InetAddress& serverAddr);
  ~Connector();

  void setNewConnectionCallback(const NewConnectionCallback& cb)
  { m_newConnectionCallBack = cb; }

  void start();// can be called in any thread
  void stop(); // can be called in any thread

private:

  enum States { kDisconnected, kConnecting, kConnected };
  static const int kMaxRetryDelayMs = 30*1000;
  static const int kInitRetryDelayMs = 500;

  void connect();
  void connecting(int sockfd);

  void handleWrite();
  void handleError();

  void retry(int sockfd);
  int removeAndResetChannel();
  void resetChannel();

  void setState(States s) { m_state = s; }
  void startInLoop();
  void stopInLoop();

  EventLoop* p_loop;
  int m_retryDelayMs;
  InetAddress m_serverAddr;

  States m_state;

  std::unique_ptr<Channel> p_channel;
  NewConnectionCallback m_newConnectionCallBack;
};

Connetor時序圖

Connector::Connector(EventLoop* loop, const InetAddress& serverAddr)
  :p_loop(loop),
  m_serverAddr(serverAddr),
  m_state(kDisconnected),
  m_retryDelayMs(kInitRetryDelayMs)
{
  LOG_DEBUG << "ctor[" << this << "]";
}

Connector::~Connector()
{
  LOG_DEBUG << "dtor[" << this << "]";
  assert(!p_channel);
}

void Connector::start()
{

  p_loop->runInLoop(std::bind(&Connector::startInLoop, this));
}

void Connector::startInLoop()
{
  p_loop->assertInLoopThread();
  assert(m_state == kDisconnected);

  connect();
}

void Connector::stop()
{
  p_loop->queueInLoop(std::bind(&Connector::stopInLoop, this));
}

void Connector::stopInLoop()
{
  p_loop->assertInLoopThread();

  if(m_state == kConnecting)
  {
    int sockfd = removeAndResetChannel();
    sockets::close(sockfd);
    setState(kDisconnected);
  }
}

void Connector::connect()
{
  int sockfd = sockets::createNonblockingOrDie(m_serverAddr.family());
  int ret = sockets::connect(sockfd, m_serverAddr.getSockAddr());
  int savedErrno = (ret == 0) ? 0 : errno;

  if(ret != 0) LOG_TRACE << "connect error ("<< savedErrno << ") : " << strerror_tl(savedErrno);

  switch(savedErrno)
  {
    case 0:
    case EINPROGRESS:      //Operation now in progress
    case EINTR:            //Interrupted system call
    case EISCONN:          //Transport endpoint is already connected
      connecting(sockfd);
      break;

    case EAGAIN:
    case EADDRINUSE:
    case EADDRNOTAVAIL:
    case ECONNREFUSED:
    case ENETUNREACH:
      retry(sockfd);
      LOG_SYSERR << "reSave Error. " << savedErrno;
      break;

    case EACCES:
    case EPERM:
    case EAFNOSUPPORT:
    case EALREADY:
    case EBADF:
    case EFAULT:
    case ENOTSOCK:
      LOG_SYSERR << "connect error in Connector::startInLoop " << savedErrno;
      sockets::close(sockfd);
      break;

    default:
      LOG_SYSERR << "Unexpected error in Connector::startInLoop " << savedErrno;
      sockets::close(sockfd);
      // connectErrorCallback_();
      break;
  }

}

void Connector::connecting(int sockfd)
{
  LOG_TRACE << "Connector::connecting] sockfd : " << sockfd;
  setState(kConnecting);
  assert(!p_channel);
  p_channel.reset(new Channel(p_loop, sockfd));
  p_channel->setWriteCallBack(std::bind(&Connector::handleWrite, this));
  //p_channel->setErrorCallback()

  //enableWriting if Channel Writeable ,Connect Success.
  p_channel->enableWriting();
}

void Connector::retry(int sockfd)
{
  sockets::close(sockfd);
  setState(kDisconnected);

  LOG_INFO << "Connector::retry - Retry connecting to " << m_serverAddr.toIpPort()
           << " in " << m_retryDelayMs << " milliseconds. ";

  p_loop->runAfter(m_retryDelayMs/1000.0, std::bind(&Connector::startInLoop, this));
  m_retryDelayMs = std::min(m_retryDelayMs * 2, kMaxRetryDelayMs);
}

int Connector::removeAndResetChannel()
{
  p_channel->disableAll();
  p_channel->remove();

  int sockfd = p_channel->fd();

  p_loop->queueInLoop(std::bind(&Connector::resetChannel, this));

  return sockfd;
}

void Connector::resetChannel()
{
  LOG_TRACE << "Connector::resetChannel()";
  p_channel.reset();
}

void Connector::handleWrite()
{
  LOG_TRACE << "Connector::handleWrite ";

  if(m_state == kConnecting)
  {
    int sockfd = removeAndResetChannel();
    int err = sockets::getSocketError(sockfd);

    if(err)
    {
      LOG_WARN << "Connector::handleWrite - SO_ERROR = "
               << err << " " << strerror_tl(err);
      retry(sockfd);
    }
    /*else if (sockets::isSelfConnect(sockfd))
    {

    }*/
    else
    {
      setState(kConnected);
      m_newConnectionCallBack(sockfd);
    }

  }
  else
  {
    assert(m_state == kDisconnected);
  }

}

void Connector::handleError()
{
  LOG_ERROR << "Connector::handleError States " << m_state;

  if(m_state == kConnecting)
  {
    int sockfd = removeAndResetChannel();
    int err = sockets::getSocketError(sockfd);
    LOG_TRACE << "SOCK_ERROR = " << err << " " << strerror_tl(err);
    retry(sockfd);
  }
}

Acceptor

相較於Connector更簡單，只要有socket可讀，便可確認鏈接創建.

系統函數accept

 #include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>

       int accept(int sockfd, struct sockaddr addr, socklen_t addrlen);

       #define _GNU_SOURCE             /* See feature_test_macros(7) */
       #include <sys/socket.h>

       int accept4(int sockfd, struct sockaddr addr,
                   socklen_t addrlen, int flags);

sockfd socket(2)建立的文件描述符, 且已被bind(2)綁定本地地址,listen(2)使能監聽.
addr 用於填充遠端套接字地址, 若是不須要知道遠端地址,能夠添NULL.
addrlen 用於填充遠端地址大小.
flags
若是flags爲0  等同於 accept.

       SOCK_NONBLOCK  在新打開的文件描述符設置 O_NONBLOCK 標記。在 fcntl(2) 中保存這個標記能夠獲得相同的效果。

       SOCK_CLOEXEC  在新打開的文件描述符裏設置 close-on-exec (FD_CLOEXEC) 標記。參看在open(2)裏關於 O_CLOEXEC標記的描述來了解這爲何有用。

  int connfd = ::accept4(sockfd, (struct sockaddr *)(addr),
                         &addrlen, SOCK_NONBLOCK | SOCK_CLOEXEC);
 
flags 會對返回的fd  connfd  設置SOCK_NONBLOCK | SOCK_CLOEXEC 標記.

若是用於監聽的文件描述符沒有設置nonblocking標誌，且監聽隊列上沒有掛起的鏈接, accept()會阻塞直到有新的鏈接到來. 若是此socket設置了nonblocking標記,accept() 會當即返回失敗並設置 error 爲 EAGAIN or EWOULDBLOCK.

Socket的封裝

Socket類封裝一個套接字 fd 析構的時候close 管理套接字的生命期.

class Socket{
public:
  explicit Socket(int sockfd) : m_sockfd(sockfd) { }
  ~Socket();

  int fd() const { return m_sockfd; }
 
  void bindAddress(const InetAddress& localaddr);
  void listen();
  int accept(int sockfd, struct sockaddr_in6* addr);

  int accept(InetAddress* peeraddr);

private:
  const int m_sockfd;
};

Acceptor的封裝

Acceptor的數據成員包含Socket和Channel，Acceptor的Socket是服務端的監聽socket，Channel用於觀察此socket上的readable事件.並回調Acceptor:: handleRead()，handleRead()會調用accept(2)來接受新鏈接， 並回調用戶callback。

class Acceptor{
public:
  typedef std::function<void (int sockfd, const InetAddress&)> NewConnectionCallBack;

  Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport = true);
  ~Acceptor();

  void listen();
  bool listenning() const { return m_listenning; } // get listen status.

  void setNewConnectionCallBack(const NewConnectionCallBack& cb) { m_newConnectionCallBack = cb; }

private:
  void handleRead(); //處理新到的鏈接.

  EventLoop* p_loop;
  Socket m_acceptSocket;
  Channel m_acceptChannel;
  NewConnectionCallBack m_newConnectionCallBack;
  bool m_listenning;
  int m_idleFd;
};

Acceptor時序圖.

Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport)
  :p_loop(loop),
  m_acceptSocket(sockets::createNonblockingOrDie(listenAddr.family())),
  m_acceptChannel(loop, m_acceptSocket.fd()),
  m_listenning(false),
  m_idleFd(::open("/dev/null", O_RDONLY | O_CLOEXEC))
{
  assert(m_idleFd >= 0);
  m_acceptSocket.setReuseAddr(true);
  m_acceptSocket.setReuseAddr(reuseport);
  m_acceptSocket.bindAddress(listenAddr);
  m_acceptChannel.setReadCallBack(
    std::bind(&Acceptor::handleRead, this));
}

Acceptor::~Acceptor()
{
  m_acceptChannel.disableAll();
  m_acceptChannel.remove();
  ::close(m_idleFd);
}

void Acceptor::listen()
{
  p_loop->assertInLoopThread();
  m_listenning = true;
  m_acceptSocket.listen();
  m_acceptChannel.enableReading();
}

void Acceptor::handleRead()
{
  p_loop->assertInLoopThread();
  InetAddress peerAddr;
  int connfd = m_acceptSocket.accept(&peerAddr);
  if(connfd >= 0)
  {
    if(m_newConnectionCallBack)
    {
      m_newConnectionCallBack(connfd, peerAddr);
    }
    else
    {
      sockets::close(connfd);
    }
  }
  else
  {
    LOG_SYSERR << "in Acceptor::handleRead";
    if(errno == EMFILE)
    {
      ::close(m_idleFd);
      m_idleFd = ::accept(m_acceptSocket.fd(), NULL, NULL);
      ::close(m_idleFd);
      m_idleFd = ::open("/dev/null", O_RDONLY | O_CLOEXEC);
    }
  }

簡單測試程序

Acceptor

void newConnetion(int sockfd, const InetAddress& peeraddr)
{
  LOG_DEBUG << "newConnetion() : accepted a new connection from";
  ::sockets::close(sockfd);
}

int main()
{
  InetAddress listenAddr(8888);
  EventLoop loop;
  Acceptor acceptor(&loop, listenAddr);
  acceptor.setNewConnectionCallBack(newConnetion);
  acceptor.listen();

  loop.loop();

}

Connctor

EventLoop* g_loop;

void newConnetion(int sockfd)
{
  LOG_DEBUG << "newConnetion() : Connected a new connection.";
  sockets::close(sockfd);
  g_loop->quit();
}

int main()
{
  EventLoop loop;
  g_loop = &loop;

  InetAddress serverAddr("127.0.0.1", 8888);
  Connector client(&loop, serverAddr);
  client.setNewConnectionCallback(newConnetion);
  client.start();

  loop.loop();

}

運行日誌