技术标签: 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开发)↓↓↓↓↓↓见下面↓↓文章底部点击免费领取↓↓
文章浏览阅读597次。/*找出10000以内的同构数同构数 376*376=141376思路:1、输入一个数num 先判断是几位数。记住数位length。 2、然后算它(num)的平方, square。 3、取square的后length位的数值temp 4、temp与num相等,则是同构数。*/#include <iostream>#include &..._小于10000的同构数
文章浏览阅读5.1k次,点赞3次,收藏26次。写了很久的语音呼叫功能、调用在线语音合成的调用系统自带的;现在客户又要求搞网页版的语音呼叫还是不带联网的。客户太难伺候了详细使用请参考 【web语音API】完整代码<!DOCTYPE><html xmlns="http://www.w3.org/1999/xhtml" lang="zh-CN"><head><title>网页文字转语音</title><meta http-equiv="Content-Typ._js tts
文章浏览阅读62次。System.out.println("100个和尚吃了100个馒头 ,100和尚有大和尚和小和尚,一个大和尚能吃3馒头,三个小和尚吃1个馒头,问大和尚和小和尚有多少个?System.out.println("大和尚有"+i+"个人");System.out.println("小和尚有"+j+"个人");System.out.println("查看答案请按回车键");
文章浏览阅读651次。在vscode中是用模块化的时候会出现报错,提示如下Access to script at ‘file:///F:/%E5%AD%A6%E4%B9%A0/%E7%BA%BF%E4%B8%8BJS/test/js./modul.js’ from origin ‘null’ has been blocked by CORS policy: Cross origin requests are only supported for protocol schemes: http, data, chrome, ch_indesssss.html:1 access to script at 'file:///i:/vscode/cheshi/tesss.js' fro
文章浏览阅读218次。为什么80%的码农都做不了架构师?>>> ..._h3c virtual converged framework切片
文章浏览阅读1.9w次,点赞44次,收藏268次。AndroidIOSHarmonyOS (鸿蒙)文档概览-HarmonyOS应用开发官网2.1.1 系统的定位搭载该操作系统的设备在系统层⾯融为⼀体、形成超级终端,让设备的硬件能⼒可以弹性 扩展,实现设备之间 硬件互助,资源共享。对消费者⽽⾔,HarmonyOS能够将⽣活场景中的各类终端进⾏能⼒整合,实现不同终端 设备之间的快速连接、能⼒互助、资源共享,匹配合适的设备、提供流畅的全场景体验。⾯向开发者,实现⼀次开发,多端部署。_鸿蒙移动应用开发
文章浏览阅读47次。定义:允许将对象组成树形结构来表现 “整体/部分” 层次结构。组合能让客户以一致的方式处理个别对象及对象组合。说白了,就是类似于树形结构。 只是它要求子节点和父节点都具备统一的接口。类图如下:示例如下:比如我们常见的电脑上的目录,目录下面有文件夹,也有文件,然后文件夹里面还有文件及文件夹。这样一层层形成了树形结构。示例代码如下:#include <iostream>#include <stdio.h>#include "string"#includ..
文章浏览阅读1.9w次,点赞26次,收藏185次。目录一.请简述下什么是kotlin?它有什么特性?二.Kotlin 中注解 @JvmOverloads 的作用?三.Kotlin中的MutableList与List有什么区别?四.kotlin实现单例的几种方式?五. kotlin中关键字data的理解?相对于普通的类有哪些特点?六.什么是委托属性?简单说一下应用场景?七.kotlin中with、run、apply、let函数的区别?一般用于什么场景?八.kotlin中Unit的应用以及和Java中void的区别?九.Ko_kotlin面试题
文章浏览阅读2.8k次。有这个想法一方面是确实很多时候会记不得一些缩写是什么意思。另外也是受 http://blog.csdn.net/lin453701006/article/details/52797415这篇博客的启发,本文主要用于自己记忆 内容主要整理自http://blog.sina.com.cn/s/blog_520811730101hmj9.html http://blog.csdn.net/feix_反量化 英文缩写
文章浏览阅读7.3k次,点赞6次,收藏36次。超级简单的Python爬虫入门教程(非常详细),通俗易懂,看一遍就会了_爬虫python入门
文章浏览阅读1.2k次。您的代码存在一些问题。首先,您在此处显示的两个模型是not等效的:尽管您将scikit-learn LogisticRegression设置为fit_intercept=True(这是默认设置),但您并没有这样做statsmodels一;来自statsmodels docs:默认情况下不包括拦截器,用户应添加。参见statsmodels.tools.add_constant。另一个问题是,尽管您处..._sm fit(method
文章浏览阅读518次。一、sfml官网下载32位的版本 一样的设置,64位的版本我没有成功,用不了。二、三、四以下这些内容拷贝过去:sfml-graphics-d.libsfml-window-d.libsfml-system-d.libsfml-audio-d.lib..._vsllfqm