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());}
}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/94591.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

treectrl类封装 2023/8/13 下午4:07:35

2023/8/13 下午4:07:35 treectrl类封装 2023/8/13 下午4:07:53 TreeCtrl 类是一个常用的图形用户界面控件,用于实现树形结构的展示和交互。以下是一个简单的 TreeCtrl 类的封装示例: python import wxclass MyTreeCtrl(wx.TreeCtrl):def __init__(self, parent):super()…

java限流

限流可以使用redis的过期时间自动过期限流的key&#xff0c;&#xff0c; 也可以使用一个定时器&#xff0c;在指定时间后清除这个key&#xff0c;&#xff0c;比如Timer Timer的使用 Timer timer new Timer() timer.schedule(timerTast,delay,period) public static void …

mysql面试题13:MySQL中什么是异步复制?底层实现?

该文章专注于面试,面试只要回答关键点即可,不需要对框架有非常深入的回答,如果你想应付面试,是足够了,抓住关键点 面试官:讲一讲mysql中什么是异步复制?底层实现? MySQL中的异步复制(Asynchronous Replication)是一种复制模式,主服务器将数据写入二进制日志后,无…

mysql面试题14:讲一讲MySQL中什么是全同步复制?底层实现?

该文章专注于面试,面试只要回答关键点即可,不需要对框架有非常深入的回答,如果你想应付面试,是足够了,抓住关键点 面试官:讲一讲mysql中什么是全同步复制?底层实现? MySQL中的全同步复制(Synchronous Replication)是一种复制模式,主服务器在写操作完成后,必须等待…

AI配套的技术: 矢量数据库的概念

一、说明 随着人工智能的快速采用和围绕大型语言模型发生的创新&#xff0c;我们需要在所有这些的中心&#xff0c;能够获取大量数据&#xff0c;将其上下文化&#xff0c;处理它&#xff0c;并使其能够有意义地搜索。 为原生整合生成式 AI 功能而构建的生成式 AI 流程和应用程…

JUC第十四讲:JUC锁: ReentrantReadWriteLock详解

JUC第十四讲&#xff1a;JUC锁: ReentrantReadWriteLock详解 本文是JUC第十四讲&#xff1a;JUC锁 - ReentrantReadWriteLock详解。ReentrantReadWriteLock表示可重入读写锁&#xff0c;ReentrantReadWriteLock中包含了两种锁&#xff0c;读锁ReadLock和写锁WriteLock&#xff…

ES6中的let、const

let ES6中新增了let命令&#xff0c;用来声明变量&#xff0c;和var类似但是也有一定的区别 1. 块级作用域 只能在当前作用域内使用&#xff0c;各个作用域不能互相使用&#xff0c;否则会报错。 {let a 1;var b 1; } console.log(a); // 会报错 console.log(b); // 1为什…

Day-05 CentOS7.5 安装docker

参考 &#xff1a; Install Docker Engine on CentOS | Docker DocsLearn how to install Docker Engine on CentOS. These instructions cover the different installation methods, how to uninstall, and next steps.https://docs.docker.com/engine/install/centos/ Doc…

iOS AVAudioSession 详解

iOS AVAudioSession 详解 - 简书 默认没有options&#xff0c;category 7种即可满足条件 - (BOOL)setCategory:(AVAudioSessionCategory)category error:(NSError **)outError API_AVAILABLE(ios(3.0), watchos(2.0), tvos(9.0)) API_UNAVAILABLE(macos); 有options&#xff…

实战型开发--3/3,clean code

编程的纯粹 hmmm&#xff0c;一开始在这个环节想聊一些具体的点&#xff0c;其实也就是《clean code》这本书中的点&#xff0c;但这个就还是更流于表面&#xff1b; 因为编码的过程&#xff0c;就更接近于运动员打球&#xff0c;艺术家绘画&#xff0c;棋手下棋的过程&#x…

Elasticsearch基础篇(四):Elasticsearch7.x的官方文档学习(Set up Elasticsearch)

Set up Elasticsearch 1 Configuring Elasticsearch(配置 Elasticsearch)1.1 Setting JVM Options(设置JVM选项)1.2 Secure Settings(安全设置)Introduction(介绍)Using the Keystore(使用密钥库)Applying Changes(应用更改)Reloadable Secure Settings(可重新加载的安全设置)R…

GPT系列论文解读:GPT-1

GPT系列 GPT&#xff08;Generative Pre-trained Transformer&#xff09;是一系列基于Transformer架构的预训练语言模型&#xff0c;由OpenAI开发。以下是GPT系列的主要模型&#xff1a; GPT&#xff1a;GPT-1是于2018年发布的第一个版本&#xff0c;它使用了12个Transformer…

软件设计师_计算机网络_学习笔记

文章目录 4.1 网路技术标准与协议4.1.1 协议4.1.2 DHCP4.1.3 DNS的两种查询方式 4.2 计算机网络的分类4.2.1 拓扑结构 4.3 网络规划与设计4.3.1 遵循的原则4.3.2 逻辑网络设计4.3.3 物理网络设计4.3.4 分层设计 4.4 IP地址与子网划分4.4.1 子网划分4.4.2 特殊IP 4.5 HTML4.6 无…

【MySQL】表的约束(二)

文章目录 一. 主键二. 自增长三. 唯一值四. 外键结束语 一. 主键 主键primary key 用来唯一的约束该字段里面的数据&#xff0c;不能重复&#xff0c;不能为空&#xff0c;一张表最多只能有一个主键&#xff0c;主键所在的列通常是整数类型 创建表时直接在字段上指定主键 mysq…

在2023年使用Unity2021从Built-in升级到Urp可行么

因为最近在做WEbgl平台&#xff0c;所以某些不可抗力原因&#xff0c;需要使用Unity2021开发&#xff0c;又由于不可明说原因&#xff0c;想用Urp&#xff0c;怎么办&#xff1f; 目录 创建RenderAsset 关联Asset 暴力转换&#xff08;Menu->Edit&#xff09; 单个文件…

栈的基本操作(数据结构)

顺序栈的基本操作 #include <stdlib.h> #include <iostream> #include <stdio.h> #define MaxSize 10typedef struct{int data[MaxSize];int top; }SqStack;//初始化栈 void InitStack(SqStack &S){S.top -1; } //判断栈空 bool StackEmpty(SqStack S)…

TM 学习记录--论文阅读1

这里可以查看所有论文。由于作者book只更新到第二章剩下的只有从论文中学习&#xff0c;但书中的目录和论文可以由于对应起来。第一二章可以对应到第一篇论文&#xff0c;这里。

【服务器】在 Linux CLI 下安装 Anaconda

【服务器】在 Linux CLI 下安装 Anaconda 1 系统环境2 下载安装包3 安装 1 系统环境 查看系统信息 cat /etc/os-release2. 查看架构 uname -a # output # Linux localhost.localdomain 4.18.0-193.28.1.el8_2.x86_64 #1 SMP Thu Oct 22 00:20:22 UTC 2020 x86_64 x86_64 x86…

华为云云耀云服务器L实例评测|Docker部署及应用

文章目录 前言&#x1f4e3; 1.前言概述&#x1f4e3; 2.服务器攻击✨ 2.1 问题描述✨ 2.2 处理方法 &#x1f4e3; 3.Docker简介&#x1f4e3; 4.安装Docker✨ 4.1 卸载旧版docker✨ 4.2 安装依赖包✨ 4.3 安装GPG证书✨ 4.4 配置仓库✨ 4.5 正式安装Docker✨ 4.6 配置用户组✨…

PG 多表连接查询

写法&#xff1a; 使用 select 表名.键名 from 表1 join表2 on 相同的主键 构造出来一张新表 多表要用表名.键名 才能知道是哪一张表 传统写法也行 类型&#xff1a; 内 而外的要这样写