diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 817104e4..75a773dc 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -36,5 +36,6 @@ Contributors: - [Linyu](https://github.com/weijinglin) - [billlib](https://github.com/billlib) - [SimonFoobar648](https://github.com/SimonFoobar648) +- [LJX1021403](https://github.com/LJX1021403) 感谢以上朋友,为CGraph项目做出的贡献,排名以贡献时间前后为顺序。 diff --git a/README.md b/README.md index 646fca28..ff778cc4 100644 --- a/README.md +++ b/README.md @@ -108,13 +108,13 @@ from pycgraph import GNode, GPipeline, CStatus class MyNode1(GNode): def run(self): - print("[{0}] {1}, enter MyNode1 run function. Sleep for 1 second ... ".format(datetime.now(), self.getName())) + print("[{0}] {1}, sleep for 1 second ... ".format(datetime.now(), self.getName())) time.sleep(1) return CStatus() class MyNode2(GNode): def run(self): - print("[{0}] {1}, enter MyNode2 run function. Sleep for 2 second ... ".format(datetime.now(), self.getName())) + print("[{0}] {1}, sleep for 2 second ... ".format(datetime.now(), self.getName())) time.sleep(2) return CStatus() diff --git a/example/BUILD b/example/BUILD index acdd5c5b..02acb701 100644 --- a/example/BUILD +++ b/example/BUILD @@ -1,4 +1,4 @@ -# test-1: E01-AutoPilot +# E01-AutoPilot cc_binary ( name = "E01-AutoPilot", srcs = ["E01-AutoPilot.cpp"], @@ -6,7 +6,7 @@ cc_binary ( deps = ["//src:CGraph",], ) -# test-2: E02-MockGUI +# E02-MockGUI cc_binary ( name = "E02-MockGUI", srcs = ["E02-MockGUI.cpp"], @@ -14,7 +14,7 @@ cc_binary ( deps = ["//src:CGraph",], ) -# test-3: E03-ThirdFlow +# E03-ThirdFlow cc_binary ( name = "E03-ThirdFlow", srcs = ["E03-ThirdFlow.cpp"], @@ -22,10 +22,26 @@ cc_binary ( deps = ["//src:CGraph",], ) -# test-4: E04-MapReduce +# E04-MapReduce cc_binary ( name = "E04-MapReduce", srcs = ["E04-MapReduce.cpp"], copts = ["-Isrc/"], deps = ["//src:CGraph",], ) + +# E05-HttpServer +cc_binary ( + name = "E05-HttpServer", + srcs = ["E05-HttpServer.cpp"], + copts = ["-Isrc/"], + deps = ["//src:CGraph",], +) + +# E06-ParallelSort +cc_binary ( + name = "E06-ParallelSort", + srcs = ["E06-ParallelSort.cpp"], + copts = ["-Isrc/"], + deps = ["//src:CGraph",], +) diff --git a/example/CMakeLists.txt b/example/CMakeLists.txt index b36c4074..ef6925a8 100644 --- a/example/CMakeLists.txt +++ b/example/CMakeLists.txt @@ -5,6 +5,7 @@ set(CGRAPH_EXAMPLE_LIST E03-ThirdFlow E04-MapReduce E05-HttpServer + E06-ParallelSort ) foreach(example ${CGRAPH_EXAMPLE_LIST}) diff --git a/example/E06-ParallelSort.cpp b/example/E06-ParallelSort.cpp new file mode 100644 index 00000000..465dce88 --- /dev/null +++ b/example/E06-ParallelSort.cpp @@ -0,0 +1,148 @@ +/*************************** +@Author: LJX1021403 +@Contact: 2898876973@qq.com +@File: E06-ParallelSort.cpp +@Time: 2026/9/2 +@Desc: 本example展示如何利用CGraph并行执行多个排序任务。 + 将一个大数组分成4块,分别交给4个SortGNode并行排序 +***************************/ + +#include +#include +#include +#include + +#include "CGraph.h" + +using namespace CGraph; + +static const int PART_COUNT = 4; // 4个并行排序节点 +static const int NUMS_PER_PART = 250000; // 每个节点处理25万个数字 +static const int TOTAL_SIZE = PART_COUNT * NUMS_PER_PART; // 总共100万个数字 +static auto PARAM_KEY = "sort-param-key"; + +// 用于在节点间共享数据的参数结构 +struct SortGParam : public GParam { + std::vector data_; // 待排序数据 + std::vector result_; + +protected: + CStatus setup() override { + result_.reserve(TOTAL_SIZE); + data_.resize(TOTAL_SIZE); + return CStatus(); + } +}; + + +// 生成随机数据的节点 +class GenerateGNode : public GNode { +public: + CStatus init() override { + return CGRAPH_CREATE_GPARAM(SortGParam, PARAM_KEY); + } + + CStatus run() override { + const auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); + std::mt19937 generator{}; + std::uniform_int_distribution distribution(0, TOTAL_SIZE); + for (auto& val : param->data_) { + val = distribution(generator); + } + return CStatus(); + } +}; + + +// 排序节点:负责对数据中的某一段进行排序 +template +class SortGNode : public GNode { +public: + CStatus run() override { + auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); + int begin = PART_TAG * NUMS_PER_PART; + int end = begin + NUMS_PER_PART; + std::sort(param->data_.begin() + begin, param->data_.begin() + end); + return CStatus(); + } +}; + + +// 合并节点:使用四路归并将四个已排序的片段合并成整体有序数组 +class MergeGNode : public GNode { +public: + CStatus run() override { + const auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); + auto& data = param->data_; + auto& result = param->result_; + + // 四个有序区间的起始索引和结束索引 + const std::vector> ranges = { + {0, NUMS_PER_PART}, + {1 * NUMS_PER_PART, 2 * NUMS_PER_PART}, + {2 * NUMS_PER_PART, 3 * NUMS_PER_PART}, + {3 * NUMS_PER_PART, TOTAL_SIZE} + }; + + // 四路归并 + std::vector indices(PART_COUNT); + std::vector ends(PART_COUNT); + for (int i = 0; i < PART_COUNT; ++i) { + indices[i] = ranges[i].first; + ends[i] = ranges[i].second; + } + + while (result.size() < TOTAL_SIZE) { + int minVal = INT32_MAX; + int minIdx = -1; + for (int i = 0; i < PART_COUNT; ++i) { + if (indices[i] < ends[i] && data[indices[i]] < minVal) { + minVal = data[indices[i]]; + minIdx = i; + } + } + if (minIdx != -1) { + result.emplace_back(minVal); + indices[minIdx]++; + } + } + return CStatus(); + } +}; + + +class CheckGNode : public GNode { +public: + CStatus run() override { + const auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); + const auto& result = param->result_; + + bool pass = std::is_sorted(result.begin(), result.end()); + std::cout << "parallel sort check result is: "<< (pass ? "PASS" : "FAIL") << std::endl; + return CStatus(); + } +}; + + +void example_parallel_sort() { + auto pipeline = GPipelineFactory::create(); + GElementPtr a, b0, b1, b2, b3, c, d = nullptr; + + CStatus status; + status += pipeline->registerGElement(&a, {}); + status += pipeline->registerGElement>(&b0, {a}); + status += pipeline->registerGElement>(&b1, {a}); + status += pipeline->registerGElement>(&b2, {a}); + status += pipeline->registerGElement>(&b3, {a}); + status += pipeline->registerGElement(&c, {b0, b1, b2, b3}); + status += pipeline->registerGElement(&d, {c}); + + pipeline->process(); + + GPipelineFactory::clear(); +} + +int main() { + example_parallel_sort(); + return 0; +} diff --git a/src/GraphCtrl/GraphDefine.h b/src/GraphCtrl/GraphDefine.h index 99fd6131..8596dacd 100644 --- a/src/GraphCtrl/GraphDefine.h +++ b/src/GraphCtrl/GraphDefine.h @@ -9,26 +9,24 @@ #ifndef CGRAPH_GRAPHDEFINE_H #define CGRAPH_GRAPHDEFINE_H -#include "../CBasic/CBasicInclude.h" - CGRAPH_NAMESPACE_BEGIN -static const char* CGRAPH_STR_PIPELINE = "pipeline"; -static const char* CGRAPH_STR_NODE = "node"; -static const char* CGRAPH_STR_CLUSTER = "cluster"; -static const char* CGRAPH_STR_REGION = "region"; -static const char* CGRAPH_STR_CONDITION = "condition"; -static const char* CGRAPH_STR_MULTI_CONDITION = "multi_condition"; -static const char* CGRAPH_STR_SOME = "some"; -static const char* CGRAPH_STR_MUTABLE = "mutable"; -static const char* CGRAPH_STR_FUNCTION = "function"; -static const char* CGRAPH_STR_SINGLETON = "singleton"; -static const char* CGRAPH_STR_DAEMON = "daemon"; -static const char* CGRAPH_STR_ASPECT = "aspect"; -static const char* CGRAPH_STR_EVENT = "event"; -static const char* CGRAPH_STR_FENCE = "fence"; -static const char* CGRAPH_STR_COORDINATOR = "coordinator"; -static const char* CGRAPH_STR_STAGE = "stage"; +static auto CGRAPH_STR_PIPELINE = "pipeline"; +static auto CGRAPH_STR_NODE = "node"; +static auto CGRAPH_STR_CLUSTER = "cluster"; +static auto CGRAPH_STR_REGION = "region"; +static auto CGRAPH_STR_CONDITION = "condition"; +static auto CGRAPH_STR_MULTI_CONDITION = "multi_condition"; +static auto CGRAPH_STR_SOME = "some"; +static auto CGRAPH_STR_MUTABLE = "mutable"; +static auto CGRAPH_STR_FUNCTION = "function"; +static auto CGRAPH_STR_SINGLETON = "singleton"; +static auto CGRAPH_STR_DAEMON = "daemon"; +static auto CGRAPH_STR_ASPECT = "aspect"; +static auto CGRAPH_STR_EVENT = "event"; +static auto CGRAPH_STR_FENCE = "fence"; +static auto CGRAPH_STR_COORDINATOR = "coordinator"; +static auto CGRAPH_STR_STAGE = "stage"; CGRAPH_NAMESPACE_END