Callback paralelo assíncrono C ++ 11

Temos um std::functionque precisa ser chamado após a conclusão de duas tarefas assíncronas diferentes. Queremos executar as duas tarefas ao mesmo tempo e não sabemos qual terminará primeiro.

Aqui está o código mais ou menos:

typedef std::function<void()> Callback;

void asyncCall(const Callback& callback)
{
    std::async(std::launch::async, [callback](){
        callback();
    });
}

void startTaskOne(const Callback& callback)
{
    asyncCall(callback);
}

void startTaskTwo(const Callback& callback)
{
    asyncCall(callback);
}

void startTwoTasks(const Callback& callback)
{
    /* ... */
}

int main()
{
    std::cout << "started" << std::endl;
    startTwoTasks([](){
        std::cout << "finished" << std::endl;
    });
    return 0;
}

Quando fui confrontado pela primeira vez com este problema em C ++, lembrei-me que na verdade é muito semelhante ao resolvido pelas várias bibliotecas javascript assíncronas que podem ser usadas em node.js , onde tudo é assíncrono e pode facilmente acabar no temido inferno de retorno de chamada.

A implementação inicial se parece com isso.

void startTwoTasks(const Callback& callback)
{
    auto counter = std::make_shared<std::atomic<int>>(2);
    Callback parallel([counter, callback](){
        --(*counter);
        if(counter->load() == 0)
        {
            callback();
        }
    });

    startTaskOne(parallel);
    startTaskTwo(parallel);
}

A ideia é criarmos um ponteiro compartilhado para um contador com a quantidade total de etapas e, em seguida, quando cada etapa for concluída, diminuiremos o contador. Quando o contador chegar a zero, todas as etapas terão terminado e podemos chamar o callback. Além disso, neste exemplo, o contador é atômico, dessa forma as etapas podem ser executadas em threads diferentes.

Isso pode ser feito de uma maneira mais agradável, escrevendo uma classe de wrapper.

struct ParallelCallbackData
{
    Callback callback;
    std::atomic<int> counter;
    std::atomic<bool> started;

    ParallelCallbackData(const Callback& callback):
    callback(callback), counter(0), started(false)
    {
    }
};

class ParallelCallback
{
private:
    typedef ParallelCallbackData Data;
    typedef std::shared_ptr<Data> DataPtr;
    DataPtr _data;

    static void step(const DataPtr& data)
    {
        if(data->counter-- == 0)
        {
            if(data->started && data->callback)
            {
                data->callback();
            }
        }
    }

public:

    ParallelCallback(const Callback& callback) :
    _data(new Data(callback))
    {
    }

    operator Callback()
    {
        assert(!_data->started);
        ++_data->counter;
        return std::bind(&ParallelCallback::step, _data);
    }

    void check()
    {
        assert(!_data->started);
        _data->started = true;
        step(_data);
    }

};

Agora o código parece muito mais limpo e é mais fácil adicionar tarefas adicionais.

void startTwoTasks(const Callback& callback)
{
    ParallelCallback parallel(callback);
    startTaskOne(parallel);
    startTaskTwo(parallel);
    parallel.check();
}

A vantagem de usar essa classe é que, ao sobrescrever o operador que converte de volta ao tipo de retorno de chamada original, podemos contar a quantidade de tarefas a serem realizadas. A desvantagem que isso apresenta é que podemos realmente acabar contando várias vezes para cima e para baixo e isso chamaria o retorno de chamada final várias vezes. Para resolver este problema, precisamos adicionar um booleano atômico adicional e um checkmétodo. Além disso, ao invés de pré-incrementar o contador como na primeira implementação, nós o postamos, saindo de uma etapa adicional para a chamada de verificação no final.

Se quiséssemos fazer isso de maneira semelhante ao javascript, poderíamos adicionalmente envolver tudo em um modelo variável.

template<class Arg, class... Args>
void operator ()(Arg&& arg, Args&&... args)
{
    arg(*this);
    operator ()(args...);
}

template<class Arg>
void operator ()(Arg&& arg)
{
    arg(*this);
    check();
}

template<class Arg, class... Args>
static void run(Arg&& arg, Args&&... args)
{
    ParallelCallback parallel(arg);
    parallel(args...);
}

Poderíamos então reduzir o código a uma linha e ele pode ser usado com quantas tarefas forem necessárias.

void startTwoTasks(const Callback& callback)
{
    ParallelCallback::run(callback, startTaskOne, startTaskTwo);
}

O código de trabalho para esses três exemplos pode ser encontrado aqui .