Как избежать условий гонки в переменной условия в VxWorks

Question

Мы программируем проприетарную встроенную платформу, сидящую поверх VxWorks 5.5. В нашем наборе инструментов есть переменная условия, которая реализуется с использованием двоичного семафора VxWorks.

Теперь POSIX предоставляет функцию ожидания, которая также принимает мьютекс. Это разблокирует мьютексы (так, чтобы какая-то другая задача могла записывать данные) и ждет, пока другая задача будет сигнализировать (это делается, записывая данные). Я считаю, что это реализует так называемый Monitor, 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 (по поставщику проприетарного аппаратного обеспечения, от того, что я понял).

Answer 1

Условия гонки можно избежать, если каждое задание ожидания ожидает отдельного двоичного семафора. Эти семафоры должны быть зарегистрированы в контейнере, который использует задача сигнализации, чтобы разблокировать все задачи ожидания. Контейнер должен быть защищен мьютексом.

Метод 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(), можно использовать пул.

Answer 2

Из описания, похоже, вы захотите реализовать (или использовать) семафор - это стандартный алгоритм CS с семантикой, подобный condvars, и есть тонны учебников о том, как их реализовать (https://www.google.com/search?q=semaphore+algorithm).

Случайный результат Google, который объясняет семафоры, находится по адресу: http://www.cs.cornell.edu/courses/cs414/2007sp/lectures/08-bakery.ppt (см. Слайд 32).

Answer 3

Это должно быть довольно просто с помощью встроенного 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 немедленно возвращается с сигналом, когда он в конечном итоге вызывается, и вы можете снова принять мьютекс в ul dtor, как указано выше.

Event-> q - это MSG_Q_ID, созданный во время создания события с вызовом функции msgQCreate, а данные в sigMsg определены вами ... но могут быть просто случайным байтом данных, или вы можете придумать более интеллектуальную структуру с информацией о том, кто сигнализирует или что-то еще, что может быть приятно узнать.

обновления для нескольких официантов, это немного сложнее: Итак, есть несколько предположений, я буду делать, чтобы упростить вещи

1. Количества задач, которые будут незавершенными известно во время создания событий и являются постоянная.
2. Будет одна задача, которая всегда отвечает за указание, когда нормально разблокировать мьютексы, все остальные задачи просто хотят уведомления, когда событие сигнализируется/завершается.

Этот подход использует счетный семафор, подобный приведенному выше только с небольшой дополнительной логики:

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; 
} 

Вы не можете быть невыразительными о numberOfWaiters: D Я скажу это снова: В numberOfWaiters должен быть правильным перед отпирать разблокирует мьютекс. Чтобы сделать его динамическим (если это необходимо), вы можете добавить функцию setNumWaiters (numOfWaiters) и вызвать это в функции wait_for перед тем, как разблокировка разблокирует мьютекс, если он всегда правильно устанавливает номер.

Теперь для последнего трюка, как указано выше, предполагается, что одна задача отвечает за разблокирование мьютекса, остальные просто ждут сигнала, а это означает, что одна и только одна задача вызовет функцию wait_for() выше , а остальные задачи просто вызывают функцию wait (event).

Имея это в виду numberOfWaiters вычисляется следующим образом:

• Количество задач, которые будут называть ожидание()
• плюс 1 для выполнения этой задачи, которая вызывает wait_for()

Конечно, вы также можете сделать это более сложным, если вам действительно нужно, но, скорее всего, это будет работать, потому что обычно одна задача запускает событие, но многие задачи хотят знать, что это завершено, и это то, что это обеспечивает.

Но ваш основной поток выглядит следующим образом:

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"); 
} 

Когда я пишу выше, я чувствую, что все концепции OO мертвы, но, надеюсь, вы получите эту идею, в реальности ожидания и wait_for должны принимать один и тот же параметр , или нет параметров, а скорее являются членами того же класса, который также имеет все данные, которые им нужно знать ... но тем не менее это обзор того, как он работает.