tbb - 入力の 1 つが空のバッファである場合、TBB join_node の入力を無視する方法は?

Question

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, unlimited,
                             [](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;
}