c++ - 如何安排一个操作在未来的时间运行

Question

我有一个包含任务队列的类 TaskManager。每次弹出并执行下一个任务。

class TaskManager
{
    TaskQueue m_queue;

    svc_tasks()
    {
         while (!m_queue.empty())
         {
             Task* task = m_queue.pop();
             task->execute();
         }
    }
};

在任务中，我想暂停某些点至少 SLEEP_TIME_MS 毫秒。在此暂停期间，我想开始执行下一个任务。当暂停结束时，我想再次将任务放入队列中。

class Task
{
    int m_phase = -1;

    execute()
    {
        m_phase++;

        switch(m_phase)
        {
         case 0:
             ...
             do_pause(SLEEP_TIME_MS);
             return;
         case 1:
             ...
             break;
        }
    }
};

std (C++ 17) 或 boost 中是否有我可以使用的调度程序，它会在 SLEEP_TIME_MS 过去时调用处理函数？

谢谢你的建议

Answer 1

您可以使用 boost::asio::high_resolution_timer 及其 async_wait 方法。

每次当你想安排将任务插入队列的操作时，你必须:

1. 创建high_resolution_timer
2. 调用 expires_after 指定到期时间 (SLEEP_TIME_MS)，即何时调用处理程序。在您的情况下，在此处理程序中，您将任务插入队列。
3. 用你的处理程序调用async_wait

---

如果我们假设 execute 方法返回 bool 指示任务是否完成(所有阶段都已执行)，它可能会被重写成这样:

     while (!m_queue.empty()) // this condition should be changed
     {
         Task* task = m_queue.pop();
         bool finished = task->execute();
         if (!finished)
            scheduler works here - start async_wait with handler
     }

如果我理解正确，你想在 SLEEP_TIME_MS 过期时将任务插入队列，所以你不能在队列为空时中断循环，因为你必须等到挂起的任务完成。您可以引入 stop 标志。并按需中断循环。

---

下面我放了一段代码，它按照你描述的方式工作(我希望):

struct Scheduler {
    Scheduler(boost::asio::io_context& io)
    : io(io) {}

    boost::asio::io_context& io;

    template<class F>
    void schedule (F&& handler) {
        auto timer = std::make_shared<boost::asio::high_resolution_timer>(io);
        timer->expires_after(std::chrono::milliseconds(5000)); // SLEEP_TIME_MS 
        timer->async_wait(
            [timer,handler](const boost::system::error_code& ec) {
                handler();
            });
    }
};

struct Task  {
    int phase = -1;

    bool execute() {
        ++phase;
        std::cout << "phase: " << phase << std::endl;
        if (phase == 0) {
            return false;
        }
        else {

        }
        return true;
    }
};

struct TaskManager {
    Scheduler s;
    std::queue<std::shared_ptr<Task>> tasks;
    std::mutex tasksMtx;
    std::atomic<bool> stop{false};

    TaskManager(boost::asio::io_context& io) : s(io) {
        for (int i = 0; i < 5; ++i)
            tasks.push(std::make_shared<Task>());
    }

    void run() {
        while (true) {
            if (stop)
                break;

            {
                std::lock_guard<std::mutex> lock{tasksMtx};
                if (tasks.empty())
                    continue;
            }

            std::shared_ptr<Task> currTask = tasks.front();
            tasks.pop();

            bool finished = currTask->execute();
            if (!finished)
                s.schedule( [this, currTask](){ insertTaskToVector(std::move(currTask)); } );
        }
    }

    template<class T>
    void insertTaskToVector(T&& t) {
        std::lock_guard<std::mutex> lock{tasksMtx};
        tasks.push(std::forward<T>(t));
    }
};

int main() {
    boost::asio::io_context io;
    boost::asio::io_context::work work{io};
    std::thread th([&io](){ io.run();});
    TaskManager tm(io);
    tm.run();