Изменение переменной-члена в одном потоке и чтение в другом потоке

Question

У меня есть две переменные-члены.

class MessageQueues {
    ....
    char* m_InputBuffer;
    uint32_t m_InputBufferSize;
    ....
};

Я обновляю их в функции в текущем потоке и даже с помощью отладчика вижу, что переменные-члены обновляются. Но в другой функции, работающей в отдельном потоке, переменные-члены содержат мусор.

Итак, у меня есть функция, которая читает два значения и записывает их в две переменные-члены, а затем запускает другой поток.

ErrorCode MessageQueue::handleIncomingMessage(char receiveBuffer[], const uint32_t bufferSize) {
    ES_TRC3("started");
    ErrorCode errorCode = ES_SUCCESS;
    pthread_mutex_lock(&mutexMsgQueueIncoming);
    m_InputBuffer = receiveBuffer;
    m_InputBufferSize = bufferSize;
    ES_TRC3("triggering and unlocking");
    pthread_cond_signal(&msgQueueCondition);
    pthread_mutex_unlock(&mutexMsgQueueIncoming);
    ES_TRC3("triggered and unlocked");
    return errorCode;
}

Затем во второй функции-члене, которая бесконечно работает в отдельном потоке и ожидает сообщений,

ErrorCode MessageQueue::runReceiver(void) {
    ES_TRC3("started");
    ErrorCode errorCode = ES_SUCCESS;
    while(true)
    {
        ES_TRC3("waiting for input messages");
        pthread_cond_wait(&msgQueueCondition, &mutexMsgQueueIncoming);
        pthread_mutex_lock(&mutexMsgQueueIncoming);
        ES_TRC3("will parse message");
        if (strlen(m_InputBuffer) > 0UL) {
            if ((errorCode = m_MsgProtocol->parseMessage(m_InputBuffer, m_InputBufferSize)) != ES_SUCCESS)
            {
                ES_TRC1("failed to parseMessage, error:%d", errorCode);
            }
        }
        ES_TRC3("message parsed");
        pthread_mutex_unlock(&mutexMsgQueueIncoming);
    }
    return errorCode;
}

С помощью отладчика (и с сообщениями журнала) я вижу, что обновление переменных и запуск происходят в правильном порядке. Поэтому я не могу понять, почему тогда во 2-й функции переменные-члены не содержат правильных значений. Я где-то читал что-то про "летучие". Но не думаю, что это правильный ответ. Я что-то здесь не так делаю?

UPDATE: Мьютексы находятся за пределами класса, но в том же исходном файле, инициализированы сверху, как этот.

pthread_mutex_t mutexMsgQueueIncoming       = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t msgQueueCondition            = PTHREAD_COND_INITIALIZER;

На самом деле, это странно. Когда я запускаю программу без отладчика, кажется, что мьютекс не может быть заблокирован в функции runReceiver (). В отладчике мьютекс заблокирован, но переменные-члены содержат мусор. Ладно, думаю, мне следует это расследовать.

Answer 1

Пара ошибок здесь:

1. Ожидание условной переменной должно выполняться, когда мьютекс заблокирован. pthread_cond_wait документацию необходимо прочитать.
2. Переменная условия ложные пробуждения должны обрабатываться циклом while.

Исправления:

class MessageQueues {
    ....
    char* m_InputBuffer;
    uint32_t m_InputBufferSize;
    uint32_t m_WriteGeneration = 0; // <--- a fix.
    uint32_t m_ReadGeneration = 0;  // <--- a fix.
    ....
};

ErrorCode MessageQueue::handleIncomingMessage(char receiveBuffer[], const uint32_t bufferSize) {
    ES_TRC3("started");
    ErrorCode errorCode = ES_SUCCESS;
    pthread_mutex_lock(&mutexMsgQueueIncoming);
    m_InputBuffer = receiveBuffer;
    m_InputBufferSize = bufferSize;
    ++m_WriteGeneration; // <--- a fix.
    ES_TRC3("triggering and unlocking");
    pthread_cond_signal(&msgQueueCondition);
    pthread_mutex_unlock(&mutexMsgQueueIncoming);
    ES_TRC3("triggered and unlocked");
    return errorCode;
}

ErrorCode MessageQueue::runReceiver(void) {
    ES_TRC3("started");
    ErrorCode errorCode = ES_SUCCESS;
    while(true)
    {
        ES_TRC3("waiting for input messages");
        pthread_mutex_lock(&mutexMsgQueueIncoming);  // <--- a fix.
        while(m_ReadGeneration == m_WriteGeneration) // <--- a fix.
            pthread_cond_wait(&msgQueueCondition, &mutexMsgQueueIncoming); // <--- a fix.
        m_ReadGeneration = m_WriteGeneration; // <--- a fix.
        ES_TRC3("will parse message");
        if (strlen(m_InputBuffer) > 0UL) {
            if ((errorCode = m_MsgProtocol->parseMessage(m_InputBuffer, m_InputBufferSize)) != ES_SUCCESS)
            {
                ES_TRC1("failed to parseMessage, error:%d", errorCode);
            }
        }
        ES_TRC3("message parsed");
        pthread_mutex_unlock(&mutexMsgQueueIncoming);
    }
    return errorCode;
}

Обратите внимание, что если handleIncomingMessage вызывается дважды, когда runReceiver разблокирует мьютекс, этот код теряет сообщение. Вам, вероятно, здесь нужна очередь сообщений вместо хранения последнего сообщения (т. Е. У вас очередь размером 1, в которой старые элементы отбрасываются при переполнении).

Код также должен проверять коды ошибок, возвращаемые функциями pthread. Это довольно утомительно, лучше использовать C ++ 11 std::mutex, std::condition_variable, std::unique_lock, которые делают проверки за вас. В C ++ 98 вы можете использовать boost эквивалентов.