pthread_cond_wait() 仅在值设置为 0 时唤醒

Question

我对线程有些陌生，但一直在阅读 pthreads 并尝试编写最终将在应用程序中使用的测试程序。这个测试应用的基本操作是：

• 产生两个线程，
• 允许用户输入存储在条件变量值中的数字（发送者线程），然后使用信号唤醒第二个线程（接收者线程）。

我正在显示代码和输出屏幕以显示测试输出的结果。 基本上，当我输入 0 以外的数字时，信号不会唤醒第二个线程。当输入 0 时，第二个线程唤醒并按预期执行（即退出）。 问题是为什么第二个线程不会为其他数字（即 1 或 2）唤醒。我用 pthread_cond_timedwait() 和 pthread_cond_wait() 都试过了，但在这两种情况下我都得到了相同的结果。 我将不胜感激任何和所有的建议。谢谢。

//============================ terminal screen =================================
enter a message to send:
0 - exit
1 - message 1
2 - message 2
1
messageSender: msg.value: 1
messageSender: signal sent!
2
messageSender: msg.value: 2
messageSender: signal sent!
0
messageSender: msg.value: 0
messageSender: signal sent!
exit pthread_cond_timedwait, testVal: 1 value: 0
0
messageReceiverThread msg.value received: 0
exiting

//================= apps/pthread_com_example.h =================================

#ifndef PTHREAD_COM_EXAMPLE_H_INCLUDED
#define PTHREAD_COM_EXAMPLE_H_INCLUDED


//  | added for thread com example RLB 04Dec2021
//  V
#if(defined PTHREAD_COM_EXAMPLE)
#include <stdio.h>       // standard I/O routines
#include <stdlib.h>      // rand() and srand() functions
// number of threads used to service requests
#define NUM_HANDLER_THREADS 2

/* format of a message structure */
typedef struct  {
    pthread_mutex_t mutex;  // message mutex
    pthread_cond_t  cond;   // message condition variable
    int testVal;            // use this as the predicate
    int value;              // value to be passed
}message_t;
//  ^
//  | added for thread com example RLB 04Dec2021
#endif


#endif // PTHREAD_COM_EXAMPLE_H_INCLUDED

//=================== pthread_com_example app ==============================
#include <iostream>
#include <pthread.h>
#define PTHREAD_COM_EXAMPLE
#include <apps/pthread_com_example.h>
#include <unistd.h>
#include <sys/time.h>     // struct timeval definition

#if(defined PTHREAD_COM_EXAMPLE)
/* global mutex for example. assignment initializes it.
   note: a RECURSIVE mutex is used, since a handler
   thread might try to lock it twice consecutively. */

#define MSG1  1    // trigger messasge 1
#define MSG2  2    // trigger message 2
#define MSGX  0    // message to exit

std::string message1 = "this is an example message in response to A";
std::string message2 = "this is an example message in response to B";

struct timeval now;
struct timespec timeout;
message_t msg = {
    // initialize the message structure.
    PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP, PTHREAD_COND_INITIALIZER, 0, 0
    };

#endif

#if(defined PTHREAD_COM_EXAMPLE)
#define DEBUG
// wait for sending thread to signal variable is ready to read
int timedWaitForCond(pthread_cond_t* pCond, pthread_mutex_t* pMutex, int timeToWait)
{
    gettimeofday(&now,NULL);
    timeout.tv_sec = now.tv_sec + timeToWait;
    timeout.tv_nsec = now.tv_usec * 1000;
    int retcode = 0;

    while((msg.testVal == 0) && (retcode != ETIMEDOUT))
    {
        // this unlocks the mutux while waiting on signal from another thread
 //       retcode = pthread_cond_timedwait(pCond, pMutex, &timeout);
        retcode = pthread_cond_wait(pCond,pMutex);
        // upon return the mutex is locked again
        std::cout << "exit pthread_cond_timedwait, testVal: " << msg.testVal << " value: " << msg.value << std::endl;
        if(retcode != 0)
        {
            switch(retcode)
            {
            case ETIMEDOUT: // exits while loop
                std::cout << "Timed Conditional Wait, Timed Out" << std::endl;
                return(-1); // conditioned variable is not ready yet
                break;
            default:        // stays in while loop
                std::cout << "Unexpected timed conditional wait result: " << retcode << std::endl;
                break;
            }
        }else
        {
            return(0);  // conditioned variable ready to read
            break;      // exit while if legitimate signal occurred
        }
    }
    return(0);
}

void outputMenu(void)
{
    std::cout << std::endl << std::endl;
    std::cout << "enter a message to send: " << std::endl;
    std::cout << " 0 - exit" << std::endl;
    std::cout << " 1 - message 1" << std::endl;
    std::cout << " 2 - message 2" << std::endl;
}
int getUserMessage(void)
{
    int rc;

    std::cin >> rc;
    return(rc);
}
/*
 * function messageReceiverThread():
 *      attempts to wait for message wait.  If result is good
 *      sends signal to waiting thread and then performs a
 *      timed conditioned wait to receive the message from the
 *      sending thread.
 *
 * algorithm:
 * input:     messageWait mutex
 * output:    character from message_t structure
 * memory:    shared message_t structure
 */
void*  messageReceiverThread(void* arg)
{
    int timeToWait = 10;        // use 10 second timeout for this example
    int rc = 0;

    // first, lock the mutex, to assure exclusive access to the conditioned variable
    rc = pthread_mutex_lock(&msg.mutex);
    if (rc)
    {   // an error has occurred
        std::cout << "messageReceiver: pthread_mutex_lock error: " << rc << std::endl;
        pthread_exit(NULL);
    }
    // mutex is now locked
    outputMenu();
    while(1)
    {
        msg.testVal = 0;    // clear signal value
        rc = timedWaitForCond( &msg.cond , &msg.mutex, timeToWait); // cond variable is unlocked while waiting
        // var is locked again after exit from timedWaitForCond()
        if(0 != rc)
        {
            // Timeout or error, no messasge yet, for this app just loop around...
            std::cout << "no message yet, rc = " << rc << std::endl;
        }else
        {
            // return from timed wait with valid signal from sender
            std::cout << msg.value << std::endl;
            std::cout << "messageReceiverThread msg.value received: " << msg.value << std::endl;
            switch(msg.value)
            {
            case MSG1:
                std::cout << message1 << std::endl;
                break;
            case MSG2:
                std::cout << message2 << std::endl;
                break;
            case MSGX:
                std::cout << "exiting" << std::endl;
                pthread_exit(arg);
                break;
            default:
                std::cout << "unrecognized message" << std::endl;
                break;
            }
            outputMenu();
        }
    }
    pthread_exit(arg);
}

/*
 * function messageSenderThread():
 *      attempts to wait for user to enter a number at the console.
 *      When the number is received this thread will attempt a lock
 *      on the message mutex and, once obtained, write the value
 *      to the msg.msgNbr.  Then this thread will send a signal to
 *      the messageReceiver thread. Once completed this thread will
 *      again wait on the user for an additional imput, until the
 *      value enterd by the user is 0, which will cause an exit.
 *
 * algorithm:
 * input:     messageWait mutex
 * output:    character from message_t structure
 * memory:    shared message_t structure
 */
void* messageSenderThread(void* arg)
{
    int rc = 0,msgCode = 0;

    while(1)
    {
        rc = pthread_mutex_lock(&msg.mutex);
        if(rc)
        {
            std::cout << "messageSender: lock request failed,result: " << rc << std::endl;
            pthread_exit(arg);
        }
        msgCode = getUserMessage();
        msg.value = msgCode;
        msg.testVal = 1;
        std::cout << "messageSender: msg.value: " << msg.value << std::endl;
        // signal the condition variable - there's a new message to handle
        rc = pthread_cond_signal(&msg.cond);
        std::cout << "messageSender: signal sent! " << std::endl;
        if(rc)
            std::cout << "messageSender: pthread_cond_signal failed,result: " << rc << std::endl;
        // unlock mutex
        rc = pthread_mutex_unlock(&msg.mutex);
        if(rc)
        {
            std::cout << "messageSender: unlock request failed,result: " << rc << std::endl;
            pthread_exit(arg);
        }

        if(!msgCode)            // user entered the exit code
            pthread_exit(arg);
    }
    pthread_exit(arg);
}

#undef DEBUG
#endif

int main(int argc, char *argv[])
{
#if(defined PTHREAD_COM_EXAMPLE)
    int        thr_id[NUM_HANDLER_THREADS];     // thread IDs
    pthread_t  thread1,thread2;  // thread's structures

    // create the message handling threads
    thr_id[0] = 0;
    (void) pthread_create(&thread1, NULL, messageReceiverThread, (void*)&thr_id[0]);
    thr_id[1] = 1;
    (void) pthread_create(&thread2, NULL, messageSenderThread, (void*)&thr_id[1]);

    // now wait for threads to exit...
    (void) pthread_join(thread1, NULL);
    (void) pthread_join(thread2, NULL);

    return 0;

#endif
}

Answer 1

这里的问题是饥饿。您的messageSenderThread() 函数实际上一直锁定互斥锁。每次它在循环底部释放互斥体时，它所做的下一件事就是（如果您不键入“0”）将互斥体重新锁定在循环顶部。 messageReceiverThread() 在锁定互斥锁的竞争中总是失败。

在您输入“退出”命令 (0) 后，消息接收者确实 打印某些内容的原因是，messageSenderThread() 解锁了互斥体并且它退出输入零后。这终于允许接收者锁定互斥体并做它的事。