TcpConnection的读写操作【深度剖析】

2023-10-04 08:15

本文主要是介绍TcpConnection的读写操作【深度剖析】,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • 前言
  • 一、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(), oldLen+remaining));}// 数据添加到缓冲区里面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());}
}

这篇关于TcpConnection的读写操作【深度剖析】的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1669

相关文章

Java深度学习库DJL实现Python的NumPy方式

《Java深度学习库DJL实现Python的NumPy方式》本文介绍了DJL库的背景和基本功能,包括NDArray的创建、数学运算、数据获取和设置等,同时,还展示了如何使用NDArray进行数据预处理... 目录1 NDArray 的背景介绍1.1 架构2 JavaDJL使用2.1 安装DJL2.2 基本操

最长公共子序列问题的深度分析与Java实现方式

《最长公共子序列问题的深度分析与Java实现方式》本文详细介绍了最长公共子序列(LCS)问题,包括其概念、暴力解法、动态规划解法,并提供了Java代码实现,暴力解法虽然简单,但在大数据处理中效率较低,... 目录最长公共子序列问题概述问题理解与示例分析暴力解法思路与示例代码动态规划解法DP 表的构建与意义动

Go中sync.Once源码的深度讲解

《Go中sync.Once源码的深度讲解》sync.Once是Go语言标准库中的一个同步原语,用于确保某个操作只执行一次,本文将从源码出发为大家详细介绍一下sync.Once的具体使用,x希望对大家有... 目录概念简单示例源码解读总结概念sync.Once是Go语言标准库中的一个同步原语,用于确保某个操

使用Python进行文件读写操作的基本方法

《使用Python进行文件读写操作的基本方法》今天的内容来介绍Python中进行文件读写操作的方法,这在学习Python时是必不可少的技术点,希望可以帮助到正在学习python的小伙伴,以下是Pyth... 目录一、文件读取:二、文件写入:三、文件追加:四、文件读写的二进制模式:五、使用 json 模块读写

五大特性引领创新! 深度操作系统 deepin 25 Preview预览版发布

《五大特性引领创新!深度操作系统deepin25Preview预览版发布》今日,深度操作系统正式推出deepin25Preview版本,该版本集成了五大核心特性:磐石系统、全新DDE、Tr... 深度操作系统今日发布了 deepin 25 Preview,新版本囊括五大特性:磐石系统、全新 DDE、Tree

Node.js 中 http 模块的深度剖析与实战应用小结

《Node.js中http模块的深度剖析与实战应用小结》本文详细介绍了Node.js中的http模块,从创建HTTP服务器、处理请求与响应,到获取请求参数,每个环节都通过代码示例进行解析,旨在帮... 目录Node.js 中 http 模块的深度剖析与实战应用一、引言二、创建 HTTP 服务器:基石搭建(一

基于UE5和ROS2的激光雷达+深度RGBD相机小车的仿真指南(五):Blender锥桶建模

前言 本系列教程旨在使用UE5配置一个具备激光雷达+深度摄像机的仿真小车,并使用通过跨平台的方式进行ROS2和UE5仿真的通讯,达到小车自主导航的目的。本教程默认有ROS2导航及其gazebo仿真相关方面基础,Nav2相关的学习教程可以参考本人的其他博客Nav2代价地图实现和原理–Nav2源码解读之CostMap2D(上)-CSDN博客往期教程: 第一期:基于UE5和ROS2的激光雷达+深度RG

韦季李输入法_输入法和鼠标的深度融合

在数字化输入的新纪元,传统键盘输入方式正悄然进化。以往,面对实体键盘,我们常需目光游离于屏幕与键盘之间,以确认指尖下的精准位置。而屏幕键盘虽直观可见,却常因占据屏幕空间,迫使我们在操作与视野间做出妥协,频繁调整布局以兼顾输入与界面浏览。 幸而,韦季李输入法的横空出世,彻底颠覆了这一现状。它不仅对输入界面进行了革命性的重构,更巧妙地将鼠标这一传统外设融入其中,开创了一种前所未有的交互体验。 想象

免费也能高质量!2024年免费录屏软件深度对比评测

我公司因为客户覆盖面广的原因经常会开远程会议,有时候说的内容比较广需要引用多份的数据,我记录起来有一定难度,所以一般都用录屏工具来记录会议内容。这次我们来一起探索有什么免费录屏工具可以提高我们的工作效率吧。 1.福晰录屏大师 链接直达:https://www.foxitsoftware.cn/REC/  录屏软件录屏功能就是本职,这款录屏工具在录屏模式上提供了多种选项,可以选择屏幕录制、窗口

动手学深度学习【数据操作+数据预处理】

import osos.makedirs(os.path.join('.', 'data'), exist_ok=True)data_file = os.path.join('.', 'data', 'house_tiny.csv')with open(data_file, 'w') as f:f.write('NumRooms,Alley,Price\n') # 列名f.write('NA