Mon implémentation d'un thread setTimeout C ++ est-elle correcte?

Question

J'ai implémenté un setTimeoutthread similaire à celui de JavaScript (nouveau dans la programmation de threads)

Dans l'exemple de cette page, je vois l'utilisation d'une variable atomique iqui, je pense, permet de s'assurer qu'aucune condition de course ne se produit sur «i», mais d'après ce que j'ai lu, je ne pense pas qu'il y ait un atomique multimap.

À partir du code, une condition de concurrence peut survenir sur le thread d'interface utilisateur au niveau queue.emplace(...)et dans le thread itérant sur le queue.

Est-ce que mon code de thread recherche généralement le travail et devrais-je en utiliser un autre condition_variablepour bloquer l' queueaccès?

---

MISE À JOUR

Je pense que j'avais vraiment besoin de sécuriser les queuefils de manipulation. Je suis tombé dans diverses impasses à ce sujet en apprenant à programmer des threads. En fin de compte en utilisant un shared_timed_mutextravaillé! Ce type de mutex peut être partagé entre les threads pour synchroniser l'accès aux données et la manipulation, par exemple vous pouvez utiliser

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

et

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

Chacun *_lockse bloquera si le mutex est actuellement verrouillé, ou vous pouvez ajouter des paramètres supplémentaires pour spécifier d'autres types de comportement. Chaque verrou est libéré après la sortie de l'oscilloscope.

---

Voici mon code d'origine:

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

Ce code crée deux délais, le premier est configuré pour déclencher 15 secondes et le second 10 secondes à partir du début, mais ils sont configurés de manière à tester le thread réveille wait_untilcorrectement le 's, ce qui fonctionne en effet:

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

Verrouillez le mutex setTimeout()

Vous avez au moins deux threads accédant queue, vous devez donc vous assurer qu'ils ne le mettent pas à jour simultanément. Vous tenez la serrure à l'intérieur WorkerThread(), mais vous devez également la tenir à l'intérieur setTimeout().

Donnez un meilleur nom à la classe

Oui, la classe utilise un thread de travail pour attendre le prochain délai d'expiration, mais c'est plus que le thread de travail. Il s'agit en fait d'une file d'attente de minuterie, où vous pouvez ajouter des minuteries qui appellent une fonction lorsqu'elles expirent.

class TimerQueue {
    ...
};

De plus, setTimeout()cela donne l'impression de définir le délai d'expiration de tout l'objet. Mais cela ajoute simplement un élément à la file d'attente. Je le nommerais donc addTimer(), ou plutôt simplement add()ou insert(), car il est clair d'après le nom TimerQueueque vous y ajouteriez des minuteries.

Évitez d'utiliser un lambda pour la fonction de thread

Ce n'est pas nécessaire. Pourquoi saisissez-vous le contexte par valeur? Saviez-vous qu'il capture toujours thispar référence? Utilisez simplement une fonction membre régulière pour cela. Vous pouvez même avoir le thread initialisé sans avoir besoin d'un constructeur, comme ceci:

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

        while (running) {
            ...
        }
    }

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

Vous avez toujours besoin d'un destructeur pour join()le thread, bien qu'en C ++ 20 ce ne soit plus nécessaire si vous utilisez un std::jthread.

Assurez-vous que le destructeur réveille le thread de travail

Votre thread de travail peut dormir jusqu'à 60 secondes s'il n'y a rien dans la file d'attente. Si vous détruisez la file d'attente du minuteur pendant cette période, vous devrez peut-être attendre longtemps pour que l'appel join()se termine. Assurez-vous de réactiver le thread dans le destructeur:

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

Une autre option consiste à mettre en file d'attente un élément spécial dans la file d'attente qui signale que le thread de travail doit s'arrêter et que le thread de travail quitte immédiatement la fonction s'il rencontre cet élément. Cela évite le besoin de la variable running.

Évitez d'utiliser system_clockpour les minuteries

Le problème avec system_clockest qu'il peut soudainement sauter, par exemple à cause des changements d'heure d'été, des secondes intercalaires et des mises à jour NTP. Vous devriez utiliser à la std::chrono::steady_clockplace. Je vous recommande de créer un alias de type pour cela:

using clock = std::chrono::steady_clock;

Et puis utilisez-le comme ceci:

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

Pensez à utiliser un std::priority_queue

C ++ a un conteneur spécialement pour garder les choses triées par priorité: std::priority_queue. Pensez à l'utiliser. Le seul inconvénient est que cela fonctionne plus comme a std::setque comme a std::map, vous devez définir une structure pour contenir à la fois un point temporel et une fonction de rappel, et le faire trier correctement:

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

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

std::priority_queue<Timer> queue;

Tu n'as pas besoin waitUntil

Vous savez déjà la prochaine fois que vous vous réveillerez en regardant le premier point dans le temps queue.

Évitez la duplication de code

À l'intérieur du thread de travail, vous traitez le cas d'une file d'attente vide et d'une file d'attente non vide. Cependant, le code dans les deux cas est identique, sauf pour le point de temps à attendre. Vous pouvez simplement écrire:

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

Déclarer les constantes en tant que telles

Vous déclarez les variables msMinet msMax, et elles ressemblent à des constantes, mais vous n'avez pas informé le compilateur de ce fait. Vous pouvez les fabriquer const, voire mieux static constexpr. Mais pour ces derniers, il faut aussi les définir dans un .cppfichier, ce qui est un peu ennuyeux. Ce problème est résolu dans C ++ 17, où vous pouvez les spécifier comme static inline constexpr.

Éviter l'invalidation d'itérateur

Lors du traitement des tâches qui ont expiré, vous appelez queue.erase(), mais vous avez déjà remarqué que vous devez faire attention à ne pas invalider l'itérateur. Le fait de compter sur l'incrémentation de l'itérateur avant d'appeler erase()n'est pas garanti de fonctionner. À la place, utilisez la valeur de retour de erase()comme itérateur vers l'élément suivant:

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

Si vous utilisez à la std::priority_queueplace, j'écrirais le code comme:

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

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