CStatus init() override {

auto status = CGRAPH_CREATE_GPARAM(NumsGParam, PARAM_KEY);

return status;

}

CStatus run() override {

std::mt19937 generator;

auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(NumsGParam, PARAM_KEY);

std::uniform_int_distribution<int> distribution(0, MAX_NUM - 1);

for (int i = 0; i < MAP_SIZE * NUMS_PER_MAP; i++) {

int num = distribution(generator);

param->random_nums_.push_back(num);

}

return CStatus();

}

CStatus run() override {

auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(NumsGParam, PARAM_KEY);

for (int i = 0; i < NUMS_PER_MAP; i++) {

int num = param->random_nums_[MAP_TAG * NUMS_PER_MAP + i];

param->map_results_[MAP_TAG][num]++; // 实际上是线程安全的

}

return CStatus();

}

CStatus run() override {

int total = 0;

auto param = CGRAPH_GET_GPARAM_WITH_NO_EMPTY(NumsGParam, PARAM_KEY);

for (int j = 0; j < MAX_NUM; j++) {

int num = 0;

for (int i = 0; i < MAP_SIZE; i++) {

num += param->map_results_[i][j];

}

total += num;

printf(" num [%d] appear [%d] times\n", j, num);

}

printf("total num is [%d]. \n", total);

return CStatus();

}

auto pipeline = GPipelineFactory::create();

GElementPtr a, b0, b1, b2, b3, c = nullptr;

CStatus status;

status += pipeline->registerGElement<GenerateGNode>(&a, {});

status += pipeline->registerGElement<MapGNode<0>>(&b0, {a});

status += pipeline->registerGElement<MapGNode<1>>(&b1, {a});

status += pipeline->registerGElement<MapGNode<2>>(&b2, {a});

status += pipeline->registerGElement<MapGNode<3>>(&b3, {a});

status += pipeline->registerGElement<ReduceGNode>(&c, {b0, b1, b2, b3});

status += pipeline->process();

if (status.isErr()) {

CGRAPH_ECHO("example_map_reduce error info is : %s", status.getInfo().c_str());

}

GPipelineFactory::clear();