Показ дописів із міткою багатопоточність. Показати всі дописи
Показ дописів із міткою багатопоточність. Показати всі дописи
пʼятниця, 29 липня 2011 р.
вівторок, 26 липня 2011 р.
Розширення Boost.Asio
/*стаття не відредагована*/
Я вже досить давно використовую бібліотеку Boost.Asio у своєму проекті, в основному для організації асинхронного введення виведення і не раз ловив себе на думці, що хотів би використовувати її не тільки для цього. Насправді це не складно виправити, asio це не стільки мережева бібліотека, скільки дуже зручний фреймверк для вашого коду =) Отже, трохи теорії. Бібліотека asio дозволяє ефективно відокремити реалізацію від інтерфейсу використовуючи принцип dependency injection, для цього існують три сутності: io_object - інтерфейс надається користувачеві; service (не плутати з io_service-ом) - об'єкт реалізує взаємодію з ОС, наприклад з системою введення-виведення; implementation - зберігає стан об'єкта (наприклад хендл сокета). Все разом це виглядає так: io_object - клас успадкований від boost:: asio:: basic_io_object, Об'єкти цього класу створюються користувачем для виконання будь-яких операцій, приклади: boost:: asio:: deadline_timer; boost:: asio:: ip:: tcp:: socket; boost:: asio:: windows:: random_access_handle. Конструктор класу basic_io_object приймає покажчик на io_service. Кожному типу io_object-а відповідає свій service. Сервіс, це об'єкт безпосередньо виконує операції, програма користувача повинна взаємодіяти з сервісом не безпосередньо а через io_object відповідного типу, який виконує роль інтерфейсу. Сервіс автоматично створюється до створення першого примірника відповідного io_object-a, наприклад, для всіх сокетів створюється один сервіс (під windows це буде win_iocp_socket_service за назвою зрозуміло, що він використовує порти завершення:), для таймерів - інший (deadline_timer_service, котрий під windows так- ж використовує IOCP). io_service містить список сервісів для всіх об'єктів (io_object) які були на ньому створені (при створенні отримали посилання на цей io_service як параметр конструктора). basic_io_objectмістить два члени класу, з допомогою яких програміст може отримати доступ до сервісу та реалізації самого об'єкта - service і implementation. service - посилання на сервіс для даного об'єкта, який має тип basic_io_object:: Service_type, implementation - об'єкт типу basic_io_object:: Implementation_type - який є членом класу basic_io_object і представляє з себе реалізацію io_object-a, часом життя цього об'єкта управляє сервіс (волаючи методи construct і destroy). Допустимо у нас є такий код: boost:: asio:: io_service io; boost:: asio:: deadline_timer timer (io); У першому рядку буде створено io_service, він не міститиме жодного сервісу і якщо ми викличемо метод io.run (), він поверне управління відразу, тому що роботи в нього немає. У другому рядку все трохи цікавіше, спочатку буде створений io_object, в даному випадку це deadline_timer, в конструкторі basic_io_object-a (базового класу) буде викликана ф-я use_service яка спробує знайти відповідний сервіс (для таймерів він буде мати тип deadline_timer_service <... >) якщо для даного io_service-a такий сервіс ще не створено, то він бует створений, далі буде викликаний метод construct тільки-що створеного сервісу, завдання якого - ініціалізація implementation-a (що має тип deadline_timer_service:: implementation_type), який є членом класу basic_io_object і містить деталі реалізації таймера на даній платформі. Але це ще не все:), після виходу з scope-a буде викликаний деструктор таймера, який викличе метод destroy сервісу, як параметр в нього буде переданий timer.implementation. Далі в деструкції io_service-a буде викликаний метод shutdown_service сервісу дедлайн таймерів, а потім видалений екземпляр класу deadline_timer_service. Як я вже згадував, все це заради того, що-б розділити реалізацію та інтерфейс. Як приклад я реалізую клас (хоча насправді їх буде три:), для роботи з повідомленнями windows, який можна буде використовувати разом в бібліотекою boost:: asio. Для початку нам потрібно клас реалізує обробку повідомлень - implementation. Він повинен дозволяти встановлювати обробники повідомлень, видаляти їх, а так-же містити метод для обробки конкретного повідомлення.
Using Namespace Boost:: tuples;
Using Boost:: function;
Class messageloop_impl {
typedef Boost:: function < void ( Const MSG &, boost:: system:: error_code)> handler_type;
typedef Boost:: unordered_map < DWORD , handler_type> table_type ; table_type table_; boost:: asio:: io_service & io_; boost:: asio:: io_service:: work work_; boost:: mutex mutex_;
public : messageloop_impl (boost:: asio:: io_service & io_service): io_ (io_service), work_ (io_service) {}
Void set_handler ( DWORD ID, handler_type handler) {boost:: mutex:: scoped_lock lock (mutex_); table_type:: iterator I;
BOOL contain_handler; tie (i, contain_handler) = find_ (id);
if (contain_handler) runhandler_ (i, MSG (), boost:: asio:: error:: eof); table_.insert (std:: make_pair (id, handler));}
Void process_message ( Const MSG & msg, int & / * out * / processed) {boost:: mutex:: scoped_lock lock (mutex_); table_type:: iterator I;
BOOL contain_handler; tie (i, contain_handler) = find_ (msg.message);
if (contain_handler) {+ + processed ; runhandler_ (i, msg, boost:: system:: error_code ());}}
Void remove_handler ( DWORD ID) {boost:: mutex:: scoped_lock lock (mutex_); table_type:: iterator I;
BOOL contain_handler; tie ( i, contain_handler) = find_ (id);
if (contain_handler) {runhandler_ (i, MSG (), boost:: asio:: error:: eof); table_.erase (i);}}
Void Clear () {boost :: mutex:: scoped_lock lock (mutex_);
for (table_type:: iterator I = table_.begin (); i! = table_.end (); + + i) runhandler_ (i, MSG (), boost:: asio :: error:: eof); table_.clear ();}
private : tuple <table_type:: iterator , bool > find_ ( DWORD ID) {table_type:: iterator I = table_. find (id);
Return tuple <table_type: : iterator , bool > (i, i! = table_.end ());}
Void runhandler_ (table_type:: iterator I, Const MSG & m, boost:: system:: error_code e) {handler_type h = i-> second ; io_.post (boost:: bind (h, m, e));}};
Конструктор messageloop_impl - приймає посилання на io_service і зберігає її всередині об'єкта класу. Обробники повідомлень реалізовані на основі boost:: function, мають сигнатуру void (const MSG &, boost:: system:: error_code). Для зберігання обробників повідомлень використовується хеш таблиця table_. Ф-я set_handler додає обробник в хеш таблицю, ф-я remove_handler відповідно видаляє, метод clear видаляє всі обробники. Метод process_message обробляє повідомлення, у випадку, якщо для даного повідомлення знайдений обробник, значення змінної processed збільшується на одиницю, а обробник передається в io_service за допомогою методу post, далі він викликається методом run, poll або run_one io_service-a. Це потрібно для того, що-б наш обробник повідомлень слідував правилу, згідно з яким усі обробники можуть викликатися тільки в тих потоках, в яких був викликаний метод run (poll або run_one) відповідного io_service-a, або під час дзвінка деструктора io_service-a. Так-же даний клас містить член work_ має тип boost:: asio:: io_service:: work. io_service містить лічильник, який инкрементируется кожен раз, коли починається будь-яка операція, і декрементируется щоразу після завершення чергової операції. Метод run io_service-a не завершується доти, поки цей лічильник не буде дорівнює нулю. Зробивши об'єкт класу io_service:: work членом класу messageloop_impl, ми гарантуємо, що цей лічильник не обнулиться до тих пір, поки хоч один примірник messageloop_impl існує. Перед видаленням обробника, він викликається за другим параметром рівним boost:: asio:: error:: eof. Тепер напишемо код нашого сервісу для обробки повідомлень.
Class basic_messageloop_service: Public Boost:: asio:: io_service:: service {
public :
Static Boost:: asio:: io_service:: id id;
typedef Boost:: shared_ptr <messageloop_impl> implementation_type;
private : boost:: mutex mutex_; std: : Set <implementation_type> processors_; std:: set < DWORD > threads_;
Static Void send_WM_QUIT_to ( DWORD Thread) {
BOOL Result =:: PostThreadMessage (thread, WM_QUIT , 0, 0);
if (! result) {boost:: system :: error_code e = boost:: system:: error_code (:: GetLastError (), boost:: system:: system_category);
Throw Boost:: system:: system_error (e);}}
public :
Explicit basic_messageloop_service (boost:: asio:: io_service & io_service): boost:: asio:: io_service:: service (io_service) {} ~ basic_messageloop_service () {}
Void CONSTRUCT (implementation_type & impl) {impl.reset ( New messageloop_impl (get_io_service ())); boost: : mutex:: scoped_lock lock (mutex_); processors_.insert (impl);}
Void Destroy (implementation_type & impl) {boost:: mutex:: scoped_lock lock (mutex_); processors_.erase (impl); impl.reset (); }
Void shutdown_service () {boost:: mutex:: scoped_lock lock (mutex_); std:: for_each (threads_.begin (), threads_.end (), & basic_messageloop_service:: send_WM_QUIT_to); threads_.clear ();}
template < Class Handler >
Void set_handler (implementation_type & impl, DWORD ID, Handler Handler) {impl-> set_handler (id, handler);}
Void remove_handler (implementation_type & impl, DWORD ID) {impl-> remove_handler (id);}
Void Clear (implementation_type & impl) {impl-> clear ();}
Void Loop () {
MSG MSG;
DWORD thread_id =:: GetCurrentThreadId (); {boost:: mutex:: scoped_lock lock (mutex_); threads_.insert (thread_id);}
while (:: GetMessage (& msg, NULL , 0, 0)) {boost:: mutex:: scoped_lock lock (mutex_);
INT proc_cnt = 0; std:: for_each (processors_.begin (), processors_.end (), boost :: bind (& messageloop_impl:: process_message, _1, msg, boost:: ref (proc_cnt)));
if (proc_cnt == 0):: DispatchMessage (& msg);} {boost:: mutex:: scoped_lock lock (mutex_) ; threads_.erase (thread_id);}}}; boost:: asio:: io_service:: id basic_messageloop_service:: id;
Отже, наш клас повинен бути спадкоємцем класу boost:: asio:: io_service:: service, а так-же мати статичний член id має тип boost:: asio:: io_service:: id, який є унікальним ідентифікатором сервісу. Так-же клас повинен визначати тип implementation_type, в даному випадку це boost:: shared_ptr, Від якого залежить тип змінної класу basic_io_object:: Implementation, тобто реалізації об'єкта обробника повідомлень. Сервіс повинен вміти ініціалізувати об'єкти мають тип implementation_type за допомогою методу construct і деініціалізіровать їх методом destroy. Так-же він містить ряд методів для управління екземплярами класу implementation_type а так-же метод loop, в якому реалізований цикл обробки повідомлень. Наприклад, метод set_handler, він приймає два параметри, перший (impl) має тип implementation_type, другий (id) - ід-р повідомлення і третій (handler) - обробник повідомлення, реалізований просто як виклик impl-> set_handler (id, handler), тоесть просто делегує виклик реалізації. Метод loop, реалізований таким чином, спочатку він запам'ятовує в безлічі threads_ ідентифікатор потоку в якому він викликаний, потім у циклі отримує повідомлення і передає їх обробникам (які сервіс запам'ятовує в безлічі processors_ під час створення кожного з них) а після отримання повідомлення WM_QUIT він видаляє з безлічі thread_ ідентифікатор потоку в якому виконувався. Завдяки цьому можна запустити кілька циклів обробки повідомлень в різних потоках і один обробник зможе отримувати повідомлення з будь-якого. Ну і останнє, метод shutdown_service, повинен привести до завершення всіх циклів обробки повідомлень, тому він просто посилає всім їм повідомлення WM_QUIT. Ну і саме останнє що нам потрібно зробити - реалізувати io_object. Це дуже просто зробити.
template < typename Service>
Class basic_messageloop: Public Boost:: asio:: basic_io_object <Service> {
public :
Explicit basic_messageloop (boost:: asio:: io_service & io_service): boost:: asio:: basic_io_object <Service> (io_service) {}
Void Loop () {
this -> service.loop ();}
template < Class Handler >
Void set_handler ( DWORD ID, Handler Handler) {
this -> service.set_handler ( this -> implementation, id, handler);}
template < Class Handler >
Void set_handler ( LPCTSTR Name, Handler Handler) {
DWORD msg_code =:: RegisterWindowMessage (name);
this -> service.set_handler ( this -> implementation, msg_code, handler);}};
typedef basic_messageloop <basic_messageloop_service> messageloop;
Єдина вимога - клас повинен бути спадкоємцем boost:: asio:: basic_io_object, Де Service - наш сервіс обробників повідомлень. У цьому класі метод set_handler просто викликає метод set_handler свого сервісу і передає в нього свою реалізацію (implementation) та додаткові параметри. Загалом цей клас - просто інтерфейс, він не повинен мати стан (хоча його батько має), а просто перенаправляти всі виклики своєму сервісу. Юзати це можна так:
boost:: asio:: io_service io; asio_aux:: messageloop message_loop1 (io); asio_aux:: messageloop message_loop2 (io); message_loop1.set_handler ( WM_CLOSE , & message_handler); message_loop1.set_handler ( L " MyMessage " , & message_handler); message_loop2. set_handler ( WM_CLOSE , & message_handler); message_loop2.set_handler ( L " MyAnotherMessage " , & message_handler); boost:: thread thr (boost:: bind (& boost:: asio:: io_service:: run, & io)); message_loop1.loop () ;
Я вже досить давно використовую бібліотеку Boost.Asio у своєму проекті, в основному для організації асинхронного введення виведення і не раз ловив себе на думці, що хотів би використовувати її не тільки для цього. Насправді це не складно виправити, asio це не стільки мережева бібліотека, скільки дуже зручний фреймверк для вашого коду =) Отже, трохи теорії. Бібліотека asio дозволяє ефективно відокремити реалізацію від інтерфейсу використовуючи принцип dependency injection, для цього існують три сутності: io_object - інтерфейс надається користувачеві; service (не плутати з io_service-ом) - об'єкт реалізує взаємодію з ОС, наприклад з системою введення-виведення; implementation - зберігає стан об'єкта (наприклад хендл сокета). Все разом це виглядає так: io_object - клас успадкований від boost:: asio:: basic_io_object, Об'єкти цього класу створюються користувачем для виконання будь-яких операцій, приклади: boost:: asio:: deadline_timer; boost:: asio:: ip:: tcp:: socket; boost:: asio:: windows:: random_access_handle. Конструктор класу basic_io_object приймає покажчик на io_service. Кожному типу io_object-а відповідає свій service. Сервіс, це об'єкт безпосередньо виконує операції, програма користувача повинна взаємодіяти з сервісом не безпосередньо а через io_object відповідного типу, який виконує роль інтерфейсу. Сервіс автоматично створюється до створення першого примірника відповідного io_object-a, наприклад, для всіх сокетів створюється один сервіс (під windows це буде win_iocp_socket_service за назвою зрозуміло, що він використовує порти завершення:), для таймерів - інший (deadline_timer_service, котрий під windows так- ж використовує IOCP). io_service містить список сервісів для всіх об'єктів (io_object) які були на ньому створені (при створенні отримали посилання на цей io_service як параметр конструктора). basic_io_objectмістить два члени класу, з допомогою яких програміст може отримати доступ до сервісу та реалізації самого об'єкта - service і implementation. service - посилання на сервіс для даного об'єкта, який має тип basic_io_object:: Service_type, implementation - об'єкт типу basic_io_object:: Implementation_type - який є членом класу basic_io_object і представляє з себе реалізацію io_object-a, часом життя цього об'єкта управляє сервіс (волаючи методи construct і destroy). Допустимо у нас є такий код: boost:: asio:: io_service io; boost:: asio:: deadline_timer timer (io); У першому рядку буде створено io_service, він не міститиме жодного сервісу і якщо ми викличемо метод io.run (), він поверне управління відразу, тому що роботи в нього немає. У другому рядку все трохи цікавіше, спочатку буде створений io_object, в даному випадку це deadline_timer, в конструкторі basic_io_object-a (базового класу) буде викликана ф-я use_service яка спробує знайти відповідний сервіс (для таймерів він буде мати тип deadline_timer_service <... >) якщо для даного io_service-a такий сервіс ще не створено, то він бует створений, далі буде викликаний метод construct тільки-що створеного сервісу, завдання якого - ініціалізація implementation-a (що має тип deadline_timer_service:: implementation_type), який є членом класу basic_io_object і містить деталі реалізації таймера на даній платформі. Але це ще не все:), після виходу з scope-a буде викликаний деструктор таймера, який викличе метод destroy сервісу, як параметр в нього буде переданий timer.implementation. Далі в деструкції io_service-a буде викликаний метод shutdown_service сервісу дедлайн таймерів, а потім видалений екземпляр класу deadline_timer_service. Як я вже згадував, все це заради того, що-б розділити реалізацію та інтерфейс. Як приклад я реалізую клас (хоча насправді їх буде три:), для роботи з повідомленнями windows, який можна буде використовувати разом в бібліотекою boost:: asio. Для початку нам потрібно клас реалізує обробку повідомлень - implementation. Він повинен дозволяти встановлювати обробники повідомлень, видаляти їх, а так-же містити метод для обробки конкретного повідомлення.
Using Namespace Boost:: tuples;
Using Boost:: function;
Class messageloop_impl {
typedef Boost:: function < void ( Const MSG &, boost:: system:: error_code)> handler_type;
typedef Boost:: unordered_map < DWORD , handler_type> table_type ; table_type table_; boost:: asio:: io_service & io_; boost:: asio:: io_service:: work work_; boost:: mutex mutex_;
public : messageloop_impl (boost:: asio:: io_service & io_service): io_ (io_service), work_ (io_service) {}
Void set_handler ( DWORD ID, handler_type handler) {boost:: mutex:: scoped_lock lock (mutex_); table_type:: iterator I;
BOOL contain_handler; tie (i, contain_handler) = find_ (id);
if (contain_handler) runhandler_ (i, MSG (), boost:: asio:: error:: eof); table_.insert (std:: make_pair (id, handler));}
Void process_message ( Const MSG & msg, int & / * out * / processed) {boost:: mutex:: scoped_lock lock (mutex_); table_type:: iterator I;
BOOL contain_handler; tie (i, contain_handler) = find_ (msg.message);
if (contain_handler) {+ + processed ; runhandler_ (i, msg, boost:: system:: error_code ());}}
Void remove_handler ( DWORD ID) {boost:: mutex:: scoped_lock lock (mutex_); table_type:: iterator I;
BOOL contain_handler; tie ( i, contain_handler) = find_ (id);
if (contain_handler) {runhandler_ (i, MSG (), boost:: asio:: error:: eof); table_.erase (i);}}
Void Clear () {boost :: mutex:: scoped_lock lock (mutex_);
for (table_type:: iterator I = table_.begin (); i! = table_.end (); + + i) runhandler_ (i, MSG (), boost:: asio :: error:: eof); table_.clear ();}
private : tuple <table_type:: iterator , bool > find_ ( DWORD ID) {table_type:: iterator I = table_. find (id);
Return tuple <table_type: : iterator , bool > (i, i! = table_.end ());}
Void runhandler_ (table_type:: iterator I, Const MSG & m, boost:: system:: error_code e) {handler_type h = i-> second ; io_.post (boost:: bind (h, m, e));}};
Конструктор messageloop_impl - приймає посилання на io_service і зберігає її всередині об'єкта класу. Обробники повідомлень реалізовані на основі boost:: function, мають сигнатуру void (const MSG &, boost:: system:: error_code). Для зберігання обробників повідомлень використовується хеш таблиця table_. Ф-я set_handler додає обробник в хеш таблицю, ф-я remove_handler відповідно видаляє, метод clear видаляє всі обробники. Метод process_message обробляє повідомлення, у випадку, якщо для даного повідомлення знайдений обробник, значення змінної processed збільшується на одиницю, а обробник передається в io_service за допомогою методу post, далі він викликається методом run, poll або run_one io_service-a. Це потрібно для того, що-б наш обробник повідомлень слідував правилу, згідно з яким усі обробники можуть викликатися тільки в тих потоках, в яких був викликаний метод run (poll або run_one) відповідного io_service-a, або під час дзвінка деструктора io_service-a. Так-же даний клас містить член work_ має тип boost:: asio:: io_service:: work. io_service містить лічильник, який инкрементируется кожен раз, коли починається будь-яка операція, і декрементируется щоразу після завершення чергової операції. Метод run io_service-a не завершується доти, поки цей лічильник не буде дорівнює нулю. Зробивши об'єкт класу io_service:: work членом класу messageloop_impl, ми гарантуємо, що цей лічильник не обнулиться до тих пір, поки хоч один примірник messageloop_impl існує. Перед видаленням обробника, він викликається за другим параметром рівним boost:: asio:: error:: eof. Тепер напишемо код нашого сервісу для обробки повідомлень.
Class basic_messageloop_service: Public Boost:: asio:: io_service:: service {
public :
Static Boost:: asio:: io_service:: id id;
typedef Boost:: shared_ptr <messageloop_impl> implementation_type;
private : boost:: mutex mutex_; std: : Set <implementation_type> processors_; std:: set < DWORD > threads_;
Static Void send_WM_QUIT_to ( DWORD Thread) {
BOOL Result =:: PostThreadMessage (thread, WM_QUIT , 0, 0);
if (! result) {boost:: system :: error_code e = boost:: system:: error_code (:: GetLastError (), boost:: system:: system_category);
Throw Boost:: system:: system_error (e);}}
public :
Explicit basic_messageloop_service (boost:: asio:: io_service & io_service): boost:: asio:: io_service:: service (io_service) {} ~ basic_messageloop_service () {}
Void CONSTRUCT (implementation_type & impl) {impl.reset ( New messageloop_impl (get_io_service ())); boost: : mutex:: scoped_lock lock (mutex_); processors_.insert (impl);}
Void Destroy (implementation_type & impl) {boost:: mutex:: scoped_lock lock (mutex_); processors_.erase (impl); impl.reset (); }
Void shutdown_service () {boost:: mutex:: scoped_lock lock (mutex_); std:: for_each (threads_.begin (), threads_.end (), & basic_messageloop_service:: send_WM_QUIT_to); threads_.clear ();}
template < Class Handler >
Void set_handler (implementation_type & impl, DWORD ID, Handler Handler) {impl-> set_handler (id, handler);}
Void remove_handler (implementation_type & impl, DWORD ID) {impl-> remove_handler (id);}
Void Clear (implementation_type & impl) {impl-> clear ();}
Void Loop () {
MSG MSG;
DWORD thread_id =:: GetCurrentThreadId (); {boost:: mutex:: scoped_lock lock (mutex_); threads_.insert (thread_id);}
while (:: GetMessage (& msg, NULL , 0, 0)) {boost:: mutex:: scoped_lock lock (mutex_);
INT proc_cnt = 0; std:: for_each (processors_.begin (), processors_.end (), boost :: bind (& messageloop_impl:: process_message, _1, msg, boost:: ref (proc_cnt)));
if (proc_cnt == 0):: DispatchMessage (& msg);} {boost:: mutex:: scoped_lock lock (mutex_) ; threads_.erase (thread_id);}}}; boost:: asio:: io_service:: id basic_messageloop_service:: id;
Отже, наш клас повинен бути спадкоємцем класу boost:: asio:: io_service:: service, а так-же мати статичний член id має тип boost:: asio:: io_service:: id, який є унікальним ідентифікатором сервісу. Так-же клас повинен визначати тип implementation_type, в даному випадку це boost:: shared_ptr, Від якого залежить тип змінної класу basic_io_object:: Implementation, тобто реалізації об'єкта обробника повідомлень. Сервіс повинен вміти ініціалізувати об'єкти мають тип implementation_type за допомогою методу construct і деініціалізіровать їх методом destroy. Так-же він містить ряд методів для управління екземплярами класу implementation_type а так-же метод loop, в якому реалізований цикл обробки повідомлень. Наприклад, метод set_handler, він приймає два параметри, перший (impl) має тип implementation_type, другий (id) - ід-р повідомлення і третій (handler) - обробник повідомлення, реалізований просто як виклик impl-> set_handler (id, handler), тоесть просто делегує виклик реалізації. Метод loop, реалізований таким чином, спочатку він запам'ятовує в безлічі threads_ ідентифікатор потоку в якому він викликаний, потім у циклі отримує повідомлення і передає їх обробникам (які сервіс запам'ятовує в безлічі processors_ під час створення кожного з них) а після отримання повідомлення WM_QUIT він видаляє з безлічі thread_ ідентифікатор потоку в якому виконувався. Завдяки цьому можна запустити кілька циклів обробки повідомлень в різних потоках і один обробник зможе отримувати повідомлення з будь-якого. Ну і останнє, метод shutdown_service, повинен привести до завершення всіх циклів обробки повідомлень, тому він просто посилає всім їм повідомлення WM_QUIT. Ну і саме останнє що нам потрібно зробити - реалізувати io_object. Це дуже просто зробити.
template < typename Service>
Class basic_messageloop: Public Boost:: asio:: basic_io_object <Service> {
public :
Explicit basic_messageloop (boost:: asio:: io_service & io_service): boost:: asio:: basic_io_object <Service> (io_service) {}
Void Loop () {
this -> service.loop ();}
template < Class Handler >
Void set_handler ( DWORD ID, Handler Handler) {
this -> service.set_handler ( this -> implementation, id, handler);}
template < Class Handler >
Void set_handler ( LPCTSTR Name, Handler Handler) {
DWORD msg_code =:: RegisterWindowMessage (name);
this -> service.set_handler ( this -> implementation, msg_code, handler);}};
typedef basic_messageloop <basic_messageloop_service> messageloop;
Єдина вимога - клас повинен бути спадкоємцем boost:: asio:: basic_io_object, Де Service - наш сервіс обробників повідомлень. У цьому класі метод set_handler просто викликає метод set_handler свого сервісу і передає в нього свою реалізацію (implementation) та додаткові параметри. Загалом цей клас - просто інтерфейс, він не повинен мати стан (хоча його батько має), а просто перенаправляти всі виклики своєму сервісу. Юзати це можна так:
boost:: asio:: io_service io; asio_aux:: messageloop message_loop1 (io); asio_aux:: messageloop message_loop2 (io); message_loop1.set_handler ( WM_CLOSE , & message_handler); message_loop1.set_handler ( L " MyMessage " , & message_handler); message_loop2. set_handler ( WM_CLOSE , & message_handler); message_loop2.set_handler ( L " MyAnotherMessage " , & message_handler); boost:: thread thr (boost:: bind (& boost:: asio:: io_service:: run, & io)); message_loop1.loop () ;
Багатопоточність: boost.threads
Минулого століття програми, написані з використанням декількох потоків виконання, зустрічалися досить рідко. Сьогодні ж Інтернет-сервери запускають безліч потоків виконання, щоб ефективно обслужити безліч з'єднань з клієнтами. З метою максимізації пропускної здатності сервери транзакцій виконують свої сервіси в окремих потоках. Графічні програми роблять тривалі операції в окремому потоці, продовжуючи чітко реагувати на дії користувача. Цей список можна продовжувати і далі. Деякі програмісти, пишуть програми на C++, використовуючи бібліотеки, надані їх операційними системами і забезпечують системну підтримку потоків. Стандарт POSIX визначає підтримку многопоточ ності в бібліотеці, відомої як PTHREADS. Вона підтримує багатопоточність для широкого діапазону зона операційних систем, включаючи Win32 посредст вом порту pthreads-win32. Однак, це C-бібліотека, вона не реалізує деякі концепції C + +, вона не доступна на деяких платформах. Boost.Threads - бібліотека, призначається для вирішення проблем нестандартності та непереносимості. Boost - це організація, заснована членами робочої групи комітету з стандартизації C++ з ціллю розробки нових бібліотек для C++. На даний час в організацію входять приблизно 2000 членів. Безліч бібліотек можна знайти в дистрибютиві Boost (www.boost.org). Щоб зробити ці бібліотеки безпечними щодо потоків і була створена бібліотеку ка Boost.Threads. Багато експертів C++ внесли свій внесок в дизайн Boost.Threads. Інтерфейс бібліотеки був спроекті ваний з чистого аркуша і не є простою обгорткою якої-небудь багатопотокового API для C. Багато можливостей C++ (такі як існування конструкторів і деструкторів, функціональних об'єктів і шаблонов) були повною мірою використані з метою створення більш гнучкого інтерфейсу. Поточна реалізація працює на платформах POSIX, Win32 і Mac OS X.
Створення потоку
Класс boost::thread представляє потік виконання в тому сенсі, в якому std::fstream представляе файл. Конструктор за замовчуванням створює екземпляр, представляючий поточний потік виконання. Пере-завантажений конструктор приймає функціональнийоб'єкт, що викликається без аргументів і нічого не повертає. Цей конструктор запускає новий потік виконання, який в свою чергу викликає функціональний об'єкт. На перший погляд може здатися, що такий дизайн менш корисний, ніж типовий підхід мови C створення потоку, коли функції, що викликається новимпотоком, передається покажчик на void, за допомогою якого передаються дані. Однак, оскільки бібліотека Boost.Threads замість покажчика на функцію використовує функціональний об'єкт, саме він і може містити дані, необхідні потоку. Такий підхід, насправді, більш гнучкий і типобеспечний. Комбінуючи з функціональними бібліотеціками, такими як Boost.Bind, цей дизайн дозволяєлегко передавати новоствореному потоку будь які обсяги даних. В даний час з об'єктом потоку можна зробити не надто багато. Фактично, можна виконати тільки дві операції. Об'єкти потоків можналегко порівняти на рівність і нерівність, використовуючи операції == і ! =, щоб визначити, чи належать вони до одного і того ж потоку виконання, і можна дочекатися завершення потоку, викликаючи boost::thread::join. Інші потокові бібліотеки дозволяють виконувати над потоками та інші операції (наприклад, встановлювати його пріоритет або навіть зупиняти). Однак так як ці бібліотеки не дуже легко відобразити непереносимі інтерфейси, все ще проводяться дослідження, метою яких є додавання відповідних можливостей і в Boost.Threads. Наступний код демонструє дуже простий приклад використування класу boost::thread. Створюється новий потік, який просто виводить "Hello World" на std::cout, а основний потік чекає його завершення.#include <boost/thread/thread.hpp> #include <iostream> void hello() { std::cout<<"Hello, World\n"; } int main() { boost::thread thrd(&hello); thrd.join(); }
Мютекси
Кожен, хто писав багатопотокову програму, розуміє, наскільки важливо безлічі потоків не звертатися до одного й того ж ресурсу, в один і той же час. Якщо один потік намагається змінити значення розділяних даних в той час, коли інший потік намагається це значення прочитати, результатом буде невизначена поведінка. Щоб цьому запобігти, використовують деякі спеціальні примітивні типи і операції. Самий фундаментальний з цих типів відомий як мютекс (Скорочення від «mutual exclusion» - взаємне виключення). Мютекс дозволяє отримати доступ до роздільного ресурсу тільки одному потоку в один момент часу, цей потік повинен «заблокувати» мютекс. Якщо інший потік вже заблокував мютекс, то ця операція чекає, поки мютекс не буде звільнений («Розблоковано»), таким чином гарантується, що тільки один потік має доступ до роздільного ресурсу в один момент часу. Мютекс може мати кілька варіацій. Дві великі категорії підтримуваних бібліотекою Boost.Threads мютекс включають прості і рекурсивні мютекси. Простий мютекс може бути заблокований лише один раз. Спроба повторного захоплення мютекса призводить до глухого кута, що виражається в нескінченному очікуванні. При використанні рекурсивного мютекса один потік може заблокувати мьютекс кілька разів і має рівно стільки ж разів його розблокувати, перш ніж інший потік отримає можливість його заблокувати. У рамках цих двох великих категорій мютекс виділяють і інші способи блокування мютекса. Потік може спробувати заблокувати мютекс трьома способами:- Спробувати заблокувати мютекс, очікуючи поки жоден інший потік не буде ним володіти.
- Спробувати заблокувати мютекс, негайно але повернувшись, якщо цей мьютекс вже кимось заблокований.
- Спробувати заблокувати мютекс, очікуючи або його звільнення іншим потоком, або закінчення зазначеного періоду часу.
- boost::mutex
- boost::try_mutex
- boost::timed_mutex
- boost::recursive_mutex
- boost::recursive_try_mutex
- boost::recursive_timed_mutex
#include <boost/thread/thread.hpp> #include <boost/thread/mutex.hpp> #include <iostream> boost::mutex io_mutex; struct count { count(int id) : id(id) {} void operator()() { for (int i = 0; i < 10; ++i) { boost::mutex::scoped_lock lock(io_mutex); std::cout<<id<< ": "<<i<<std::endl; } } int id; }; int main(int argc, char* argv[]) { boost::thread thrd1(count(1)); boost::thread thrd2(count(2)); thrd1.join(); thrd2.join(); return 0; }Багато користувачів помітять, що передача даних потоку вимагає ручного кодування функціонального об'єкта. Хоча цей код і тривіальний, писати його кожного разу досить нудно. Є й більш просте рішення. Функціональні бібліотеки дозволяють створити нові функціональні об'єкти, пов'язуючи (bind) інші функціональні об'єкти з даними, які при виклику будуть їм передані. У наступному прикладі показано, як при використанні бібліотеки Boost.Bind можна спростити код попереднього прикладу прикладу, відмовившись від ручного кодування функціонального об'єкта.
/** *Ця програма ідентична програмі *з попереднього прикладу, крім того, що *використовує Boost.Bind *при створені потоку, *що приймає параметры. */ #include <boost/thread/thread.hpp> #include <boost/thread/mutex.hpp> #include <boost/bind.hpp> #include <iostream> boost::mutex io_mutex; void count(int id) { for (int i = 0; i < 10; ++i) { boost::mutex::scoped_lock lock(io_mutex); std::cout<<id<<": "<<i<<std::endl; } } int main(int argc, char* argv[]) { boost::thread thrd1( boost::bind(&count, 1)); boost::thread thrd2( boost::bind(&count, 2)); thrd1.join(); thrd2.join(); return 0; }
Умовні змінні
Іноді недостатньо просто встановити блок і скористатися роздільним ресурсом. Іноді необхІдно, щоб цей ресурс перед використанням знаходився в деякому спеціальному стані. Наприклад, потік може намагатися витягти дані з стека, чекаючи, коли в ньому вони з'являться, якщо в цей момент стек порожній. Для реалізації такого виду синхронізації, мьютекса недостатньо. У такому разі можна використовувати інший механізм синхронізації, відомий як умовна змінна. Умовна змінна завжди використовується в зв'язці з мьютексом та ресурсом (або ресурсами). Потік насамперед блокує мьютекс, а потім перевіряє, чи знаходиться ресурс в стані, що дозволяє його безпечне використання необхідним чином. Якщо він не в потрібному стані, потік викликає для умовної змінної операцію очікування. Ця операція призводить до розблокування мьютекса під час очікування, так що інший потік отримує можливість змінити стан роздільного ресурсу. Вона також гарантує, що при поверненні потоку після очікування мьютекс виявиться заблокованим. Коли інший потік змінює стан поділюваного ресурсу, він має повідомити потоки, які можуть чекати умовну змінну, дозволити їм завершити очікування. Наступний приклад ілюструє застосування класу boost::condition. Визначено клас, який реалізує обмежений буфер - контейнер фіксованого розміру з підтримкою введення/виводу в порядку черги (FIFO). Цей буфер зроблений потокобезпечним завдяки boost::mutex. Операції put і get використовують умовну змінну, щоб переконатися в тому, що потік буде чекати, поки буфер не опиниться в стані, необхідному для завершення операції. Створюються два потоки, один поміщає в цей буфер сто цілих, а інший їх же витягує. Обмежений буфер в кожен момент часу може містити тільки 10 цілих, тому кожному з потоків доводиться періодично чекати інший потік. Для перевірки того, що це дійсно відбувається, операції put і get виводять в std::cout діагностичні повідомлення. Нарешті, основний потік чекає завершення обох потоків.#include <boost/thread/thread.hpp> #include <boost/thread/mutex.hpp> #include <boost/thread/condition.hpp> #include <iostream> const int BUF_SIZE = 10; const int ITERS = 100; boost::mutex io_mutex; class buffer { public: typedef boost::mutex::scoped_lock scoped_lock; buffer() : p(0), c(0), full(0) {} void put(int m) { scoped_lock lock(mutex); if (full == BUF_SIZE) { { boost::mutex::scoped_lock lock(io_mutex); std::cout << "Buffer is full. Waiting..." << std::endl; } while (full == BUF_SIZE) cond.wait(lock); } buf[p] = m; p = (p+1) % BUF_SIZE; ++full; cond.notify_one(); } int get() { scoped_lock lk(mutex); if (full == 0) { { boost::mutex::scoped_lock lock(io_mutex); std::cout << "Buffer is empty. Waiting..." << std::endl; } while (full == 0) cond.wait(lk); } int i = buf[c]; c = (c+1) % BUF_SIZE; --full; cond.notify_one(); return i; } private: boost::mutex mutex; boost::condition cond; unsigned int p, c, full; int buf[BUF_SIZE]; }; buffer buf; void writer() { for (int n = 0; n < ITERS; ++n) { { boost::mutex::scoped_lock lock(io_mutex); std::cout << "sending: " << n << std::endl; } buf.put(n); } } void reader() { for (int x = 0; x < ITERS; ++x) { int n = buf.get(); { boost::mutex::scoped_lock lock(io_mutex); std::cout << "received: " << n << std::endl; } } } int main(int argc, char* argv[]) { boost::thread thrd1(&reader); boost::thread thrd2(&writer); thrd1.join(); thrd2.join(); return 0; }
Локальна пам'ять потоку
Багато функцій не розраховані на рентабельність. Це означає, що небезпечно викликати таку функцію в момент, коли її викликає інший потік. Нереентерабельні функції зберігають статичні дані між успішними викликами або повертають покажчик на статичні дані. Наприклад, функція std::strtok нереентерабельна, оскільки використовує статичні дані для зберігання рядка, що розбивається на токени. Така функція може бути зроблена реєнтерабельним за допомогою двох підходів. Один підхід полягає в зміні її інтерфейсу, так що функція приймає покажчик або посилання на тип даних, який можна використовувати замість використовуваної перш статичної області даних. Наприклад, стандарт POSIX визначає strtok_r, реєнтерабельним версію std::strtok, яка приймає додатковий параметр типу char**, що використовується замість статичної пам'яті. Це просте рішення, до того ж воно забезпечує найбільшу можливу продуктивність, а проте, воно означає зміну відкритого інтерфейсу, що потенційно може призвести до зміни великого обсягу коду. Другий підхід залишає інтерфейс без зміни і замінює статичну пам'ять на локальну пам'ять потоку (thread local storage) (іноді її називають потік-специфічною пам'яттю thread-specific storage). Локальна пам'ять потоку - це дані, пов'язані з конкретним (поточним) потоком. Багатопотокові бібліотеки надають доступ до локальної пам'яті потоку за допомогою інтерфейсу, що дозволяє отримати доступ до екземпляра даних, що належать поточному потоку. Кожен потік отримує свій власний екземпляр цих даних, тобто тут не виникає проблема спільного доступу. Однак, доступ до локальної пам'яті потоку повільніший, ніж до локальних або статичних данних, так що це не завжди найкраще рішення. Тим не менш, це єдине можливе рішення в разі, коли зміна відкритого інтерфейсу неприйнятна. Бібліотека Boost.Threads забезпечує доступ до локальної пам'яті потоку за допомогою «розумного» покажчика (smart pointer) boost::thread_specific_ptr. При першому зверненні кожного потоку до цього «розумному» вказівником, він має нульове значення, так що код повинен перевірити цей факт і забезпечити ініціалізацію покажчика при першому використанні. Бібліотека Boost.Threads гарантує, що дані, які зберігаються в локальній пам'яті потоку, будуть «зачищені» при виході з потоку. Наступний приклад ілюструє дуже просте застосування класу boost::thread_specific_ptr. Створюються два потоки, в них ініціалізується локальна пам'ять потоку, а потім у циклі 10 разів значення цілого, розташованого за адресою «розумного» покажчика інкрементується, а результат виводиться в std::cout (який синхронізований за допомогою мьютекса, так як є роздільним ресурсом) . Основний потік чекає завершення цих двох потоків. Висновок в цьому прикладі ясно показано, що кожен потік оперує зі своїм екземпляром даних, хоча обидва вони використовують один і той же boost::thread_specific_ptr.#include <boost/thread/thread.hpp> #include <boost/thread/mutex.hpp> #include <boost/thread/tss.hpp> #include <iostream> boost::mutex io_mutex; boost::thread_specific_ptr<int> ptr; struct count { count(int id) : id(id) { } void operator()() { if (ptr.get() == 0) ptr.reset(new int(0)); for (int i = 0; i < 10; ++i) { (*ptr)++; boost::mutex::scoped_lock lock(io_mutex); std::cout << id << ": " << *ptr << std::endl; } } int id; }; int main(int argc, char* argv[]) { boost::thread thrd1(count(1)); boost::thread thrd2(count(2)); thrd1.join(); thrd2.join(); return 0; }
Одноразові функції
Залишається розібратися з одним питанням: як зробити функції ініціалізації (такі як конструктори) потокобезпечна. Наприклад, коли «глобальний» екземпляр об'єкта створюється як сінглетон рівня програми (єдиний існуючий з програми такого типу), існує проблема порядку інстанціюванія, тому використовується функція, яка повертає статичний екземпляр, яка гарантує, що при першому зверненні до неї цей екземпляр буде створений. Проблема в тому, що якщо кілька потоків одночасно викличуть цю функцію, конструктор для статичного об'єкта також може бути викликаний кілька разів, і результати можуть виявитися поганими. Рішення проблеми в так званих «одноразові функції» (once routine). Така функція викликається в програмі тільки один раз. Якщо кілька потоків спробують її викликати одночасно, тільки один з них отримає таку можливість, а в цей час всі інші потоки будуть чекати, поки виконання функції не завершиться. Щоб гарантувати однократне виконання, така функція викликається побічно через іншу функцію, якій передається покажчик на вихідну і посилання на спеціальний прапор (flag), що сигналізує про факт виклику функції. Цей прапор ініціалізується статично, що гарантує ініціалізацію в період компіляції, а не в період виконання. І таким чином не представляє проблеми для багатопотокової ініціалізації. Бібліотека Boost.Threads надає можливість одноразового виклику функції за допомогою boost::call_once, а також визначає тип для прапора boost::once_flag і спеціальну макропідстановку, використовувану для статичної ініціалізації прапора, BOOST_ONCE_INIT. Наступний приклад показує дуже простий приклад використання boost::call_once. Глобальне ціле статично ініціалізується нулем, а примірник boost::once_flag статично ініціалізується за допомогою BOOST_ONCE_INIT. Основний потік запускає два потоки, кожен з яких намагається «ініціалізувати» глобальне ціле, викликаючи boost::call_once з покажчиком на функцію, інкрементуючу ціле. Потім основний потік чекає завершення обох потоків і виводить кінцеве значення цілого в std::cout. Висновок демонструє, що функція дійсно була викликана тільки одного разу, так як значення цілого - одиниця.#include <boost/thread/thread.hpp> #include <boost/thread/once.hpp> #include <iostream> int i = 0; boost::once_flag flag = BOOST_ONCE_INIT; void init() { ++i; } void thread() { boost::call_once(&init, flag); } int main(int argc, char* argv[]) { boost::thread thrd1(&thread); boost::thread thrd2(&thread); thrd1.join(); thrd2.join(); std::cout << i << std::endl; return 0; }
Підписатися на:
Дописи (Atom)