Minha implementação de um thread setTimeout C ++ está correta?

Question

Eu implementei um setTimeoutthread semelhante ao do JavaScript (novo na programação de threads)

No exemplo desta página , vejo o uso de uma variável atômica ique acho que é para garantir que nenhuma condição de corrida ocorra em 'i', mas pelo que li não acho que haja uma atômica multimap.

A partir do código, uma condição de corrida pode surgir no encadeamento da IU queue.emplace(...)e no encadeamento iterando sobre o queue.

O meu código de thread se adapta ao trabalho em geral, e devo usar outro condition_variablepara bloquear o queueacesso?

---

ATUALIZAR

Acho que definitivamente precisava tornar queueseguro o thread de manipulação. Eu entrei em vários becos sem saída enquanto aprendia a programar threads. No final, usando um shared_timed_mutextrabalhado! Este tipo de mutex pode ser compartilhado entre threads para sincronizar o acesso e manipulação de dados, por exemplo, você pode usar

{
   unique_lock<shared_timed_mutex> lock(shared_m);  // for writing
   // write data to whatever...
}

e

{
   shared_lock<shared_timed_mutex> lock(shared_m);  // for reading
   // read data from wherever...
}

Cada um *_lockserá bloqueado se o mutex estiver bloqueado no momento, ou você pode adicionar parâmetros adicionais para especificar outros tipos de comportamento. Cada bloqueio é liberado depois que o escopo é encerrado.

---

Este é meu código original:

WorkerThread.hpp:

using namespace std;
using namespace chrono;

class WorkerThread
{
public:
    typedef chrono::milliseconds Millis;
    typedef function<void(void)> Function;

    bool running = false;
    
    thread t;
    multimap<time_point<system_clock>, Function> queue;  // function queue (sorted)
    condition_variable cv;
    mutex cv_m;
    
    Millis msMin = 1ms;  // lowest sleep time allowed
    Millis msMax = 5ms;  // highest execution time preferred
    time_point<system_clock> waitUntil;  // next wake up time
        
    void setTimeout(Millis ms, Function f) {
    
        // is this line risky? what if the thread is processing queue?
        auto taskTime = system_clock::now() + ms;
        queue.emplace(taskTime, f);

        if(taskTime < waitUntil) {
            cout << "this task is earlier than previously added tasks" << endl;
            cv.notify_all();  // wake up waits in case this timeout task is more recent
        }
    }

    WorkerThread() {
        running = true;
        
        t = thread([=]() {
            std::unique_lock<std::mutex> lk(cv_m);

            while (running == true) {
                
                if(queue.empty()){
                    cout << "empty queue, sleep 60000ms" << endl;

                    // wake up in a minute if there's nothing to do
                    waitUntil = system_clock::now() + 60000ms;

                    // nothing to do, except if woken up
                    if(cv.wait_until(lk, waitUntil) == cv_status::timeout)
                        cout << "thread timed out" << endl;
                    else
                        cout << "thread woken up - earlier task identified !" << endl;
                }
                else {
                    // sleep until next task is ready ("up to" minimum permissible time)
                    waitUntil = max((*queue.begin()).first, system_clock::now() + msMin);

                    cout << "sleeping until next task: " << waitUntil.time_since_epoch().count() << endl;

                    // wait until next task, unless woken up
                    if(cv.wait_until(lk, waitUntil) == cv_status::timeout)
                        cout << "thread timed out" << endl;
                    else
                        cout << "thread woken up - earlier task identified !" << endl;
                }

                // process all available tasks up to maximum execution time
                auto maxtime = system_clock::now() + msMax;

                for(auto task = queue.begin(); task != queue.end(); ) {
                    if((*task).first <= maxtime) {
                        cout << "running task at: " << (*task).first.time_since_epoch().count() << endl;
                        (*task).second();  // run the task

                        // delete the task (the safe way)
                        auto taskSaved = task;
                        task++;
                        queue.erase(taskSaved);
                    }
                    else break; // max exec time reached, exit the for loop
                }
            }
        });
    }

    void stop()
    {
        running = false;
        t.join();
    }
};

A Principal:

    t = new WorkerThread();
    this_thread::sleep_for(1000ms);

    t->setTimeout(15000ms, []() { cout << "Hello from 2" << endl; } );
    cout << "added timeout 1" << endl;

    this_thread::sleep_for(6000ms);
    t->setTimeout(4000ms, []() { cout << "Hello from 1" << endl; } );
    cout << "added timeout 2" << endl;
    
    this_thread::sleep_for(100000ms);
    t->stop();

Este código cria dois tempos limite, o primeiro é definido para acionar 15 segundos e o segundo 10 segundos a partir do início, mas eles são configurados de forma a testar se o encadeamento desperta o do wait_untilcorretamente, o que de fato funciona:

empty queue, sleep 60000ms
this task is earlier than previously added tasks
added timeout 1
thread woken up - earlier task identified !
sleeping until next task: 1600855233135593
this task is earlier than previously added tasks
thread woken up - earlier task identified !added timeout 2

sleeping until next task: 1600855228137566
thread timed out
running task at: 1600855228137566
Hello from 1
sleeping until next task: 1600855233135593
thread timed out
running task at: 1600855233135593
Hello from 2
empty queue, sleep 60000ms

Answer 1

Trave o mutex em setTimeout()

Você tem pelo menos dois threads acessando queue, então você deve garantir que eles não os atualizem simultaneamente. Você está segurando a fechadura por dentro WorkerThread(), mas também deve mantê-la por dentro setTimeout().

Dê um nome melhor para a classe

Sim, a classe usa um thread de trabalho para esperar até o próximo tempo limite, mas é mais do que apenas o thread de trabalho. Na verdade, é uma fila de cronômetro, onde você pode adicionar cronômetros que chamam uma função quando atingem o tempo limite.

class TimerQueue {
    ...
};

Além disso, setTimeout()parece que define o tempo limite de todo o objeto. Mas isso apenas adiciona um elemento à fila. Então, eu o nomearia addTimer(), ou melhor, apenas add()ou insert(), já que é claro pelo nome TimerQueueque você adicionaria temporizadores a ele.

Evite usar um lambda para a função thread

Não é necessário. Por que você está capturando o contexto por valor? Você sabia que ele ainda captura thispor referência? Basta usar uma função de membro regular para isso. Você pode até mesmo ter o thread inicializado sem precisar de um construtor, assim:

class TimerQueue {
    void worker() {
        std::unique_lock<std::mutex> lk(cv_m);

        while (running) {
            ...
        }
    }

    thread workerThread{&TimerQueue::worker, this};
    ...
};

Você ainda precisa de um destruidor para join()o thread, embora no C ++ 20 isso não seja mais necessário se você usar um std::jthread.

Certifique-se de que o destruidor acorde o thread de trabalho

Seu thread de trabalho pode ficar inativo por até 60 segundos se não houver nada na fila. Se você destruir a fila do cronômetro durante esse período, poderá ter que esperar muito até que a chamada join()termine. Certifique-se de ativar o thread no destruidor:

~TimerQueue() {
    std::lock_guard<std::mutex> lk(cv_m);
    running = false;
    cv.notify_one();
    workerThread.join();
}

Outra opção é enfileirar um item especial na fila que sinaliza que o thread de trabalho deve parar e fazer com que o thread de trabalho saia imediatamente da função se encontrar esse item. Isso evita a necessidade da variável running.

Evite usar system_clockpara temporizadores

O problema system_clocké que ele pode pular repentinamente, por exemplo, devido a mudanças no horário de verão, segundos bissextos e atualizações de NTP. Você deve usar em seu std::chrono::steady_clocklugar. Recomendo que você crie um alias de tipo para ele:

using clock = std::chrono::steady_clock;

E então use-o assim:

multimap<clock::time_point, Function> queue;
clock::time_point waitUntil;
...
waitUntil = clock::now() + ...;

Considere usar um std::priority_queue

C ++ tem um recipiente especificamente para manter as coisas organizadas por prioridade: std::priority_queue. Considere usar isso. A única desvantagem é que ele funciona mais como um do std::setque um std::map, você deve definir alguma estrutura para conter um ponto de tempo e uma função de retorno de chamada, e classificá-la corretamente:

struct Timer {
    clock::time_point deadline;
    Function callback;

    bool operator<(const Timer &other) const {
        return other.deadline < deadline;
    }
};

std::priority_queue<Timer> queue;

Você não precisa waitUntil

Você já sabe a próxima vez que acordar olhando o ponto de tempo mais antigo em queue.

Evite duplicação de código

Dentro do thread de trabalho, você lida com o caso de uma fila vazia e uma fila não vazia. No entanto, o código em ambos os casos é idêntico, exceto pelo ponto de tempo de espera até. Você poderia apenas escrever:

waitUntil = clock::now() + queue.empty() ? 60000ms : queue.front().deadline;
cv.wait_until(lk, waitUntil);

Declarar constantes como tais

Você declara as variáveis msMine msMax, e elas parecem constantes, mas não informa o compilador sobre esse fato. Você pode torná-los const, ou até melhores static constexpr. Mas para o último, você também precisa defini-los em um .cpparquivo, o que é um pouco chato. Isso foi corrigido no C ++ 17, onde você pode especificá-los como static inline constexpr.

Evite a invalidação do iterador

Ao processar tarefas que expiraram, você chama queue.erase(), mas já percebeu que deve ter cuidado para não invalidar o iterador. Confiar no incremento do iterador antes da chamada erase()não tem garantia de funcionamento. Em vez disso, use o valor de retorno de erase()como o iterador para o próximo elemento:

for (auto task = queue.begin(); ...) {
    if (...) {
        ...
        task = queue.erase(task);
    } else {
        break;
    }
}

Se você usar um std::priority_queue, eu escreveria o código como:

while (!queue.empty()) {
    auto timer = queue.top();

    if (timer.deadline < maxtime) {
        timer.callback();
        queue.pop();
    } else {
        break;
    }
}