技术标签: spring SPDK c++无锁队列 C++后端开发 网络编程 C++Linux后端 Thread
Reactor – 单个CPU Core抽象,主要包含了:
Thread – 线程,但它是spdk抽象出来的线程,主要包含了:
对象g_reactor_state有五个状态对应了应用中reactors运行运行状态,
enum spdk_reactor_state {
SPDK_REACTOR_STATE_INVALID = 0,
SPDK_REACTOR_STATE_INITIALIZED = 1,
SPDK_REACTOR_STATE_RUNNING = 2,
SPDK_REACTOR_STATE_EXITING = 3,
SPDK_REACTOR_STATE_SHUTDOWN = 4,
};
本文福利, 免费领取C++学习资料包、技术视频/代码,1000道大厂面试题,内容包括(C++基础,网络编程,数据库,中间件,后端开发,音视频开发,Qt开发)↓↓↓↓↓↓见下面↓↓文章底部点击免费领取↓↓
初始情况下是:
SPDK_REACTOR_STATE_INVALID状态,在spdk app(任意一个target,比如nvmf_tgt)启动时,即调用了spdk_app_start方法,会调用spdk_reactors_init,在这个方法中将会初始化所有需要被初始化的reactors(可以在配置文件中指定需要使用的Core,CPU Core 和reactor是一对一的)。并且会将g_reactor_state设置为SPDK_REACTOR_STATE_INITIALIZED。具体代码如下:
Int spdk_reactors_init(void)
{
// 初始化所有的event mempool
g_spdk_event_mempool = spdk_mempool_create(…);
// 为g_reactors分配内存,g_reactors是一个数组,管理了所有的reactors
posix_memalign((void **)&g_reactors, 64, (last_core + 1) * sizeof(struct spdk_reactor));
// 这里设置了reactor创建线程的方法,之后需要初始化线程的时候将会调用该方法
spdk_thread_lib_init(spdk_reactor_schedule_thread, sizeof(struct spdk_lw_thread));
// 对于每一个启动的reactor,将会初始化它们
// 初始化reactor过程,即为绑定lcore,初始化spdk ring、threads,对rusage无操作
SPDK_ENV_FOREACH_CORE(i) {
reactor = spdk_reactor_get(i);
spdk_reactor_construct(reactor, i);
}
// 设置好状态返回
g_reactor_state = SPDK_REACTOR_STATE_INITIALIZED;
return 0;
}
在进入SPDK_REACTOR_STATE_INITIALIZED状态且spdk_app_start在创建了自己的线程并绑定到了reactors后,会调用spdk_reactors_start方法并将g_reactor_state设置为SPDK_REACTOR_STATE_RUNNING状态并会创建所有reactor的线程且轮询。
Void spdk_reactors_start(void) {
SPDK_ENV_FOREACH_CORE(i) {
if (i != current_core) { // 在非master reactor中
reactor = spdk_reactor_get(i); // 得到相应的reactor
// 设置好线程创建后的一个消息,该消息为轮询函数
rc = spdk_env_thread_launch_pinned(reactor->lcore, _spdk_reactor_run, reactor);
// reactor创建好线程并且会自动执行第一个消息
spdk_thread_create(thread_name, tmp_cpumask);
}
}
// 当前CPU core得到reactor,并且开始轮询
reactor = spdk_reactor_get(current_core);
_spdk_reactor_run(reactor);
}
之前提到spdk_reactors_init方法中调用了spdk_thread_lib_init方法传入了创建thread的spdk_reactor_schedule_thread方法,在调用spdk_thread_create会回调该方法。这个方法它主要的功能就是告诉这个新创建的线程绑定创建该线程的reactor。
spdk_reactor_schedule_thread(struct spdk_thread *thread)
{
// 得到该线程设置的cpu mask
cpumask = spdk_thread_get_cpumask(thread);
for (i = 0; i < spdk_env_get_core_count(); i++) {
…. // 遍历cpu core
// 通过cpu mask找到对应的核心,并产生event
if (spdk_cpuset_get_cpu(cpumask, core)) {
evt = spdk_event_allocate(core, _schedule_thread, lw_thread, NULL);
break;
}
}
// 传递该event,即对应的reatcor会调用_schedule_thread方法,
spdk_event_call(evt);
}
_schedule_thread(void *arg1, void *arg2)
{
struct spdk_lw_thread *lw_thread = arg1;
struct spdk_reactor *reactor;
// 消息传递到对应的reactor后将该thread加入到reactor中
reactor = spdk_reactor_get(spdk_env_get_current_core());
TAILQ_INSERT_TAIL(&reactor->threads, lw_thread, link);
}
在SPDK_REACTOR_STATE_RUNNING后,此时所有reactor就进入了轮询状态。_spdk_reactor_run函数为线程提供了轮询方法:
static int _spdk_reactor_run(void *arg) {
while (1) {
// 处理reactor上的event消息,消息会在之后讲到
_spdk_event_queue_run_batch(reactor);
// 每一个reactor上注册的thread进行遍历并且处理poller事件
TAILQ_FOREACH_SAFE(lw_thread, &reactor->threads, link, tmp) {
rc = spdk_thread_poll(thread, 0, now);
}
// 检查reactor的状态
if (g_reactor_state != SPDK_REACTOR_STATE_RUNNING) {
break;
}
}
}
而当spdk app被调用spdk_app_stop方法后将会相应的通知每一个reactor调用spdk_reactors_stop方法,将g_reactor_state赋值为SPDK_REACTOR_STATE_EXITING,即开始退出了。回到_spdk_reactor_run函数中,轮询将会被跳出,并且执行销毁线程的代码。
static int _spdk_reactor_run(void *arg) {
….. // 轮询
TAILQ_FOREACH_SAFE(lw_thread, &reactor->threads, link, tmp) {
thread = spdk_thread_get_from_ctx(lw_thread);
TAILQ_REMOVE(&reactor->threads, lw_thread, link);
spdk_set_thread(thread);
spdk_thread_exit(thread);
spdk_thread_destroy(thread);
}
}
在这之后,主线程的_spdk_reactor_run会返回到spdk_reactors_start中,并将g_reactor_state赋值为SPDK_REACTOR_STATE_SHUTDOWN,返回到spdk_app_start中等待应用退出。
最后,总结一下reactors和CPU core以及spdk thread关系应该如图1所示
图1 CPU cores、reactors和thread关系图
Reactor生命周期流程图则如图2所示
图2 reactor生命周期流程图
当Reactors进行轮询时,除了处理自己的事件消息之外,还会调用注册在该reactor下面的每一个线程进行轮询。不过通常一个reactor只有一个thread,在spdk应用中,更多的是注册多个poller而不是注册多个thread。具体的轮询方法为:
Int spdk_thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now) {
// 首先先处理ring传递过来的消息
msg_count = _spdk_msg_queue_run_batch(thread, max_msgs);
// 调用非定时poller中的方法
TAILQ_FOREACH_REVERSE_SAFE(poller, &thread->active_pollers,
active_pollers_head, tailq, tmp) {
// 调用poller注册的方法之前,会对poller状态检测且转换
if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) {
TAILQ_REMOVE(&thread->active_pollers, poller, tailq);
free(poller);
continue;
}
poller->state = SPDK_POLLER_STATE_RUNNING;
// 调用poller注册的方法
poller_rc = poller->fn(poller->arg);
// poller转换状态
poller->state = SPDK_POLLER_STATE_WAITING;
}
// 调用定时poller中的方法
TAILQ_FOREACH_SAFE(poller, &thread->timer_pollers, tailq, tmp) {
// 类似非定时poller过程,不过会检查是否到了预定的时间
if (now < poller->next_run_tick) break;
}
// 最后统计时间
}
Io_device 和 io_channel在thread中也是非常重要的概念。它们的实现都在thread.c中,io_device是设备的抽象,io_channel是对该设备通道的抽象。一个线程可以创建多个io_channel . io_channel只能和一个io_device绑定,并且这个io_channel是别的线程使用不了的。
图 3 io_device、io_channel和线程关系图
Io_device结构
struct io_device {
void *io_device; // 抽象的device指针
char name[SPDK_MAX_DEVICE_NAME_LEN + 1]; // 名字
spdk_io_channel_create_cb create_cb; // io_channel创建的回调函数
spdk_io_channel_destroy_cb destroy_cb; // io_channel销毁的回调函数
spdk_io_device_unregister_cb unregister_cb; // io_device解绑的回调函数
struct spdk_thread *unregister_thread; // 不使用该device线程
uint32_t ctx_size; // ctx的大小,将会传给io_channel处理
uint32_t for_each_count; // io_channel的数量
TAILQ_ENTRY(io_device) tailq; // device队列头
uint32_t refcnt; // 计数器
bool unregistered; // 是否该device被注册
};
可以看到,io_device实际上只提供了一些自身io_device的操作和io_channel相关的方法,具体的io_device实体其实是那个名字叫io_device的void指针。因为thread中的io_device只提供了thread这一层接口,具体的io操作每一个设备很难被抽象出来,所以这一层的接口只负责管理io_channel的创建、销毁和绑定等。
Io_channel的结构
struct spdk_io_channel {
struct spdk_thread *thread; // 绑定的线程
struct io_device *dev; // 绑定的io_device
uint32_t ref; // io_channel引用计数
uint32_t destroy_ref; // destroy前被引用的次数
TAILQ_ENTRY(spdk_io_channel) tailq; // io_channel 队列头
spdk_io_channel_destroy_cb destroy_cb; // io_channel销毁的回调函数
};
虽然io_channel看起来是很简单的结构体,实际上在创建一个io_device的时候,会要求使用者传入一个io_channel_ctx的大小作为调用的参数,而在给io_channel分配内存的时候,除了分配本身io_channel结构体的大小外,还会额外分配一个io_channel_ctx的大小,这个context可以理解成一个void指针,当用户在使用io_channel的时候,实际上还是通过context的部分去访问io_device。
nvmf_tgt 是spdk中一个重要的模块,这里详细的写一下它作为一个target实例是如何使用thread、io_device以及io_channel的。
在spdk应用刚启动的时候,reactor模块就会自动加载起来,然后在加载nvmf subsystem的时候,会调用spdk_nvmf_subsystem_init(lib/event/subsystems/nvmf/nvmf_tgt.c)方法,nvmf_tgt其实也是有生命周期,并且有一个状态机去管理它的生命周期。
enum nvmf_tgt_state {
NVMF_TGT_INIT_NONE = 0, // 最初的状态
NVMF_TGT_INIT_PARSE_CONFIG, // 解析配置文件
NVMF_TGT_INIT_CREATE_POLL_GROUPS, // 创建poll groups
NVMF_TGT_INIT_START_SUBSYSTEMS, // 启动subsystem
NVMF_TGT_INIT_START_ACCEPTOR, // 开始接收
NVMF_TGT_RUNNING, // running
NVMF_TGT_FINI_STOP_SUBSYSTEMS,
NVMF_TGT_FINI_DESTROY_POLL_GROUPS,
NVMF_TGT_FINI_STOP_ACCEPTOR,
NVMF_TGT_FINI_FREE_RESOURCES,
NVMF_TGT_STOPPED,
NVMF_TGT_ERROR,
};
首先在NVMF_TGT_INIT_PARSE_CONFIG状态中,nvmf_tgt会去解析启动时传入的配置文件,当解析了[nvmf]这个label后,会调用spdk_nvmf_tgt_create这个方法,这个方法将初始化了全局的g_nvmf_tgt变量,同时也将tgt注册成了一个io_device。
1 spdk_io_device_register(tgt,
2 spdk_nvmf_tgt_create_poll_group,
3 spdk_nvmf_tgt_destroy_poll_group,
4 sizeof(struct spdk_nvmf_poll_group),
5 "nvmf_tgt");
spdk_nvmf_tgt_create_poll_group和spdk_nvmf_tgt_destroy_poll_group是io_channel创建和销毁的回调方法(在spdk_get_io_channel时调用 create_cb)。第三个参数是io_channel_ctx的size,既然这里传入了spdk_nvmf_poll_group的大小,那么很明显说明在nvmf中io_channel_ctx对象就是spdk_nvmf_poll_group。
当config文件解析完了之后,nvmf_tgt状态到了NVMF_TGT_INIT_CREATE_POLL_GROUPS,这个状态下会为每一个线程都创建相应的poll group。
spdk_for_each_thread(nvmf_tgt_create_poll_group,
NULL,
nvmf_tgt_create_poll_group_done);
static void nvmf_tgt_create_poll_group(void *ctx)
{
struct nvmf_tgt_poll_group *pg;
….
pg->thread = spdk_get_thread();
pg->group = spdk_nvmf_poll_group_create(g_spdk_nvmf_tgt);
….
}
再看spdk_nvmf_poll_group_create中,
struct spdk_nvmf_poll_group * spdk_nvmf_poll_group_create(struct spdk_nvmf_tgt *tgt)
{
struct spdk_io_channel *ch;
ch = spdk_get_io_channel(tgt);
….
return spdk_io_channel_get_ctx(ch);
}
在spdk_get_io_channel中,会先去检查传入的io_device是不是已经注册好了的,如果已经注册了,将会创建一个新的io_channel返回,创建的过程会回调在注册io_device时注册的io_channel创建方法(即方法spdk_nvmf_tgt_create_poll_group)。
static int spdk_nvmf_tgt_create_poll_group(void *io_device, void *ctx_buf)
{
….. // 初始化transport 、nvmf subsystem等
// 注册一个poller
group->poller = spdk_poller_register(spdk_nvmf_poll_group_poll, group, 0);
group->thread = spdk_get_thread();
return 0;
}
在spdk_nvmf_poll_group_poll中,因为spdk_nvmf_poll_group对象中有transport的poll group,所以它会调用对应的transport的poll_group_poll方法,比如rdma的poll_group_poll就会轮询rdma注册的poller处理每个在相应的qpair来的请求,进入rdma的状态机将请求处理好。
然后这个状态就结束了,之后再初始化好了nvmf subsystem相关的东西之后,到了状态NVMF_TGT_INIT_START_ACCEPTOR。在这个状态中,只注册了一个poller。
1 g_acceptor_poller = spdk_poller_register(acceptor_poll, g_spdk_nvmf_tgt,
2 g_spdk_nvmf_tgt_conf->acceptor_poll_rate);
这个poller调用的transport的方法,不断的监听是不是有新的fd连接进来,如果有就调用new_qpair的回调。
spdk thread 模型是spdk无锁化的基础,在一个线程中,当分配一个任务后,一直会运行到任务结束为止,这确保了不需要进行线程之间的切换而带来额外的损耗。同时,高效的spdk ring提供了不同线程之间的消息传递,这就使得任务结束的结果可以高效的传递给别的处理线程。而io_device和io_channel的设计保证了资源的抽象访问以及独立的路径不去争抢资源池,并且块设备由于是对块进行操作的所以也十分适合抽象成io_device。正是因为以上几点才让spdk线程模型能够达到无锁化且为多个target提供了基础线程框架的支持。
本文福利, 免费领取C++学习资料包、技术视频/代码,1000道大厂面试题,内容包括(C++基础,网络编程,数据库,中间件,后端开发,音视频开发,Qt开发)↓↓↓↓↓↓见下面↓↓文章底部点击免费领取↓↓
文章浏览阅读2.5k次,点赞2次,收藏5次。迟到的文章,就当库存发出来吧~_jangow01
文章浏览阅读1.7w次,点赞2次,收藏5次。默认情况下RDD的transformation是lazy形式,实际计算只有在ation时才会进行,而且rdd的计算结果默认都是临时的,用过即丢弃,每个action都会触发整个DAG的从头开始计算,因此在迭代计算时都会想到用cache或persist进结果进行缓存。敝人看到很多资料或书籍有的说是persist或cache会触发transformation真正执行计算,有的说是不会!敝人亲自实验了一把..._spark cache和persist不生效
文章浏览阅读2.4k次。HTML之marquee(文字滚动)详解语法:以下是一个最简单的例子:代码如下:Hello, World下面这两个事件经常用到:onMouseOut=“this.start()” :用来设置鼠标移出该区域时继续滚动onMouseOver=“this.stop()”:用来设置鼠标移入该区域时停止滚动代码如下:onMouseOut=“this.start()” :用来设置鼠标移出该区域时继续滚动 onMouseOver=“this.stop()”:用来设置鼠标移入该区域时停止滚动这是一个完_html滚动
leecode++理解:本文介绍了一些LeetCode题目的解析和思路,包括两数相加、寻找有序数组的中位数、最长字串、整数反转等。还讨论了一些下标规律和正则表达式匹配问题。
文章浏览阅读8.5k次,点赞2次,收藏28次。UDPServer.javaimport java.io.IOException;import java.net.DatagramPacket;import java.net.DatagramSocket;import java.net.InetAddress;/* * 服务器端,实现基于UDP的用户登陆 */public class UDPServer { public s_java xsocket jar包,实现udp服务端
文章浏览阅读692次,点赞2次,收藏3次。规划采购管理是记录项目采购决策、明确采购方法,及识别潜在卖方的过程。本过程的主要作用 是,确定是否从项目外部获取货物和服务,如果是,则还要确定将在什么时间、以什么方式获取什 么货物和服务。货物和服务可从执行组织的其他部门采购,或者从外部渠道采购。本过程仅开展一 次或仅在项目的预定义点开展。图 12-2 描述本过程的输入、工具与技术和输出。图 12-3 是本过程的 数据流向图。..._pmp项目管理实施
文章浏览阅读2.3k次,点赞3次,收藏28次。数据挖掘实践(金融风控-贷款违约预测)(三):特征工程目录数据挖掘实践(金融风控-贷款违约预测)(三):特征工程1.引言2.特征预处理2.1缺失值填充2.2时间格式处理2.3类别特征处理3.异常值处理3.1 检测异常的方法一:正态分布法3.2 检测异常的方法二:箱型图3.3异常值的处理方法4.数据分桶5.特征交互6.特征编码6.1 labelEncode 直接放入树模型中6.2 逻辑回归等模型要单独增加的特征工程7.特征选择7.1 Filter7.2 Wrapper (Recursive feature _金融风控(大数据)特征工程
文章浏览阅读1.1k次,点赞7次,收藏11次。2️⃣若失败消息提示需要更换http,去阿里云中找到相关镜像 测试连接通过后重新下载驱动。3️⃣上述方法还是不行 可以关闭data grip后再来几遍。1️⃣首先将电脑连接的Wi-Fi关闭 连接手机热点。_datagrip插件下载失败
文章浏览阅读162次。一、介绍Npm包管理工具,JS开发者能够更方便的分享和复用以及更新代码,被复用的代码被称为包或者模块,一个模块中包含了一到多个js文件。在模块中一般还会包含一个package.json的文件,该文件中包含了该模块的配置信息。一个完整的项目,需要依赖很多个模块。一个完整的npm包含三部分npm网站-----用于预览npm管理的包注册机制-----用于上传包,使用数据库来维护包与上传者..._h5页面能使用npm包管理工具么
文章浏览阅读154次。如果是通过chrome调用摄像头的话:navigator.mediaDevices.enumerateDevices().then(gotDevices).catch(errorCallback);...function gotDevices(deviceInfos) {...for (var i = 0; i !== deviceInfos.length; ++i) {var deviceInf..._navigator.mediadevices.enumeratedevices() 将摄像头渲染到 video
文章浏览阅读337次。TPM(Trusted Platform Module)安全芯片,是指符合TPM(可信赖平台模块)标准的安全芯片。标准由TCG(可信赖计算组织,Trusted Computing Group)提出,目前最新版本为2.0。符合TPM的芯片首先必须具有产生加解密密钥的功能,此外还必须能够进行高速的资料加密和解密,以及充当保护BIOS和操作系统不被修改的辅助处理器。TPM的可信基础来源于可信根,可信根(..._tpm 总线监听
文章浏览阅读6.7w次。也就更容易获得更高的浏览量。,我们称为午高峰,这个时间段主要是针对一二线城市的上班族,因为玩抖音的一二线城市的人比较多,所以这个时间段他们基本都是下班的时间段,刷抖音的人也很多。,我们成为晚高峰,这个时间段的人基本都忙完工作在休息了,这个时间段可以说是一天中抖音流量最高的时间段,是高峰中的高峰。,这一个时间段的人大都是刚刚睡醒,躺在床上刷一刷抖音,或者在上班的路上没事看看抖音,坐公交的路上刷着抖音。,我们称之为午高峰,这个时间段是人们的午休时间,这个时间段刷抖音的人也很多,吃完饭午休,拿着手机刷刷抖音。_几点发抖音浏览量最高