如果输入之一是空缓冲区，如何忽略TBB join_node中的输入？

如何解决如果输入之一是空缓冲区，如何忽略TBB join_node中的输入？

我正在尝试修改以下示例 https://software.intel.com/content/www/us/en/develop/blogs/a-feature-detection-example-using-the-intel-threading-building-blocks-flow-graph.html

我的想法是我有一个“预处理”功能应用于每个输入，还有一个“处理”功能应用于每个对的预处理输入，以扫描流。在我的示例中，我们只是简单地计算了连续平方数之间的差。

下面的示例不起作用。我相信这是因为缓冲区一开始是空的。在博客示例中，它们有时会填充缓冲区。调用buffer.try_put(0)似乎无法为我解决。我想念什么？还有什么更好的方法可以让进程自然地工作，只需忽略由于缓冲区为空而无法完全处理的前几个元素？

#include <cstring>
#include <iostream>

#include "tbb/flow_graph.h"

using namespace tbb;
using namespace tbb::flow;

const int N = 13;


template<typename T>
class source_body {

    unsigned my_count;
    int *ninvocations;

public:

    source_body() : ninvocations(NULL) { my_count = 0; }

    source_body(int &_inv) : ninvocations(&_inv) { my_count = 0; }

    bool operator()(T &v) {
        v = (T) my_count++;
        if (ninvocations) ++(*ninvocations);
        if ((int) v < N)
            return true;
        else
            return false;
    }

};

int main() {
    graph g;

    typedef std::tuple<int32_t,int32_t> resource_tuple;

    queue_node<int> buffer(g);
    join_node<resource_tuple,reserving> resource_join(g);

    tbb::flow::source_node<int> src3(g,source_body<int>());
    src3.activate();

    function_node<int,int>
            preprocess_function(g,unlimited,[](const int &a) -> int {

                                    return a * a;
                                }
    );

    make_edge(src3,preprocess_function);

    make_edge(preprocess_function,input_port<1>(resource_join));
    make_edge(preprocess_function,buffer);
    make_edge(buffer,input_port<0>(resource_join));

    function_node<resource_tuple,int>
            process_function(g,[](const resource_tuple &a) -> int {
                                 return std::get<1>(a) - std::get<0>(a);
                             }
    );
    make_edge(resource_join,process_function);


    function_node<int> printint(g,serial,[](const int &t) -> void {
                                   std::cout << t << '\n';
                               }
    );
    make_edge(process_function,printint);

    buffer.try_put(0);
    g.wait_for_all();

    return 0;
}

解决方法

以下代码有效。用reserving代替queueing可以解决问题。但是，我并没有真正忽略无法处理的第一个元素。仍然很想知道在TBB图中实现这种“扫描”的最佳方法是什么。

#include <cstring>
#include <iostream>

#include "tbb/flow_graph.h"

using namespace tbb;
using namespace tbb::flow;

const int N = 13;


template<typename T>
class source_body {

    unsigned my_count;
    int *ninvocations;

public:

    source_body() : ninvocations(NULL) { my_count = 0; }

    source_body(int &_inv) : ninvocations(&_inv) { my_count = 0; }

    bool operator()(T &v) {
        v = (T) my_count++;
        if (ninvocations) ++(*ninvocations);
        if ((int) v < N)
            return true;
        else
            return false;
    }

};

int main() {
    graph g;

    typedef std::tuple<int32_t,int32_t> resource_tuple;

    queue_node<int> buffer(g);
    join_node<resource_tuple,queueing> resource_join(g);

    tbb::flow::source_node<int> src3(g,source_body<int>());
    src3.activate();

    function_node<int,int>
            preprocess_function(g,unlimited,[](const int &a) -> int {

                                    return a * a;
                                }
    );

    make_edge(src3,preprocess_function);

    make_edge(preprocess_function,input_port<1>(resource_join));
    make_edge(preprocess_function,buffer);
    make_edge(buffer,input_port<0>(resource_join));

    function_node<resource_tuple,int>
            process_function(g,[](const resource_tuple &a) -> int {
                                 return std::get<1>(a) - std::get<0>(a);
                             }
    );
    make_edge(resource_join,process_function);

    function_node<int> printint(g,serial,[](const int &t) -> void {
                                    std::cout << t << '\n';
                                }
    );

    make_edge(process_function,printint);

    buffer.try_put(0);
    g.wait_for_all();

    return 0;
}