• Главная
• В избранное
• Наш E-MAIL
• Добавить материал
• Нашёл ошибку
• Вниз

Операциям ввода и вывода присуща более медленная скорость выполнения по сравнению с другими видами обработки. Причиной такого замедления являются следующие факторы:

• Задержки, обусловленные затратами времени на поиск нужных дорожек и секторов на устройствах произвольного доступа (диски, компакт-диски).

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

• Задержки при передаче данных по сети с использованием файловых, серверов, хранилищ данных и так далее.

Во всех предыдущих примерах операции ввода/вывода выполняются синхронно с потоком, поэтому весь поток вынужден простаивать, пока они не завершатся.

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

Некоторые из этих методик используются в таймерах ожидания, которые также описываются в настоящей главе.

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

В Windows выполнение асинхронного ввода/вывода обеспечивается в соответствии с тремя методиками.

• Многопоточный ввод/вывод (Multihreaded I/O). Каждый из потоков внутри процесса или набора процессов выполняет обычный синхронный ввод/вывод, но при этом другие потоки могут продолжать свое выполнение.

• Перекрывающийся ввод/вывод (Overlapped I/O). Запустив операцию чтения, записи или иную операцию ввода/вывода, поток продолжает свое выполнение. Если потоку для продолжения выполнения требуются результаты ввода/вывода, он ожидает, пока не станет доступным соответствующий дескриптор или не наступит заданное событие. В Windows 9x перекрывающийся ввод/вывод поддерживается только для последовательных устройств, например именованных каналов.

• Процедуры завершения (или расширенный ввод/вывод) (Completion routines (extended I/O)). Когда наступает завершение операций ввода/вывода, система вызывает специальную процедуру завершения, выполняющуюся внутри потока. Расширенный ввод/вывод для дисковых файлов в Windows 9x не поддерживается.

Многопоточный ввод/вывод с использованием именованных каналов применен в сервере с многопоточной поддержкой, который рассматривался в главе 11. Программа grepMT (программа 7.1) управляет параллельным выполнением операций ввода/вывода с участием нескольких файлов. Таким образом, мы уже располагаем рядом программ, которые выполняют многопоточный ввод/вывод и тем самым обеспечивают одну из форм асинхронного ввода/вывода.

Перекрывающийся ввод/вывод является предметом рассмотрения следующего раздела, а в приведенных в нем примерах, реализующих преобразование файлов (из ASCII в UNICODE), эта методика применена для иллюстрации возможностей последовательной обработки файлов. С этой целью используется видоизмененный вариант программы 2.4. Вслед за перекрывающимся вводом/выводом рассматривается расширенный ввод/вывод, использующий процедуры завершения.

Примечание

Методы перекрывающегося и расширенного ввода/вывода часто оказываются сложными в реализации, редко обеспечивают какие-либо преимущества в отношении производительности, иногда даже становясь причиной ее ухудшения, а в случае файлового ввода/вывода способны работать лишь под управлением Windows NT. Эти проблемы преодолеваются с помощью потоков, поэтому, вероятно, многие читатели захотят сразу же перейти к разделам, посвященным таймерам ожидания и портам завершения ввода/вывода, возвращаясь к этому разделу по мере необходимости. С другой стороны, элементы асинхронного ввода/вывода присутствуют как в устаревших, так и в новых технологиях, в связи с чем эти методы все-таки стоит изучить.

Так, технология СОМ на платформе NT5 поддерживает асинхронный вызов методов, поэтому указанная методика может пригодиться многим читателям, которые используют или собираются использовать технологию СОМ. Кроме того, много общего с расширенным вводом/выводом имеют операции асинхронного вызова процедур (глава 10), и хотя лично я предпочитаю использовать потоки, другие могут отдать предпочтение именно этому механизму.

Первое, что необходимо сделать для организации асинхронного ввода/вывода, будь то перекрывающегося или расширенного, — это установить атрибут перекрывания (overlapped attribute) для файлового или иного дескриптора. Для этого при вызове CreateFile или иной функции, в результате которого создается файл, именованный канал или иной дескриптор, следует указать флаг FILE_FLAG_OVERLAPPED.

В случае сокетов (глава 12), независимо от того, были они созданы с использованием функции socket или accept, атрибут перекрывания устанавливается по умолчанию в Winsock 1.1, но должен устанавливаться явным образом в Winsock 2.0. Перекрывающиеся сокеты могут использоваться в асинхронном режиме во всех версиях Windows.

До этого момента структуры OVERLAPPED использовались нами совместно с функцией LockFileEx, а также в качестве альтернативы использованию функции SetFilePointer (глава 3), но они также являются существенным элементом перекрывающегося ввода/вывода. Эти структуры выступают в качестве необязательных параметров при вызове четырех приведенных ниже функций, которые могут блокироваться при завершении операций.

ReadFile

WriteFile

TransасtNamedPipe

ConnectNamedPipe

Вспомните, что при указании флага FILE_FLAG_OVERLAPPED в составе параметра dwAttrsAndFlags (в случае функции CreateFile) или параметра dwOpen-Mode (в случае функции CreateNamedPipe) соответствующие файл или канал могут использоваться только в режиме перекрывания. С анонимными каналами перекрывающийся ввод/вывод не работает.

Примечание

В документации по функции CreateFile есть упоминание о том, что использование флага FILE_FLAG_NO_BUFFERING улучшает характеристики быстродействия перекрывающегося ввода/вывода. Эксперименты показывают лишь незначительное повышение производительности (примерно на 15%, что может быть проверено путем экспериментирования с программой 14.1), но вы должны убедиться в том, что суммарный размер считываемых данных при выполнении операций ReadFile или WriteFile, кратен размеру сектора диска. 

Одним из наиболее важных нововведений в Windows Sockets 2.0 (глава 12) является стандартизация перекрывающегося ввода/вывода. В частности, сокеты уже не создаются автоматически как дескрипторы файлов с перекрытием. Функция socket создает неперекрывающийся дескриптор. Чтобы создать перекрывающийся сокет, следует вызвать функцию WSASocket, явно запросив создание перекрывающегося совета путем указания значения WSA_FLAG_OVERLAPPED для параметра dwFlags функции WSASocket. 

SOCKET WSAAPI WSASocket(int iAddressFamily, int iSocketType, int iProtocol, LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags);

 

Для создания сокета используйте вместо функции socket функцию WSASocket. Любой сокет, возвращенный функцией accept, будет иметь те же свойства, что и аргумент.

Перекрывающийся ввод/вывод выполняется в асинхронном режиме. Это имеет несколько следствий.

• Операции перекрывающегося ввода/вывода не блокируются. Функции ReadFile, WriteFile, TransactNamedPipe и ConnectNamedPipe осуществляют возврат, не дожидаясь завершения операции ввода/вывода.

• Возвращаемое функцией значение не может быть использовано в качестве критерия успешности или неудачи ее выполнения, поскольку операция ввода/вывода к этому моменту еще не успевает завершиться. Для индикации состояния выполнения ввода/вывода требуется привлечение другого механизма.

• Возвращенное значение количества переданных байтов также приносит мало пользы, поскольку передача данных могла не завершиться до конца. Для получения такого рода информации Windows должна предоставить другой механизм.

• Программа может многократно предпринимать попытки чтения или записи с использованием одного и того же перекрывающегося дескриптора файла. Поэтому незначащим оказывается и указатель файла, соответствующий такому дескриптору. Следовательно, должен быть предусмотрен дополнительный метод, обеспечивающий указание позиции в файле для каждой операции чтения или записи. В случае именованных каналов, в силу присущего им последовательного характера обработки данных, это не является проблемой.

• Для программы должна быть обеспечена возможность ожидания (синхронизации) завершения ввода/вывода. При наличии нескольких незавершенных операций ввода/вывода, связанных с одним и тем же дескриптором, программа должна быть в состоянии определить, какие из операций уже завершились. Операции ввода/вывода вовсе не обязательно завершаются в том же порядке, в каком они начинали выполняться.

Для преодоления двух последних из перечисленных выше трудностей используются структуры OVERLAPPED.

С помощью структуры OVERLAPPED (указываемой, например, параметром lpOverlapped функции ReadFile) можно указывать следующую информацию:

• Позицию в файле (64 бита), с которой должно начинаться выполнение операции чтения или записи в соответствии с обсуждением, которое содержится в главе 3.

• Событие (сбрасываемое вручную), которое будет переходить в сигнальное состояние по завершении соответствующей операции.

Ниже приводится определение структуры OVERLAPPED. 

typedef struct_OVERLAPPED {

 DWORD Internal;

 DWORD InternalHigh;

 DWORD Offset;

 DWORD OffsetHigh;

 HANDLE hEvent;

} OVERLAPPED

 

Для задания позиции в файле (указателя) должны использоваться оба поля Offset и OffsetHigh, хотя старшая часть указателя (OffsetHigh) во многих случаях равна 0. Не следует использовать поля Internal и InternalHigh, зарезервированные для системных нужд.

Параметр hEvent — дескриптор события (созданного посредством функции CreateEvent). Это событие может быть как именованным, так и неименованным, но оно должно быть обязательно сбрасываемым вручную (см. главу 8), если используется для перекрывающегося ввода/вывода; причины этого будут вскоре объяснены. По завершении операции ввода/вывода событие переходит в сигнальное состояние. 

В другом возможном варианте его использования дескриптор hEvent имеет значение NULL; в этом случае программа может ожидать перехода в сигнальное состояние дескриптора файла, который также может выступать в роли объекта синхронизации (см. приведенные далее предостережения). Система использует для отслеживания завершения операций сигнальные состояния дескриптора файла, если дескриптор hEvent равен NULL, то есть объектом синхронизации в этом случае является дескриптор файла.

Примечание

В целях удобства термин "дескриптор файла" ("file handle"), используемый по отношению к дескрипторам, указываемым при вызове функций ReadFile, WriteFile и так далее, будет применяться нами даже в тех случаях, когда речь идет о дескрипторах именованного канала или устройства, а не файла.

При выполнении вызова функций ввода/вывода это событие сразу же сбрасывается системой (устанавливается в несигнальное состояние). Когда операция ввода/вывода завершается, событие устанавливается в сигнальное состояние и остается в нем до тех пор, пока не будет использовано другой операцией ввода/вывода. Событие должно быть сбрасываемым вручную, если его перехода в сигнальное состояние могут ожидать несколько потоков (хотя в наших примерах используется всего один поток), и на момент завершения операции они могут не находиться в состоянии ожидания.

Даже если дескриптор файла является синхронным (то есть созданным без флага FILE_FLAG_OVERLAPPED), структура OVERLAPPED может послужить в качестве альтернативы функции SetFilePointer для указания позиции в файле. В этом случае возврат после вызова функции ReadFile или иного вызова не происходит до тех пор, операция ввода/вывода пока не завершится. Этой возможностью мы уже воспользовались в главе 3. Также обратите внимание на то, что незавершенные операции ввода/вывода однозначно идентифицируются комбинацией дескриптора файла и соответствующей структуры OVERLAPPED.

Ниже перечислены некоторые предостережения, которые следует принимать во внимание.

• Не допускайте повторного использования структуры OVERLAPPED в то время, когда связанная с ней операция ввода/вывода, если таковая имеется, еще не успела завершиться.

• Аналогичным образом, избегайте повторного использования события, указанного в структуре OVERLAPPED.

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

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

Возврат из функций ReadFile и WriteFile, а также двух указанных выше функций, относящихся к именованным каналам, в случаях, когда они используются для выполнения перекрывающихся операций ввода вывода, осуществляется немедленно. В большинстве случаев операция ввода/вывода к этому моменту завершена не будет, и возвращаемым значением при чтении и записи будет FALSE. Функция GetLastError возвратит в этой ситуации значение ERROR_IO_PENDING.

По окончании ожидания перехода объекта синхронизации (события или, возможно, дескриптора файла) в сигнальное состояние, свидетельствующее о завершении операции, вы должны выяснить, сколько байтов было передано. В этом и состоит основное назначение функции GetOverlappedResult. 

BOOL GetOverlappedResult(HANDLE hFile, LPOVERLAPPED lpOverlapped, LPWORD lpcbTransfer, BOOL bWait)

Указание конкретной операции ввода/вывода обеспечивается сочетанием дескриптора и структуры OVERLAPPED. Значение TRUE параметра bWait указывает на то, что до завершения операции функция GetOverlappedResult должна находиться в состоянии ожидания; в противном случае возврат из функции должен быть немедленным. В любом случае эта функция будет возвращать значение TRUE только после успешного завершения операции. Если возвращаемым значением функции GetOverlappedResult является FALSE, то функция GetLastError возвратит значение ERROR_IO_INCOMPLETE, что позволяет вызывать эту функцию для опроса завершения ввода/вывода.

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

Булевская функция CancelIO позволяет отменить выполнение незавершенных операций перекрывающегося ввода/вывода, связанных с указанным дескриптором (у этой функции имеется всего лишь один параметр). Отменяется выполнение всех инициированных вызывающим потоком операций, использующих данный дескриптор. На операции, инициированные другими потоками, вызов этой функции никакого влияния не оказывает. Отмененные операции завершаются С ошибкой ERROR OPERATION ABORTED.

Перекрывающийся ввод/вывод очень удобно и просто реализуется в тех случаях, когда может существовать только одна незавершенная операция. Тогда для целей синхронизации программа может использовать не событие, а дескриптор файла.

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

OVERLAPPED ov = { 0, 0, 0, 0, NULL /* События не используются. */ };

HANDLE hF;

DWORD nRead;

BYTE Buffer[BUF_SIZE];

…

hF = CreateFile( …, FILE_FLAG_OVERLAPPED, … );

ReadFile(hF, Buffer, sizeof(Buffer), &nRead, &ov);

/* Выполнение других видов обработки. nRead не обязательно достоверно.*/

/* Ожидать завершения операции чтения. */

WaitForSingleObject(hF, INFINITE);

GetOverlappedResult(hF, &ov, &nRead, FALSE);

Программа 2.4 (atou) осуществляла преобразование ASCII-файла к кодировке UNICODE путем последовательной обработки файла, а в главе 5 было показано, как выполнить такую же последовательную обработку с помощью отображения файлов. В программе 14.1 (atouOV) та же самая задача решается с использованием перекрывающегося ввода/вывода и множественных буферов, в которых хранятся записи фиксированного размера.

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

Сначала в программе выполняется инициализация всех элементов структур OVERLAPPED, определяющих события и позиции в файлах. Для каждого входного и выходного буферов предусмотрена отдельная структура OVERLAPPED. После этого для каждого из входных буферов инициируется операция перекрывающегося чтения. Далее с помощью функции WaitForMultipleObjects в программе организуется ожидание одиночного события, указывающего на завершение чтения или записи. При завершении операции чтения входной буфер копируется и преобразуется в соответствующий выходной буфер, после чего инициируется операция записи. При завершении записи инициируется следующая операция чтения. Заметьте, что события, связанные с входными и выходными буферами размещаются в единственном массиве, который используется в качестве аргумента при вызове функции WaitForMultipleObjects. 

Рис. 14.1. Модель асинхронного обновления файла

/* Глава 14. atouOV

   Преобразование файла из кодировки ASCII в кодировку Unicode с использованием перекрывающегося ввода/вывода. Программа работает только в Windows NT. */

#include "EvryThng.h"

#define MAX_OVRLP 4 /* Количество перекрывающихся операций ввода/вывода.*/ 

#define REC_SIZE 0x8000 /* 32 Кбайт: Минимальный размер записи, обеспечивающий приемлемую производительность. */

#define UREC_SIZE 2 * REC_SIZE

int _tmain(int argc, LPTSTR argv[]) {

 HANDLE hInputFile, hOutputFile;

 /* Каждый из элементов определенных ниже массивов переменных */

 /* и структур соответствует отдельной незавершенной операции */

 /* перекрывающегося ввода/вывода. */

 DWORD nin[MAX_OVRLP], nout[MAX_OVRLP], ic, i;

 OVERLAPPED OverLapIn[MAX_OVRLP], OverLapOut[MAX_OVRLP];

 /* Необходимость использования сплошного, двумерного массива */

 /* диктуется Функцией WaitForMultipleObjects. */

 /* Значение 0 первого индекса соответствует чтению, значение 1 – записи.*/

 HANDLE hEvents[2][MAX_OVRLP];

 /* В каждом из определенных ниже двух буферных массивов первый индекс */

 /* нумерует операции ввода/вывода. */

 CHAR AsRec[MAX_OVRLP][REC_SIZE];

 WCHAR UnRec[MAX_OVRLP][REC_SIZE];

 LARGE_INTEGER CurPosIn, CurPosOut, FileSize;

 LONGLONG nRecord, iWaits;

 hInputFile = CreateFile(argv[1], GENERIC_READ, 0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);

 hOutputFile = CreateFile(argv[2], GENERIC_WRITE, 0, NULL, CREATE_ALWAYS, FILE_FLAG_OVERLAPPED, NULL);

 /* Общее количество записей, подлежащих обработке, вычисляемое */

 /* на основе размера входного файла. Запись, находящаяся в конце, */

 /* может быть неполной. */

 FileSize.LowPart = GetFileSize(hInputFile, &FileSize.HighPart);

 nRecord = FileSize.QuadPart / REC_SIZE;

 if ((FileSize.QuadPart % REC_SIZE) != 0) nRecord++;

 CurPosIn.QuadPart = 0;

 for (ic = 0; ic < MAX_OVRLP; ic++) {

  /* Создать события чтения и записи для каждой структуры OVERLAPPED.*/

  hEvents[0][ic] = OverLapIn[ic].hEvent /* Событие чтения.*/

   = CreateEvent(NULL, TRUE, FALSE, NULL);

  hEvents[1][ic] = OverLapOut[ic].hEvent /* Событие записи. */

   = CreateEvent(NULL, TRUE, FALSE, NULL);

  /* Начальные позиции в файле для каждой структуры OVERLAPPED. */

  OverLapIn[ic].Offset = CurPosIn.LowPart;

  OverLapIn[ic].OffsetHigh = CurPosIn.HighPart;

  /* Инициировать перекрывающуюся операцию чтения для данной структуры OVERLAPPED. */

  if (CurPosIn.QuadPart < FileSize.QuadPart) ReadFile(hInputFile, AsRec[ic], REC_SIZE, &nin[ic], &OverLapIn[ic]); 

  CurPosIn.QuadPart += (LONGLONG)REC_SIZE;

 }

 /* Выполняются все операции чтения. Ожидать завершения события и сразу же сбросить его. События чтения и записи хранятся в массиве событий рядом друг с другом. */

 iWaits =0; /* Количество выполненных к данному моменту операций ввода/вывода. */

 while (iWaits < 2 * nRecord) {

  ic = WaitForMultipleObjects(2 * MAX_OVRLP, hEvents[0], FALSE, INFINITE) – WAIT_OBJECT_0;

  iWaits++; /* Инкрементировать счетчик выполненных операций ввода вывода.*/

  ResetEvent(hEvents[ic / MAX_OVRLP][ic % MAX_OVRLP]);

  if (ic < MAX_OVRLP) {

   /* Чтение завершено. */

   GetOverlappedResult(hInputFile, &OverLapIn[ic], &nin[ic], FALSE);

   /* Обработать запись и инициировать операцию записи. */

   CurPosIn.LowPart = OverLapIn[ic].Offset;

   CurPosIn.HighPart = OverLapIn[ic].OffsetHigh;

   CurPosOut.QuadPart = (CurPosIn.QuadPart / REC_SIZE) * UREC_SIZE;

   OverLapOut[ic].Offset = CurPosOut.LowPart;

   OverLapOut[ic].OffsetHigh = CurPosOut.HighPart;

   /* Преобразовать запись из ASCII в Unicode. */

   for (i =0; i < REC_SIZE; i++) UnRec[ic][i] = AsRec[ic][i];

   WriteFile(hOutputFile, UnRec[ic], nin[ic] * 2, &nout[ic], &OverLapOut[ic]);

   /* Подготовиться к очередному чтению, которое будет инициировано после того, как завершится начатая выше операция записи. */

   CurPosIn.QuadPart += REC_SIZE * (LONGLONG)(MAX_OVRLP);

   OverLapIn[ic].Offset = CurPosIn.LowPart;

   OverLapIn[ic].OffsetHigh = CurPosIn.HighPart;

  } else if (ic < 2 * MAX_OVRLP) { /* Операция записи завершилась. */

   /* Начать чтение. */

   ic –= MAX_OVRLP; /* Установить индекс выходного буфера. */

   if (!GetOverlappedResult (hOutputFile, &OverLapOut[ic], &nout[ic], FALSE)) ReportError(_T("Ошибка чтения."), 0, TRUE);

   CurPosIn.LowPart = OverLapIn[ic].Offset;

   CurPosIn.HighPart = OverLapIn[ic].OffsetHigh;

   if (CurPosIn.QuadPart < FileSize.QuadPart) {

    /* Начать новую операцию чтения. */

    ReadFile(hInputFile, AsRec[ic], REC_SIZE, &nin[ic], &OverLapIn[ic]);

   }

  }

 } 

 /* Закрыть все события. */

 for (ic = 0; ic < MAX_OVRLP; ic++) {

  CloseHandle(hEvents[0][ic]);

  CloseHandle(hEvents[1][ic]);

 }

 CloseHandle(hInputFile);

 CloseHandle(hOutputFile);

 return 0;

}

Программа 14.1 способна работать только под управлением Windows NT. Средства асинхронного ввода/вывода Windows 9x не позволяют использовать дисковые файлы. В приложении В приведены результаты и комментарии, свидетельствующие о сравнительно низкой производительности программы atouOV. Как показали эксперименты, для достижения приемлемой производительности размер буфера должен составлять, по крайней мере, 32 Кбайт, но даже и в этом случае обычный синхронный ввод/вывод работает быстрее. К тому же, производительность этой программы не повышается и в условиях SMP, поскольку в данном примере, в котором обрабатываются всего лишь два файла, ЦП не является критическим ресурсом.

Существует также другой возможный подход к использованию объектов синхронизации. Вместо того чтобы заставлять поток ожидать поступления сигнала завершения от события или дескриптора, система может инициировать вызов определенной пользователем процедуры завершения сразу же по окончании выполнения операции ввода/вывода. Далее процедура завершения может запустить очередную операцию ввода/вывода и выполнить любые необходимые действия по учету использования системных ресурсов. Эта косвенно вызываемая (callback) процедура завершения аналогична асинхронному вызову процедуры, который применялся в главе 10, и требует использования состояний дежурного ожидания (alertable wait states).

Каким образом процедура завершения может быть указана в программе? Среди параметров или структур данных функций ReadFile и WriteFile не остается таких, которые можно было бы использовать для хранения адреса процедуры завершения. Однако существует семейство расширенных функций ввода/вывода, которые обозначаются суффиксом "Ех" и содержат дополнительный параметр, предназначенный для передачи адреса процедуры завершения. Функциями чтения и записи являются, соответственно, ReadFileEx и WriteFileEx. Кроме того, требуется использование одной из указанных ниже функций дежурного ожидания.

• WaitForSingleObjectEx

• WaitForMultipleObjectsEx 

• SleepEx

• SignalObjectAndWait

• MsgWaitForMultipleObjectsEx

Расширенный ввод/вывод иногда называют дежурным вводом/выводом (alertable I/O). О том, как использовать расширенные функции, рассказывается в последующих разделах.

Примечание

Под управлением Windows 9x расширенный ввод/вывод не может работать с дисковыми файлами и коммуникационными портами. В то же время, средства расширенного ввода/вывода Windows 9x способны работать с именованными каналами, почтовыми ящиками, сокетами и последовательными устройствами.

Расширенные функции чтения и записи могут использоваться совместно с дескрипторами открытых файлов, именованных каналов и почтовых ящиков, если соответствующий объект открывался (создавался) с установленным флагом FILE_FLAG_OVERLAPPED. Заметьте, что этот флаг устанавливает атрибут дескриптора, и хотя перекрывающийся и расширенный ввод/вывод отличаются друг от друга, к дескрипторам обоих типов асинхронного ввода/вывода применяется один и тот же флаг.

Перекрывающиеся сокеты (глава 12) могут использоваться совместно с функциями ReadFileEx и WriteFileEx во всех версиях Windows. 

BOOL ReadFileEx(HANDLE hFile, LPVOID lpBuffer, DWORD nNumberOfBytesToRead, LPOVERLAPPED lpOverlapped, LPOVERLAPPED_COMPLETION_ROUTINE lpcr)

BOOL WriteFileEx(HANDLE hFile, LPVOID lpBuffer, DWORD nNumberOfBytesToWrite, LPOVERLAPPED lpOverlapped, LPOVERLAPPED_COMPLETION_ROUTINE lpcr)

 

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

Каждой из функций необходимо предоставлять структуру OVERLAPPED, но надобность в указании элемента hEvent этой структуры отсутствует; система игнорирует его. Вместе с тем, этот элемент оказывается очень полезным для передачи такой, например, информации, как порядковый номер, используемый для различения отдельных операций ввода/вывода, что демонстрируется в программе 14.2.

Сравнивая с функциями ReadFile и WriteFile, можно заметить, что расширенные функции не требуют параметров для хранения количества переданных байтов. Эта информация передается функции завершения, которая должна включаться в программу.

В функции завершения предусмотрены параметры для счетчика байтов, кода ошибки и адреса структуры OVERLAPPED. Последний из названных параметров требуется для того, чтобы процедура завершения могла определить, какая именно из невыполненных операций завершилась. Заметьте, что ранее высказанные предостережения относительно повторного использования или уничтожения структур OVERLAPPED справедливы здесь в той же мере, что и в случае перекрывающегося ввода/вывода. 

VOID WINAPI FileIOCompletionRoutine(DWORD dwError, DWORD cbTransferred, LPOVERLAPPED lpo)

 

Как и в случае функции CreateThread, при вызове которой также указывается имя некоторой функции, имя FileIOCompletionRoutine является заменителем, а не фактическим именем процедуры завершения.

Значения параметра dwError ограничены 0 (успешное завершение) и ERROR_HANDLE_EOF (при попытке выполнить чтение с выходом за пределы файла). Структура OVERLAPPED — это та структура, которая использовалась завершившимся вызовом ReadFileEx или WriteFileEx.

Прежде чем процедура завершения будет вызвана системой, должны произойти две вещи:

1. Должна завершиться операция ввода/вывода.

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

Каким образом поток переходит в состояние дежурного ожидания? Он должен выполнить явный вызов одной из функций дежурного ожидания, описанных в следующем разделе. Тем самым поток создает условия, делающие преждевременное выполнение процедуры завершения невозможным. В состоянии дежурного ожидания поток может находиться только на протяжении того времени, пока длится вызов функции дежурного ожидания; после возврата из этой функции поток выходит из указанного состояния. 

Если оба эти условия удовлетворены, выполняются процедуры завершения, помещенные в очередь в результате завершения операций ввода/вывода. Процедуры завершения выполняются в том же потоке, который выполнил первоначальный вызов функции ввода/вывода и находится в состоянии дежурного ожидания. Поэтому поток должен переходить в состояние дежурного ожидания только тогда, когда для выполнения процедур завершения существуют безопасные условия.

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

DWORD WaitForSingleObjectEx(HANDLE hObject, DWORD dwMilliseconds, BOOL bAlertable)

DWORD WaitForMultipleObjectsEx(DWORD cObjects, LPHANDLE lphObjects, BOOL fWaitAll, DWORD dwMilliseconds, BOOL bAlertable)

DWORD SleepEx(DWORD dwMilliseconds, BOOL bAlertable)

 

В каждой из функций дежурного ожидания имеется флаг bAlertable, который в случае асинхронного ввода/вывода должен устанавливаться в TRUE. Приведенные выше функции являются расширением знакомых вам функций Wait и Sleep.

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

• Дескриптор (дескрипторы) переходит (переходят) в сигнальное состояние, чем удовлетворяются стандартные требования двух из функций ожидания.

• Истекает интервал ожидания.

• Все процедуры завершения, находящиеся в очереди потока, прекращают свое выполнение, а значение параметра bAlertable равно TRUE. Процедура завершения помещается в очередь тогда, когда завершается соответствующая ей операция ввода/вывода (рис. 14.2). 

Заметьте, что со структурами OVERLAPPED в функциях ReadFileEx и WriteFileEx не связаны никакие события, поэтому ни один из дескрипторов, указываемых при вызове функции ожидания, не связывается непосредственно с какой-либо определенной операцией ввода/вывода. В то же время, функция SleepEx не связана с объектами синхронизации, и поэтому ее проще всего использовать. В случае функции SleepEx в качестве длительности интервала ожидания обычно указывают значение INFINITE, поэтому возврат из этой функции произойдет только после того, как закончится выполнение одной или нескольких процедур завершения, которые в настоящий момент находятся в очереди.

По окончании выполнения операции расширенного ввода/вывода связанная с ней процедура завершения со своими аргументами, определяющими структуру OVERLAPPED, счетчик байтов и код ошибки, помещается в очередь для выполнения.

Все процедуры завершения, находящиеся в очереди потока, начинают выполняться тогда, когда поток переходит в состояние дежурного ожидания. Они выполняются поочередно, но не обязательно в той же последовательности, в которой завершились операции ввода/вывода. Возврат из функции дежурного ожидания происходит только после того, как осуществят возврат процедуры завершения. Эту особенность важно учитывать для обеспечения правильного функционирования большинства программ, поскольку при этом предполагается, что процедуры завершения получают возможность подготовиться к очередному использованию структуры OVERLAPPED и выполнить другие необходимые действия для перевода программы в известное состояние, прежде чем будет осуществлен возврат из состояния дежурного ожидания.

Если возврат из функции SleepEx обусловлен выполнением одной или нескольких процедур завершения, находящихся в очереди, то возвращаемым значением функции будет WAIT_TO_COMPLETION, и это же значение будет возвращено функцией GetLastError, вызванной после выполнения возврата одной из функций ожидания.

В заключение отметим два момента:

1. При вызове любой из функций дежурного ожидания в качестве значения параметра интервала ожидания используйте INFINITE. В отсутствие возможности истечения интервала ожидания возврат из функций будет осуществляться лишь после того, как закончится выполнение всех процедур завершения или дескрипторы перейдут в сигнальное состояние.

2. Для передачи информации процедуре завершения общепринято использовать элемент данных hEvent структуры OVERLAPPED, поскольку это поле игнорируется ОС. 

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

Рис. 14.2. Асинхронный ввод/вывод с использованием процедур завершения

Программа 14.3 (atouEX) представляет собой переработанную версию программы 14.1. Эти программы иллюстрируют различие между двумя методами асинхронного ввода/вывода. Программа atouEx аналогична программе 14.1, но большая часть кода, предназначенного для упорядочения ресурсов, перемещена в ней в процедуру завершения, а многие переменные сделаны глобальными, чтобы процедура завершения могла иметь к ним доступ. Вместе с тем, в приложении В показано, что в отношении быстродействия программа atouEx вполне может конкурировать с другими методами, в которых не используется отображение файлов, тогда как программа atouOV работает медленнее. 

/* Глава 14. atouEX

   Преобразование файла из ASCII в Unicode средствами РАСШИРЕННОГО ВВОДА/ВЫВОДА. */

/* atouEX файл1 файл2 */

#include "EvryThng.h"

#define MAX_OVRLP 4

#define REC_SIZE 8096 /* Размер блока не имеет столь важного значения в отношении производительности, как в случае atouOV. */

#define UREC_SIZE 2 * REC_SIZE

static VOID WINAPI ReadDone(DWORD, DWORD, LPOVERLAPPED);

static VOID WINAPI WriteDone(DWORD, DWORD, LPOVERLAPPED);

/* Первая структура OVERLAPPED предназначена для чтения, а вторая — для записи. Структуры и буферы распределяются для каждой предстоящей операции. */

OVERLAPPED OverLapIn[MAX_OVRLP], OverLapOut [MAX_OVRLP];

CHAR AsRec[MAX_OVRLP][REC_SIZE];

WCHAR UnRec[MAX_OVRLP][REC_SIZE];

HANDLE hInputFile, hOutputFile;

LONGLONG nRecord, nDone;

LARGE_INTEGER FileSize;

int _tmain(int argc, LPTSTR argv[]) {

 DWORD ic;

 LARGE_INTEGER CurPosIn;

 hInputFile = CreateFile(argv[1], GENERIC_READ, 0, NULL, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);

 hOutputFile = CreateFile(argv[2], GENERIC_WRITE, 0, NULL, CREATE_ALWAYS, FILE_FLAG_OVERLAPPED, NULL);

 FileSize.LowPart = GetFileSize(hInputFile, &FileSize.HighPart);

 nRecord = FileSize.QuadPart / REC_SIZE;

 if ((FileSize.QuadPart % REC_SIZE) != 0) nRecord++;

 CurPosIn.QuadPart = 0;

 for (ic = 0; ic < MAX_OVRLP; ic++) {

  OverLapIn[ic].hEvent = (HANDLE)ic; /* Перегрузить событие. */

  OverLapOut[ic].hEvent = (HANDLE)ic; /* Поля. */

  OverLapIn[ic].Offset = CurPosIn.LowPart;

  OverLapIn[ic].OffsetHigh = CurPosIn.HighPart;

  if (CurPosIn.QuadPart < FileSize.QuadPart) ReadFileEx(hInputFile, AsRec[ic], REC_SIZE, &OverLapIn [ic], ReadDone);

  CurPosIn.QuadPart += (LONGLONG)REC_SIZE;

 }

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

 nDone = 0; 

 while (nDone < 2 * nRecord) SleepEx(INFINITE, TRUE);

 CloseHandle(hInputFile);

 CloseHandle(hOutputFile);

 _tprintf(_T("Преобразование из ASCII в Unicode завершено.\n"));

 return 0;

}

static VOID WINAPI ReadDone(DWORD Code, DWORD nBytes, LPOVERLAPPED pOv) {

 /* Чтение завершено. Преобразовать данные и инициировать запись. */

 LARGE_INTEGER CurPosIn, CurPosOut;

 DWORD ic, i;

 nDone++;

 /* Обработать запись и инициировать операцию записи. */

 ic = (DWORD)(pOv->hEvent);

 CurPosIn.LowPart = OverLapIn[ic].Offset;

 CurPosIn.HighPart = OverLapIn[ic].OffsetHigh;

 CurPosOut.QuadPart = (CurPosIn.QuadPart / REC_SIZE) * UREC_SIZE;

 OverLapOut[ic].Offset = CurPosOut.LowPart;

 OverLapOut[ic].OffsetHigh = CurPosOut.HighPart;

 /* Преобразовать запись из ASCII в Unicode. */

 for (i = 0; i < nBytes; i++) UnRec[ic][i] = AsRec[ic][i];

 WriteFileEx(hOutputFile, UnRec[ic], nBytes*2, &OverLapOut[ic], WriteDone);

 /* Подготовить структуру OVERLAPPED для следующего чтения. */

 CurPosIn.QuadPart += REC_SIZE * (LONGLONG)(MAX_OVRLP);

 OverLapIn[ic].Offset = CurPosIn.LowPart;

 OverLapIn[ic].OffsetHigh = CurPosIn.HighPart;

 return;

}

static VOID WINAPI WriteDone(DWORD Code, DWORD nBytes, LPOVERLAPPED pOv) {

 /* Запись завершена. Инициировать следующую операцию чтения. */

 LARGE_INTECER CurPosIn;

 DWORD ic;

 nDone++;

 ic = (DWORD)(pOv->hEvent);

 CurPosIn.LowPart = OverLapIn[ic].Offset;

 CurPosIn.HighPart = OverLapIn[ic].OffsetHigh;

 if (CurPosIn.QuadPart < FileSize.QuadPart) {

  ReadFileEx(hInputFile, AsRec[ic], REC_SIZE, &OverLapIn[ic], ReadDone);

 }

 return;

}

Перекрывающийся и расширенный ввод/вывод позволяют добиться асинхронного выполнения операций ввода/вывода в пределах единственного потока, хотя для поддержки этой функциональности ОС создает собственные потоки. В том или ином виде методы этого типа часто используются во многих ранних ОС для поддержки ограниченных форм выполнения асинхронных операций в однопоточных системах.

Однако Windows обеспечивает многопоточную поддержку, поэтому становится возможным достижение того же эффекта за счет выполнения синхронных операций ввода/вывода в нескольких, выполняемых независимо потоках. Ранее эти возможности уже были продемонстрированы на примере многопоточных серверов и программы grepMT (глава 7). Кроме того, потоки обеспечивают концептуально последовательный и, предположительно, гораздо более простой способ выполнения асинхронных операций ввода/вывода. В качестве альтернативы методам, используемым в программах 14.1 и 14.2, можно было бы предоставить каждому потоку собственный дескриптор файла, и тогда каждый из потоков мог бы обрабатывать в синхронном режиме каждую четвертую запись.

Такой способ использования потоков продемонстрирован в программе atouMT, которая в книге не приводится, но включена в материал, размещенный на Web-сайте. Программа atouMT не только способна выполняться под управлением любой версии Windows, но и более проста по сравнению с любым из двух вариантов программ асинхронного ввода/вывода, поскольку учет использования ресурсов в этом случае менее сложен. Каждый поток просто поддерживает собственные буферы в собственном стеке и выполняет в цикле последовательность синхронных операций чтения, преобразования и записи. При этом производительность программы остается на достаточно высоком уровне.

Примечание

В программе atouMT.с, которая находится на Web-сайте, содержатся комментарии по поводу нескольких возможных "ловушек", которые могут поджидать вас при организации доступа одновременно нескольких потоков к одному и тому же файлу. В частности, все отдельные дескрипторы файлов должны создаваться с помощью функции CreateHandle, а не функции DuplicateHandle.

Лично я предпочитаю использовать многопоточную обработку файлов, а не асинхронные операции ввода/вывода. Потоки легче программировать, и в большинстве случаев они обеспечивают более высокую производительность.

Существуют два исключения из этого общего правила. Первое из них, как было показано ранее в этой главе, касается ситуаций, в которых может быть только одна невыполненная операция, и в целях синхронизации можно использовать дескриптор файла. Второе, более важное исключение встречается в случае портов завершения асинхронного ввода/вывода, о чем будет говориться в конце настоящей главы. 

Windows NT поддерживает таймеры ожидания (waitable timers), являющихся одним из типов объектов ядра, осуществляющих ожидание.

Вы всегда можете создать собственный сигнал синхронизации, создав синхронизирующий поток, который устанавливает событие в результате пробуждения после вызова функции Sleep. В программе serverNP (программа 11.3) сервер также использует синхронизирующий поток для периодической широковещательной рассылки имени своего канала. Поэтому таймеры ожидания обеспечивают хотя и несколько избыточный, но удобный способ организации выполнения задач на периодической основе или в соответствии с определенным расписанием. В частности, таймер ожидания можно настроить таким образом, чтобы сигнал был сгенерирован в строго определенное время.

Таймер ожидания может быть либо синхронизирующим (synchronization timer), либо сбрасываемым вручную уведомляющим (manual-reset notification timer) таймером. Синхронизирующий таймер связывается с функцией косвенного вызова, аналогичной процедуре завершения расширенного ввода/вывода, тогда как для синхронизации по сбрасываемому вручную уведомляющему таймеру используется функция ожидания.

Для начала потребуется создать дескриптор таймера, используя для этого функцию CreateWaitableTimer. 

HANDLE CreateWaitableTimer(LPSECURITY_ATTRIBUTES lpTimerAttributes, BOOL bManualReset, LPCTSTR lpTimerName);

Второй параметр, bManualReset, определяет, таймер какого типа должен быть создан — синхронизирующий или уведомляющий. В программе 14.3 используется синхронизирующий таймер, но, изменив комментарии и настройку параметра, вы легко превратите его в уведомляющий таймер. Заметьте, что существует также функция OpenWaitableTimer, которая может использовать необязательное имя, предоставляемое третьим аргументом.

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

BOOL SetWaitableTimer(HANDLE hTimer, const LARGE_INTEGER *pDueTime, LONG IPeriod, PTIMERAPCROUTINE pfnCompletionRoutine, LPVOID lpArgToCompletionRoutine, BOOL fResume);

 

hTimer — действительный дескриптор таймера, созданного с использованием функции CreateWaitableTimer.

Второй параметр, на который указывает указатель pDueTime, может принимать либо положительные значения, соответствующие абсолютному времени, либо отрицательные, соответствующие относительному времени, причем фактические значения выражаются в единицах времени длительностью 100 наносекунд, а их формат описывается структурой FILETIME. Переменные типа FILETIME были введены в главе 3 и уже использовались нами в главе 6 в программе timep (программа 6.2).

Величина интервала между сигналами, указываемая в третьем параметре, выражается в миллисекундах. Если это значение установлено равным 0, то таймер переводится в сигнальное состояние только один раз. При положительных значениях этого параметра таймер является периодическим и срабатывает периодически до тех пор, пока его действие не будет прекращено вызовом функции CancelWaitableTimer. Отрицательные значения указанного интервала не допускаются.

Четвертый параметр, pfnCompletionRoutine, применяется в случае синхронизирующего таймера и указывает адрес процедуры завершения, которая вызывается при переходе таймера в сигнальное состояние и при условии, что поток переходит в состояние дежурного ожидания. При вызове этой процедуры в качестве одного из аргументов используется указатель, определяемый пятым параметром, plArgToComplretionRoutine.

Установив синхронизирующий таймер, вы можете перевести поток в состояние дежурного ожидания путем вызова функции SleepEx, чтобы обеспечить возможность вызова процедуры завершения. В случае сбрасываемого вручную уведомляющего таймера следует организовать ожидание перехода дескриптора таймера в сигнальное состояние. Дескриптор будет оставаться в сигнальном состоянии до следующего вызова функции SetWaitableTimer. Полная версия программы 14.3, находящаяся на Web-сайте, предоставляет вам возможность проводить собственные эксперименты, используя таймер выбранного типа в сочетании с процедурой завершения или ожиданием перехода дескриптора таймера в сигнальное состояние, что в итоге дает четыре различные комбинации.

Последний параметр, fResume, связан с режимами энергосбережения. Для получения более подробной информации по этому вопросу обратитесь к справочной документации.

Функция CancelWaitableTimer используется для отмены действия вызванной перед этим функции SetWaitableTimer, но при этом не изменяет сигнальное состояние таймера. Чтобы это сделать, необходимо в очередной раз вызвать функцию SetWaitableTimer.

В программе 14.3 демонстрируется применение таймера ожидания для генерации периодических сигналов.

/* Глава 14. TimeBeep.с. Периодическое звуковое оповещение.  */

/* Использование: TimeBeep период (в миллисекундах). */

#include "EvryThng.h"

static BOOL WINAPI Handler(DWORD CntrlEvent);

static VOID APIENTRY Beeper(LPVOID, DWORD, DWORD);

volatile static BOOL Exit = FALSE;

HANDLE hTimer;

int _tmain(int argc, LPTSTR argv[]) {

 DWORD Count = 0, Period;

 LARGE_INTEGER DueTime;

 /* Перехват нажатия комбинации клавиш <Ctrl-c> для прекращения операции. См. главу 4. */

 SetConsoleCtrlHandler(Handler, TRUE);

 Period = _ttoi(argv[1]) * 1000;

 DueTime.QuadPart = –(LONGLONG)Period * 10000;

 /* Параметр DueTime отрицателен для первого периода ожидания и задается относительно текущего времени. Период ожидания измеряется в мс (10^-3 с), a DueTime — в единицах по 100 нc (10^-7 с) для согласования с типом FILETIME. */

 hTimer = CreateWaitableTimer(NULL, FALSE /* "Таймер синхронизации" */, NULL);

 SetWaitableTimer(hTimer, &DueTime, Period, Beeper, &Count, TRUE);

 while (!Exit) {

  _tprintf(_T("Count = %d\n"), Count);

  /* Значение счетчика увеличивается в процедуре таймера. */

  /* Войти в состояние дежурного ожидания. */

  SleepEx(INFINITE, TRUE);

 }

 _tprintf(_T("Завершение. Счетчик = %d"), Count);

 CancelWaitableTimer(hTimer);

 CloseHandle(hTimer);

 return 0;

}

static VOID APIENTRY Beeper(LPVOID lpCount, DWORD dwTimerLowValue, DWORD dwTimerHighValue) {

 *(LPDWORD)lpCount = *(LPDWORD)lpCount + 1; 

 _tprintf(_T("Генерация сигнала номер: %d\n"), *(LPDWORD) lpCount);

 Веер(1000 /* Частота. */, 250 /* Длительность (мс). */);

 return;

}

BOOL WINAPI Handler(DWORD CntrlEvent) {

 Exit = TRUE;

 _tprintf(_T("Завершение работы\n"));

 return TRUE;

}

Исходя из типа таймера и используя либо процедуру завершения, либо ожидание перехода дескриптора в сигнальное состояние, можно образовать четыре различных комбинации. Программа 14.3 иллюстрирует использование процедуры завершения и синхронизирующего таймера. Вы сможете тестировать каждую из четырех возможных комбинаций, изменяя комментарии в версии программы TimeBeep.с, доступной на Web-сайте.

Порты завершения ввода/вывода, поддерживаемые лишь на NT-платформах, объединяют в себе возможности перекрывающегося ввода/вывода и независимых потоков и используются чаще всего в серверных программах. Чтобы выяснить, какими требованиями это может диктоваться, обратимся к серверам, построенным в главах 11 и 12, где каждый клиент поддерживался отдельным рабочим потоком, связанным с сокетом или экземпляром именованного канала. Это решение хорошо работает лишь в тех случаях, когда число клиентов невелико.

Посмотрим, однако, что произойдет, если число клиентов достигнет 1000. В имеющейся модели для этого потребуется 1000 потоков, для каждого из которых необходимо выделить значительный объем виртуальной памяти. Так, по умолчанию каждому потоку выделяется 1 Мбайт стекового пространства, так что для 1000 потоков потребуется 1 Гбайт, и переключение контекстов потоков может увеличить задержки, обусловленные ошибками из-за отсутствия страниц.[35] Кроме того, потоки будут состязаться между собой за право владения общими ресурсами как на уровне планировщика, так и внутри процесса, и это, как было показано в главе 9, может приводить к снижению производительности. В связи с этим требуется механизм, позволяющий небольшому пулу рабочих потоков обслуживать большое количество клиентов.

Искомое решение обеспечивается портами завершения ввода/вывода, которые предоставляют возможность создавать ограниченное количество серверных потоков в пуле потоков, имея очень большое количество дескрипторов именованных каналов (или сокетов). При этом дескрипторы не соединяются попарно с отдельными рабочими серверными потоками; серверный поток может обслуживать любой дескриптор, данные которого нуждаются в обработке.

Итак, порт завершения ввода/вывода — это набор перекрывающихся дескрипторов, и потоки ожидают перехода порта в сигнальное состояние. Когда завершается операция чтения или записи с участием какого-либо дескриптора, один из потоков пробуждается и принимает данные и результаты выполнения операции ввода/вывода. Далее поток может обработать данные и вновь перейти в состояние ожидания перехода порта в сигнальное состояние.

Прежде всего необходимо создать порт завершения ввода/вывода и присоединить к нему перекрывающиеся дескрипторы.

Для создания порта и присоединения к нему дескрипторов используется одна и та же функция — CreateCompletionPort. Необходимость выполнения этой функцией двух разных задач соответственно усложняет использование ее параметров. 

HANDLE CreateIoCompletionPort(HANDLE FileHandle, HANDLE ExistingCompletionPort, DWORD CompletionKey, DWORD NumberOfConcurrentThreads);

Порт завершения ввода/вывода представляет собой совокупность дескрипторов файлов, открытых в режиме OVERLAPPED. Параметр FileHandle — это перекрывающийся дескриптор, присоединяемый к порту. Если задать его значение равным INVALID_DESCRIPTOR_HANDLE, то функция создаст новый порт завершения ввода/вывода и возвратит его дескриптор. В этом случае следующий параметр, ExistingCompletionPort, должен быть установлен в NULL.

ExistingCompletionPort — порт, созданный при первом вызове функции, к которому должен быть присоединен дескриптор, указанный в первом параметре. В случае успешного выполнения функция возвращает дескриптор порта, иначе — NULL.

CompletionKey — указывает ключ, который будет включен в пакет завершения для дескриптора FileHandle. Обычно в качестве ключа используется значение индекса массива структур данных, содержащих тип операции, дескриптор и указатель на буфер данных. 

NumberOfConcurrentThreads — предельно допустимое количество потоков, которым разрешено параллельное выполнение. При наличии других потоков сверх этого количества, ожидающих перехода порта в сигнальное состояние, они будут оставаться блокированными, даже если существует дескриптор с доступными данными. Если этот параметр установлен равным 0, то в качестве предела используется количество процессоров, установленных в системе.

Количество перекрывающихся дескрипторов, которые могут быть связаны с одним портом завершения ввода/вывода, ничем не ограничивается. Первоначальный вызов функции CreateCompletionPort используется для создания порта и указания максимального количества потоков. Для каждого очередного перекрывающегося дескриптора, подлежащего связыванию с данным портом, следует повторно вызывать эту же функцию. К сожалению, способов, позволяющих удалить дескриптор из порта завершения, не существует, и это упущение значительно ограничивает гибкость программ.

Дескрипторы, связанные с портом не должны использоваться совместно с функциями ReadFileEx и WriteFileEx. В документации Microsoft не рекомендуется разделять файлы и объекты иного типа, используя другие открытые дескрипторы.

Для выполнения ввода/вывода с участием дескрипторов, связанных с портом, используются функции ReadFile и WriteFile со структурами OVERLAPPED (дескрипторы событий не требуются). Далее операция ввода/вывода помещается в очередь порта завершения.

Поток ожидает завершения перекрывающейся операции ввода/вывода, находящейся в очереди, не путем ожидания события, а путем вызова функции GetQueueCompletionStatus с указанием порта завершения (completion port). Когда вызывающий поток пробуждается, функция возвращает ключ, который был указан при первоначальном присоединении к порту дескриптора, чья операция завершилась, и этот ключ может указывать количество переданных байтов и идентификационные данные фактического дескриптора, связанного с завершившейся операцией.

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

Имеется также возможность использовать конечный интервал ожидания (time-out). 

BOOL GetQueuedCompletionStatus(HANDLE CompletionPort, LPDWORD lpNumberOfBytesTransferred, LPDWORD lpCompletionKey, LPOVERLAPPED *lpOverlapped, DWORD dwMilliseconds);

 

Иногда может оказаться удобным, чтобы операция не помещалась в очередь порта завершения ввода/вывода. В этом случае поток может ожидать наступления перекрывающегося события, как показано в программе 14.4 и дополнительном примере, atouMTCP, который находится на Web-сайте. Для указания того, что перекрывающаяся операция не должна помещаться в очередь порта завершения, вы должны установить младший бит дескриптора события (hEvent) в структуре OVERLAPPED; тогда вы получите возможность ожидать наступления события для данной конкретной операции. Такое решение является довольно странным, однако оно документировано, хотя особо и не подчеркивается.

Поток может отправить в порт событие завершения вместе с ключом, чтобы завершить остающийся невыполненным вызов функции GetQueueCompletionStatus. Вся необходимая для этого информация предоставляется функцией PostQueueCompletionStatus. 

BOOL PostQueuedCompletionStatus(HANDLE CompletionPort, DWORD dwNumberOfBytesTransferred, DWORD dwCompletionKey, LPOVERLAPPED lpOverlapped);

 

Для пробуждения ожидающих потоков даже в условиях отсутствия завершившихся операций иногда используют метод, суть которого заключается в предоставлении фиктивного значения ключа, например, –1. Ожидающие потоки должны проверять значения ключей, и эта методика может использоваться, например, для того, чтобы сигнализировать потоку о необходимости завершить работу.

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

Эту же методику мы могли бы применить и в серверах serverSK (программа 12.2) и serverNP (программа 11.3). Все, что для этого требуется — это организовать ожидание перехода семафора в сигнальное состояние после завершения запроса на чтение, выполнение этого запроса, создание ответа и освобождение семафора перед тем, как записать ответ. Такое решение гораздо проще того, которое реализовано в примере с портом завершения ввода/вывода, приведенном в следующем разделе. Единственная проблема состоит в том, что потоков может оказаться очень много, и для каждой из них требуется собственное стековое пространство, что приведет к большому расходу виртуальной памяти. Остроту этой проблемы можно несколько ослабить, тщательно распределяя необходимые объемы стекового пространства. Упражнение 14.6 включает в себя выполнение экспериментов с альтернативным решением подобного рода, а реализация соответствующего примера находится на Web-сайте.

Существует еще одна возможность, которую можно использовать при создании масштабируемых серверов. Выборка пакетов рабочих заготовок (work items) из очереди (см. главу 10) может осуществляться с использованием ограниченного количества потоков. Поступающие рабочие заготовки могут помещаться в очередь одной или несколькими главными потоками, как показано в программе 10.5.

Программа 14.4 представляет видоизмененный вариант программы serverNP (программа 11.3), в котором используются порты завершения ввода/вывода. Этот сервер создает небольшой пул серверных потоков и больший пул дескрипторов перекрывающихся каналов, а также ключей завершения, по одному для каждого дескриптора. Перекрывающиеся дескрипторы присоединяются к порту завершения, а затем вызывается функция ConnectNamedPipe. Серверные потоки ожидают сигналов завершения, связанных как с подключениями клиентов, так и с операциями чтения. Когда регистрируется операция чтения, обрабатывается соответствующий клиентский запрос, и результаты возвращаются без использования порта завершения. Вместо этого серверный поток ожидает наступления события после выполнения операции записи, причем младший бит дескриптора события в структуре OVERLAPPED устанавливается в 1.

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

Поскольку с большей частью кода вы уже знакомы по предыдущим примерам, она здесь не приводится.

/* Глава 14. ServerCP. Многопоточный сервер.

   Версия на основе именованного канала, пример ПОРТА ЗАВЕРШЕНИЯ.

   Использование: Server [ИмяПользователя ИмяГруппы]. */

#include "EvryThng.h"

#include "ClntSrvr.h"

/* Здесь определяются сообщения запроса и ответа. */

typedef struct { /*Структуры, на которые указывают ключи портов завершения*/

 HANDLE hNp; /* и которые представляют еще не выполненные операции */

 REQUEST Req; /* ReadFile и ConnectNamedPipe. */

 DWORD Type; /* 0 – ConnectNamedPipe; 1 – ReadFile. */

 OVERLAPPED Ov;

} CP_KEY;

static CP_KEY Key[MAX_CLIENTS_CP]; /* Доступно всем потокам. */

/* … */

_tmain(int argc, LPTSTR argv[]) {

 HANDLE hCp, hMonitor, hSrvrThread[MAXCLIENTS];

 DWORD iNp, iTh, MonitorId, ThreadId;

 THREAD_ARG ThArgs[MAX_SERVER_TH];

 /*…*/

 hCp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, MAX_SERVER_TH);

 /* Создать перекрывающийся именованный канал для каждого потенциального */

 /* клиента, добавить порт завершения и ожидать соединения. */

 /* Предполагается, что максимальное количество клиентов намного */

 /* превышает количество серверных потоков. */

 for (iNp = 0; iNp < MAX_CLIENTS_CP; iNp++) {

  memset(&Key[iNp], 0, sizeof(CP_KEY));

  Key[iNp].hNp = CreateNamedPipe(SERVER_PIPE, PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED, PIPE_READMODE_MESSAGE | PIPE_TYPE_MESSAGE | PIPE_WAIT, MAX_CLIENTS_CP, 0, 0, INFINITE, pNPSA);

  CreateIoCompletionPort(Key[iNp].hNp, hCp, iNp, MAX_SERVER_TH + 2);

  Key[iNp].Ov.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);

  ConnectNamedPipe(Key[iNp].hNp, &Key[iNp].Ov);

 }

 /* Создать рабочие серверные потоки и имя временного файла для каждой из них.*/

 for (iTh = 0; iTh < MAX_SERVER_TH; iTh++) {

  ThArgs[iTh].hCompPort = hCp;

  ThArgs[iTh].ThreadNo = iTh;

  GetTempFileName(_T("."), _T("CLP"), 0, ThArgs[iTh].TmpFileName); 

  hSrvrThread[iTh] = (HANDLE)_beginthreadex (NULL, 0, Server, &ThArgs[iTh], 0, &ThreadId);

 }

 /* Дождаться завершения всех потоков и "убрать мусор". */

 /* … */

 return 0;

}

static DWORD WINAPI Server(LPTHREAD_ARG pThArg)

/* Функция потока сервера.

   Имеется по одному потоку для каждого потенциального клиента. */

{

 HANDLE hCp, hTmpFile = INVALID_HANDLE_VALUE;

 HANDLE hWrEvent = CreateEvent(NULL, TRUE, FALSE, NULL);

 DWORD nXfer, KeyIndex, ServerNumber;

 /* … */

 BOOL Success, Disconnect, Exit = FALSE;

 LPOVERLAPPED pOv;

 OVERLAPPED ovResp = {0, 0, 0, 0, hWrEvent}; /*Для ответных сообщений.*/

 /* Чтобы избежать помещения перекрывающейся операции в очередь порта завершения, должен быть установлен младший бит события. Несмотря на всю странность этого способа, он документирован. */

 ovResp.hEvent = (HANDLE)((DWORD)hWrEvent | 0x1);

 GetStartupInfo(&StartInfoCh);

 hCp = pThArg->hCompPort;

 ServerNumber = pThArg->ThreadNo;

 while(!ShutDown && !Exit) __try {

  Success = FALSE; /* Устанавливается только в случае успешного завершения всех операций. */

  Disconnect = FALSE;

  GetQueuedCompletionStatus(hCp, &nXfer, &KeyIndex, &pOv, INFINITE);

  if (Key [KeyIndex].Type == 0) { /* Соединение установлено. */

   /* Открыть временный файл с результатами для этого соединения. */

   hTmpFile = CreateFile(pThArg->TmpFileName, /* … */);

   Key[KeyIndex].Type = 1;

   Disconnect = !ReadFile(Key[KeyIndex].hNp, &Key[KeyIndex].Req, RQ_SIZE, &nXfer, &Key[KeyIndex].Ov) && GetLastError () == ERROR_HANDLE_EOF; /* Первая операция чтения. */

   if (Disconnect) continue;

   Success = TRUE;

  } else {

   /* Чтение завершилось. Обработать запрос. */

   ShutDown = ShutDown || (_tcscmp (Key[KeyIndex].Req.Record, ShutRqst) == 0);

   if (ShutDown) continue;

   /* Создать процесс для выполнения команды. */

   /* … */ 

   /* Отвечать по одной строке за один раз. На данном этапе удобно использовать функции библиотеки С для работы со строками. */

   fp = _tfopen(pThArg->TmpFileName, _T("r"));

   Response.Status = 0;

   /* Поскольку младший бит события установлен, ответные сообщения в очередь порта завершения не помещаются. */

   while(_fgetts(Response.Record, MAX_RQRS_LEN, fp) != NULL) {

    WriteFile(Key [KeyIndex].hNp, &Response, RS_SIZE, &nXfer, &ovResp);

    WaitForSingleObject(hWrEvent, INFINITE);

   }

   fclose(fp);

   /* Уничтожить содержимое временного файла. */

   SetFilePointer(hTmpFile, 0, NULL, FILE_BEGIN);

   SetEndOfFile(hTmpFile);

   /* Отправить признак конца ответа. */

   Response.Status = 1;

   strcpy(Response.Record, "");

   WriteFile(Key[KeyIndex].hNp, &Response, RS_SIZE, &nXfer, &ovResp);

   WaitForSingleObject(hWrEvent, INFINITE);

   /* Конец основного командного цикла. Получить следующую команду.*/

   Disconnect = !ReadFile(Key[KeyIndex].hNp, &Key[KeyIndex].Req, RQ_SIZE, &nXfer, &Key[KeyIndex].Ov) && GetLastError() == ERROR_HANDLE_EOF; /* Следующее чтение */

   if (Disconnect) continue;

   Success = TRUE;

  }

 } __finally {

  if (Disconnect) {

   /* Создать еще одно соединение по этому каналу. */

   Key[KeyIndex].Type = 0;

   DisconnectNamedPipe(Key[KeyIndex].hNp);

   ConnectNamedPipe(Key[KeyIndex].hNp, &Key[KeyIndex].Ov);

  }

  if (!Success) {

   ReportError(_T("Ошибка сервера"), 0, TRUE);

   Exit = TRUE;

  }

 }

 FlushFileBuffers(Key[KeyIndex].hNp);

 DisconnectNamedPipe(Key[KeyIndex].hNp);

 CloseHandle(hTmpFile);

 /* … */

 _endthreadex(0);

 return 0;

 /* Подавление предупреждающих сообщений компилятора. */

}

Для выполнения асинхронных операций ввода/вывода в Windows предусмотрены три метода. Самой распространенной и наиболее простой является методика, основанная на использовании потоков, которая, в отличие от двух остальных, способна работать даже под управлением Windows 9x. Каждый из потоков отвечает за выполнение определенной последовательности действий, состоящей из одной или нескольких последовательно выполняющихся, блокирующихся операций ввода/вывода. Кроме того, каждый поток должен располагать собственным дескриптором файла или канала.

Перекрывающийся ввод/вывод обеспечивает возможность выполнения асинхронных операций одним потоком с использованием одного дескриптора файла, но каждой отдельной операции вместо пары "поток—дескриптор файла" должен предоставляться дескриптор события. При этом требуется организовать ожидание завершения выполнения каждой конкретной операции ввода/вывода по отдельности, а затем очищать системные ресурсы или выполнять любые другие действия, необходимые для управления последовательностью выполнения операций.

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

Одним неоспоримым преимуществом перекрывающегося ввода/вывода является то, что он предоставляет возможность создания портов завершения ввода/вывода, однако, о чем ранее уже говорилось и что иллюстрируется программой atouMTCP, которая находится на Web-сайте, но ценность и этого преимущества несколько снижается из-за того, что для ограничения количества активных потоков в пуле рабочих потоков могут быть использованы семафоры. Дополнительным недостатком портов завершения является то, что они не допускают удаления присоединенных к ним дескрипторов. 

UNIX обеспечивает поддержку потоков средствами Pthreads, что ранее уже обсуждалось.

В System V UNIX асинхронный ввод/вывод ограничивается потоками и не может использоваться для выполнения операций с файлами и каналами.

В версии BSD 4.3 для указания события, связанного с дескриптором файла, и выбора функции с целью определения состояния готовности дескрипторов файлов используется комбинация сигналов (SIGIO). Для дескрипторов файлов должен устанавливаться режим O_ASYNC. Такой подход может использоваться только с терминалами и в сетевых коммуникациях.

Глава 15 завершает наше обсуждение Windows API демонстрацией методов обеспечения безопасности объектов Windows. Основное внимание уделяется защите файлов, но те же самые методы можно применять и к другим объектам, например, именованным каналам или процессам. 

14.1. Воспользуйтесь асинхронным вводом/выводом для слияния нескольких отсортированных файлов в один отсортированный файл большего размера.

14.2. Приводит ли использование флага FILE_FLAG_MO_BUFFERING к повышению производительности программ atouOV и atouEX, как того можно было бы ожидать в соответствии с утверждениями, содержащимися в описании функции CreateFile? Существуют ли какие-либо ограничения, касающиеся размера файлов?

14.3. Модифицируйте программу timebeep (программа 14.3), введя в нее сбрасываемый вручную уведомляющий таймер.

14.4. Модифицируйте клиент именованного канала в программе clientNP (программа 11.2), введя в него перекрывающийся ввод/вывод, чтобы клиент мог продолжать работу после отправки запроса. В результате этого один клиент сможет иметь нескольких невыполненных запросов.

14.5. Перепишите программу serversk (программа 12.2), представляющую сервер на базе сокетов, введя в нее порты завершения ввода/вывода.

14.6. Перепишите одну из программ serverSK или serverNP таким образом, чтобы количество готовых к выполнению рабочих потоков ограничивалось семафором. Выполните эксперименты с большим пулом потоков, чтобы выяснить, насколько эффективен такой альтернативный вариант. Находящаяся на Web-сайте программа serverSM является модифицированным вариантом программы serverNP. С увеличением объемов доступной физической памяти и распространением платформы Win64 относительная ценность этого подхода и портов завершения может варьироваться.

14.7. Используйте программу управления заданиями JobShell (программа 6.3) для работы с большим количеством клиентов и исследуйте сравнительную способность к реагированию серверов serverNP и serverCP. Дополнительную нагрузку могут составить сетевые клиенты. Определите оптимальный интервал значений для количества активных потоков. 

Примечания:

3

Замечания, сделанные в адрес UNIX, в равной степени относятся также к Linux и некоторым другим системам, поддерживающим POSIX API.

35

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