| 373 | } |
| 374 | |
| 375 | void process_distributed_node(std::vector<bmf_engine::NodeConfig> &nodes) { |
| 376 | NodeConfig *upstream_node = nullptr; |
| 377 | int nodes_index = 0; |
| 378 | while ((nodes.begin() + nodes_index) != nodes.end()) { |
| 379 | NodeConfig *node = &nodes[nodes_index]; |
| 380 | if (!(node->get_dist_nums() > 1)) { |
| 381 | upstream_node = node; |
| 382 | } else if (upstream_node) { |
| 383 | int dist_nums = node->get_dist_nums(); |
| 384 | // preallocate space for new nodes |
| 385 | // two internal nodes and dist_nums - 1 dist nodes |
| 386 | nodes.reserve(nodes.size() + dist_nums + 1); |
| 387 | // repoint to memory address after allocation |
| 388 | node = &nodes[nodes_index]; |
| 389 | upstream_node = nullptr; |
| 390 | // create and insert split node |
| 391 | auto split_node = create_split_node(nodes.size(), |
| 392 | node->get_input_streams()[0], |
| 393 | nodes.size(), dist_nums); |
| 394 | // split_node.set_output_manager("split"); |
| 395 | nodes.push_back(split_node); |
| 396 | |
| 397 | node->change_input_stream_identifier(split_node.output_streams[0]. |
| 398 | get_identifier()); |
| 399 | // store input streams for assemble node |
| 400 | std::vector<StreamConfig> assemble_input_streams; |
| 401 | assemble_input_streams.push_back(node->output_streams[0]); |
| 402 | |
| 403 | // creat and insert copies of the current node |
| 404 | for (int i = 1; i < dist_nums; ++i) { |
| 405 | auto new_node = NodeConfig(*node); |
| 406 | new_node.set_id(nodes.size()); |
| 407 | new_node.change_input_stream_identifier(split_node.output_streams[i]. |
| 408 | get_identifier()); |
| 409 | new_node.change_output_stream_identifier(); |
| 410 | new_node.set_dist_nums(1); |
| 411 | new_node.set_scheduler(new_node.get_id()); |
| 412 | assemble_input_streams.push_back(new_node.output_streams[0]); |
| 413 | nodes.push_back(new_node); |
| 414 | } |
| 415 | node->set_dist_nums(1); |
| 416 | |
| 417 | // creat and insert assemble node |
| 418 | auto assemble_node = create_assemble_node(nodes.size(), |
| 419 | assemble_input_streams, |
| 420 | nodes.size(), 1); |
| 421 | nodes.push_back(assemble_node); |
| 422 | |
| 423 | // link downstream node's inputstream and assemble node's outputstream |
| 424 | for (auto &tem_node : nodes) { |
| 425 | for (auto &input_stream : tem_node.input_streams) { |
| 426 | if (input_stream.get_identifier() == |
| 427 | node->output_streams[0].get_identifier() && |
| 428 | tem_node.get_id() != assemble_node.get_id()) { |
| 429 | tem_node.change_input_stream_identifier((assemble_node. |
| 430 | get_output_streams())[0]. |
| 431 | get_identifier()); |
| 432 | } |
no test coverage detected