¿Es correcta mi implementación de un hilo setTimeout de C ++?
Implementé un setTimeouthilo similar al de JavaScript (nuevo en la programación de hilos)
En el ejemplo de esta página , veo el uso de una variable atómica ique creo que es para asegurarme de que no ocurran condiciones de carrera en 'i', pero por lo que he leído, no creo que haya una atómica multimap.
A partir del código, puede surgir una condición de carrera en el subproceso de la interfaz de usuario en queue.emplace(...)y en el subproceso que itera sobre queue.
¿Mi código de hilo busca el trabajo en general, y debería usar otro condition_variablepara bloquear el queueacceso?
ACTUALIZAR
Creo que definitivamente necesitaba hacer que las queuemanipulaciones fueran seguras. Bajé por varios callejones sin salida en esto mientras aprendía a programar hilos. ¡Al final usando un shared_timed_mutextrabajado! Este tipo de mutex se puede compartir entre subprocesos para sincronizar el acceso y la manipulación de datos, por ejemplo, puede usar
{
unique_lock<shared_timed_mutex> lock(shared_m); // for writing
// write data to whatever...
}
y
{
shared_lock<shared_timed_mutex> lock(shared_m); // for reading
// read data from wherever...
}
Cada uno *_lockse bloqueará si el mutex está actualmente bloqueado, o puede agregar parámetros adicionales para especificar otros tipos de comportamiento. Cada bloqueo se libera después de salir del osciloscopio.
Aquí está mi 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();
}
};
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 crea dos tiempos de espera, el primero está configurado para disparar 15 segundos y el segundo 10 segundos desde el principio, pero están configurados de tal manera que prueban que el hilo despierta wait_untilcorrectamente, lo que de hecho 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
Respuestas
Bloquear el mutex en setTimeout()
Tiene al menos dos subprocesos accediendo queue, por lo que debe asegurarse de que no lo actualicen simultáneamente. Mantienes el candado adentro WorkerThread(), pero también debes mantenerlo adentro setTimeout().
Dale a la clase un nombre mejor
Sí, la clase usa un hilo de trabajo para esperar hasta el próximo tiempo de espera, pero es más que solo el hilo de trabajo. En realidad, es una cola de temporizadores, donde puede agregar temporizadores que llaman a una función cuando se agota el tiempo de espera.
class TimerQueue {
...
};
Además, setTimeout()parece que establece el tiempo de espera de todo el objeto. Pero solo agrega un elemento a la cola. Así que lo nombraría addTimer(), o más bien simplemente add()o insert(), ya que está claro por el nombre TimerQueueque le agregarías temporizadores.
Evite usar una lambda para la función de hilo
No es necesario. ¿Por qué captura el contexto por valor? ¿Sabías que todavía captura thispor referencia? Solo use una función miembro regular para esto. Incluso puede inicializar el hilo sin necesidad de un constructor, así:
class TimerQueue {
void worker() {
std::unique_lock<std::mutex> lk(cv_m);
while (running) {
...
}
}
thread workerThread{&TimerQueue::worker, this};
...
};
Todavía necesita un destructor para join()el hilo, aunque en C ++ 20 esto ya no es necesario si usa un archivo std::jthread.
Asegúrese de que el destructor despierte el hilo de trabajo
Su hilo de trabajador puede dormir hasta 60 segundos si no hay nada en la cola. Si destruye la cola del temporizador durante ese tiempo, es posible que deba esperar mucho tiempo hasta join()que finalice la llamada . Asegúrese de despertar el hilo en el destructor:
~TimerQueue() {
std::lock_guard<std::mutex> lk(cv_m);
running = false;
cv.notify_one();
workerThread.join();
}
Otra opción es poner en cola un elemento especial en la cola que indica que el hilo de trabajo debe detenerse y hacer que el hilo de trabajo salga inmediatamente de la función si encuentra ese elemento. Esto evita la necesidad de la variable running.
Evite el uso system_clockde temporizadores
El problema system_clockes que puede saltar repentinamente, por ejemplo, debido a los cambios de horario de verano, los segundos intercalares y las actualizaciones de NTP. Deberías usar std::chrono::steady_clocken su lugar. Te recomiendo que crees un alias de tipo para ello:
using clock = std::chrono::steady_clock;
Y luego úsalo así:
multimap<clock::time_point, Function> queue;
clock::time_point waitUntil;
...
waitUntil = clock::now() + ...;
Considere usar un std::priority_queue
C ++ tiene un contenedor específicamente para mantener las cosas ordenadas por prioridad: std::priority_queue. Considere usar eso. El único inconveniente es que funciona más como a std::setque a std::map, tienes que definir alguna estructura para contener tanto un punto de tiempo como una función de devolución de llamada, y hacer que se ordene correctamente:
struct Timer {
clock::time_point deadline;
Function callback;
bool operator<(const Timer &other) const {
return other.deadline < deadline;
}
};
std::priority_queue<Timer> queue;
No necesitas waitUntil
Ya sabes la próxima vez que te despiertas mirando el punto de tiempo más temprano en queue.
Evite la duplicación de código
Dentro del hilo de trabajo, se ocupa del caso de una cola vacía y una cola no vacía. Sin embargo, el código en ambos casos es idéntico, excepto por el punto de tiempo para esperar. Podrías simplemente escribir:
waitUntil = clock::now() + queue.empty() ? 60000ms : queue.front().deadline;
cv.wait_until(lk, waitUntil);
Declarar constantes como tales
Declaras las variables msMiny msMax, y parecen constantes, pero no le dijiste al compilador sobre ese hecho. Puedes hacerlos const, o incluso mejores static constexpr. Pero para este último, también debe definirlos en un .cpparchivo, lo cual es un poco molesto. Esto se corrige en C ++ 17, donde puede especificarlos como static inline constexpr.
Evite la invalidación del iterador
Cuando procesa tareas que han expirado, llama queue.erase(), pero ya notó que debe tener cuidado de no invalidar el iterador. erase()No se garantiza que funcione si se depende de incrementar el iterador antes de llamar . En su lugar, use el valor de retorno de erase()como iterador del siguiente elemento:
for (auto task = queue.begin(); ...) {
if (...) {
...
task = queue.erase(task);
} else {
break;
}
}
Si usa un std::priority_queueen su lugar, escribiría el código como:
while (!queue.empty()) {
auto timer = queue.top();
if (timer.deadline < maxtime) {
timer.callback();
queue.pop();
} else {
break;
}
}