使用管道在 C++ 中创建调度队列/线程处理程序：FIFO 溢出

Question

创建和使用线程需要大量资源，因此通常会为异步任务重用线程池。将任务打包，然后“发布”到代理，该代理将把任务排入下一个可用线程。

这是调度队列（即 Apple 的 Grand Central Dispatch）和线程处理程序（Android 的 Looper 机制）背后的理念。

现在，我正在尝试自己动手。事实上，我正在填补 Android 中的一个空白，即有一个用于在 Java 中发布任务的 API，但在本机 NDK 中没有。但是，我会尽可能保持这个问题平台的独立性。

管道是我的方案的理想选择。我可以轻松地在我的工作线程上轮询pipe(2) 的读取端的文件描述符，并通过写入写入端将来自任何其他线程的任务排入队列。看起来是这样的：

int taskRead, taskWrite;

void setup() {
    // Create the pipe
    int taskPipe[2];
    ::pipe(taskPipe);
    taskRead = taskPipe[0];
    taskWrite = taskPipe[1];

    // Set up a routine that is called when task_r reports new data
    function_that_polls_file_descriptor(taskRead, []() {
        // Read the callback data
        std::function<void(void)>* taskPtr;
        ::read(taskRead, &taskPtr, sizeof(taskPtr));

        // Run the task - this is unsafe! See below.
        (*taskPtr)();

        // Clean up
        delete taskPtr;
    });
}

void post(const std::function<void(void)>& task) {
    // Copy the function onto the heap
    auto* taskPtr = new std::function<void(void)>(task);

    // Write the pointer to the pipe - this may block if the FIFO is full!
    ::write(taskWrite, &taskPtr, sizeof(taskPtr));
}

此代码将std::function 放在堆上，并将指针传递给管道。 function_that_polls_file_descriptor 然后调用提供的表达式来读取管道并执行函数。请注意，此示例中没有安全检查。

这在 99% 的情况下都很好用，但有一个主要缺点。管道的大小是有限的，如果管道被填满，那么对post() 的调用将挂起。这本身并不是不安全的，直到对post() 的调用在一项任务中进行。

auto evil = []() {
    // Post a new task back onto the queue
    post({});
    // Not enough new tasks, let's make more!
    for (int i = 0; i < 3; i++) {
        post({});
    }

    // Now for each time this task is posted, 4 more tasks will be added to the queue.
});

post(evil);
post(evil);
...

如果发生这种情况，则工作线程将被阻塞，等待写入管道。但是管道的 FIFO 已满，工作线程没有从中读取任何内容，因此整个系统处于死锁状态。

如何确保从工作线程发出的对post() 的调用始终成功，从而允许工作线程在队列已满时继续处理队列？

Answer 1

感谢这篇文章中的所有 cmets 和其他答案，我现在有一个解决这个问题的可行解决方案。

我采用的技巧是通过检查哪个线程正在调用post() 来确定工作线程的优先级。这是粗略的算法：

pipe ← NON-BLOCKING-PIPE()
overflow ← Ø
POST(task)
    success ← WRITE(task, pipe)
    IF NOT success THEN
        IF THREAD-IS-WORKER() THEN
            overflow ← overflow ∪ {task}
        ELSE
            WAIT(pipe)
            POST(task)

然后在工作线程上：

LOOP FOREVER
    task ← READ(pipe)
    RUN(task)

    FOR EACH overtask ∈ overflow
        RUN(overtask)

    overflow ← Ø

使用pselect(2) 执行等待，改编自@Sigismondo 的回答。

这是在我的原始代码示例中实现的算法，该算法适用于单个工作线程（尽管我在复制粘贴后没有对其进行测试）。通过为每个线程设置一个单独的溢出队列，它可以扩展为适用于线程池。

int taskRead, taskWrite;

// These variables are only allowed to be modified by the worker thread
std::__thread_id workerId;
std::queue<std::function<void(void)>> overflow;
bool overflowInUse;

void setup() {
    int taskPipe[2];
    ::pipe(taskPipe);
    taskRead = taskPipe[0];
    taskWrite = taskPipe[1];

    // Make the pipe non-blocking to check pipe overflows manually
    ::fcntl(taskWrite, F_SETFL, ::fcntl(taskWrite, F_GETFL, 0) | O_NONBLOCK);

    // Save the ID of this worker thread to compare later
    workerId = std::this_thread::get_id();
    overflowInUse = false;

    function_that_polls_file_descriptor(taskRead, []() {
        // Read the callback data
        std::function<void(void)>* taskPtr;
        ::read(taskRead, &taskPtr, sizeof(taskPtr));

        // Run the task
        (*taskPtr)();
        delete taskPtr;

        // Run any tasks that were posted to the overflow
        while (!overflow.empty()) {
            taskPtr = overflow.front();
            overflow.pop();

            (*taskPtr)();
            delete taskPtr;
        }

        // Release the overflow mechanism if applicable
        overflowInUse = false;
    });
}

bool write(std::function<void(void)>* taskPtr, bool blocking = true) {
    ssize_t rc = ::write(taskWrite, &taskPtr, sizeof(taskPtr));

    // Failure handling
    if (rc < 0) {
        // If blocking is allowed, wait for pipe to become available
        int err = errno;
        if ((errno == EAGAIN || errno == EWOULDBLOCK) && blocking) {
            fd_set fds;
            FD_ZERO(&fds);
            FD_SET(taskWrite, &fds);

            ::pselect(1, nullptr, &fds, nullptr, nullptr, nullptr);

            // Try again
            return write(tdata);
        }

        // Otherwise return false
        return false;
    }

    return true;
}

void post(const std::function<void(void)>& task) {
    auto* taskPtr = new std::function<void(void)>(task);

    if (std::this_thread::get_id() == workerId) {
        // The worker thread gets 1st-class treatment.
        // It won't be blocked if the pipe is full, instead
        // using an overflow queue until the overflow has been cleared.
        if (!overflowInUse) {
            bool success = write(taskPtr, false);
            if (!success) {
                overflow.push(taskPtr);
                overflowInUse = true;
            }
        } else {
            overflow.push(taskPtr);
        }
    } else {
        write(taskPtr);
    }
}

Answer 2

使管道写入文件描述符非阻塞，这样当管道已满时，write 会以 EAGAIN 失败。

---

一项改进是增加管道缓冲区大小。

另一个是使用 UNIX 套接字/套接字对并增加套接字缓冲区大小。

另一种解决方案是使用许多工作线程可以读取的 UNIX 数据报套接字，但只有一个可以获取下一个数据报。换句话说，您可以将数据报套接字用作线程调度程序。

Answer 3

您可以使用旧好的select 来确定文件描述符是否已准备好用于写入：

会观察 writefds 中的文件描述符，看是否 空间可用于写入（尽管大写入可能仍会阻塞）。

由于您正在编写指针，因此您的 write() 根本不能归类为大。

显然，您必须准备好处理帖子可能失败的事实，然后准备稍后重试...否则您将面临无限增长的管道，直到您的系统再次崩溃。

或多或少（未测试）：

bool post(const std::function<void(void)>& task) {
    bool post_res = false;

    // Copy the function onto the heap
    auto* taskPtr = new std::function<void(void)>(task);

    fd_set wfds;
    struct timeval tv;
    int retval;

    FD_ZERO(&wfds);
    FD_SET(taskWrite, &wfds);

    // Don't wait at all
    tv.tv_sec = 0;
    tv.tv_usec = 0;

    retval = select(1, NULL, &wfds, NULL, &tv);
    // select() returns 0 when no FD's are ready
    if (retval == -1) {
      // handle error condition
    } else if (retval > 0) {
      // Write the pointer to the pipe. This write will succeed
      ::write(taskWrite, &taskPtr, sizeof(taskPtr));
      post_res = true;
    }
    return post_res;
}

Answer 4

如果您只使用管道查看 Android/Linux 并不是艺术的开始，但使用事件文件描述符和 epoll 是可行的方法。