From ba89a94b9cbf66c8b4088c32a7f1dbf5254b25c3 Mon Sep 17 00:00:00 2001 From: LJX1021403 <2898876973@qq.com> Date: Wed, 2 Sep 2026 17:58:23 +0800 Subject: [PATCH 1/2] Add parallel sort example (#220) (#606) * Add parallel sort example (#220) * Improve parallel sort example: use 4-way merge and compare with serial sort --- CONTRIBUTORS.md | 1 + example/CMakeLists.txt | 1 + example/E06-ParallelSort.cpp | 168 +++++++++++++++++++++++++++++++++++ 3 files changed, 170 insertions(+) create mode 100644 example/E06-ParallelSort.cpp 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/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..3598c362 --- /dev/null +++ b/example/E06-ParallelSort.cpp @@ -0,0 +1,168 @@ +/*************************** +@Author: LJX1021403 +@Contact: 2898876973@qq.com +@File: E06-ParallelSort.cpp +@Time: 2026/9/2 +@Desc: 本example展示如何利用CGraph并行执行多个排序任务。 + 将一个大数组分成4块，分别交给4个SortGNode并行排序， + 最后通过四路归并合并结果，并与串行std::sort对比性能。 +***************************/ + +#include +#include +#include +#include +#include +#include +#include + +#include "CGraph.h" + +using namespace CGraph; + +std::mutex g_cout_mutex; + +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 const char* PARAM_KEY = "sort-param-key"; + +// 用于在节点间共享数据的参数结构 +struct SortGParam : public GParam { + std::vector data_; // 待排序数据 +}; + +// 生成随机数据的节点 +class GenerateGNode : public GNode { +public: + CStatus init() override { + return CGRAPH_CREATE_GPARAM(SortGParam, PARAM_KEY); + } + + CStatus run() override { + auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); + std::mt19937 generator; + std::uniform_int_distribution distribution(0, 1000000); + param->data_.resize(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 { + auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); + auto& data = param->data_; + + // 四个有序区间的起始索引和结束索引 + std::vector> ranges = { + {0, NUMS_PER_PART}, + {NUMS_PER_PART, 2 * NUMS_PER_PART}, + {2 * NUMS_PER_PART, 3 * NUMS_PER_PART}, + {3 * NUMS_PER_PART, TOTAL_SIZE} + }; + + // 四路归并 + std::vector merged; + merged.reserve(TOTAL_SIZE); + std::vector indices(4); + std::vector ends(4); + for (int i = 0; i < 4; ++i) { + indices[i] = ranges[i].first; + ends[i] = ranges[i].second; + } + + while (merged.size() < TOTAL_SIZE) { + int minVal = INT32_MAX; + int minIdx = -1; + for (int i = 0; i < 4; ++i) { + if (indices[i] < ends[i] && data[indices[i]] < minVal) { + minVal = data[indices[i]]; + minIdx = i; + } + } + if (minIdx == -1) break; + merged.push_back(minVal); + indices[minIdx]++; + } + + std::copy(merged.begin(), merged.end(), data.begin()); + + bool sorted = std::is_sorted(data.begin(), data.end()); + std::cout << "MergeGNode finished. Data sorted: " << (sorted ? "YES" : "NO") << std::endl; + return CStatus(); + } +}; + +void example_parallel_sort() { + // 用于串行排序基准的数据 + std::vector initial_data(TOTAL_SIZE); + std::mt19937 generator; + std::uniform_int_distribution distribution(0, 1000000); + for (auto& val : initial_data) { + val = distribution(generator); + } + auto serial_data = initial_data; // 拷贝一份用于串行排序 + + // 1. 串行排序基准 + auto start_serial = std::chrono::steady_clock::now(); + std::sort(serial_data.begin(), serial_data.end()); + auto end_serial = std::chrono::steady_clock::now(); + double serial_ms = std::chrono::duration(end_serial - start_serial).count(); + + // 2. 并行排序 + auto pipeline = GPipelineFactory::create(); + GElementPtr a, b0, b1, b2, b3, c = 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}); + + if (status.isErr()) { + CGRAPH_ECHO("register error: %s", status.getInfo().c_str()); + return; + } + + auto start_parallel = std::chrono::steady_clock::now(); + status += pipeline->process(); + auto end_parallel = std::chrono::steady_clock::now(); + double parallel_ms = std::chrono::duration(end_parallel - start_parallel).count(); + + if (status.isErr()) { + CGRAPH_ECHO("process error: %s", status.getInfo().c_str()); + return; + } + + // 输出性能对比 + std::cout << "Serial sort time: " << serial_ms << " ms" << std::endl; + std::cout << "Parallel sort (CGraph) time: " << parallel_ms << " ms" << std::endl; + std::cout << "Speedup: " << serial_ms / parallel_ms << "x" << std::endl; + + GPipelineFactory::clear(); +} + +int main() { + example_parallel_sort(); + return 0; +} \ No newline at end of file From 05510a375b7935890ce0a068729094443a280606 Mon Sep 17 00:00:00 2001 From: Chunel Date: Wed, 2 Sep 2026 22:07:48 +0800 Subject: [PATCH 2/2] [chron] simple change E06 --- README.md | 4 +- example/BUILD | 24 ++++++-- example/E06-ParallelSort.cpp | 114 +++++++++++++++-------------------- src/GraphCtrl/GraphDefine.h | 34 +++++------ 4 files changed, 85 insertions(+), 91 deletions(-) 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/E06-ParallelSort.cpp b/example/E06-ParallelSort.cpp index 3598c362..465dce88 100644 --- a/example/E06-ParallelSort.cpp +++ b/example/E06-ParallelSort.cpp @@ -4,34 +4,37 @@ @File: E06-ParallelSort.cpp @Time: 2026/9/2 @Desc: 本example展示如何利用CGraph并行执行多个排序任务。 - 将一个大数组分成4块，分别交给4个SortGNode并行排序， - 最后通过四路归并合并结果，并与串行std::sort对比性能。 + 将一个大数组分成4块，分别交给4个SortGNode并行排序 ***************************/ -#include -#include #include -#include #include -#include -#include +#include +#include #include "CGraph.h" using namespace CGraph; -std::mutex g_cout_mutex; - 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 const char* PARAM_KEY = "sort-param-key"; +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: @@ -40,10 +43,9 @@ class GenerateGNode : public GNode { } CStatus run() override { - auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); - std::mt19937 generator; - std::uniform_int_distribution distribution(0, 1000000); - param->data_.resize(TOTAL_SIZE); + 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); } @@ -51,6 +53,7 @@ class GenerateGNode : public GNode { } }; + // 排序节点：负责对数据中的某一段进行排序 template class SortGNode : public GNode { @@ -64,72 +67,66 @@ class SortGNode : public GNode { } }; + // 合并节点：使用四路归并将四个已排序的片段合并成整体有序数组 class MergeGNode : public GNode { public: CStatus run() override { - auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); + const auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(SortGParam, PARAM_KEY); auto& data = param->data_; + auto& result = param->result_; // 四个有序区间的起始索引和结束索引 - std::vector> ranges = { + const std::vector> ranges = { {0, NUMS_PER_PART}, - {NUMS_PER_PART, 2 * 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 merged; - merged.reserve(TOTAL_SIZE); - std::vector indices(4); - std::vector ends(4); - for (int i = 0; i < 4; ++i) { + 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 (merged.size() < TOTAL_SIZE) { + while (result.size() < TOTAL_SIZE) { int minVal = INT32_MAX; int minIdx = -1; - for (int i = 0; i < 4; ++i) { + 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) break; - merged.push_back(minVal); - indices[minIdx]++; + if (minIdx != -1) { + result.emplace_back(minVal); + indices[minIdx]++; + } } - - std::copy(merged.begin(), merged.end(), data.begin()); - - bool sorted = std::is_sorted(data.begin(), data.end()); - std::cout << "MergeGNode finished. Data sorted: " << (sorted ? "YES" : "NO") << std::endl; return CStatus(); } }; -void example_parallel_sort() { - // 用于串行排序基准的数据 - std::vector initial_data(TOTAL_SIZE); - std::mt19937 generator; - std::uniform_int_distribution distribution(0, 1000000); - for (auto& val : initial_data) { - val = distribution(generator); + +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(); } - auto serial_data = initial_data; // 拷贝一份用于串行排序 +}; - // 1. 串行排序基准 - auto start_serial = std::chrono::steady_clock::now(); - std::sort(serial_data.begin(), serial_data.end()); - auto end_serial = std::chrono::steady_clock::now(); - double serial_ms = std::chrono::duration(end_serial - start_serial).count(); - // 2. 并行排序 +void example_parallel_sort() { auto pipeline = GPipelineFactory::create(); - GElementPtr a, b0, b1, b2, b3, c = nullptr; + GElementPtr a, b0, b1, b2, b3, c, d = nullptr; CStatus status; status += pipeline->registerGElement(&a, {}); @@ -138,26 +135,9 @@ void example_parallel_sort() { status += pipeline->registerGElement>(&b2, {a}); status += pipeline->registerGElement>(&b3, {a}); status += pipeline->registerGElement(&c, {b0, b1, b2, b3}); + status += pipeline->registerGElement(&d, {c}); - if (status.isErr()) { - CGRAPH_ECHO("register error: %s", status.getInfo().c_str()); - return; - } - - auto start_parallel = std::chrono::steady_clock::now(); - status += pipeline->process(); - auto end_parallel = std::chrono::steady_clock::now(); - double parallel_ms = std::chrono::duration(end_parallel - start_parallel).count(); - - if (status.isErr()) { - CGRAPH_ECHO("process error: %s", status.getInfo().c_str()); - return; - } - - // 输出性能对比 - std::cout << "Serial sort time: " << serial_ms << " ms" << std::endl; - std::cout << "Parallel sort (CGraph) time: " << parallel_ms << " ms" << std::endl; - std::cout << "Speedup: " << serial_ms / parallel_ms << "x" << std::endl; + pipeline->process(); GPipelineFactory::clear(); } @@ -165,4 +145,4 @@ void example_parallel_sort() { int main() { example_parallel_sort(); return 0; -} \ No newline at end of file +} 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