Graph Partitioning Based on Flow Marks

Description

The front-end representation supports only IR. If you want to express a flow between homogeneous engines, you can use "_flow_attr" to set the flow attribute of Op or SubGraphOp. After partitioning, different flow graphs are generated, performing graph building, deployment, and execution. For subgraphs, the PartitionedCall operator can be used to implement subgraph nesting.

To set "_flow_attr", use either of the following methods:

Use SetAttr to set the flow attributes for the operator. In this case, all inputs and outputs of that operator are flows, and the connected operators will be partitioned into different FlowGraphs.
Use SetInputAttr() or SetOutputAttr() to set the flow attributes for the input or output of the operator. In this case, the input or output of that operator is a flow, and the connected operators will be partitioned into different FlowGraphs.

In the queue-driven scenario, queues are inserted between asynchronous graphs. The queue attributes are as follows:

"_flow_attr": an attribute of the bool type. The value true indicates that the flow attribute exists, and false indicates that the flow attribute does not exist.
_flow_attr_depth: an attribute of the INT type that specifies the queue depth. The value must be greater than or equal to 2 and defaults to 128.
"_flow_attr_enqueue_policy": an attribute of the string type that specifies the enqueuing policy. Only "FIFO" and "OVERWRITE" are supported. "FIFO" means data is enqueued in order and the process will wait for dequeue operations when the queue is full; "OVERWRITE" means no waiting is performed during enqueuing and data will be overwritten circularly. If this attribute is not configured, FIFO is adopted as the default policy.
FlowGraph partitioning is not supported for the V1 control operator structure of while or loop.

Usage

IR graph construction examples:

    auto data1 = op::Data("Data1").set_attr_index(0);
    auto data2 = op::Data("Data2").set_attr_index(1);
    auto data3 = op::Data("Data3").set_attr_index(2);
    auto data4 = op::Data("Data4").set_attr_index(3);

    float inputData1[2] = {1,1};
    Tensor input1(TensorDesc(ge::Shape({2}), FORMAT_ND, DT_FLOAT), (uint8_t *)inputData1, 2 * sizeof(float));  
    input.push_back(input1);

    float inputData2[2] = {1,1};
    Tensor input2(TensorDesc(ge::Shape({2}), FORMAT_ND, DT_FLOAT), (uint8_t *)inputData2, 2 * sizeof(float));
    input.push_back(input2);

    float inputData3[2] = {1,1};
    Tensor input3(TensorDesc(ge::Shape({2}), FORMAT_ND, DT_FLOAT), (uint8_t *)inputData3, 2 * sizeof(float));  
    input.push_back(input3);

    float inputData4[2] = {1,1};
    Tensor input4(TensorDesc(ge::Shape({2}), FORMAT_ND, DT_FLOAT), (uint8_t *)inputData4, 2 * sizeof(float));
    input.push_back(input4);

    auto OP_Reciprocal = op::Reciprocal("Reciprocal").set_input_x(data1);
    // Set the flow attributes for the operator.
    
    const std::string ATTR_NAME_FLOW_ATTR = "_flow_attr";
    const std::string ATTR_NAME_FLOW_ATTR_DEPTH= "_flow_attr_depth";
    const std::string ATTR_NAME_FLOW_ATTR_ENQUEUE_POLICY = "_flow_attr_enqueue_policy";

    OP_Reciprocal.SetAttr(ATTR_NAME_FLOW_ATTR.c_str(), true);
    AscendString policy("FIFO");
    OP_Reciprocal.SetAttr(ATTR_NAME_FLOW_ATTR_ENQUEUE_POLICY.c_str(), policy);

    auto OP_Square = op::Square("Square").set_input_x(data2);
    // Set the flow attributes for the operator.
    OP_Square.SetAttr(ATTR_NAME_FLOW_ATTR.c_str(), true);
    OP_Square.SetAttr(ATTR_NAME_FLOW_ATTR_DEPTH.c_str(), static_cast<int32_t>(16));
    AscendString policy1("");
    OP_Square.SetAttr(ATTR_NAME_FLOW_ATTR_ENQUEUE_POLICY.c_str(), policy1);

    auto OP_Add = op::Add("Add1").set_input_x1(OP_Reciprocal).set_input_x2(OP_Square);
    // Set the flow attributes for the operator.
    OP_Add.SetAttr(ATTR_NAME_FLOW_ATTR.c_str(), true);
    OP_Add.SetAttr(ATTR_NAME_FLOW_ATTR_DEPTH.c_str(), static_cast<int32_t>(16));
    AscendString policy2("OVERWRITE");
    OP_Add.SetAttr(ATTR_NAME_FLOW_ATTR_ENQUEUE_POLICY.c_str(), policy2);
    
    auto OP_Sub = op::Sub("Sub").set_input_x1(data3).set_input_x2(data4);

    auto OP_Mul = op::Mul("Mul").set_input_x1(OP_Add).set_input_x2(OP_Sub);
    auto partition1 = op::PartitionedCall("partitio_001")
	.create_dynamic_input_args(1)
	.set_dynamic_input_args(0, OP_Mul)  
	.create_dynamic_output_output(1)
	.set_subgraph_builder_f(build_graph1);
    // Use PartitionedCall to set the flow attributes.
    partition1.SetAttr(ATTR_NAME_FLOW_ATTR.c_str(), true);
    partition1.SetAttr(ATTR_NAME_FLOW_ATTR_DEPTH.c_str(), static_cast<int32_t>(16));
    AscendString policy3("FIFO");
    partition1.SetAttr(ATTR_NAME_FLOW_ATTR_ENQUEUE_POLICY.c_str(), policy3);
    
    auto result_output = op::Add("Add2").set_input_x1(partition1).set_input_x2(partition1);
    // Set the flow attributes for input and output operators.
    result_output.SetInputAttr(0,ATTR_NAME_FLOW_ATTR.c_str(), true);
    result_output.SetInputAttr(0,ATTR_NAME_FLOW_ATTR_DEPTH.c_str(), static_cast<int32_t>(32));
    AscendString policy4("OVERWRITE");
    result_output.SetInputAttr(0,ATTR_NAME_FLOW_ATTR_ENQUEUE_POLICY.c_str(), policy4);
    result_output.SetOutputAttr(0,ATTR_NAME_FLOW_ATTR.c_str(), true);
    result_output.SetOutputAttr(0,ATTR_NAME_FLOW_ATTR_DEPTH.c_str(), static_cast<int32_t>(32));
    AscendString policy5("OVERWRITE");
    result_output.SetOutputAttr(0,ATTR_NAME_FLOW_ATTR_ENQUEUE_POLICY.c_str(), policy5);
    
    std::vector<Operator> inputs{data1,data2,data3,data4};

    std::vector<Operator> outputs{result_output};
    graph.SetInputs(inputs).SetOutputs(outputs);
    ge::Session* session = new Session(options);
    session->AddGraph(graphId, graph);
    session->RunGraph(graphId, input, output);
    session->RemoveGraph(graphId);
    GEFinalize();

build_graph is defined as follows:

Graph build_graph1(){
Graph graph("build_graph1");
std::vector<Tensor> input;
std::vector<Tensor> output;
auto data1 = op::Data("Data1").set_attr_index(0);
float inputData1[2] = {1,1};
Tensor input1(TensorDesc(ge::Shape({2}), FORMAT_ND, DT_INT32), (uint8_t *)inputData1, 2 * sizeof(float));
input.push_back(input1);
auto output_result=op::Cast("Cast_01").set_input_x(data1).set_attr_dst_type(DT_FLOAT);
std::vector<Operator> inputs{data1};
std::vector<Operator> outputs{output_result};
graph.SetInputs(inputs).SetOutputs(outputs);
return graph;
}

DataFlow graph construction example:

// DataFlow graph construction
auto node0 = dflow::FlowNode("node0", 2, 1)         // Create a FlowOperator instance of the FlowNode type.
  .SetInput(0, data0)                     // Set data0 as the first input of FlowNode.
  .SetInput(1, data1);                    // Set data1 as the second input of FlowNode.
// Set flow attributes.
node0.SetAttr("_flow_attr", true);
node0.SetAttr("_flow_attr_depth", static_cast<int32_t>(108));
AscendString policy1("FIFO");
node0.SetAttr("_flow_attr_enqueue_policy", policy1);

Parent topic: Special Topics