使用 Poco 多线程时如何使用信号量来保证顺序？

Question

我正在编写一个简单的程序来演示信号量的使用。 （后来测试自定义编写的信号量是否有效）。

我有 4 个线程同时运行一个函数。每个函数等待一段随机的时间，然后打印：Hello, world! This is thread n I slept for uSuS。

如您所料，消息以随机顺序打印到标准输出。这表明线程是并发运行的，因为如果它们按顺序执行，它们将按顺序出现。

我想在此演示中使用信号量来强制执行命令。但是，它目前不起作用。

这是我的代码：

sem = sem_open("mutex" , O_CREAT | O_RDWR , S_IRWXU | S_IRWXG | S_IRWXO, 1); // name, oflag, mode, initial value

Poco::Thread thread[5];

class HelloRunnable: public Poco::Runnable
{
public:
    HelloRunnable(int arg) //constructor for the runnable
    {
        n = arg;
    }

    int n;

    virtual void run() //entry point for the threads
    {
        sem_wait(sem); //the semaphore
            timeval t;
            gettimeofday(&t, NULL);
            srand(t.tv_usec * t.tv_sec);
            int uS = rand()%100000;
            usleep(uS); //sleep for random length of time

            std::cout << "Hello, world! This is thread " << n << " I slept for "<< uS << "uS" <<std::endl;
        sem_post(sem);
        return;
    }
};



int main()
{
    HelloRunnable runnable1(1); //construct a runnable with arg = 1
    thread[1].start(runnable1); //execute that runnable

    HelloRunnable runnable2(2); //construct a runnable with arg = 2
    thread[2].start(runnable2); //execute that runnable

    HelloRunnable runnable3(3); //...
    thread[3].start(runnable3);

    HelloRunnable runnable4(4);
    thread[4].start(runnable4);

    //wait for all threads to finish
    thread[1].join();
    thread[2].join();
    thread[3].join();
    thread[4].join();

    return 0;
}

但是，线程仍然以随机顺序将消息打印到标准输出。例如：

//Hello, world! This is thread 2 I slept for 15001uS
//Hello, world! This is thread 1 I slept for 51124uS
//Hello, world! This is thread 4 I slept for 60884uS
//Hello, world! This is thread 3 I slept for 86137uS

我应该在代码中的什么位置放置信号量，以确保消息按顺序打印？抱歉，如果这很简单。我不是编码背景。

编辑

我将sem_wait 移到了usleep 之前。现在它工作得更好，但不是所有的时间。它按大约 %45 的时间顺序打印，大约 45% 的时间以相反的顺序打印，大约 %10 的时间以随机顺序打印。这是为什么！？

Answer 1

原因是 sem_wait 中阻塞的线程以不可预知的顺序被唤醒。此外，您添加了一些随机睡眠时间，增加了更多的混乱。

事实上，每个线程需要一个信号量。 线程 n 应该 sem_post 信号量 n + 1 然后 sem_wait 信号量 n。 最后一个线程应该 sem_post 信号量 0 等等。 只需在一个数组中创建信号量，第一个初始值为 1，另一个为 0。线程参数可以是访问线程中正确信号量的索引。

#include <Poco/Thread.h>

#include <iostream>

#include <semaphore.h>
#include <sys/time.h>
#include <unistd.h>

#define THREADS 5
#define LOOPS 5

sem_t semaphores[THREADS];

class HelloRunnable: public Poco::Runnable {
public:
    HelloRunnable(unsigned int index) {
        _index = index;
        // Seed random generator
        timeval time;
        gettimeofday(&time, NULL);
        srand(time.tv_usec * time.tv_sec);
    }

    virtual void run() {
        for (unsigned int loop = 0; loop < LOOPS; ++loop) {
            // Wait
            sem_wait(&semaphores[_index]);
            // Sleep random time
            int sleep = rand() % 100000;
            usleep(sleep);
            // Output
            std::cout << "Hello, world! This is thread " << _index
                    << " in loop " << loop << " I slept for " << sleep << "µS"
                    << std::endl;
            // Unlock next thread
            sem_post(&semaphores[(_index + 1) % THREADS]);
        }
        return;
    }
private:
    unsigned int _index;
};

int main() {
    HelloRunnable *runnables[THREADS];
    Poco::Thread *threads[THREADS];

    // Initialize semaphores and create threads and runnables
    for (unsigned int index = 0; index < THREADS; ++index) {
        sem_init(&semaphores[index], 0, index ? 0 : 1);
        threads[index] = new Poco::Thread();
        runnables[index] = new HelloRunnable(index);
    }

    // Start threads
    for (unsigned int index = 0; index < THREADS; ++index) {
        threads[index]->start(*runnables[index]);
    }

    // Wait for all threads to finish
    for (unsigned int index = 0; index < THREADS; ++index) {
        threads[index]->join();
    }

    // Cleanup
    for (unsigned int index = 0; index < THREADS; ++index) {
        free(runnables[index]);
        free(threads[index]);
    }
    return 0;
}