tbb 플로우 그래프에서 노드와 그 자식의 실행을 중단하는 방법

Question

현재 tbb의 플로우 그래프 기능을 테스트 중입니다. 그래프를 사용하려면 그래프에 의존하는 모든 자식을 포함하지만 실행에 의존하지 않는 다른 자식은 그대로두고 그래프의 일부 노드 실행을 중단 할 수 있어야합니다. 본문에서 예외를 throw하거나 task :: cancel_group_execution()을 호출하면 모든 노드의 실행이 중단됩니다. 당신이 그래프의 실행의 일부를 취소 할 수있게하려면

#include <cstdio> 
#include "tbb/flow_graph.h" 

using namespace tbb::flow; 

struct body 
{ std::string my_name; 
    body(const char *name) : my_name(name) 
    { 
    } 
    void operator()(continue_msg) const 
    { if (my_name == "B") 
      tbb::task::self().group()->cancel_group_execution(); 
     else 
     { sleep(1); 
      printf("%s\n", my_name.c_str()); 
     } 
    } 
}; 

int main() 
{ 
    graph g; 

    broadcast_node<continue_msg> start(g); 
    continue_node<continue_msg> a(g, body("A")); 
    continue_node<continue_msg> b(g, body("B")); 
    continue_node<continue_msg> c(g, body("C")); 
    continue_node<continue_msg> d(g, body("D")); 
    continue_node<continue_msg> e(g, body("E")); 

    make_edge(start, a); 
    make_edge(start, b); 
    make_edge(a, c); 
    make_edge(b, c); 
    make_edge(c, d); 
    make_edge(a, e); 

    for (int i = 0; i < 3; ++i) 
     try 
     { start.try_put(continue_msg()); 
      g.wait_for_all(); 
     } catch (...) 
     { printf("Caught exception\n"); 
     } 
    return 0; 
} 

Answer 1

continue_msg 대신에 bool을 사용하여 중단 상태를 나타낼 수 있습니다. 각 process_node은 사용 가능한 경우 선행 노드 상태 및 처리 작업을 수신하고 업데이트 된 중단 상태를 후속 노드로 보냅니다.

struct body 
{ std::string my_name; 
    body(const char *name) : my_name(name) 
    { 
    } 
    bool operator()(bool avail) const 
    { if (!avail) 
      printf("%s skipped\n", my_name.c_str()); 
     else 
      if (my_name == "B") 
      { printf("%s fail\n", my_name.c_str()); 
       avail = false; // fail task 
      } 
      else 
      { sleep(1); 
       printf("%s\n", my_name.c_str()); 
      } 
     return avail; 
    } 
}; 

int main() 
{ 
    graph g; 

    typedef function_node<bool, bool> process_node; 
    typedef std::tuple<bool,bool> bool_pair; 
    broadcast_node<bool> start(g); 
    process_node a(g, unlimited, body("A")); 
    process_node b(g, unlimited, body("B")); 
    process_node c(g, unlimited, body("C")); 
    join_node<bool_pair> join_c(g); 
    function_node<bool_pair, bool> and_c(g, unlimited, [](const bool_pair& in)->bool { 
     return std::get<0>(in) && std::get<1>(in); 
    }); 
    process_node d(g, unlimited, body("D")); 
    process_node e(g, unlimited, body("E")); 

    /* 
    * start -+-> A -+-> E 
    *  |  \ 
    *  |  \ 
    *  |   join_c -> and_c -> C -> D 
    *  |  /
    *  |  /
    *  +-> B -- 
    */ 
    make_edge(start, a); 
    make_edge(start, b); 
    make_edge(a, input_port<0>(join_c)); 
    make_edge(b, input_port<1>(join_c)); 
    make_edge(join_c, and_c); 
    make_edge(and_c, c); 
    make_edge(c, d); 
    make_edge(a, e); 

    for (int i = 0; i < 3; ++i) 
     try 
     { start.try_put(true); 
      g.wait_for_all(); 
     } catch (...) 
     { printf("Caught exception\n"); 
     } 
    return 0; 
} 

Answer 2

, 당신은 task_group_contexts를 사용해야합니다.

#include "tbb/task.h" 

을 다음에 메인 프로그램을 변경 : 다음을 포함 추가

int main() 
{ 
    tbb::task_group_context tgc1; 
    tbb::task_group_context tgc2; 
    graph g1(tgc1); 
    graph g2(tgc2); 
    printf("Constructing graph\n"); 
    broadcast_node<continue_msg> start(g1); 
    continue_node<continue_msg> a(g1, body("A")); 
    continue_node<continue_msg> b(g2, body("B")); 
    continue_node<continue_msg> c(g2, body("C")); 
    continue_node<continue_msg> d(g2, body("D")); 
    continue_node<continue_msg> e(g1, body("E")); 

    make_edge(start, a); 
    make_edge(start, b); 
    make_edge(a, c); 
    make_edge(b, c); 
    make_edge(c, d); 
    make_edge(a, e); 

    for (int i = 0; i < 3; ++i) { 
     try 
     { 
      printf("broadcasting graph %d\n", i); 
      start.try_put(continue_msg()); 
      g1.wait_for_all(); 
      g2.wait_for_all(); 
     } catch (...) 
     { printf("Caught exception\n"); 
     } 
     g1.wait_for_all(); 
     g1.reset(); 
     g2.reset(); 
    } 
    return 0; 
} 

각 task_group_context는 (기본값) 부모 컨텍스트의 서브 컨텍스트입니다. g2를 취소해도 g1에는 영향을주지 않습니다. 취소하는 대신 B가 throw하는 경우 catch는 예외가 부모에게 전달되지 않도록합니다. 당신이 예외를 catch하지 않으면 두 그래프를 기다릴 필요는 A에 대한 상황 및 E.

공지 사항을 완료하는 바와 같이, 부모 컨텍스트는 또한 취소됩니다. 또한 reset() 그래프를 사용하여 continue_nodes '카운터를 재설정해야합니다. 사실, 예외가 발생하여 잡히면 catch(...)이 완료된 후 g1이 완료된다는 보장이 없으므로 try/catch 외부에서 g1.wait_for_all()을 수행해야합니다. 그것을 보여주기 위해 코드를 편집했습니다. 대신 계산의 일부를 정지 취소를 사용

하면 continue_msg의 입력 및 continue_msg 단일 출력 B A multifunction_node 만들 수는 :

:

typedef multifunction_node<continue_msg, tuple<continue_msg> > mf_type; 

struct mf_body { 
    std::string my_name; 
    mf_body(const char *name) : my_name(name) {} 
    void operator()(continue_msg, mf_type::output_ports_type &op) { 
     if(my_name == "B") { 
      printf("%s returning without sending\n", my_name.c_str()); 
      return; 
     } 
     sleep(1); 
     get<0>(op).try_put(continue_msg()); 
     return; 
    } 
}; 

는 그러면 노드 B를 만들

및 C에 대한 B의 에지는

mf_type b(g, unlimited, mf_body("B")); 

과 같이 설정된다 :

make_edge(output_port<0>(b), c); 

이 경우 그래프를 두 개의 하위 그래프로 나눌 필요가 없습니다. 노드 B가 취소 된 경우 대신 후속 노드에 continue_msg을 전달하지 않고 반환합니다. 노드 B가 메시지를 전달하지 않으면 두 개의 노드가 시작되어야하기 때문에 노드 C는 실행되지 않습니다. C의 수를 재설정하려면 그래프를 다시 설정해야합니다.

multifunction_node에는 메시지 전달 여부를 선택할 수있는 장점이 있습니다. 여기서주의해야 할 점은 이 continue_msg 인 입력이 continue_node과 다르다는 것입니다. continue_node은 전신자 (구성시 초기화 값 포함)를 가지고 있으므로 continue_msgs만큼 많은 숫자가 필요합니다. multifunction_node 본문은 얼마나 많은 선행자가 있더라도 continue_msg을 받으면 실행됩니다. 그래서 그래프를 위해서 모든 노드를 만들 수는 없습니다. multifunction_nodes.