Ефективна багатопоточність

Олексій Ширшов

Вступ

Отже, знову багатопоточність. Ви скажете, яка побита тема, вже скільки можна про це писати! Так, написано про неї чимало. Практично кожен програміст, який з нею стикався (тобто хоч раз в житті викликав функцію CreateThread), може заявити, що він про неї знає все або майже все. Але це глибока помилка. Створення ефективних багатопоточних серверів (а робити багатопотокового клієнта особливого сенсу немає) - справа складна і потребує хороших знань системних механізмів: багатопоточності, синхронізації, асинхронного введення/виводу і багато іншого. У цій статті я торкнуся теми організації пулу потоків для ефективної обробки клієнтських запитів.

Навіщо це потрібно

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

При приході клієнтського запиту у сервера є кілька варіантів дій:

Обробляти всі запити в одному потоці;

Обробляти кожен запит в окремому потоці;

Організувати пул потоків.

Розглянемо кожен із сценаріїв.

Обробка всіх запитів в одному потоці

Відразу зрозуміло, що це рішення підходить тільки для дуже обмеженого числа випадків, в яких кількість клієнтів невелике, і звертаються вони до сервера не часто. Ця найпростіша схема роботи: мінімум потоків, мінімум ресурсів, і не потрібно нічого синхронізувати. Головне, що потрібно зробити - побудувати чергу вхідних запитів, щоб вони не губилися при послідовній обробці. Це нескладно, до того ж можна взяти вже готові рішення: наприклад, СОМ-сервер STA singleton.

Обробка кожного запиту в окремому потоці

Це, мабуть, найпопулярніша у розробників схема. У ній для кожного клієнтського запиту створюється окремий потік. Рішення це просте і для багатьох випадків задовільний, тому що при цьому потрібно піклуватися, за великим рахунком, тільки про синхронізацію загальних змінних потоків (а їх може і не бути). Така схема працює в такий спосіб: первинний потік програми прослуховує клієнтські запити і під час вступу кожного створює новий потік, передаючи йому клієнтський пакет (дані або команду). Створений потік виконує відповідну обробку, передає результати назад клієнтові, або ж поміщає їх у БД (або ще куди-небудь), і завершує своє існування.

Давайте замислимося, що станеться, якщо клієнтів виявиться занадто багато. Сервер для кожного з них буде створювати потік, а це, з точки зору системи, непроста операція, що вимагає певного часу і ресурсів. Віртуальне адресний простір процесу також зменшується як мінімум на прийнятий за замовчуванням для потоку розмір стека. Все це дуже погано. Сервер витрачає час та ресурси на створення потоку, який обробляє клієнтський запит за все за долі секунди і потім знищується. При цьому ми повинні враховувати, що фізично одночасно можуть виконуватися тільки кількість потоків, що не перевищує кількості процесорів на комп'ютері. На ОС Windows NT/2000 при 100 одночасно запущених потоках наш сервер буде працювати дуже неоптимально, що негативно позначиться на часі обробки запиту.

Основні недоліки такої моделі:

часте створення і завершення потоків;

малий час роботи потоку;

нерегульованому кількість потоків;

в більшості випадків відсутність черги клієнтських запитів;

велика кількість перемикань контекстів робочих потоків.

Для вирішення цих проблем і призначений пул потоків.

Організація пулу потоків

Що таке пул потоків? У житті ми дуже часто зустрічаємося з організацією пулу. Наприклад, коли ви йдете в їдальню, ви зустрічаєтеся з пулом підносів. Так-так, не смійтеся. Підноси організовані в пул (спробуйте пояснити це кухарям:)); клієнтів може бути набагато менше, ніж таць, і навпаки. Коли таць багато, вони лежать без діла, коли таць мало, клієнти чекають, поки вони звільняться. Число таць, тобто розмір пулу, заздалегідь визначається так, щоб в більшості випадків клієнти не чекали таць. Однак трапляються години пік, коли клієнтів дуже багато. Просто нереально виділити окремий піднос кожного клієнта, та й не потрібно це. Клієнт все одно буде стояти в черзі до каси, так що витрати на підноси не принесуть реальних вигод. Це, звичайно, дуже далека і недосконала аналогія, але вона показує, що в природі та житті пул чого-небудь дуже часто використовується як наіеффектівнейшая схема обслуговування запитів.

Розглянемо механізм роботи пулу потоків. Є головний потік програми, прослуховували клієнтські запити. Пул потоків створюється заздалегідь або під час першого запиту. Мінімальний розмір пулу звичайно вибирається рівним 1, однак це не принципово. Під час отримання запиту головний потік вибирає потік з пулу і передає йому запит. Якщо кількість потоків в пулі досягло максимуму, запит поміщається в чергу. Якщо кількість потоків менше максимального, і всі вони зайняті обробкою, створюється новий потік, який отримує клієнтський пакет на обробку. Якщо кількість потоків одно максимальному і всі потоки займаються обробкою, тобто активні, пакет ставиться в чергу і чекає звільнення одного з потоків. Алгоритми долучення потоків в пул і визначення оптимального розміру пулу сильно залежать від розв'язуваної задачі. Більш докладно про це буде сказано пізніше.

При використанні RPC-транспорту (у випадку з СОМ-серверами) про кулю потоків піклуватися не потрібно. СОМ-сервер MTA singleton -- краще рішення для СОМ в тому сенсі, що нічого не треба робити з приводу організації пулу потоків. Система (точніше СОМ-runtime) все робить сама. Однак, якщо ви використовуєте чистий RPC, вам доведеться все організовувати самому.

Примітиви операційної системи

Відразу скажу, що в якості операційної системи я буду розглядати Windows NT версії 3.1 і вище. Для функцій, які з'явилися пізніше, версія ОС буде обговорюватися окремо. Лінійка Windows 9x НЕ надає жодних засобів для організації пулу потоків.

В операційній системі є три механізму організації черги запитів (черга запитів - невід'ємна частина пулу): DPC - deferred procedure call (відкладений виклик процедури), APC - asynchronous procedure call (асинхронний виклик процедури) і об'єкт ядра queue (черга), яку можна додаткам режиму користувача (user mode) у вигляді більш складного об'єкта "порт завершення введення/виводу". DPC використовується тільки в режимі ядра (kernel mode) в основному драйверами пристроїв для більш ефективної обробки запитів вводу/виводу. DPC ми розглядати не будемо, тому що ця тема більше стосується програмування драйверів пристроїв, а ми збираємося писати прикладну програму користувацького режиму. APC, на відміну від DPC, завжди виконується в контексті будь-якого потоку (з кожним потоком асоційована своя чергу APC-запитів) і може генерувати сторінкові помилки (page faults), очікувати переходу об'єкта ядра в сигнальний стан, і так далі.        

ПРИМІТКА   

А чому функції DPC не можуть   генерувати сторінкові помилки? Справа в тому, що DPC і APC ставляться в чергу   системою за допомогою програмного переривання і обробляються на певному   рівні переривань IRQL - interrupt request level. IRQL DPC збігається з IRQL   dispatch, на якому обробляються сторінкові помилки (він навіть називається   DPC/dispatch, щоб відобразити це). Як тільки система піднімає поточний   рівень до DPC/dispatch, всі переривання з меншим або рівним рівнем   маскуються (блокуються). Після обробки DPC система знижує рівень і,   якщо в черзі перебувати ще одна DPC-запит, знову генерується програмне   переривання. Якщо при обробці DPC-запиту трапиться часткою сторінки   пам'яті, не знаходиться у фізичній пам'яті, система не зможе «підкачати» цю   сторінку з диска. Рівень переривання IRQL APC нижче DPC/dispatch, так що APC   можуть вільно насолоджуватися всією красою віртуального адресного   простору процесу.     

APC бувають двох видів: режиму ядра і призначеного для користувача режиму. APC режиму ядра відрізняється від APC користувацького режиму тим, що система може перервати роботу потоку для виклику процедури без його відома, тоді як для виконання APC режиму користувача потік повинен знаходиться в спеціальному «тривожному» (alertable) очікуванні, як би даючи згоду на виконання процедури. Об'єкт "чергу" і його похідний об'єкт "порт завершення введення/виводу" спеціально призначені для організації пулу і, крім черги запитів, можуть управляти асоційованими з ними потоками. Давайте розглянемо APC режиму користувача і порт завершення введення/виводу більш докладно.

APC режиму користувача

Цей механізм можна використовувати, якщо потрібно виконати будь-яку операцію (функцію) в контексті певного потоку. Для виконання функції потік повинен «дати згоду», перейшовши в стан тривожного очікування (alertable wait state). Якщо потік перебуває в такому стані, то, як тільки ми поставимо в чергу APC-запит із зазначенням адреси функції та довільного параметра для неї, потік перейде до виконання даної функції, після чого вийде зі стану очікування. APC призначеного для користувача режиму можуть використовувати функції ReadFileEx, WriteFileEx, а також SetWaitableTimer, про яку ми поговоримо окремо. Опції ReadFileEx і WriteFileEx призначені спеціально для асинхронних операцій - для них ви зобов'язані відкривати файл (файл у самому загальному сенсі) в асинхронному режимі, вказуючи прапор FILE_FLAG_OVERLAPPED, а також для кожної операції створювати структуру OVERLAPPED. Як останній параметра обидві функції беруть адресу спеціальної функції завершення -- FileIOCompletionRoutine. Після завершення асинхронної операції, якщо потік знаходиться в тривожному очікуванні, ця функція буде викликана за допомогою механізму APC. У тривожне очікування потік може перейти за допомогою «розширених» функцій очікування, які закінчуються на Ex. Це SleepEx, WaitForSingleObjectEx, WaitForMultipleObjectsEx та інші. Для того, щоб вручну помістити APC-запит в чергу потоку, потрібно скористатися функцією QueueUserAPC. Ось її прототип:        

DWORD QueueUserAPC (   

PAPCFUNC pfnAPC,// APC функція   

HANDLE hThread,// хендл потоку   

ULONG_PTR dwData// параметр APC функції   

);     

Розглянемо невеликий приклад її використання (перевірка помилок усунена для підвищення наочності).        

const int _SOME_MAGIC_VALUE = 5;      

DWORD CALLBACK trd1 (LPVOID p)   

(   

HANDLE hEvent = (HANDLE) p;   

SetEvent (hEvent);      

int i = 0;   

while (i <   _SOME_MAGIC_VALUE) (   

SleepEx (INFINITE, true);   

cout <   

)   

return 0;   

)      

VOID CALLBACK APCProc (ULONG_PTR dwParam)   

(   

cout << "APC Proc   # "<   

cout << "threadid   : "<   

)      

int main ()   

(   

HANDLE hEvent = CreateEvent (0,   false, false, NULL);      

DWORD trd_id = 0;   

HANDLE hThread =   CreateThread (0, 0, trd1, hEvent, 0, & trd_id);   

cout << "Thread id   is 0x "<      

WaitForSingleObject (hEvent,   INFINITE);      

for (int i = 0; i <   _SOME_MAGIC_VALUE; I ++){   

QueueUserAPC (APCProc,   hThread, i);   

)      

WaitForSingleObject (hThread,   1000);   

CloseHandle (hThread);   

return 0;   

)     

Незважаючи на простоту, приклад досить складний, і не завжди можна передбачити, що буде на екрані після його завершення. Давайте розберемо його у формі питання-відповідь.

Чому я Синхронізація потоки за допомогою події?

Якщо цього не зробити, то всі п'ять APC-запитів виконуватися ще до того, як функція потоку trd1 отримає управління. Це відбудеться тому, що сама система в процесі створення потоку використовує механізм APC-викликів для ініціалізації потоку. З його допомогою, наприклад, відбувається виклик всіх функцій DllMain з параметром DLL_THREAD_ATTACH, якщо, звичайно, ви не викликали DisableThreadLibraryCalls для будь-якої бібліотеки.

Чому на екран виводиться дивний результат?        

Thread id is 0x68c   

APC Proc # 0 threadid: 68c   

0   

APC Proc # 1 threadid: 68c   

APC Proc # 2 threadid: 68c   

APC Proc # 3 threadid: 68c   

APC Proc # 4 threadid: 68c   

1     

Як я вже казав, для кожного потоку система організовує чергу з APC-запитів, так що в момент обробки першого запиту потоком система встигає додати в чергу всі інші запити, які і виконуються при наступному виклику SleepEx. Результат сильно залежить від завантаженості системи, так що у вас він може бути іншим: наприклад, всі запити встигнуть виконатися за одну ітерацію.

Чому, якщо закоментувати тіло APCProc, на екран виводиться наступне?        

Thread id is 0x7b4   

0   

1   

2   

3   

4     

Так як тепер ця процедура фактично нічого не робить, система не встигає додати новий запит в чергу до завершення обробки попереднього, так що кожен SleepEx обробляє «свій» APC-запит.

Тепер вам повинно бути зрозуміло, як використовувати даний механізм для організації пулу потоків. Ось приблизний сценарій для фіксованої кількості потоків в пулі: в головному потоці програми створюються кілька робочих потоків, кожен з яких відразу переходить у стан тривожного очікування спеціально створеного основним потоком події. Коли приходить клієнтський запит, головний потік передає APC-запит одного з робітників потоків. Робочий потік прокидається і виконує функцію, яку було поставлено в чергу головним потоком. При цьому він не залишає функції WaitForSingleObjectEx. Тобто виконання APC-запиту здійснюється як би всередині функції WaitForSingleObjectEx. Після завершення виконання запиту управління передається функції WaitForSingleObjectEx, яка, в свою чергу, передає управління основному коду потоку, повертаючи WAIT_IO_COMPLETION.

При отриманні керування робочий потік повинен проаналізувати значення, повернене цією функцією. Якщо воно дорівнює WAIT_IO_COMPLETION, то причиною виходу з функції WaitForSingleObjectEx було завершення обробки APC-запиту - потік при цьому повинен знову перейти в стан очікування події. Якщо ж повертається значення WAIT_OBJECT_0, то причиною виходу була установка події в сигнальний стан головним потоком додатки. При цьому робочий потік повинен завершитися.

Це дуже проста схема (наприклад, робочі потоки замість очікування можуть виконувати якусь іншу корисну роботу), але вона досить непогано пояснює механізм використання APC для організації пулу.

SetWaitableTimer

Ця функція з'явилася з версії 4.0. Вона дозволяє активувати таймер, який через заданий період часу в 100-наносекундних інтервалах або при настанні заданого абсолютного часу переходить в сигнальне стан. Крім цього, можна вказати процедуру завершення, яка буде викликана за допомогою APC-запиту в цьому потоці. Для завершення процедури можна вказати додатковий параметр. Функція хороша тим, що вона не прив'язана до вікон і циклу вибірки повідомлень, як, наприклад, SetTimer. З її допомогою можна використовувати таймери в будь-яких додатках, включаючи консольні та сервіси. Однак у SetWaitableTimer є і деякі недоліки:

функція завершення завжди викликається в потоці, що викликала SetWaitableTimer;

потік повинен бути в стані тривожного очікування, щоб обробити APC-запит;

другу APC-запит почне оброблятися тільки після закінчення обробки попереднього запиту, тобто запити обробляються послідовно.

Всі ці проблеми вирішує об'єкт "черга таймерів ", про який мова піде пізніше.

Порт завершити введення/виводу

Це, безумовно, одна з наймогутніших і складних об'єктів виконавчої системи. Він спеціально призначений для оптимізації обробки клієнтських запитів в серверних застосуваннях. Він не тільки організовує чергу запитів, а й ефективно управляє їх обробкою.

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

Як працює порт

При створенні порту вказується максимальна кількість активних потоків, здатних обробляти клієнтські запити паралельно. Так як кількість реально працюючих паралельно потоків на комп'ютері одно кількості процесорів, то вказівка більшого максимальної кількості активних потоків не вигідно. Чому? Справа в тому, що для виконання декількох потоків на одному процесорі, системі доводиться постійно перемикати процесварок між потоками, емуліруя, таким чином, паралельність, однак це перемикання, зване перемиканням контекстів - досить дорога операція. Уникнути її можна тільки одним способом - не створювати паралельно працюють потоки в кількості більшій, ніж кількість процесорів. Таким чином, при створенні порту, здавалося б, треба вказувати в якості максимальної кількості активних потоків число процесорів в системі, але тут є одна тонкість. Припустимо, у нас однопроцесорний комп'ютер і, відповідно, клієнтські запити ми обробляємо в одному потоці. Що буде, якщо клієнтський запит прийде в момент виконання синхронної операції з диском або в момент очікування якого-небудь об'єкта цим потоком? Він буде чекати, поки потік не закінчить свою роботу, але адже процесор у цей час не діє, тому що потік заблокований на синхронної операції або на який-небудь об'єкт. Коли процесор не діє, а клієнтський запит не обробляється - це погано. Ми приходимо до висновку про те, що завжди повинен існувати резервний потік, який підхоплював б запити в момент, коли «основний» потік виконує блокують операції, і процесор не діє.

Робота з файлами (в самому широкому сенсі слова) дуже тісно пов'язана з багатопоточність і обробкою запитів на сервері. Сокет або pipe - це теж файли. Щоб обробляти запити через ці канали паралельно, потрібний порт. Давайте розглянемо функцію створення порту і зв'язку його з файлом (зачем-то розробники з Microsoft об'єднали дві ці функції в одну; в виконавчій системі ці дві функції виконують сервіси NtCreateIoCompletion і NtSetInformationFile, відповідно).        

HANDLE CreateIoCompletionPort (   

HANDLE FileHandle,// хендл файлу   

HANDLE ExistingCompletionPort,   //Хендл порту завершити введення/виводу   

ULONG_PTR CompletionKey,// ключ завершення   

DWORD NumberOfConcurrentThreads//   Максимальна кількість паралельних потоків   

);     

Для простого створення порту потрібно як перший параметра передати INVALID_HANDLE_VALUE, а в якості другого і третього - 0. Для зв'язування файлу з портом потрібно вказати перші три параметри і проігнорувати четвертий.

Після того, як файл (під файлом тут мається на увазі об'єкт підсистеми Win32, який реалізується за допомогою об'єкту "файл виконавчої системи ", до таких відносяться файли, сокети, поштові ящики, іменовані канали та ін.) пов'язаний з портом, закінчення всіх асинхронних запитів вводу/виводу потрапляють в чергу порту і можуть бути оброблені пулом потоків. Наступні функції можуть бути використані з портом для завершення обробки асинхронних операцій введення/виводу:

ConnectNamedPipe - очікує підключення клієнта до іменовані канали.

DeviceIoControl - низькорівневий введення/виведення.

LockFileEx - блокування регіону файлу.

ReadDirectoryChangesW - очікування змін до директорії.

ReadFile - читання файлу.

TransactNamedPipe - Комбіноване читання і запис за іменовані канали, які здійснюються за одну мережеву операцію.

WaitCommEvent - очікування події послідовного інтерфейсу (СОМ-порт).

WriteFile - запис у файл.

Якщо ви не хочете, щоб закінчення асинхронного введення/виводу оброблялося портом (наприклад, коли вам не важливий результат операції), потрібно використати такий трюк [1]. Потрібно встановити поле hEvent структури OVERLAPPED рівним описувач події до встановленого першого бітом. Робиться це приблизно так:        

OVERLAPPED   ov = (0);   

ov.hEvent   = CreateEvent (...);   

ov.hEvent   = (HANDLE) ((DWORD_PTR) (ov.hEvent) | 1);     

І не забувайте скидати молодший біт при закритті Хендли події.

Додавати потік до пулу (підключати його до обробки запитів) можна за допомогою наступної функції:        

BOOL   GetQueuedCompletionStatus (   

// хендл порту завершити введення/виводу   

HANDLE CompletionPort,   

// кількість переданих байт   

LPDWORD lpNumberOfBytes,   

// ключ завершення   

PULONG_PTR lpCompletionKey,   

// структура OVERLAPPED   

LPOVERLAPPED * lpOverlapped,   

// значення таймауту   

DWORD dwMilliseconds   

);     

Ця функція блокує потік до тих пір, поки порт не передасть потоку пакет запиту або не закінчиться таймаут.

Помістити пакет запиту в порт можна за допомогою функції PostQueuedCompletionStatus.        

BOOL PostQueuedCompletionStatus (   

  HANDLE CompletionPort,   //Хендл порту завершити введення/виводу   

  DWORD dwNumberOfBytesTransferred,// кількість переданих байт   

  ULONG_PTR dwCompletionKey,   //Ключ завершення   

  LPOVERLAPPED lpOverlapped   //Структура OVERLAPPED   

);     

Пакет запиту не обов'язково повинен бути структурою OVERLAPPED або похідною від неї [2].

Давайте зберемо всю інформацію воєдино. Порт завершення - об'єкт, що організує кілька черг з клієнтських запитів і потоків, їх обробляють. Потік додається в чергу очікують на запит потоків порту при виконанні функції GetQueuedCompletionStatus. При надходженні запиту порт розблокує перший потік в черзі чекають потоків і передає йому цей запит (у вигляді структури OVERLAPPED і ключа завершення). Потік при цьому переміщається в чергу активних потоків (число активних потоків збільшується на 1). Припустимо, у нас Максимальна кількість активних потоків дорівнює 1, тоді при надходженні запиту інший потік з черги чекають активований не буде. Після обробки клієнтського запиту потік знову викликає GetQueuedCompletionStatus і ставиться в початок переліку адрес потоків. Чому потік ставиться саме в початок списку? Справа в тому, що потоки беруться з початку списку, і при низькій активності можуть використовуватися не всі потоки. При цьому стеки і контексти не використовуваних потоків можуть бути вивантажені на диск за непотрібність.

Якщо в процесі обробки запиту потік звернувся до блокує функції, число активних потоків зменшується на 1, як якщо б потік перейшов знову в чергу очікують потоків. Це дає можливість при приході наступного клієнтського запиту задіяти наступний потік з черги що очікують. Коли перший потік закінчить блокує операцію, число активних потоків перевищить максимальний, і при наступному виклику функції GetQueuedCompletionStatus один з цих потоків заблокується, а друга отримає пакет запиту (якщо він є).        

Черга         

Запис додається при:         

Запис видаляється за:             

Список пристроїв,   асоційованих з портом         

виклик   CreateIoCompletionPort         

закриття хенду файлу             

Черга клієнтських запитів   (FIFO)         

завершення асинхронної   операції файлу, що асоціюється з портом, або виклику функції   PostQueuedCompletionStatus         

передачі портом запиту   потоку на обробку             

Черга очікують потоків         

виконанні функції   GetQueuedCompletionStatus         

початку обробки   клієнтського запиту потоком             

Список працюючих потоків         

початку обробки   клієнтського запиту потоком         

виклик потоком   GetQueuedCompletionStatus або будь-яку блокує функції             

Список призупинених   потоків         

виклик потоком якої-небудь   блокує функції         

виході потоку з будь-якої   блокує функції     

Таблиця 1. Список черг порту завершення введення/виводу [1].

недокументовані можливості порту і його низькорівневе пристрій

Як завжди це буває в Microsoft, порт завершення володіє багатьма недокументовані можливості:

У порту завершити введення/виводу може бути ім'я, і відповідно, він доступний для інших процесів. Абсолютно незрозуміло, чому розробники вирішили приховати цю, на мій погляд, потрібну особливість порту. Назва можна задати в параметрі ObjectAttributes функції NtCreateIoCompletion.

Друга особливість випливає з першої: з портом може бути пов'язаний дескриптор безпеки, який також задається в параметрі ObjectAttributes функції NtCreateIoCompletion.

Відкривається порт за допомогою функції 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 можуть нам запропонувати. У Windows2000 з'явилися нові функції, які повністю беруть на себе всю чорнову роботу по створенню та видаленню потоків в пулі. Про них -- наступний розділ.

Вбудована підтримка пулу потоків

У 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         

Запит поміщається в потік,   який