Строительные блоки Intel Threading, source_node типа std :: shared_ptr

Короче говоря, у меня есть источник данных, который предоставляет тип shared_ptr. Этот указатель, похоже, выходит из области видимости при перегрузке оператора source_node. Я добавил полностью упрощенный пример, демонстрирующий проблему.

У меня вопрос: что за элегантность зачем преодолевать эту проблему?

#include <exception>
#include <tbb/concurrent_queue.h>
#include <tbb/flow_graph.h>
#include <condition_variable>
#include <thread>
#include <chrono>
#include <memory>
#include <iostream>

using namespace tbb::flow;

// Plain Data Object
class Data
{
public:
size_t _data;
Data(size_t d) : _data(d){}

};// simpel counting node
class CountingNode
{
private:
size_t _count;
public:
CountingNode():_count(0){}
size_t operator()(std::shared_ptr<Data> data)
{
if (static_cast<bool>(data)) // <-- PROBLEM always false
_count += data->_data;

std::cout << _count << std::endl;
return _count;
}
size_t count() { return _count;}
};

// Source,
// exhausted when a empty/null shared_ptr is encountered
// our datasource provides a shared_ptr to a data object
class SourceNode
{
public:

std::shared_ptr<tbb::concurrent_bounded_queue<std::shared_ptr<Data>>> _queue;
std::shared_ptr<std::mutex> _mtx;
bool _started;

SourceNode() : _started(false)
{
_queue = std::shared_ptr<tbb::concurrent_bounded_queue<std::shared_ptr<Data>>>(new tbb::concurrent_bounded_queue<std::shared_ptr<Data>>());
_mtx = std::shared_ptr<std::mutex>(new std::mutex());
}

SourceNode(const SourceNode& other)
{
_queue = other._queue;
_mtx = other._mtx;
_started = other._started;
}

void push(std::shared_ptr<Data> data)
{
_queue->push(data);
}

bool operator ()(std::shared_ptr<Data> data)
{
std::unique_lock<std::mutex> ul(*_mtx);
{
if (!_started)
{
_started = true;
for(size_t idx = 0; idx != 10; ++idx)
_queue->push(std::shared_ptr<Data>(new Data(idx)));

_queue->push(std::shared_ptr<Data>()); // no more data
}
}

_queue->pop(data);
return static_cast<bool>(data);
}

void close()
{
_queue->push(std::shared_ptr<Data>());
}
};int main(int argc, char* argv[])
{
graph g;
source_node<std::shared_ptr<Data>> source(g,SourceNode(),false);
function_node<std::shared_ptr<Data>,int> sink(g,tbb::flow::serial, CountingNode());
make_edge(source, sink );

source.activate();
g.wait_for_all();

return 0;
}

Если необходимо предоставить более подробную информацию, просто дайте мне знать.

С наилучшими пожеланиями, Ауке-Дирк