Как избежать состояния гонки в условной переменной в VxWorks
Мы программируем на проприетарной встраиваемой платформе, стоящей поверх VxWorks 5.5. В нашем наборе инструментов есть переменная условия, которая реализована с использованием двоичного семафора VxWorks.
Теперь POSIX предоставляет функцию ожидания, которая также принимает мьютекс. Это разблокирует мьютекс (так что какая-то другая задача может записать данные) и будет ждать, пока другая задача выдаст сигнал (запись данных завершена). Я считаю, что это реализует то, что называется Монитор, ICBWT.
Нам нужна такая функция ожидания, но реализовать ее сложно. Простой подход сделал бы это:
bool condition::wait_for(mutex& mutex) const {
unlocker ul(mutex); // relinquish mutex
return wait(event);
} // ul's dtor grabs mutex again
Тем не менее, это спортивное состояние гонки, потому что оно позволяет другой задаче выгрузить эту задачу после разблокировки и перед ожиданием. Другая задача может записывать дату после того, как она была разблокирована, и сигнализировать о состоянии, прежде чем эта задача начнет ожидать семафор. (Мы проверили это, и это действительно происходит и блокирует задачу ожидания навсегда.)
Учитывая, что VxWorks 5.5, похоже, не предоставляет API для временного освобождения семафора во время ожидания сигнала, есть ли способ реализовать это поверх предоставленных подпрограмм синхронизации?
Замечания: Это очень старая версия VxWorks, которая была скомпилированный без поддержки POSIX (от производителя проприетарного оборудования, насколько я понял).
Решение
Можно избежать условий гонки, если каждая задача ожидания ожидает отдельный двоичный семафор.
Эти семафоры должны быть зарегистрированы в контейнере, который используется задачей сигнализации для разблокирования всех ожидающих задач. Контейнер должен быть защищен мьютексом.
wait_for() Метод получает двоичный семафор, ожидает его и, наконец, удаляет.
void condition::wait_for(mutex& mutex) {
SEM_ID sem = semBCreate(SEM_Q_PRIORITY, SEM_EMPTY);
{
lock l(listeners_mutex); // assure exclusive access to listeners container
listeners.push_back(sem);
} // l's dtor unlocks listeners_mutex again
unlocker ul(mutex); // relinquish mutex
semTake(sem, WAIT_FOREVER);
{
lock l(listeners_mutex);
// remove sem from listeners
// ...
semDelete(sem);
}
} // ul's dtor grabs mutex again
signal() Метод перебирает все зарегистрированные семафоры и разблокирует их.
void condition::signal() {
lock l(listeners_mutex);
for_each (listeners.begin(), listeners.end(), /* call semGive()... */ )
}
Такой подход гарантирует, что wait_for() никогда не пропустит сигнал. Недостатком является необходимость дополнительных системных ресурсов.
Чтобы избежать создания и уничтожения семафоров для каждого wait_for() вызов, пул может быть использован.
Другие решения
Это должно быть довольно просто с нативной vxworks, здесь требуется очередь сообщений. Ваш метод wait_for может использоваться как есть.
bool condition::wait_for(mutex& mutex) const
{
unlocker ul(mutex); // relinquish mutex
return wait(event);
} // ul's dtor grabs mutex again
но код ожидания (события) будет выглядеть так:
wait(event)
{
if (msgQRecv(event->q, sigMsgBuf, sigMsgSize, timeoutTime) == OK)
{
// got it...
}
else
{
// timeout, report error or something like that....
}
}
и ваш код сигнала будет выглядеть примерно так:
signal(event)
{
msgQSend(event->q, sigMsg, sigMsgSize, NO_WAIT, MSG_PRI_NORMAL);
}
Таким образом, если сигнал сработает до того, как вы начнете ждать, то msgQRecv сразу же вернется с сигналом, когда он в конечном итоге будет вызван, и вы можете снова взять мьютекс в ultor, как указано выше.
Событие-> q — это MSG_Q_ID, который создается во время создания события с помощью вызова msgQCreate, а данные в sigMsg определяются вами … но могут быть просто случайным байтом данных или вы можете получить более интеллектуальная структура с информацией о том, кто сигнализировал или что-то еще, что может быть приятно узнать.
Обновление для нескольких официантов, это немного сложно: так что я сделаю пару предположений, чтобы упростить вещи
- Количество задач, ожидающих выполнения, известно во время создания события и является постоянным.
- Будет одна задача, которая всегда отвечает за указание, когда можно разблокировать мьютекс, все остальные задачи просто хотят уведомления, когда событие сигнализируется / завершается.
В этом подходе используется счетный семафор, похожий на описанный выше, с небольшой дополнительной логикой:
wait(event)
{
if (semTake(event->csm, timeoutTime) == OK)
{
// got it...
}
else
{
// timeout, report error or something like that....
}
}
и ваш код сигнала будет выглядеть примерно так:
signal(event)
{
for (int x = 0; x < event->numberOfWaiters; x++)
{
semGive(event->csm);
}
}
Создание события происходит примерно так, помните, что в этом примере число официантов постоянно и известно во время создания события. Вы можете сделать это динамическим, но ключ в том, что каждый раз, когда происходит событие, numberOfWaiters должно быть правильным, прежде чем разблокировщик разблокирует мьютекс.
createEvent(numberOfWaiters)
{
event->numberOfWaiters = numberOfWaiters;
event->csv = semCCreate(SEM_Q_FIFO, 0);
return event;
}
Вы не можете быть неуверенными в отношении числа Ожиданий: D Я повторю еще раз: число Должных пассажиров должно быть правильным, прежде чем разблокировщик разблокирует мьютекс. Чтобы сделать его динамическим (если это необходимо), вы можете добавить функцию setNumWaiters (numOfWaiters) и вызвать ее в функции wait_for, прежде чем разблокировщик разблокирует мьютекс, при условии, что он всегда правильно устанавливает число.
Теперь для последнего трюка, как указано выше, предполагается, что одна задача отвечает за разблокировку мьютекса, остальные просто ждут сигнала, а это означает, что одна и только одна задача будет вызывать функцию wait_for () выше, а остальные из задач просто вызовите функцию ожидания (события).
Учитывая это, числоOfWaiters вычисляется следующим образом:
- Количество задач, которые будут вызывать wait ()
- плюс 1 за задачу, которая вызывает wait_for ()
Конечно, вы также можете сделать это более сложным, если вам действительно нужно, но есть вероятность, что это сработает, потому что обычно 1 задача вызывает событие, но многие задачи хотят знать, что оно завершено, и это то, что это обеспечивает.
Но ваш основной поток выглядит следующим образом:
init()
{
event->createEvent(3);
}
eventHandler()
{
locker l(mutex);
doEventProcessing();
signal(event);
}
taskA()
{
doOperationThatTriggersAnEvent();
wait_for(mutex);
eventComplete();
}
taskB()
{
doWhateverIWant();
// now I need to know if the event has occurred...
wait(event);
coolNowIKnowThatIsDone();
}
taskC()
{
taskCIsFun();
wait(event);
printf("event done!\n");
}
Когда я пишу вышеупомянутое, я чувствую, что все концепции ОО мертвы, но, надеюсь, вы поняли идею, в действительности, wait и wait_for должны принимать один и тот же параметр или не иметь никакого параметра, а скорее быть членами того же класса, который также имеет все данные, которые они нужно знать … но тем не менее это обзор того, как это работает.
Из описания выглядит, что вы, возможно, захотите реализовать (или использовать) семафор — это стандартный алгоритм CS с семантикой, аналогичной condvars, и существует множество учебников о том, как их реализовать (https://www.google.com/search?q=semaphore+algorithm).
Случайный результат Google, который объясняет семафоры, находится по адресу: http://www.cs.cornell.edu/courses/cs414/2007sp/lectures/08-bakery.ppt(См. Слайд 32).