C++ bị trì hoãn nhiệm vụ

Question

Tôi muốn thực hiện một cách để lên lịch cho một tác vụ được thực thi sau này. Giao diện sẽ tương tự như setTimeout(function, milliseconds) của JavaScript.

Trong ứng dụng của tôi, một số tài nguyên được sở hữu bởi một chuỗi. Để tránh điều kiện chủng tộc, chúng phải luôn được truy cập từ cùng một chuỗi đó. Nếu các luồng khác muốn truy cập tài nguyên, chúng phải gửi một đối tượng nhiệm vụ tới luồng tài nguyên.

Vì vậy, hai vấn đề tôi cần phải giải quyết là:

1. cử một nhiệm vụ to a thread
2. chậm trễ invocation

Vấn đề đầu tiên là nhanh chóng cố định bằng cách sử dụng một hàng đợi lock-free trong đó có luồng tài nguyên ở phía tiêu thụ. (Tôi sử dụng concurrent_bounded_queue của TBB.) Tuy nhiên, vấn đề thứ hai không quá rõ ràng đối với tôi. Tôi có thể nghĩ đến hai chiến lược:

1. Bắt đầu một chuỗi mới cho mỗi tác vụ. Chủ đề này sẽ ngủ chậm trễ yêu cầu, sau đó gửi nhiệm vụ đến hàng đợi đồng thời.
2. Chỉ bắt đầu một chuỗi chạy vòng lặp lặp lại các tác vụ đã lên lịch và gọi chúng nếu thời gian chờ của chúng hết hạn.

Tôi đã thử nghiệm với cả hai cách tiếp cận và tôi có xu hướng ưu tiên đầu tiên bởi vì nó đơn giản và đáng tin cậy, trong khi thứ hai có xu hướng dễ bị lỗi tinh vi hơn. Cách tiếp cận đầu tiên ủy quyền cho bộ lập lịch trình chủ đề OS.

Tuy nhiên, giải pháp đầu tiên tạo ra nhiều chuỗi thời gian tồn tại ngắn, trong khi tôi thường nghe đề xuất sử dụng lại các chuỗi.

Answer 1

Triển khai thủ công sẽ giống như dưới đây.

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