c++ - 将使用 CreateEvent 和 WaitForMultipleObjects 的程序移植到 Linux

Question

我需要将使用 Windows API 函数 SetEvent、CreateEvent 和 WaitForMultipleObjects 的多进程应用程序移植到 Linux。我发现了很多关于这个问题的帖子，但没有一个为我的问题提供合理的解决方案。

我有一个应用程序，它分为三个进程并通过这些事件管理一个进程的线程工作池。

对于这个问题，我有多种解决方案。一种是在 Linux 上使用 mkfifo 在 Linux 上创建 FIFO 特殊文件，并使用 select 语句唤醒线程。问题在于此解决方案的运行方式与 WaitForMultipleObjects 不同。例如，如果 workerpool 的 10 个线程将等待该事件，并且我调用 SetEvent 五次，正好有五个 workerthreads 将被唤醒并完成工作，当在 Linux 中使用 FIFO 变体时，它会在每个线程，我在 select 语句中等待数据放入 fifo。最好的描述方式是，Windows API 的工作方式类似于计数为 1 的全局信号量。

我还考虑过使用 pthreads 和 condition variables 来重新创建它并通过共享内存共享变量(shm_open 和 mmap )，但我在这里遇到了同样的问题!

在 Linux 上重现此行为的合理方法是什么？我找到了一些在单个进程内执行此操作的解决方案，但是在多个进程之间执行此操作呢？

欢迎任何想法(注意:我不希望有完整的实现，我只需要更多的想法来让自己开始解决这个问题)。

Answer 1

您可以使用信号量 ( sem_init )，它们在共享内存上工作。如果你想从不同的进程初始化它们，还有命名的信号量 ( sem_open )。如果您需要与工作人员交换消息，例如将实际任务传递给他们，那么解决此问题的一种方法是使用 POSIX message queues .它们被命名并在进程间工作。这是一个简短的例子。请注意，只有第一个工作线程实际初始化消息队列，其他线程使用现有线程的属性。此外，它(可能)保持持久性，直到使用 mq_unlink 明确删除，为简单起见，我在此处跳过。

带有工作线程的接收器:

// Link with -lrt -pthread

#include <fcntl.h>
#include <mqueue.h>
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>

void *receiver_thread(void *param) {
    struct mq_attr mq_attrs = { 0, 10, 254, 0 };

    mqd_t mq = mq_open("/myqueue", O_RDONLY | O_CREAT, 00644, &mq_attrs);
    if(mq < 0) {
        perror("mq_open");
        return NULL;
    }
    char msg_buf[255];
    unsigned prio;

    while(1) {
        ssize_t msg_len = mq_receive(mq, msg_buf, sizeof(msg_buf), &prio);
        if(msg_len < 0) {
            perror("mq_receive");
            break;
        }
        msg_buf[msg_len] = 0;
        printf("[%lu] Received: %s\n", pthread_self(), msg_buf);
        sleep(2);
    }
}

int main() {
    pthread_t workers[5];
    for(int i=0; i<5; i++) {
        pthread_create(&workers[i], NULL, &receiver_thread, NULL);
    }
    getchar();
}

发件人:

#include <fcntl.h>
#include <stdio.h>
#include <mqueue.h>
#include <unistd.h>

int main() {
    mqd_t mq = mq_open("/myqueue", O_WRONLY);
    if(mq < 0) {
        perror("mq_open");
    }
    char msg_buf[255];
    unsigned prio;

    for(int i=0; i<255; i++) {
        int msg_len = sprintf(msg_buf, "Message #%d", i);
        mq_send(mq, msg_buf, msg_len, 0);
        sleep(1);
    }
}