Рассмотрим пример потока, модули 1 и 3 которого поддерживают управление потоком данных, а модуль 2 — нет. Другими словами, модуль 2 не имеет процедуры
<i>xx</i>service()
. Когда сообщение достигает модуля 3, вызывается его функция
<i>xx</i>put()
. После необходимой обработки сообщения, оно помещается в очередь модуля 3 с помощью функции
putq(9F). Если при этом число сообщений в очереди превышает верхнюю ватерлинию,
putq(9F) устанавливает специальный флаг, сигнализирующий о том, что очередь переполнена:
mod1put(queue_t* q, mblk_t* mp) {
/* Необходимая обработка сообщения */
...
putq(q, mp);
}
Через некоторое время ядро автоматически запускает процедуру
<i>xx</i>service()
модуля 3. Для каждого сообщения очереди
xxput()
вызывает функцию
canput(9F), которая проверяет заполненность очереди следующего по потоку модуля. Функция
canput(9F) имеет вид:
#include <sys/stream.h>
int canput(queue_t* q);
Заметим, что canput(9F) проверяет заполненность очереди следующего модуля, реализующего механизм управления передачей данных, т.е. производящего обработку очереди с помощью процедуры
<i>xx</i>service()
. В противном случае, как уже говорилось, очередь модуля не принимает участия в передаче данных. В нашем примере,
canput(9F) проверит заполненность очереди записи модуля 1. Функция возвращает истинное значение, если очередь может принять сообщение, и ложное — в противном случае. В зависимости от результата проверки процедура
<i>xx</i>service()
либо передаст сообщение следующему модулю (в нашем примере — модулю 2, который после необходимой обработки сразу же передаст его модулю 1), либо вернет сообщение обратно в очередь, если следующая очередь переполнена.
Описанная схема показана на рис. 5.20. Ниже приведен скелет процедуры
<i>xx</i>service()
модуля 3, иллюстрирующий описанный алгоритм передачи сообщений с использованием механизма управления передачей данных.
Рис. 5.20. Управление потоком данных
mod1service(queue_t *q) {
mblk_t* mp;
while ((mp = getq(q)) != NULL) {
if (canput(q->q_next))
putnext(q, mp);
else {
putbq(q, mp);
break;
}
}
В этом примере функция getq(9F) используется для извлечения следующего сообщения из очереди, а функция putbq(9F) — для помещения сообщения в начало очереди. Если модуль 1 блокирует передачу, т.е. canput(9F) вернет "ложно", процедура
<i>xx</i>service()
завершает свою работу, и сообщения начинают буферизоваться в очереди модуля 3. При этом очередь временно исключается из списка очередей, ожидающих обработки, и процедура
<i>xx</i>service()
для нее вызываться не будет. Данная ситуация продлится до тех пор, пока число сообщений очереди записи модуля 1 не станет меньше нижней ватерлинии.
Пока существует возникшая блокировка передачи, затор будет постепенно распространяться вверх по потоку, последовательно заполняя очереди модулей, пока, в конечном итоге, не достигнет головного модуля. Поскольку передачу данных в головной модуль (вниз по потоку) инициирует приложение, попытка передать данные в переполненный головной модуль вызовет блокирование процесса[61] и переход его в состояние сна.
В конечном итоге, модуль 1 обработает сообщения своей очереди, и их число станет меньше нижней ватерлинии. Как только очередь модуля 1 станет готовой к приему новых сообщений, планировщик STREAMS автоматически вызовет процедуры
<i>xx</i>service()
для модулей, ожидавших освобождения очереди модуля в нашем примере — для модуля 3.
Управление передачей данных в потоке требует согласованной работы всех модулей. Например, если процедура
<i>xx</i>put()
буферизует сообщения для последующей обработки
<i>xx</i>service()
, такой алгоритм должен выполняться для всех сообщений.
[62] В противном случае, это может привести к нарушению порядка сообщений, и как следствие, к потере данных.
Когда запускается процедура
<i>xx</i>service()
, она должна обработать все сообщения очереди. "Уважительной" причиной прекращения обработки является переполнение очереди следующего по потоку модуля. В противном случае нарушается механизм управления передачей, и очередь может навсегда лишиться обработки.
Драйвер
Драйверы и модули очень похожи, они используют одинаковые структуры данных (
streamtab
,
qinit
,
module_info
) и одинаковый интерфейс (
<i>xx</i>open()
,
<i>xx</i>put()
,
<i>xx</i>service()
и
<i>xx</i>close()
). Однако между драйверами и модулями существуют различия.
Во-первых, только драйверы могут непосредственно взаимодействовать с аппаратурой и отвечать за обработку аппаратных прерываний. Поэтому драйвер должен зарегистрировать в ядре соответствующий обработчик прерываний. Аппаратура обычно генерирует прерывания при получении данных. В ответ на это драйвер копирует данные от устройства, формирует сообщение и передает его вверх по потоку.
Во-вторых, к драйверу может быть подключено несколько потоков. Как уже обсуждалось, на мультиплексировании потоков построены многие подсистемы ядра, например, поддержка сетевых протоколов. В качестве мультиплексора может выступать только драйвер. Несмотря на то что драйвер в этом случае не является оконечным модулем (см., например, рис. 5.15), размещение драйверов существенным образом отличается от встраивания модулей.
Наконец, процесс инициализации драйверов и модулей различен. Функция
<i>xx</i>open()
драйвера вызывается при открытии потока, в то время как инициализация модуля происходит при встраивании.
Головной модуль
Обработку системных вызовов процессов осуществляет головной модуль. Головной модуль потока является единственным местом, где возможно блокирование обработки и, соответственно, процесса, в контексте которого осуществляется операция ввода/вывода. Головной модуль является внешним интерфейсом потока, и хотя его структура похожа на структуру обычного модуля, функции обработки здесь обеспечиваются самой подсистемой STREAMS. В отличие от точек входа в модуль или драйвер потока, реализующих специфическую для данного устройства обработку, функции головного модуля выполняют ряд общих для всех потоков задач, включающих:
