Андрей Робачевский - Операционная система UNIX

Как уже обсуждалось, передача данных в потоке происходит в виде сообщений. Процесс инициирует передачу данных с помощью системных вызовов write(2) и putmsg(2) , которые непосредственно взаимодействуют с головным модулем. Головной модуль формирует сообщение, копируя в него прикладные данные, и передает его следующему модулю вниз по потоку. В конечном итоге сообщение принимается драйвером, который выполняет необходимые операции с конкретным устройством. В случае, когда драйвер получает данные от устройства, он также передает их в виде сообщений вверх по потоку. Процесс имеет возможность получить данные с помощью системных вызовов read(2) или getmsg(2) . Если в головном модуле данные отсутствуют, процесс блокируется и переходит в состояние сна.

Сообщения передаются модулями с помощью системной функции putnext(9F) :

#include

#include

int putnext(queue_t *q, mblk_t *mp);

Эта функция адресует очередь следующего модуля параметром qи вызывает процедуру xx put()этой очереди, передавая ей сообщение mp. Не поощряется непосредственный вызов функции xx put()следующего модуля, поскольку это может вызвать определенные проблемы переносимости.

Передача данных внутри потока осуществляется асинхронно и не может блокировать процесс. Блокирование процесса возможно только при передаче данных между процессом и головным модулем. Таким образом, функции обработки данных потока — xx put()и xx service()не могут блокироваться. Если процедура xx put()не может передать данные следующему модулю, она помещает сообщение в собственную очередь, откуда оно может быть передано позже процедурой xx service(). Если и процедура xx service()не может осуществить передачу сообщения, например, из-за переполнения очереди следующего модуля, она не будет ожидать изменения ситуации, а вернет сообщение обратно в собственную очередь и завершит выполнение. Попытка передачи повторится, когда ядро через некоторое время опять запустит xx service().

Процедура xx service()вызывается в системном контексте, а не в контексте процесса, который инициировал передачу данных. Таким образом, блокирование процедуры xx service()может заблокировать (перевести в состояние сна) независимый процесс, что может привести к непредсказуемым результатам и потому недопустимо. Решение этой проблемы заключается в запрещении процедурам xx put()и xx service()блокирования своего выполнения.

Блокирование недопустимо и для драйвера. Обычно прием данных драйвером осуществляется с использованием прерываний. Таким образом процедура xx put()вызывается в контексте прерывания и не может блокировать свое выполнение.

Когда процедура xx put()не может передать сообщение следующему модулю, она вызывает функцию putq(9F) , имеющую следующий вид:

#include

int putq(queue_t *q, mblk_t *mp);

Функция putq(9F) помещает сообщение mp в очередь q, где сообщение ожидает последующей передачи, и заносит очередь в список очередей, нуждающихся в обработке. Для таких очередей ядро автоматически вызывает процедуру xx service(). Планирование вызова процедур xx service()производится функцией ядра runqueues(). [59] Система планирования STREAMS использует собственные функции и не имеет отношения к планированию процессов в UNIX. Функция runqueues()вызывается ядром в двух случаях:

□ Когда какой-либо процесс выполняет операцию ввода/вывода над потоком.

□ Непосредственно перед переходом какого-либо процесса из режима ядра в режим задачи.

Заметим, что планирование обслуживания очередей не связано с конкретным процессом и производится для всей подсистемы STREAMS в целом.

Функция runqueue()производит поиск всех потоков, нуждающихся в обработке очередей. При наличии таковых просматривается список очередей, ожидающих обработки, и для каждой из них вызывается соответствующая функция xx service(). Каждая процедура xx service(), в свою очередь, пытается передать все сообщения очереди следующему модулю. Если для каких-либо сообщений это не удается, они остаются в очереди, ожидая следующего вызова runqueue(), после чего процесс повторяется.

Деление процесса передачи данных на два этапа, выполняемых, соответственно, функциями xx put()и xx service(), позволяет реализовать механизм управления передачей данных.

Как уже упоминалось, обязательной для модуля является лишь функция xx put(). Рассмотрим ситуацию, когда модули потока не содержат процедур xx service(). В этом случае, проиллюстрированном на рис. 5.19, каждый предыдущий модуль вызывает функцию xx put()следующего, передавая ему сообщение, с помощью функции ядра putnext(9F) . Функция xx put()немедленно вызывает putnext(9F) и т.д.:

xxput(queue_t *q, mblk_t *mp) {

putnext(q, mp);

}

Рис. 5.19. Передача данных без управления потоком

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

Хотя такой вариант может применяться для некоторых драйверов (как правило, для псевдоустройств, например, /dev/null), в общем случае устройство не может быть все время готово к обработке данных, а потеря данных из-за занятости устройства недопустима. Таким образом, в потоке может происходить блокирование передачи данных [60] Блокирование передачи может происходить не только в драйвере (оконечном модуле) потока из-за занятости устройства. Возможна ситуация, когда отдельный модуль вынужден отложить обработку сообщений до наступления некоторого события. , и эта ситуация не должна приводить к потере сообщений, во избежание которой необходим согласованный между модулями механизм управления потоком. Для этого сообщения обрабатываются и буферизуются в соответствующей очереди модуля, а их передача возлагается на функцию xx service(), вызываемую ядром автоматически. Для каждой очереди определены две ватерлинии — верхняя и нижняя, которые используются для контроля заполненности очереди. Если число сообщений превышает верхнюю ватерлинию, очередь считается переполненной, и передача сообщений блокируется, пока их число не станет меньше нижней ватерлинии.