Attività ritardate C++

Question

Desidero implementare un modo per pianificare un'attività da eseguire in un secondo momento. L'interfaccia sarebbe simile a JavaScript setTimeout(function, milliseconds).

Nella mia applicazione alcune risorse sono di proprietà di una discussione. Per evitare condizioni di gara, è sempre necessario accedervi dallo stesso thread. Se altri thread vogliono accedere alla risorsa, devono inviare un oggetto task al thread di risorse.

Così due problemi che devo risolvere sono:

1. inviare un compito ad un filo
2. ritardo l'invocazione

Il primo problema è risolto rapidamente utilizzando una coda senza blocchi che ha il filo della risorsa dal lato dei consumatori. (Io uso TBB's concurrent_bounded_queue.) Il secondo problema, tuttavia, non è così ovvio per me. Posso pensare a due strategie:

1. Avviare un nuovo thread per ogni attività. Questo thread interromperà il ritardo richiesto, quindi invierà l'attività alla coda concorrente.
2. Avvia solo un thread che esegue un ciclo che itera le attività pianificate e li richiama se il loro tempo di attesa è scaduto.

Ho sperimentato con entrambi gli approcci e tendo a preferire il primo perché è semplice e affidabile, mentre il secondo tende ad essere più incline a bug sottili. Il primo approccio lo delega allo scheduler dei thread del sistema operativo.

Tuttavia, la prima soluzione crea molti thread di breve durata, mentre di solito ascolto la raccomandazione di riutilizzare i thread.

Answer 1

L'implementazione manuale sarà simile al seguente.

struct myrunnable { 
    uint64_t id_; 
    uint64_t stamp_; 
    std::function<void()> runnable_; 
    uint64_t id() { return id_; } 
    uint64_t stamp() { return stamp_; } 
    void execute() { if (runnable_) runnable_(); } 
}; 

typedef std::shared_ptr<myrunnable> task_t; 
// timestamp_cmp_t - a comparator by timestamp + incrementing task id 
typedef tbb::concurrent_blocking_queue<task_t> queue_t; 
typedef std::priority_queue<task, timestamp_cmp_t> schedule_t; 

uint64_t now(); // a wrapper around gettimeofday(), write yourself 

queue_t queue; // inbound concurrent blocking queue not bound in size 
schedule_t schedule; // priority queue, a scheduler 
// queue_t sink; // optional sink concurrent queue if you don't 
       // want to execute tasks in the scheduler thread context 

// now() - a wrapper around gettimeofday(), write yourself 
for(;;) { // "termination mark" comments below - exit points 
    while (!schedule.empty() && schedule.top().stamp() <= now()) { 
    task_t task = schedule.pop(); 
    task .execute(); 
    // alternatively sink.push(task) to offload scheduler thread 
    } 

    if (schedule.empty()) { 
    task_t task = queue.pop(); // block on the input queue 
    if (!task) return; // scheduler termination mark, empty task 
    schedule.push(task); 
    } else { 
    // Here we are driven by our latency/cpu balance requirements 
    // in this example we are ultra low latency and are just spinning CPU 
    // and on Linux such thread may need extra tuning to perform consistently. 
    // To pace it down one can use TBB's sleep_for() or select() system call 

    while (schedule.top().stamp() > now()) { 
     task_t task; 
     if (queue.try_pop(task)) { 
     if (!task) return; // scheduler termination mark, empty task 
     schedule.push(task); 
     } 
    } 
    } 
}