В даній статті, ми розглянемо, як створювати потоки у власній програмі.
Створення потоків у Юнікс-подібних ОС
Для того, щоб створювати потоки у Юнікс-подібних системах, нам знадобиться інтерфейс POSIX Threads, функції якої знаходяться в бібліотеці pthread. Це означає, що нам необхідно компонувати об'єктний файл нашої програми разом з бібліотекою pthread, використовуючи параметр компілятора -lpthread (звісно, якщо ви використовуєте компілятор GCC).
Отже, щоб створити потік, і запустити його виконання, необхідно мати спеціально створену для цього функцію. В загальному вигляді її оголошення виглядає наступним чином:
void* my_thread_function_name (void*) ;
Оскільки функція створення нового потоку (себто thread_create), не може “знати” усі можливі комбінації параметрів функції потоку і її типу поверненого значення, розробники даного інтерфейсу були змушені спроектувати інтерфейс так, щоб функція потоку приймала один параметр — вказівник на довільний тип (void*). Повернене значення функції потоку також повинно бути вказівником на довільний тип (void*). Це призводить до того, що нам необхідно перетворювати вказівник “void*” на до необхідного нам типу, що не завжди безпечно. Окрім небезпечності перетворення типів вказівників, ми змушені передавати усю множину необхідних нам параметрів в якості єдиного параметру. Тобто без створення спеціальної структури для збереження усіх необхідних значень не обійтися. Іншим рішенням проблеми множини параметрів також може слугувати використання глобальних змінних. Але це також небезпечно з точки зору багатопоковості, оскільки в довільний момент часу до однієї змінної може звернутися декілька потоків і своїми операціями зруйнувати дані, або ж банально їх спотворити (дана проблема вирішується спеціальними сутностями — мютексами, про які ми поговоримо далі в статті). Усі ці проблеми також справедливі і для API інтерфейсу ОС Windows.
Функція, яка створює і запускає на виконання потік називається pthread_create і її оголошення виглядає наступним чином:
int pthread_create (pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg) ;
Першим параметром передається вказівник на тип pthread_t, який можна вважати дескриптором потоку. У разі успішного створення потоку, в ньому буде зберігатися ідентифікаційна інформація про створений потік, через яку, до даного нового потоку, можна звернутися за необхідною додатковою інформацією. Другим параметром слугує вказівник на спеціальну структуру pthread_attr_t, через яку, ми передаємо деякі бажані атрибути створення нового потоку. Цей параметр може бути NULL, який означає, що потоку присвоюється атрибути по замовчуванню. Третім параметром передається вказівник на функцію потоку, оголошення якої ми розглядали вище. І четвертим — параметр функції нового потоку, в якості вказівника на довільний тип (void*).
Якщо функції pthread_create вдалося успішно створити новий потік, вона повертає нуль (в якості значення типу int). В іншому випадку, вона повертає код помилки, і значення у області пам'яті вказівника thread є невизначеним (тобто його непотрібно інтерпретувати).
Розглянемо просту програму, яка створює новий потік:
#include <pthread.h> /* API POSIX Threads */
#include <iostream> /* стандартний вивід */
using namespace std ; /* використовуємо простір імен std*/
/* оголошуємо і реалізовуємо функцію потоку */
void* new_thread (void*)
{
/* виводимо на екран "Привіт, Світ" */
cout << "Hallo, World!!!" << endl ;
/* повертаємо NULL */
return NULL ;
}
/* головна функція програми */
int main (int argc, char** argv)
{
/* змінна-дескриптор нового потоку */
pthread_t desc ;
/* створюємо новий потік і зберігаємо повернене значення pthread_create */
int pthc_ret = pthread_create(&desc, NULL, new_thread, NULL) ;
/* перевіряємо чи функція pthread_create успішно виконала роботу */
if (pthc_ret!=0)
{
/* в разі неуспішного завершення, тут - обробник помилки */
cerr << "fail to create thread!\n" ;
}
else
{
/* за успішного створення потоку - очікуємо завершення потоку
** оскільки головний потік може завершитися швидше, ніж
** новий потік виконає свою роботу */
if (pthread_join (desc, NULL)!=0)
{
/* якщо не вдалося зачекати завершення виконання потоку-
** тут код обробника помилки */
cerr << "fail to wait for thread\n" ;
}
}
/* потік або не був створений, або успішно завершив свою роботу */
/* вихід з програми */
return 0 ;
}
Дану програму можна скомпілювати наступною командою (за умови, що її код знаходиться в файлі “main.cxx”, а виконуваний файл ми бажаємо назвати “file”):
g++ main.cxx -o file -lpthread
Після виконання даної програми у вікні терміналу, ми можемо побачити наступне:
Дана програма створює новий потік з функції “new_thread” і очікує завершення нового потоку, після чого, завершує своє виконання. Безпосередньо функція “new_thread” виводить на екран повідомлення “Hallo, World!!!” і завершує своє виконання.
Розглянемо складніший приклад: створимо два нових потоки, які будуть змінювати значення спільної глобальної змінної (в циклі збільшуючи її значення на одиницю). Для цього нам знадобиться тільки одна функція потоку (в якості іншого потоку, ми будемо використовувати головний потік програми), одна глобальна цілочисельна змінна, і одна спеціальна змінна синхронізації доступу — мютекс. Ось код програми:
#include <pthread.h> /* API POSIX Threads */
#include <iostream> /* стандартний вивід */
using namespace std ; /* використовуємо простір імен std*/
/* глобальна змінна - лічильник */
unsigned int GlobalCounter = 0 ;
/* мютекс - синхронізатор доступу */
pthread_mutex_t GlobalCounter_mutex ;
/* функція-процедура для інкрементування значення змінної */
/* не пишемо один і той самий код в двох місцях */
inline void increment_variable ()
{
/* цикл для інкрементування значення GlobalCounter */
for (unsigned int iter=0; iter<100; ++iter)
{
/* займаємо мютекс - це гарантує, щи ми єдині в даний момент
** зможемо редагувати значення змінної GlobalCounter */
pthread_mutex_lock (&GlobalCounter_mutex) ;
/* збільшуємо значення на одиницю */
++ GlobalCounter ;
/* звільняємо мютекс - щоб інший потік
** також зміг виконати свої операції */
pthread_mutex_unlock (&GlobalCounter_mutex) ;
}
}
/* оголошуємо і реалізовуємо функцію потоку */
void* new_thread (void*)
{
increment_variable () ;
/* повертаємо NULL */
return NULL ;
}
/* головна функція програми */
int main (int argc, char** argv)
{
/* підготовлюємо до використання мютекс */
pthread_mutex_init (&GlobalCounter_mutex, NULL) ;
/* змінна-дескриптор нового потоку */
pthread_t desc ;
/* створюємо новий потік і зберігаємо повернене значення pthread_create */
int pthc_ret = pthread_create (&desc, NULL, new_thread, NULL) ;
/* перевіряємо чи функція pthread_create успішно виконала роботу */
if (pthc_ret!=0)
{
/* в разі неуспішного завершення, тут - обробник помилки */
cerr << "fail to create thread!\n" ;
return 0 ;
}
/* використаємо головний потік, як тестовий для інкрементування
** значення глобальної змінної */
increment_variable () ;
/* тестовий потік може ще не завершити свою роботу
** очікування його завершення гарантує коректну роботу програми */
if (pthread_join (desc, NULL)!=0)
{
/* якщо не вдалося зачекати завершення виконання потоку-
** тут код обробника помилки */
cerr << "fail to wait for thread\n" ;
}
/* виводимо результат інкрементування у вікно терміналу */
cout << "GlobalCounter value is: " << GlobalCounter << endl ;
/* вихід з програми */
return 0 ;
}
Після компіляції програми і виконання у вікні терміналу з'явиться наступний текст:
Дана програма створює новий потік з функції new_thread, після чого, разом з ним, асинхронно розпочинає інкрементувати значення цілочисельної беззнакової змінної GlobalCounter. Мютекс GlobalCounter_mutex, в даному випадку, виступає синхронізатором доступу до змінної GlobalCounter, без якого, дані можуть бути спотвореними.
В даному випадку, при обмеженні у 200 ітерацій проблема спотворення даних може і не виникнути, оскільки для процесора виконати 100 ітерацій для кожного потоку, втискаються в виділений час на кожний потік, тобто кожен потік встигає виконати свої інтерації ще до того, як інший їх розпочне.
Але, наприклад, якщо ми збільшимо кількість ітерацій до двох мільйонів (по мільйону на кожен потік) і закоментуємо виклики функцій “pthread_mutex_lock” (для монопольного займання мютекса) і “pthread_mutex_unlock” (для його вивільнення потоком), ось що може статись з значенням GlobalCounter:
Як ми бачимо, замість значення в 2 000 000, програма виводить менші значення. В двох останніх випадках майже вдвічі менші ніж потрібні. Тому не забувайте синхронізувати доступ до змінних за допомогою мютексів.
Створення потоків у ОС Windows
Для створення потоку за допомогою API ОС Windows, нам знадобиться функція “CreateThread” з заголовкового файлу “windows.h”. ЇЇ оголошення виглядає наступним чином:
HANDLE WINAPI CreateThread(
_In_opt_ LPSECURITY_ATTRIBUTES lpThreadAttributes,
_In_ SIZE_T dwStackSize,
_In_ LPTHREAD_START_ROUTINE lpStartAddress,
_In_opt_ LPVOID lpParameter,
_In_ DWORD dwCreationFlags,
_Out_opt_ LPDWORD lpThreadId
);
В якості параметрів, її передаються
- Вказівник на структуру атрибутів безпеки (може бути NULL);
- dwStackSize — початковий розмір стеку нового потоку;
- lpStartAddress — адреса функції нового потоку;
- lpParameter — параметр, який передається функції (void*)
- dwCreationFlags — прапорці, які модифікують початковий стан потоку (може бути 0 - для стану за замовчуванням);
- lpThread — цілочисельний ідентифікатор потоку (може бути NULL, якщо це значення не потрібне).
Повертає дана функція тип HANDLE, який являється дескриптором потоку.
Проста програма, на подобі вищенаведеної для Юнікс-систем, прийме наступний вигляд (замість pthread_join ми будемо використовувати стандартну функцію API “WaitForSingleObject”):
#include <windows.h> /* API POSIX Threads */
#include <stdio.h> /* стандартний вивід */
using namespace std ; /* використовуємо простір імен std*/
/* оголошуємо і реалізовуємо функцію потоку */
void* new_thread (void*)
{
/* виводимо на екран "Привіт, Світ" */
printf("Hallo, World!!!\n") ;
/* повертаємо NULL */
return NULL ;
}
/* головна функція програми */
int main (int argc, char** argv)
{
/* створюємо новий потік і зберігаємо повернений дескриптор */
HANDLE thhandle = CreateThread (NULL,
0,
(LPTHREAD_START_ROUTINE) new_thread,
NULL,
0,
NULL) ;
/* перевіряємо чи успішно створений потік */
if (!thhandle)
{
/* в разі неуспішного завершення, тут - обробник помилки */
printf("fail to create thread!\n") ;
}
else
{
/* за успішного створення потоку - очікуємо завершення потоку
** оскільки головний потік може завершитися швидше, ніж
** новий потік виконає свою роботу */
WaitForSingleObject (thhandle, INFINITE) ;
}
/* потік або не був створений, або успішно завершив свою роботу */
/* вихід з програми */
return 0 ;
}
Після компілювання і виконання програми у вікні терміналу ОС Windows можна побачити приблизно наступне:
Для складнішої версії програми, яка наведена вище у розділі Юнікс-подібних ОС, вигляд для ОС Windows прийме наступний вигляд:
#include <windows.h> /* API POSIX Threads */
#include <stdio.h> /* стандартний вивід */
/* глобальна змінна - лічильник */
unsigned int GlobalCounter = 0 ;
/* мютекс - синхронізатор доступу */
HANDLE GlobalCounter_mutex ;
/* функція-процедура для інкрементування значення змінної */
/* не пишемо один і той самий код в двох місцях */
inline void increment_variable ()
{
/* цикл для інкрементування значення GlobalCounter */
for (unsigned int iter=0; iter<100; ++iter)
{
/* займаємо мютекс - це гарантує, щи ми єдині в даний момент
** зможемо редагувати значення змінної GlobalCounter */
WaitForSingleObject (GlobalCounter_mutex, INFINITE) ;
/* збільшуємо значення на одиницю */
++ GlobalCounter ;
/* звільняємо мютекс - щоб інший потік
** також зміг виконати свої операції */
ReleaseMutex (GlobalCounter_mutex) ;
}
}
/* оголошуємо і реалізовуємо функцію потоку */
void* new_thread (void*)
{
increment_variable () ;
/* повертаємо NULL */
return NULL ;
}
/* головна функція програми */
int main (int argc, char** argv)
{
/* підготовлюємо до використання мютекс */
GlobalCounter_mutex = CreateMutex (NULL, FALSE, NULL) ;
/* створюємо новий потік і зберігаємо повернене значення pthread_create */
/* створюємо новий потік і зберігаємо повернений дескриптор */
HANDLE thhandle = CreateThread (NULL,
0,
(LPTHREAD_START_ROUTINE) new_thread,
NULL,
0,
NULL) ;
/* перевіряємо чи функція pthread_create успішно виконала роботу */
if (!thhandle)
{
/* в разі неуспішного завершення, тут - обробник помилки */
printf ("fail to create thread!\n") ;
return 0 ;
}
/* використаємо головний потік, як тестовий для інкрементування
** значення глобальної змінної */
increment_variable () ;
/* тестовий потік може ще не завершити свою роботу
** очікування його завершення гарантує коректну роботу програми */
WaitForSingleObject (thhandle, INFINITE) ;
/* виводимо результат інкрементування у вікно терміналу */
printf ("GlobalCounter value is: %i", GlobalCounter) ;
/* вихід з програми */
return 0 ;
}
Результат виконання наступний:
Для чого потрібні потоки?
В основному потоки необхідні для розпаралелювання виконання завдань. Наприклад, якщо у Вас є великий масив даних і два ядра процесора, Ви можете обробити його вдвічі швидше за допомогою потоків. Також потоки використовуються серверними програмами, тобто сервер створює новий потік для кожного підключення клієнта, в результаті чого, кожен з них може виконувати роботу незалежно від іншого.
Ще одним хорошим призначенням потоків являється розбиття програми на потоки відображення (вікно) і потік обробки даних. Наприклад, якщо у вас є достатньо великий масив даних і програма має візуальний інтерфейс користувача, тоді при запуску обробки цих даних, інтерфейс перестає оновлюватися і реагувати на події вводу, що може змусити користувача вважати програму “завислою”.
На сьогодні все. Не забудьте написати деякі власні програми для закріплення матеріалу. А також прочитайте більше документації по потокових API для Вашої системи.