La mia implementazione di un thread setTimeout C ++ è corretta?

Question

Ho implementato un setTimeoutthread simile a quello in JavaScript (nuovo per la programmazione dei thread)

Nell'esempio in questa pagina vedo l'uso di una variabile atomica iche penso sia per assicurarsi che non si verifichino condizioni di gara su "i", ma da quello che ho letto non penso che ci sia un atomico multimap.

Dal codice potrebbe sorgere una condizione di competizione sul thread dell'interfaccia utente in queue.emplace(...)e nel thread che iterano su queue.

Il codice del mio thread cerca in generale il lavoro e dovrei usarne un altro condition_variableper bloccarne l' queueaccesso?

---

AGGIORNARE

Penso di aver assolutamente bisogno di rendere queuesicure le manipolazioni. Mi sono imbattuto in vari vicoli ciechi su questo mentre sto imparando a programmare i thread. Alla fine usando un shared_timed_mutexlavorato! Questo tipo di mutex può essere condiviso tra i thread per sincronizzare l'accesso e la manipolazione dei dati, ad esempio è possibile utilizzare

{
   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...
}

Ciascuno *_locksi bloccherà se il mutex è attualmente bloccato, oppure puoi aggiungere parametri aggiuntivi per specificare altri tipi di comportamento. Ogni blocco viene rilasciato dopo l'uscita dall'ambito.

---

Ecco il mio codice originale:

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();
    }
};

Principale:

    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();

Questo codice crea due timeout, il primo è impostato per innescare 15 secondi e il secondo 10 secondi dall'inizio, ma sono impostati in modo tale da testare che il thread riattivi wait_untilcorrettamente, il che funziona davvero:

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

Blocca il mutex setTimeout()

Hai almeno due thread che accedono queue, quindi devi assicurarti che non lo aggiornino contemporaneamente. Stai tenendo il lucchetto all'interno WorkerThread(), ma dovresti anche tenerlo all'interno setTimeout().

Dai alla classe un nome migliore

Sì, la classe utilizza un thread di lavoro per attendere il timeout successivo, ma è più del semplice thread di lavoro. In realtà è una coda di timer, in cui è possibile aggiungere timer che chiamano una funzione quando scade.

class TimerQueue {
    ...
};

Inoltre, setTimeout()sembra che imposti il ​​timeout dell'intero oggetto. Ma aggiunge solo un elemento alla coda. Quindi lo chiamerei addTimer(), o meglio solo add()o insert(), poiché è chiaro dal nome TimerQueueche aggiungeresti dei timer ad esso.

Evita di usare un lambda per la funzione thread

Non è necessario. Perché catturi il contesto in base al valore? Sapevi che cattura ancora thisper riferimento? Basta usare una normale funzione membro per questo. Puoi persino inizializzare il thread senza bisogno di un costruttore, in questo modo:

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

        while (running) {
            ...
        }
    }

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

Hai ancora bisogno di un distruttore per join()il thread, anche se in C ++ 20 questo non è più necessario se usi un file std::jthread.

Assicurati che il distruttore riattivi il thread di lavoro

Il thread di lavoro può dormire fino a 60 secondi se non c'è nulla nella coda. Se si distrugge la coda del timer durante questo periodo, potrebbe essere necessario attendere molto tempo prima che la chiamata termini join(). Assicurati di riattivare il thread nel distruttore:

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

Un'altra opzione è accodare un elemento speciale nella coda che segnala che il thread di lavoro deve arrestarsi e fare in modo che il thread di lavoro esca immediatamente dalla funzione se incontra quell'elemento. Ciò evita la necessità della variabile running.

Evita di usare i system_clocktimer

Il problema system_clockè che può saltare improvvisamente, ad esempio a causa delle modifiche all'ora legale, dei secondi intercalari e degli aggiornamenti NTP. Dovresti usare std::chrono::steady_clockinvece. Ti consiglio di creare un alias di tipo per questo:

using clock = std::chrono::steady_clock;

E poi usalo in questo modo:

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

Considera l'idea di utilizzare un file std::priority_queue

C ++ ha un contenitore appositamente per mantenere le cose ordinate in base alla priorità: std::priority_queue. Considera l'idea di usarlo. L'unico inconveniente è che funziona più come a std::setche a std::map, devi definire una struttura per contenere sia un punto temporale che una funzione di callback, e ordinarla correttamente:

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

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

std::priority_queue<Timer> queue;

Non hai bisogno waitUntil

Sai già la prossima volta che ti sveglierai guardando il primo momento queue.

Evita la duplicazione del codice

All'interno del thread di lavoro, gestisci il caso di una coda vuota e di una coda non vuota. Tuttavia, il codice in entrambi i casi è identico, ad eccezione del punto temporale da attendere fino a. Potresti semplicemente scrivere:

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

Dichiara le costanti come tali

Dichiarate le variabili msMine msMax, e sembrano costanti, ma non ne avete parlato al compilatore. Puoi farli const, o anche meglio static constexpr. Ma per quest'ultimo, devi effettivamente definirli anche in un .cppfile, il che è un po 'fastidioso. Questo problema è stato risolto in C ++ 17, dove è possibile specificarli come static inline constexpr.

Evita l'invalidazione dell'iteratore

Quando elabori attività scadute, chiami queue.erase(), ma hai già notato che devi stare attento a non invalidare l'iteratore. Affidarsi all'incremento dell'iteratore prima di chiamare erase()non è garantito che funzioni. Utilizza invece il valore restituito di erase()come iteratore per l'elemento successivo:

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

Se std::priority_queueinvece usi a , scriverei il codice come:

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

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