45817 (Эффективная многопоточность), страница 3
Описание файла
Документ из архива "Эффективная многопоточность", который расположен в категории "". Всё это находится в предмете "информатика" из , которые можно найти в файловом архиве . Не смотря на прямую связь этого архива с , его также можно найти и в других разделах. Архив можно найти в разделе "рефераты, доклады и презентации", в предмете "информатика, программирование" в общих файлах.
Онлайн просмотр документа "45817"
Текст 3 страницы из документа "45817"
Открывается порт с помощью функции NtOpenIoCompletion. При вызове функции нужно указать имя порта и уровень доступа. В качестве уровня доступа можно указывать все стандартные и следующие специальные права [2] (таблица 2).
| Символическое обозначение | Константа | Описание |
| IO_COMPLETION_QUERY_STATE | 1 | Необходим для запроса состояния объекта "порт" |
| IO_COMPLETION_MODIFY_STATE | 2 | Необходим для изменения состояния объекта "порт" |
Таблица 2.
У порта можно запрашивать количество необработанных запросов с помощью функции NtQueryIoCompletion. Хотя в [3] утверждается, что эта функция определяет, находится ли порт в сигнальном состоянии, на самом деле она возвращает количество клиентских запросов в очереди. Это довольно важная информация, которую почему-то опять решили от нас скрыть.
Давайте более детально рассмотрим, как создается и функционирует порт завершения ввода/вывода [4].
При создании порта функцией CreateIoCompletionPort вызывается внутренний сервис NtCreateIoCompletion. Объект "порт" представлен следующей структурой [5]:
| typedef stuct _IO_COMPLETION { KQUEUE Queue; } IO_COMPLETION; |
То есть, по существу, объект "порт завершения" является объектом "очередь исполнительной системы" (KQUEUE). Вот как представлена очередь:
| typedef stuct _KQUEUE { DISPATCHER_HEADER Header; LIST_ENTRY EnrtyListHead; //очередь пакетов DWORD CurrentCount; DWORD MaximumCount; LIST_ENTRY ThreadListHead; //очередь ожидающих потоков } KQUEUE; |
Итак, для порта выделяется память, и затем происходит его инициализация с помощью функции KeInitializeQueue. (все, что касается такого супернизкого устройства порта, взято из [4], остальное – из DDK и [3]).
Когда происходит связывание порта с объектом "файл", Win32-функция CreateIoCompletionPort вызывает NtSetInformationFile. Класс информации для этой функции устанавливается как FileCompletionInformation, а в качестве параметра FileInformation передается указатель на структуру IO_COMPLETION_CONTEXT [5] или FILE_COMPLETION_INFORMATION [3].
| typedef struct _IO_COMPLETION_CONTEXT { PVOID Port; PVOID Key; } IO_COMPLETION_CONTEXT; typedef struct _FILE_COMPLETION_INFORMATION { HANDLE IoCompletionHandle; ULONG CompletionKey; } FILE_COMPLETION_INFORMATION, *PFILE_COMPLETION_INFORMATION; |
Указатель на эту структуру заносится в поле CompletionConext структуры FILE_OBJECT (смещение 0x6C).
После завершения асинхронной операции ввода/вывода для ассоциированного файла диспетчер ввода/вывода проверяет поле CompletionConext и, если оно не равно 0, создает пакет запроса (из структуры OVERLAPPED и ключа завершения) и помещает его в очередь с помощью вызова KeInsertQueue. Когда поток вызывает функцию GetQueuedCompletionStatus, на самом деле вызывается функция NtRemoveIoCompletion. NtRemoveIoCompletion проверяет параметры и вызывает функцию KeRemoveQueue, которая блокирует поток, если в очереди отсутствуют запросы, или поле CurrentCount структуры KQUEUE больше или равно MaximumCount. Если запросы есть, и число активных потоков меньше максимального, KeRemoveQueue удаляет вызвавший ее поток из очереди ожидающих потоков и увеличивает число активных потоков на 1. При занесении потока в очередь ожидающих потоков поле Queue структуры KTHREAD (смещение 0xE0) устанавливается равным адресу очереди (порта завершения). Зачем это нужно? Когда вызываются функции блокировки потока (WaitForSingleObject и др.), планировщик проверяет это поле, и если оно не равно 0, вызывает функцию KeActivateWaiterQueue, которая уменьшает число активных потоков порта на 1. Когда поток пробуждается после вызова блокирующих функций, планировщик выполняет те же действия, только вызывает при этом функцию KeUnwaitThread, которая увеличивает счетчик активных потоков на 1.
Когда вы помещаете запрос в порт завершения функцией PostQueuedCompletionStatus, на самом деле вызывается функция NtSetIoCompletion, которая после проверки параметров и преобразования хендла порта в указатель, вызывает KeInsertQueue.
Организуем пул
Итак, мы знаем, как работает порт завершения ввода/вывода, когда потоки добавляются в пул и когда удаляются. Но сколько потоков должно быть в пуле? В два раза больше, чем число процессоров. Это очень общая рекомендация, и для некоторых задач она не подходит. По большому счету имеется только два критерия, по которым можно определять, нужно создавать новый поток или нет. Эти критерии – загруженность процессора и число пакетов запросов. Если число пакетов превышает определенное количество, и загруженность процессора невысока, есть смысл создать новый поток. Если пакетов мало, или процессор занят более чем на 90 процентов, дополнительный поток создавать не следует. Удалять поток из пула нужно, если он давно не обрабатывал клиентские запросы (просто подсчитать, сколько раз GetQueuedCompletionStatus вернула управление по таймауту). При удалении потока нужно следить, чтобы закончились все асинхронные операции ввода/вывода, начатые этим потоком.
Надо сказать, что определение загруженности процессора, количества пакетов в очереди порта и наличия у потока незавершенных операций ввода/вывода – задачи не самые простые. Например, вы можете использовать WMI для определения загруженности процессора, но при этом не сможете определить, есть ли у потока незавершенные операции ввода/вывода. Ниже я приведу функции получения вышеперечисленных показателей только недокументированными способами (здесь используется заголовочный файл ntdll.h из [3]):
| // Функция получения загруженности процессора double GetCPUUsage() { #define Li2Double(x) ((double)((x).HighPart) * 4.294967296E9 \ + (double)((x).LowPart)) typedef NTSTATUS (NTAPI ZwQuerySystemInformation_t)( IN NT::SYSTEM_INFORMATION_CLASS SystemInformationClass, OUT PVOID SystemInformation, IN ULONG SystemInformationLength, OUT PULONG ReturnLength OPTIONAL ); static ZwQuerySystemInformation_t* ZwQuerySystemInformation = 0; if(!ZwQuerySystemInformation) { ZwQuerySystemInformation = (ZwQuerySystemInformation_t*)GetProcAddress( GetModuleHandle(_T("ntdll.dll")), _T("NtQuerySystemInformation")); } double dbIdleTime = 0; static NT::LARGE_INTEGER liOldIdleTime = {0, 0}; static NT::LARGE_INTEGER liOldSystemTime = {0, 0}; // Получаем число процессоров NT::SYSTEM_BASIC_INFORMATION sysinfo = {0}; NT::NTSTATUS status = ZwQuerySystemInformation(NT::SystemBasicInformation, &sysinfo, sizeof sysinfo, 0); if(status != NO_ERROR) return -1; // Получаем системное время NT::SYSTEM_TIME_OF_DAY_INFORMATION timeinfo = {0}; status = ZwQuerySystemInformation(NT::SystemTimeOfDayInformation, &timeinfo, sizeof timeinfo, 0); if(status!=NO_ERROR) return -1; // Получаем время простоя NT::SYSTEM_PERFORMANCE_INFORMATION perfinfo = {0}; status = ZwQuerySystemInformation(NT::SystemPerformanceInformation, &perfinfo, sizeof perfinfo, 0); if(status != NO_ERROR) return -1; // если это первый вызов, значение вычислить нельзя if(liOldIdleTime.QuadPart != 0) { // Время простоя dbIdleTime = Li2Double(perfinfo.IdleTime) - Li2Double(liOldIdleTime); // Системное время const double dbSystemTime = Li2Double(timeinfo.CurrentTime) - Li2Double(liOldSystemTime); dbIdleTime = dbIdleTime / dbSystemTime; dbIdleTime = 100.0 - dbIdleTime * 100.0 / (double)sysinfo.NumberProcessors + 0.5; } // сохраняем полученные значения liOldIdleTime = perfinfo.IdleTime; liOldSystemTime = timeinfo.CurrentTime; // Если это первый вызов, получаем загруженность CPU за последние // 200 милисекунд if(dbIdleTime == 0) { Sleep(200); dbIdleTime = GetCPUUsage(); } return dbIdleTime; } // Возвращает true, если поток имеет незавершенные операции ввода/вывода bool HasThreadIoPending(HANDLE hThread = GetCurrentThread()) { typedef NTSTATUS (NTAPI ZwQueryInformationThread_t)( IN HANDLE ThreadHandle, IN NT::THREADINFOCLASS ThreadInformationClass, OUT PVOID ThreadInformation, IN ULONG ThreadInformationLength, OUT PULONG ReturnLength OPTIONAL ); static ZwQueryInformationThread_t* ZwQueryInformationThread = 0; if(!ZwQueryInformationThread) { ZwQueryInformationThread = (ZwQueryInformationThread_t*)GetProcAddress( GetModuleHandle(_T("ntdll.dll")), _T("NtQueryInformationThread")); } ULONG io = 0; ZwQueryInformationThread(hThread, NT::ThreadIsIoPending, &io, 4, 0); return io > 0; } // Возвращает количество необработанных запросов в очереди порта DWORD GetIoCompletionLen(HANDLE hIoPort) { typedef NTSTATUS (NTAPI ZwQueryIoCompletion_t)( IN HANDLE IoCompletionHandle, IN NT::IO_COMPLETION_INFORMATION_CLASS IoCompletionInformationClass, OUT PVOID IoCompletionInformation, IN ULONG IoCompletionInformationLength, OUT PULONG ResultLength OPTIONAL ); static ZwQueryIoCompletion_t* ZwQueryIoCompletion = 0; if(!ZwQueryIoCompletion) { ZwQueryIoCompletion = (ZwQueryIoCompletion_t*)GetProcAddress( GetModuleHandle(_T("ntdll.dll")), _T("NtQueryIoCompletion")); } NT::IO_COMPLETION_BASIC_INFORMATION ioinfo = {0}; DWORD dwRetLen = 0; ZwQueryIoCompletion(hIoPort, NT::IoCompletionBasicInformation, &ioinfo, sizeof ioinfo, &dwRetLen); return ioinfo.SignalState; } |
Как видите, не простое это дело – создавать эффективный пул потоков, однако кое-что ребята из Microsoft могут нам предложить. В Windows 2000 появились новые функции, которые полностью берут на себя всю черновую работу по созданию и удалению потоков в пуле. О них – следующий раздел.
Встроенная поддержка пула потоков
В Windows 2000 появились новые функции, которые условно можно разделить на четыре группы:
помещение запроса в очередь;
вызов функции при окончании асинхронной операции ввода/вывода;
периодический вызов функции;
вызов функции при переходе объекта в сигнальное состояние.
Рассмотрим их по порядку.
Помещение запроса в очередь
Передать на выполнение потоку из пула какую-либо функцию можно с помощью сервиса QueueUserWorkItem. Эта с виду простая функция делает очень много: она создает порт завершения ввода/вывода, создает и уничтожает потоки в пуле и многое другое. Вот ее описание:
| BOOL QueueUserWorkItem( LPTHREAD_START_ROUTINE Function, // адрес функции PVOID Context, // произвольный параметр ULONG Flags // флаги выполнения ); |
QueueUserWorkItem помещает пакет запроса в виде адреса функции и произвольного параметра в очередь запросов порта завершения и сразу же возвращает управление. Вот как выглядит функция, которая будет вызвана одним из потоков в пуле:
| DWORD WINAPI ThreadProc( LPVOID lpParameter // произвольный параметр ); |
Ее прототип ничем не отличается от стартовой процедуры потока, так что здесь вам все должно быть ясно. Гораздо интереснее знать, что скрывается внутри функции QueueUserWorkItem. Давайте разбираться.
При первом помещении запроса количество потоков в пуле равно нулю, так что QueueUserWorkItem приходится создавать поток и порт завершения. Затем в порт помещается пакет запроса, а поток вызывает функцию GetQueuedCompletionStatus. После обработки запроса поток не разрушается, а остается еще некоторое время в пуле, так что следующий запрос обработается намного быстрее. Если вы отправляете запросы слишком часто, и количество необработанных пакетов увеличивается, QueueUserWorkItem создаст для вызова функции новый поток. Максимальное количество потоков в пуле равно количеству процессоров, что не очень хорошо, но есть способ заставить функцию всегда создавать новый поток.
| ПРИМЕЧАНИЕ Те из вас, кто читал статью Дж. Рихтера «New Windows 2000 Pooling Functions Greatly Simplify Thread Management» из апрельского MSJ за 1999 год, могут поспорить со мной насчет размера пула. В статье указывается, что количество потоков в нем равно удвоенному количеству процессоров в системе, однако это не так. Вы можете собственноручно в этом убедиться, поставив breakpoint на функцию _RtlpInitializeWorkerThreadPool (адрес 0x77FA95CD на Windows 2000 Professional SP3) и вызвав функцию QueueUserWorkItem. |
Рассмотрим флаги функции QueueUserWorkItem.
| Константа | Значение | Описание |
| WT_EXECUTEDEFAULT | 0 | Запрос помещается в простой рабочий поток |
| WT_EXECUTEINIOTHREAD | 1 | Запрос помещается в поток ввода/вывода |
| WT_EXECUTEINPERSISTENTTHREAD | 0x80 | Запрос помещается в поток, который не завершается после обработки запроса, поэтому он может сохранять свое состояние, например в TLS. |
| WT_EXECUTELONGFUNCTION | 0x10 | Запрос с данным флагом всегда помещается в новый поток |
Таблица 3. Флаги функции QueueUserWorkItem.
Если вы не выполняете асинхронных запросов ввода/вывода в функции ThreadProc, не используете TLS (Thread Local Storage) или функций, которые его используют, а продолжительность выполнения операции невелика – указывайте флаг WT_EXECUTEDEFAULT.
Предположим, вы начали асинхронную операцию ввода/вывода в своей функции ThreadProc. Для того чтобы она завершилась, поток в котором она началась, не должен быть разрушен. Однако флаг WT_EXECUTEDEFAULT этого не гарантирует. С этим флагом поток может быть удален, даже если у него имеются незавершенные асинхронные операции. Для того чтобы поток завершался только после окончания всех начатых асинхронных операций, нужно указать флаг WT_EXECUTEINIOTHREAD.
При указании флага WT_EXECUTEINPERSISTENTTHREAD пакет запроса помещается в поток, который никогда не удаляется, так что вы спокойно можете использовать TLS. Так как поток всего один, не рекомендуется выполнять в нем продолжительных операций.
