这是一个我个人对于CGraph的重写,目的是为了方便自己学习CGraph,并且方便自己以后使用。 经过一个多月的学习,了解了CGraph的一些原理,以及一些常用的算法,如今终于开始从零开始写一个CGraph了。 目前已经完成了一个简单的图算法,并且已经完成了一个简单的图算法的测试。 之后这里会记录一些学习到的经验以及思考的内容。
0721更新: 目前已经完成本项目的所有主要功能,基于C11少部分C17(未来可能会修正到C11),完成了一套图流程执行框架,内部支持提供包含依赖元素依次执行、非依赖元素并发执行的功能,支持参数读写传递,循环执行,条件选择执行,相关使用测试在test文件夹中。采用线程池技术,使用RAII和互斥锁等技术保证线程安全,包含核心线程和辅助线程,批量处理,动态扩缩容,任务封装,任务盗窃,分支预测等,相关请看V1.30更新,未来将进行性能测试和优化补充。
在该版本中,实现了基本的图节点依赖,搭建了最基本的框架,最初使用list实现图结构的依赖关系,希望能够通过list方便的进行节点的顺序执行(最开始是像邻接表),在v0.10版本改为使用set
认知: [[nodiscard]]是 C++17 引入的一个标准属性(attribute),用于告诉编译器:该函数的返回值不应被忽略。
在该版本中,实现了基本的线程池
认知: explicit 是一个关键字,用于防止单参数构造函数(或可以接受单参数的多参数构造函数,C++11 起)触发隐式类型转换。
std::future<T> 表示一个将来会返回 T 类型结果的异步返回值。
在之前的linux的RPC框架开发中,个人通常使用pthread_mutex_t, semaphore.h等库来自行实现互斥锁,该版本的开发中使用c++11引入的lock_guard 来实现管理锁,更方便使用更安全,在使用锁的区域使用{}标注作用域
当前 GraphThreadPool 在任务调度上采用逐个提交、多次加锁的方式,存在性能瓶颈; 而在RPC框架开发中 EventLoop 多线程模型通过批量提交 + 单次唤醒的方式,在任务调度效率上更具优势; 将来可以通过引入批量任务接口或无锁队列等方式进行优化,显著提升性能
当前版本中的依赖注册方式过于繁琐,GraphNode 进行节点的依赖关系注册,之后Graphic又将节点注册到自己的图化中,内存重复占用
由于本项目需要频繁进行节点的注册以及拷贝或移动已存在的对象,本项目中采用emplace_back而不是push_back,是考虑到性能的高效,避免额外的拷贝和移动操作,提高项目的性能
在该版本中,优化了节点的依赖关系注册以及实现循环执行,并且是真正意义上的能够用来执行一些具有图依赖的功能
主要新增registerGraphNode函数用于注册节点,addDependNodes函数用于添加依赖关系,checkFinalStatus函数用于检查和重置便于循环。修改前驱和后继的数据结构为set,便于快速查找和获取大小
认知: cout并不是线程安全的,所以在run函数中直接使用cout会混乱,并且在使用中实现run函数时要注意线程安全,测试代码中已经添加简单的锁,已在v1.10版本中将锁规范化
由于set(std::initializer_list<T> list),可以使用initializer_list实现多个节点到set的转换
该版本准备进行版本重构,主要为实现图循环逻辑实现结构重组并设计以下模块, element(元素) 是所有被执行结构的基类,可以派生出node,group两种类型。无实际意义,且不可被执行。
node(节点) 是最小粒度的算子。node本身无法执行,但所有有具体功能的功能节点,都继承自node。
functionNode(功能节点) 是最小粒度的可执行算子,继承自node类,相当于是node的功能实现类。与node不同的是,functionNode有具体功能,且可以被执行。至于具体功能是什么,可以是输出字母a,也可以是去挖比特币。总之,需要自己去实现。
group(组) 是多个functionNode的组合。自身不可以执行,但可以派生出cluster和region等组合逻辑。
cluster(簇) 继承自group,由多个functionNode线性组合而成。执行cluster的时候,内部的node依次顺序执行。简而言之就是可以依次(按注册入pipeline的先后顺序)完成多个功能。
region(区域) 继承自group,也是由多个functionNode组合而成。与cluster的区别是,region中的加入的node需要指定相互依赖关系。如果不指定依赖的话,就相当于是并发执行了,因为没有任何需要依赖信息。
pipeline(流水线) 是以上信息运行的地方。所有的functionNode、cluster、region信息,都需要注册到pipeline中,并且设定相互依赖关系。注册了以上三种信息的pipeline,实际上就对应了一个dag图,执行pipeline的过程,就是图执行的过程。 在pipeline中,所有的节点信息都视为element来通过GelementManager类进行统一管理,而在管理类中又可以设置依赖关系,循环次数,串行节点自动解析为cluster。又因为都是继承自element,所以节点之间可以相互套娃,如region内是几个并行的cluster,cluster内是几个串行功能节点。
认知: std::all_of 是 C++ 标准库 <algorithm> 中的一个函数模板,用于检查给定范围内的所有元素是否都满足某个条件。如果范围内所有元素都满足条件,则返回 true;否则返回 false,该条件可使用lambda表达式来自由设定。类似的还有std::any_of, std::none_of
is_base_of在c++11引入,用来判断first是否是second的基类,is_same_v在C++17引入,用来判断两个类型是否相同。
像这种大型版本更新应该开一个新的分支的,新旧代码杂糅在一起让人抓狂。
当前版本重构完成第一阶段,模块均已设计完成,加入了第一个简单测试用例,已经能够实现基本功能,但仍未仔细检查,可能存在未知错误等待检查。
认知: 锁相关:采用RAII模式的锁管理方式。
锁对象:c++11引入std::mutex(互斥锁)与c++17std::shared_mutex(读写锁);(降到C11的主要修改点)
锁管理对象:
std::lock_guard构造时加锁,离开作用域析构解锁,不支持手动解锁或重新加锁,无额外状态存储,开销更小 不可移动或复制。
std::unique_lock用于代替互斥对象的成员函数,管理互斥对象,支持延迟加锁defer_lock、手动解锁unlock()和重新加锁lock(),更灵活。适合需要条件变量或复杂锁管理的场景,需要维护锁状态(如是否已加锁),略有性能损耗,可通过移动语义转移锁的所有权。
要谨防自赋值行为,深层复制时释放自身资源后再尝试访问它们,导致未定义行为有资源重复释放、内存泄漏等风险。
目前GElementManager类中的Analyze函数会重复添加相交的cluster,目前做了去重但是较为复杂且开销大,待完善。
使用 new(std::nothrow) 可以避免程序因内存不足而崩溃,特别是在资源受限的环境中或需要处理大量内存分配的情况下非常有用,可以通过检查返回值是否为 nullptr 来判断内存分配是否成功。
完成功能性测试,目前功能已经基本完成,但似乎存在一些bug,如在T04测试时出现过一次运行错误,重复执行了已经执行过的节点,但已无法复现。
GParam 模块用于参数传递,创建和获取逻辑分离,任何算子节点都可以通过名字获取param类(内部通过哈希表)进行参数传递,为保证多线程安全,使用读写锁提高读速度,独占锁提高写速度,另外可以通过reset方法将所有参数重置为默认值。
新建了分支开发condition模块,意外触发了循环依赖,正在修复中。尝试引入include-what-you-used辅助修复,需要进行环境更换引入clangd和llvm,还需要Virtual studio,网上写的教程也是很垃圾,本身那个项目的README.md写的也不明不白,最后实际上没什么用还降低编译效率,而且为了引入该工具还费尽周折,算是知道为啥网上的教程垃圾了。
引入的clangd功能还行,就是比较严格把之前的一些不影响编译的错误都报了,修复一处Ctime的不安全使用(感觉没啥必要)。代码补全和参数获取也还不错,就是参数太多了会看得眼花
将参数的管理下从Gnode沉到了GElement,这样就可以支持所有类型的参数了,也就能够支持condition的运行设定。
完成了condition模块的编写测试。
在T06的测试过程中,最开始参数存储功能还没完全下沉到GElement,不清楚是哪里的私有泄漏,导致参数获取失败,后来发现是测试程序的哈希查询键值中p的大小写的问题,已修复。
将大部分.inl 文件都移到.h 文件中,虽然我挺喜欢原本的写法将模板函数定义在.inl 文件中在.h末尾引入,但是这样在.inl文件中必须要引入.h文件不然编辑器报未定义错误,引入后clangd会报循环依赖错误,所以只能将.inl 文件中的内容全部移到.h 文件中。
将pipeline模块的依赖添加也下沉到了GElement,目前还未仔细测试,未来版本测试修正
添加了PipelineFactory模块,提供Pipeline的创建方法,可以管理多个Pipeline,为线程池的开发做准备。
逐步实现线程池,目前实现功能有:
UTaskWrapper将任务封装好,使用模板和forward实现完美转发,支持任意可调用对象,只允许使用移动语义的零拷贝实现高性能,内部的任务使用unique_ptr管理防止内存泄漏。
UAtomicQueue构建了一个原子队列,实现弹出和传入任务操作,内部使用RAII锁管理,不会被其他线程中断操作。所有队列均支持获取单个或多个任务来增加扇出。
UWorkStealingQueue构建了一个可盗窃的双端队列。不采用RAII锁而是互斥锁保证线程多次进入的高性能,在push自旋过程中调用 yield(),避免线程持续占用 CPU,同时保持快速响应任务的能力,在pop与原子队列相似,另外有盗窃节点功能将调用时将任务从末尾传出。
UThreadBase维护了一个线程和存放封装任务的UAtomicQueue原子队列。派生PT和ST。
UThreadPrimary是核心线程简称PT,在初始化时创建,内部维护了一个私有的UWorkStealingQueue队列和线程池pool公用的UAtomicQueue队列指针,先执行自己任务队列中的task;如果自身的queue为空,则执行pool中的task;如果仍没有任务,就从相邻的n个PT中去偷窃任务执行(不会放入偷窃队列,这意味着如果偷窃到了执行时间超长的几个任务,只能由自己执行,从而导致任务执行时间过长,所以保守的偷窃任务数量不多)。
UThreadSecondary是辅助线程简称ST,在PT都在执行任务(繁忙执行任务)时创建,但PT+ST不会超过CPU核心*2+1,在自己空闲一段时间后会自动销毁,与PT不同,ST会从全局的UAtomicQueue队列中获取任务执行。
UThreadPool构建了一个线程池,维护一个公用的UAtomicQueue和所有的PT和ST,线程池中PT线程数量为CPU核数,另外创建了一个监控线程监控PT和ST线程,负责动态创建销毁ST实现负载均衡;分发任务依次轮询放入PT私有队列,超过PT数量时放入线程池队列(可以提供给辅助线程执行),超过最大线程数时从头轮询。
另外说明likely(x) __builtin_expect(!!(x), 1)是用来进行执行分支预测,提高执行效率。
已经基本测试完成线程池功能。 目前T08存在一个bug,似乎时平台使用的编译器与库不匹配的问题。
并发使用std::cout打印时数字混乱,与std::cout作用机制有关,具体来说,T08的并发打印100个数字,在极短时间内便可完成多个数字打印任务的commit,但是在使用std::cout<<"2"<<endl时std::cout<<endl是在std::cout<<"2"之后才commit的,这意味着在盗窃队列中的打印顺序可能是1 2 3 4 endl 5 endl endl ,实际上是符合目前的设计期望的。设计上也是希望本线程产生新任务时在本线程中完成,而不是在别的线程中完成。
补充了CmakeLists.txt在linux下编译缺少线程库的错误。
添加了两张T05-Param的火焰图,在实际的测试中500w次读写任务根据核心线程数和最大线程数有所不同,在centos8 四核八G二并行任务运行结果为:130s左右(T05Perf-1.svg),固定核心线程和最大线程数为2时为96s左右。固定核心线程和最大线程数为4时为108s左右(T05Perf-4.svg)。个人认为多余的线程对于并行数较少的任务来说,在火焰图中可以观察到std::yield占据了大部分时间,频繁的上下文切换会增加执行时间。目前的设计中虽然设计了condition_variable来等待任务,但实际并没有使用,主要是为了能够获取多个任务增加扇出。