muduo 간단한 echo 서버 분석

앞의 두 편은muduo 네트워크 프레임워크 스레드 처리를 썼는데, 이 두 편은 작은 echo 서버를 통해 이 네트워크 이벤트 처리의 절차를 완전하게 설명한다.echo는 muduo가 가지고 있는 예입니다. 매우 간단합니다.

int main()
{
  LOG_INFO << "pid = " << getpid();
  muduo::net::EventLoop loop;
  muduo::net::InetAddress listenAddr(2007);
  EchoServer server(&loop, listenAddr);
  server.start();
  loop.loop();
}

EchoServer::EchoServer(muduo::net::EventLoop* loop,
                       const muduo::net::InetAddress& listenAddr)
  : server_(loop, listenAddr, "EchoServer")
{
  server_.setConnectionCallback(
      std::bind(&EchoServer::onConnection, this, _1));
  server_.setMessageCallback(
      std::bind(&EchoServer::onMessage, this, _1, _2, _3));
}

void EchoServer::start()
{
  server_.start();
}

void EchoServer::onConnection(const muduo::net::TcpConnectionPtr& conn)
{
  LOG_INFO << "EchoServer - " << conn->peerAddress().toIpPort() << " -> "
           << conn->localAddress().toIpPort() << " is "
           << (conn->connected() ? "UP" : "DOWN");
}

void EchoServer::onMessage(const muduo::net::TcpConnectionPtr& conn,
                           muduo::net::Buffer* buf,
                           muduo::Timestamp time)
{
  muduo::string msg(buf->retrieveAllAsString());
  LOG_INFO << conn->name() << " echo " << msg.size() << " bytes, "
           << "data received at " << time.toString();
  conn->send(msg); 
}

muduo의 메인 라인은 창고에 표시된 EventLoop 실례가 있습니다. EchoServer에 TcpServer 형식의 조합 변수 server_가 있는 것을 보았습니다.(muduo가 파생적인 방식이 아니라 조합 방식을 제창하는 것에 주의), server_이 EventLoop 인스턴스가 참조됩니다.더 나아가 서버_start를 시작합니다.start에서 스레드 탱크도 start를 실행하기 시작했습니다. 스레드 탱크의 start 분석은 앞에서 말했듯이 여기서는 더 이상 군더더기를 하지 않습니다.이어서 소켓 대상인 Acceptor의 감청 작업을 감청하는 것은 말할 것이 없다.왜 loop_->run In Loop의 방식은 다음과 같습니다.
새로운 연결이 오면 이벤트를 터치하고 TcpServer::newConnection을 호출합니다. 이것은 TcpServer 구조 함수에 귀속된 것입니다. 함수는 다음과 같습니다.

void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{
  loop_->assertInLoopThread();
  EventLoop* ioLoop = threadPool_->getNextLoop();
  char buf[64];
  snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
  ++nextConnId_;
  string connName = name_ + buf;

  InetAddress localAddr(sockets::getLocalAddr(sockfd));
  // FIXME poll with zero timeout to double confirm the new connection
  // FIXME use make_shared if necessary
  TcpConnectionPtr conn(new TcpConnection(ioLoop,
                                          connName,
                                          sockfd,
                                          localAddr,
                                          peerAddr));
  connections_[connName] = conn;
  conn->setConnectionCallback(connectionCallback_);
  conn->setMessageCallback(messageCallback_);
  conn->setWriteCompleteCallback(writeCompleteCallback_);
  conn->setCloseCallback(
      std::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe
  ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished, conn));
}

앞에서 말했듯이 새로운 연결 이벤트 처리는 작업 라인의 EventLoop에 있습니다.새 스레드를 가져오는 방법,threadPool_-> 호출getNextLoop()은 간단한 round-robin 방식으로 얻을 수 있습니다.이어서 TcpConnection 실례를 만들고 각종 리셋 함수를 설정합니다. 예를 들어 연결이 성공한 후의 리셋 함수, 메시지를 받은 후의 리셋 함수, 그들은 이미 EchoServer 구조 함수에 귀속되어 있습니다.그리고 ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished,conn)을 실행합니다.

void EventLoop::runInLoop(Functor cb)
{
  if (isInLoopThread())
  {
    cb();
  }
  else
  {
    queueInLoop(std::move(cb));
  }
}

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

  if (!isInLoopThread() || callingPendingFunctors_)
  {
    wakeup();
  }
}

runInLoop의 목적은 새 연결이 EventLoop이 메인 라인에 있으면 이 연결된 함수를 직접 실행하는 것이다. 그렇지 않으면 EventLoop은 미결 함수인 PendingFunctors라고 불리는 리셋 함수를 잠시 저장한다.언제 호출합니까?앞에서 말한 새로운 라인은 epoll의 막힘 속에 있을 수 있다.그래서 새로운 라인을 깨우치거나 이미 처리하기 전의 리셋 함수 상태를 신속하게 처리하기 위해 새로운 리셋 함수를 다시 깨우는 이벤트를 보냅니다.
깨우면 EventLoop이 계속 순환하여 미결 함수 doPendingFunctors를 처리합니다.TcpConnection::connectEstablished() 함수가 호출됩니다.

void TcpConnection::connectEstablished()
{
  loop_->assertInLoopThread();
  assert(state_ == kConnecting);
  setState(kConnected);
  channel_->tie(shared_from_this());
  channel_->enableReading();

  connectionCallback_(shared_from_this());
}

그가 해야 할 일은 epoll의 감독에 추가하는 것이다.그리고 리셋 함수가 트리거됩니다. 앞에서 EchoServer에 귀속되었다고 말했습니다.데이터가 오면 읽기 리셋 함수를 터치합니다. (이것은 채널 채널 채널 클래스에 귀속되어 있으며, 다른 글을 써서 설명합니다.)

void TcpConnection::handleRead(Timestamp receiveTime)
{
  loop_->assertInLoopThread();
  int savedErrno = 0;
  ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
  if (n > 0)
  {
    messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
  }
  else if (n == 0)
  {
    handleClose();
  }
  else
  {
    errno = savedErrno;
    LOG_SYSERR << "TcpConnection::handleRead";
    handleError();
  }
}

messageCallback_EchoServer에 귀속되어 있으면 클라이언트에게 받은 메시지를 그대로 보냅니다.사실 메시지를 수신하고 발송할 때 관련된 캐시 처리는 좀 복잡하기 때문에 다시 분석할 기회가 있습니다.
이로써 전체 절차는 끝났다.