常州网站建设方案策划,企业管理软件价格,wordpress 购物商城,物流网站建设平台分析文章目录 前言一、TcpConnection的读二、TcpConnection的写三、TcpConnection的关闭 前言
今天总结TcpConnection类的读写事件。
一、TcpConnection的读
当Poller检测到套接字的Channel处于可读状态时#xff0c;会调用Channel的回调函数#xff0c;回调函数中根据不同激活… 文章目录 前言一、TcpConnection的读二、TcpConnection的写三、TcpConnection的关闭 前言
今天总结TcpConnection类的读写事件。
一、TcpConnection的读
当Poller检测到套接字的Channel处于可读状态时会调用Channel的回调函数回调函数中根据不同激活原因调用不同的函数这些函数都由TcpConnection在创建Channel之初提供当可读时调用TcpConnection的可读函数handleRead而在这个函数中读缓冲区就会从内核的tcp缓冲区读取数据。 void TcpConnection::handleRead(Timestamp receiveTime)
{int savedErrno 0;ssize_t n inputBuffer_.readFd(channel_-fd(), savedErrno);if (n 0){// 已建立连接的用户有可读事件发生了调用用户传入的回调操作onMessagemessageCallback_(shared_from_this(), inputBuffer_, receiveTime);}else if (n 0){handleClose();}else{errno savedErrno;LOG_ERROR(TcpConnection::handleRead);handleError();}
}TcpConnection::handleRead( )函数首先调用Buffer_.readFd(channel_-fd(), saveErrno)该函数底层调用Linux的函数readv( )将Tcp接收缓冲区数据拷贝到用户定义的缓冲区中inputBuffer_。如果在读取拷贝的过程中发生了什么错误这个错误信息就会保存在savedErrno中。 对于缓冲区 Buffer::readFd()函数之前的文章已经剖析过了。
二、TcpConnection的写
TcpConnection::send() 方法是用户调用发送接口会调用TcpConnection::sendInLoop() 方法来处理具体的发送操作。如果在当前线程直接发送就会调用 sendInLoop() 方法处理否则需要把发送任务加入到事件循环中等待对应的线程处理。
//给用户提供的 发送接口
void TcpConnection::send(const std::string buf)
{if (state_ kConnected){if (loop_-isInLoopThread()){// 判断当前的线程 是不是在对应的线程里面// 有一些情况sendInLoop(buf.c_str(), buf.size());}else{loop_-runInLoop(std::bind(TcpConnection::sendInLoop,this,buf.c_str(),buf.size()));}}
}
在TcpConnection::sendInLoop() 方法的实现中接通过系统调用 write() 发送数据如果有剩余未发送的数据则会将数据添加到发送缓冲区中并注册 channel 的可写事件等待事件循环通知空闲后再进行发送。 /*** 发送数据 应用写的快 而内核发送数据慢 需要把待发送数据写入缓冲区* 而且设置了水位回调*/
void TcpConnection::sendInLoop(const void* data, size_t len)
{ssize_t nwrote 0;// remaining是没发送完的数据size_t remaining len;bool faultError false;// 之前调用过该connection的shutdown不能再进行发送了if (state_ kDisconnected){LOG_ERROR(disconnected, give up writing!);return;}// 表示channel_第一次开始写数据而且缓冲区没有待发送数据if (!channel_-isWriting() outputBuffer_.readableBytes() 0){// 返回的是具体发送的 数据nwrote ::write(channel_-fd(), data, len);if (nwrote 0){remaining len - nwrote;// 如果放松完了 并且注册了 发送完回调函数if (remaining 0 writeCompleteCallback_){// 既然在这里数据全部发送完成就不用再给channel设置epollout事件了loop_-queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));}}else // nwrote 0{nwrote 0;if (errno ! EWOULDBLOCK){LOG_ERROR(TcpConnection::sendInLoop);if (errno EPIPE || errno ECONNRESET) // SIGPIPE RESET{faultError true;}}}}// 说明当前这一次write并没有把数据全部发送出去// 剩余的数据需要保存到缓冲区当中然后给channel// 注册epollout事件poller发现tcp的发送缓冲区有空间// 因为是lt模式 如果缓存区空余 就会不断地提醒// 会通知相应的sock-channel调用writeCallback_回调方法也就是hanldwrite方法// 也就是调用TcpConnection::handleWrite方法把发送缓冲区中的数据全部发送完成if (!faultError remaining 0) {// 目前发送缓冲区剩余的待发送数据的长度size_t oldLen outputBuffer_.readableBytes();if (oldLen remaining highWaterMark_ oldLen highWaterMark_ highWaterMarkCallback_){loop_-queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLenremaining));}// 数据添加到缓冲区里面outputBuffer_.append((char*)data nwrote, remaining);if (!channel_-isWriting()){// 这里一定要注册channel的写事件否则poller不会给channel通知epolloutchannel_-enableWriting(); }}
}
发送缓冲区中有数据时TcpConnection::handleWrite() 方法会被调用来处理具体的发送操作。在该方法中首先会判断 channel 是否可写如果可写则通过系统调用 writeFd() 将发送缓冲区中的数据写入到套接字中。如果写入成功就会从发送缓冲区中删除已经发送的数据并判断是否还有剩余数据如果没有则禁用 channel 的写事件并执行可写回调函数。如果还有剩余数据则会继续等待事件循环通知空闲后再次进行发送。 // 对outputBuffer_ 进行发送
void TcpConnection::handleWrite()
{if (channel_-isWriting()){int savedErrno 0;ssize_t n outputBuffer_.writeFd(channel_-fd(), savedErrno);if (n 0){// 有数据发送成功 n个数据已经处理过了 把readable 向右移outputBuffer_.retrieve(n);if (outputBuffer_.readableBytes() 0){// 已经发送完成了 编程不可写 执行回调写完回调writeCompleteCallback_channel_-disableWriting();if (writeCompleteCallback_){// 唤醒loop_对应的thread线程执行回调// 唤醒线程 执行写完之后的回调事件loop_-queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));}if (state_ kDisconnecting){// 如果还有数据但是 就调用了 shutdown// state_就变成了 kDisconnecting// 但是 需要等待 数据传输完成 再调用shutdownInLoopshutdownInLoop();}}}else{LOG_ERROR(TcpConnection::handleWrite);}}else{LOG_ERROR(TcpConnection fd%d is down, no more writing \n, channel_-fd());}
}
这里的细节问题就是如果想要关闭连接那么通常是先关闭读端等到将写缓冲区所有数据都写到tcp缓冲区后再关闭写端否则这些数据就不能发送给对端了
三、TcpConnection的关闭
需要关闭时候setState(kDisconnecting);把状态设置为kDisconnecting但是没有立即关闭而是判断是否还有数据可写。
// 关闭连接
void TcpConnection::shutdown()
{if (state_ kConnected){setState(kDisconnecting);loop_-runInLoop(std::bind(TcpConnection::shutdownInLoop, this));}
}
void TcpConnection::shutdownInLoop()
{// 如果buffer还有数据这个就是writing状态// 会一直发 知道发完 然后监控到状态是kDisconnecting // 再次调用这个函数 就会关闭了// 保证数据发送完if (!channel_-isWriting()) // 说明outputBuffer中的数据已经全部发送完成{socket_-shutdownWrite(); // 关闭写端}
}如果buffer还有数据这个就是writing状态会一直被epoll提醒发送直到发完 然后监控到状态是kDisconnecting 再次调用这个函数 就会关闭了 保证数据发送完。 // 对outputBuffer_ 进行发送
void TcpConnection::handleWrite()
{if (channel_-isWriting()){int savedErrno 0;ssize_t n outputBuffer_.writeFd(channel_-fd(), savedErrno);if (n 0){// 有数据发送成功 n个数据已经处理过了 把readable 向右移outputBuffer_.retrieve(n);if (outputBuffer_.readableBytes() 0){// 已经发送完成了 编程不可写 执行回调写完回调writeCompleteCallback_channel_-disableWriting();if (writeCompleteCallback_){// 唤醒loop_对应的thread线程执行回调// 唤醒线程 执行写完之后的回调事件loop_-queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));}if (state_ kDisconnecting){// 如果还有数据但是 就调用了 shutdown// state_就变成了 kDisconnecting// 但是 需要等待 数据传输完成 再调用shutdownInLoopshutdownInLoop();}}}else{LOG_ERROR(TcpConnection::handleWrite);}}else{LOG_ERROR(TcpConnection fd%d is down, no more writing \n, channel_-fd());}
}
