Ist meine Implementierung eines C ++ setTimeout-Threads korrekt?

Question

Ich habe einen setTimeoutThread ähnlich dem in JavaScript implementiert (neu in der Thread-Programmierung).

In dem Beispiel auf dieser Seite sehe ich die Verwendung einer atomaren Variablen, idie meiner Meinung nach dazu dient, um sicherzustellen, dass auf 'i' keine Rennbedingungen auftreten, aber nach dem, was ich gelesen habe, glaube ich nicht, dass es eine atomare Variable gibt multimap.

Aus dem Code kann eine Race-Bedingung für den UI-Thread an queue.emplace(...)und in dem Thread entstehen, der über den queue.

Schaut mein Thread-Code im Allgemeinen nach dem Job und sollte ich einen anderen verwenden condition_variable, um den queueZugriff zu blockieren ?

---

AKTUALISIEREN

Ich denke, ich musste definitiv den queueManipulationsfaden sicher machen. Ich bin in verschiedene Sackgassen geraten, als ich lernte, wie man Threads programmiert. Am Ende mit einem shared_timed_mutexfunktionierten! Diese Art von Mutex kann von mehreren Threads gemeinsam genutzt werden, um den Datenzugriff und die Datenmanipulation zu synchronisieren, z. B. die Sie verwenden können

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

und

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

Jeder *_lockblockiert, wenn der Mutex derzeit gesperrt ist, oder Sie können zusätzliche Parameter hinzufügen, um andere Verhaltensweisen anzugeben. Jede Sperre wird aufgehoben, nachdem der Bereich verlassen wurde.

---

Hier ist mein Originalcode:

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

Main:

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

Dieser Code erzeugt zwei Zeitüberschreitungen, die erste ist so eingestellt, dass sie 15 Sekunden und die zweite 10 Sekunden von Anfang an auslöst, aber sie sind so eingerichtet, dass der Thread getestet wird, der die wait_untilordnungsgemäß aufweckt , was in der Tat funktioniert:

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

Sperren Sie den Mutex ein setTimeout()

Sie haben mindestens zwei Threads, die darauf zugreifen queue, also müssen Sie sicherstellen, dass sie nicht gleichzeitig aktualisiert werden. Sie halten das Schloss im Inneren WorkerThread(), aber Sie sollten es auch im Inneren halten setTimeout().

Geben Sie der Klasse einen besseren Namen

Ja, die Klasse verwendet einen Arbeitsthread, um bis zum nächsten Timeout zu warten, aber es ist mehr als nur der Arbeitsthread. Es ist eigentlich eine Timer-Warteschlange, in der Sie Timer hinzufügen können, die eine Funktion aufrufen, wenn sie eine Zeitüberschreitung aufweisen.

class TimerQueue {
    ...
};

Klingt auch so, setTimeout()als würde das Zeitlimit für das gesamte Objekt festgelegt. Es wird jedoch nur ein Element zur Warteschlange hinzugefügt. Also würde ich es benennen addTimer(), oder eher nur add()oder insert(), da aus dem Namen klar hervorgeht, TimerQueuedass Sie Timer hinzufügen würden.

Vermeiden Sie die Verwendung eines Lambda für die Thread-Funktion

Es ist nicht nötig. Warum erfassen Sie den Kontext nach Wert? Wussten Sie, dass es immer noch als thisReferenz erfasst ? Verwenden Sie dazu einfach eine reguläre Member-Funktion. Sie können den Thread sogar initialisieren lassen, ohne einen Konstruktor zu benötigen.

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

        while (running) {
            ...
        }
    }

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

Sie benötigen noch einen Destruktor für join()den Thread, obwohl dies in C ++ 20 nicht mehr erforderlich ist, wenn Sie a verwenden std::jthread.

Stellen Sie sicher, dass der Destruktor den Arbeitsthread aufweckt

Ihr Worker-Thread kann bis zu 60 Sekunden lang schlafen, wenn sich nichts in der Warteschlange befindet. Wenn Sie die Timer-Warteschlange während dieser Zeit zerstören, müssen Sie möglicherweise lange warten, join()bis der Anruf beendet ist. Stellen Sie sicher, dass Sie den Thread im Destruktor aufwecken:

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

Eine andere Möglichkeit besteht darin, ein spezielles Element in die Warteschlange zu stellen, das signalisiert, dass der Arbeitsthread gestoppt werden soll, und den Arbeitsthread die Funktion sofort beenden zu lassen, wenn er auf dieses Element stößt. Dies vermeidet die Notwendigkeit der Variablen running.

Vermeiden Sie die Verwendung system_clockfür Timer

Das Problem dabei system_clockist, dass es plötzlich springen kann, beispielsweise aufgrund von Änderungen der Sommerzeit, Schaltsekunden und NTP-Updates. Sie sollten std::chrono::steady_clockstattdessen verwenden. Ich empfehle Ihnen, einen Typalias dafür zu erstellen:

using clock = std::chrono::steady_clock;

Und dann benutze es so:

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

Erwägen Sie die Verwendung von a std::priority_queue

C ++ verfügt speziell über einen Container, um die Dinge nach Priorität zu sortieren : std::priority_queue. Erwägen Sie, das zu verwenden. Der einzige Nachteil ist, dass es eher wie a std::setals wie a funktioniert. std::mapSie müssen eine Struktur definieren, die sowohl einen Zeitpunkt als auch eine Rückruffunktion enthält, und sie richtig sortieren lassen:

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

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

std::priority_queue<Timer> queue;

Du brauchst nicht waitUntil

Sie wissen bereits, wann Sie das nächste Mal aufwachen müssen, indem Sie sich den frühesten Zeitpunkt ansehen queue.

Vermeiden Sie Codeduplikationen

Innerhalb des Arbeitsthreads behandeln Sie den Fall einer leeren Warteschlange und einer nicht leeren Warteschlange. Der Code ist jedoch in beiden Fällen identisch, mit Ausnahme des Zeitpunkts, bis zu dem gewartet werden muss. Sie könnten einfach schreiben:

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

Deklarieren Sie Konstanten als solche

Sie deklarieren die Variablen msMinund msMaxund sie sehen aus wie Konstanten, aber Sie haben dem Compiler nichts davon erzählt. Sie können sie machen const, oder noch besser static constexpr. Für letztere müssen Sie sie jedoch auch in einer .cppDatei definieren , was etwas ärgerlich ist. Dies ist in C ++ 17 behoben, wo Sie sie als angeben können static inline constexpr.

Vermeiden Sie die Ungültigmachung des Iterators

Wenn Sie abgelaufene Aufgaben verarbeiten, rufen Sie an queue.erase(), aber Sie haben bereits bemerkt, dass Sie vorsichtig sein müssen, um den Iterator nicht ungültig zu machen. Es erase()ist nicht garantiert, dass das Inkrementieren des Iterators vor dem Aufruf funktioniert. Verwenden Sie stattdessen den Rückgabewert von erase()als Iterator für das nächste Element:

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

Wenn Sie std::priority_queuestattdessen a verwenden, würde ich den Code wie folgt schreiben:

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

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