Zadania opóźnione w C++

Question

Chcę zaimplementować sposób zaplanowania zadania, które ma zostać wykonane w późniejszym czasie. Interfejs będzie podobny do JavaScriptu setTimeout(function, milliseconds).

W mojej aplikacji niektóre zasoby są własnością wątku. Aby uniknąć warunków wyścigowych, muszą być zawsze dostępne z tej samej nitki. Jeśli inne wątki chcą uzyskać dostęp do zasobu, muszą wysłać obiekt zadania do wątku zasobów.

więc dwa problemy muszę rozwiązać to:

1. wysłanie zadania do wątku
2. opóźnienia inwokacja

Pierwszy problem szybko rozwiązać używając kolejkę blokady wolne który ma wątek zasobów po stronie konsumującej. (Używam TBB's concurrent_bounded_queue.) Drugi problem nie jest jednak dla mnie tak oczywisty. Mogę wymyślić dwie strategie:

1. Rozpocznij nowy wątek dla każdego zadania. Ten wątek przesłoni wymagane opóźnienie, a następnie wywoła zadanie do kolejki współbieżnej.
2. Uruchomić tylko jeden wątek, który uruchamia pętlę, która iteruje zaplanowane zadania i wywołuje je, jeśli upłynął ich czas oczekiwania.

Eksperymentowałem z obydwoma podejściami i preferuję pierwsze, ponieważ jest proste i niezawodne, podczas gdy drugie jest bardziej podatne na subtelne błędy. Pierwsze podejście deleguje to do programu planującego wątki systemu operacyjnego.

Jednak pierwsze rozwiązanie tworzy wiele krótkotrwałych wątków, a ja zazwyczaj słyszę zalecenie ponownego użycia wątków.

Answer 1

Implementacja ręczna będzie wyglądać jak poniżej.

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