У мережі гуляє багато рішень для емуляції багатопоточності в php. Найчастіше вони ґрунтуються на форках, але є й варіації на тему із використанням curl, proc_open тощо.
Усі зустріті варіанти з тих чи інших причин мене не влаштували і довелося написати своє рішення.
Набір вимог у мене був наступний:
- використання форків;
- Синхронний режим із збереженням інтерфейсу за відсутності необхідних розширень;
- багаторазове використання дочірніх процесів;
- Повноцінний обмін даними між процесами. Тобто. запуск з аргументами та отримання результату після завершення;
- Можливість обміну подіями між дочірнім процесом-«потоком» та основним процесом під час роботи;
- Робота з пулом потоків із збереженням багаторазового використання, передачі аргументів та отримання результатів;
- Обробка помилок виконання;
- Таймаути виконання роботи, очікування роботи потоком, ініціалізацію;
- Максимум продуктивності;
Для нетерплячих одразу посилання на вихідники:
github.com/Anizoptera/AzaThread
Опис
AzaThread надає простий інтерфейс створення класів-потоків. Які насправді використовують окремі процеси для асинхронної роботи, але це вас не повинно хвилювати. Ви можете посилати події з потоку, повертати результати, використовувати один потік безліч разів передаючи йому аргументи запуску або створити пул з 16 потоків, що розгрібають ваші завдання як гарячі пиріжки, не звертаючи уваги на те, що робота відбувається в різних процесах.Крім цього, ви можете легко протестувати продуктивність бібліотеки в різних режимах, вибравши оптимальну кількість потоків і варіант передачі даних між процесами спеціально для вашої конфігурації.
Для повноцінної роботи потрібні такі розширення: libevent, posixі pcntl.
Бібліотека використовує LibEvent та парні сокети для спілкування між процесами. Підтримує 5 варіантів передачі даних (аргументів, результатів та даних подій)!
Варіанти наводжу одночасно з даними продуктивності. Тестувалося з пулом із восьми потоків на Intel Core i7 2600K 3.40 Ghz (Ubuntu 11.04 на віртуалці VMware). Наведено середні результати за 10 повторів тесту в jps (jobs per second - кількість завдань, що просто отримують аргументи і віддають дані в секунду).
Автоматично вибирається розширення для роботи із сокетами. Якщо доступно, то використовується розширення socketsщо дає поліпшення продуктивності. В іншому випадку задіюється stream.
У дочірньому процесі слухаються всі доступні сигнали. За замовчуванням на всі (крім SIGWINCH і SIGINFO) слід завершити роботу. Але це легко можна перевизначити, створивши в класі потоку метод з ім'ям сигналу. Наприклад sigWinch.
У батьківському процесі за умовчанням теж перехоплюються всі сигнали. Це можна змінити виставивши у класу параметр listenMasterSignalsу false. У цьому випадку оброблятиметься лише SIGCHLD. Свої власні обробники можна легко додати, створивши статичний метод з назвою m< имя сигнала > . Наприклад mSigTerm.
Якщо дочірній процес помре з якихось причин, клас автоматично форкнеться під час запуску нового завдання. Це відбувається непомітно і можна взагалі про це не замислюватись. Просто інстанс не потрібно перетворювати у разі будь-якої помилки.
Дочірній процес іноді перевіряє існування батьківського. Якщо він раптом помре, то автоматичний дочірній завершиться.
Всі ресурси, що використовуються потоком або пулом потоків, автоматично очищаються при виклику деструктора. Але їх можна очистити примусово, якщо викликати метод cleanup. У цьому випадку потік/пул не можна використовувати.
При стандартних налаштуваннях потік ініціалізується заздалегідь, відразу під час створення класу. Якщо встановити параметр preforkу false, то форк відбуватиметься лише у момент запуску завдання.
Взагалі настроюваних параметрів досить багато. Зміна імені дочірнього процесу після форки (параметр pNameконструктора), таймують на час виконання завдання ( timeoutWork), таймують на максимальний час очікування завдань дочірнім процесом ( timeoutMaxWait), таймують на час пре-ініціалізації ( timeoutInit), розміри буферів для читання сокетів ( pipeReadSize, pipeMasterReadSize).
Можна вимкнути режим мультизадачності потоків ( multitask). У цьому випадку щоразу після завершення завдання дочірній процес помиратиме і форкатиметься заново для наступного запуску. Це помітно зменшить продуктивність.
Код покритий тестами та детально документований, приклади використання можна подивитися та запустити у файлі example.php.
Складніші приклади з обробкою помилок можна побачити в коді юніт тесту.
Є режим налагодження, в якому виводиться дуже докладна інформація про те, що саме і де відбувається.
Приклади використання
Основна фіча – максимальна простота. Якщо ви хочете просто запустити що-небудь в окремому потоку достатньо наступного коду:class ExampleThread extends Thread (protected function process() ( // Some work here)) $thread = new ExampleThread(); $thread->wait()->run();
Якщо є все необхідне для повноцінної роботи, завдання буде виконано асинхронно. Якщо ні, то все буде працювати, але в синхронному режимі.
З передачею параметра та отриманням результату код виглядатиме лише трохи складніше:
class ExampleThread extends Thread ( protected function process() ( return $this->getParam(0); ) ) $thread = new ExampleThread(); $thread->wait()->run(123); $result = $thread->wait()->getResult();
Аналогічно легким помахом руки додаємо обробку подій із потоку:
class ExampleThread extends Thread ( const EV_PROCESS = "process"; protected function process() ( $events = $this->getParam(0); for ($i = 0; $i< $events; $i++) {
$event_data = $i;
$this->trigger(self::EV_PROCESS, $event_data); ) ) ) // Додатковий аргумент. $additionalArgument = 123; $thread->bind(ExampleThread::EV_PROCESS, function($event_name, $event_data, $additional_arg) ( // обробка події ), $additionalArgument); $events = 10; // кількість подій, що згенерує потік // Щоб не викликати вручну очікування потоку перед першим викликом, // можна перевизначити властивість preforkWait у TRUE у класі-спадкоємці $thread->wait(); $thread = New ExampleThread(); $thread->run($events)->wait();
І нарешті використання пулу з восьми потоків з обробкою помилок виконання:
$ Threads = 8 // Число потоків $ Pool = New ThreadPool ("ExampleThread", $ Threads); $num = 25; //Кількість завдань $left = $num; // Кількість завдань, що залишилися do ( // Якщо в кулі є вільні потоки // І у нас залишилися завдання для виконання while ($pool->hasWaiting() && $left > 0) ( // При запуску отримуємо id потоку $threadId = $pool->run(); $left-; Результат можна ідентифікувати // по id потоку ($threadId) $num--; таймують виконання завдання foreach ($failed as $threadId) ( $left++; ) ) ) while ($num > 0); // Завершуємо всі дочірні процеси. Очищаємо ресурси, що використовуються пулом. $pool->cleanup();
Результати тестування продуктивності
Тести запускав на двох машинах із Ubuntu 11.04.
Перша - Intel Core i3 540 3.07 Ghz
Друга – Intel Core i7 2600K 3.40 Ghz (убунту стоїть на VMware віртуалці)
Результати наводжу просто, щоб можна було оцінити зростання продуктивності.
Знову ж таки це середні результати за серію з 10 повторів тесту в jps (jobs per second - кількість завдань за секунду).
Як завдання потоки виконують наступну фігню:
for ($i = 0; $i< 1000; $i++) {
$r = mt_rand(0, PHP_INT_MAX) * mt_rand(0, PHP_INT_MAX);
}
Перший результат вказаний для синхронного режиму (без форків).
18 і 20 потоків на першій конфігурації я не пробував, тому що вже для 12 почалося падіння продуктивності.
| Число потоків | Перша конфігурація | Друга |
|---|---|---|
| 0 | 553 | 763 |
| 1 | 330 | 669 |
| 2 | 580 | 1254 |
| 4 | 1015 | 2188 |
| 8 | 1040 | 2618 |
| 10 | 1027 | 2719 |
| 12 | 970 | 2739 |
| 16 | 958 | 2904 |
| 18 | - | 2830 |
| 20 | - | 2730 |
Тобто продуктивність піднімається в 2-4 і більше рази в залежності від процесора!
Код, який виконує серію тестів із потрібними параметрами, лежить у файлі examples/speed_test.php. Так що ви легко можете протестувати продуктивність і вибрати оптимальну кількість потоків у себе.
02.10.2011, 21:27
За те, що побила мужика і повісила його футболку на дерево.
Я у великій компанії пристала до дівчини і сказала що її хлопець на розпродажі купив а ще їла банани і шкірку всім в маршкутці на коліна розкладала а ще постійно в барах влаштовую лесбі шоу одного разу під спідниці до офіціанток у віп клубі заглядала і комента роздавала а взагалі я погана п'яна відразу забуваю що люблю людей приказка якщо прокинувся і тобі соромно але не пам'ятаєш за що точно про мене
Після останньої п'янки мені здитий:
1. репетувала з балкона мужику "відсмоктувати"
2. зжерла цигарку
3. кидалася яблуками
4. дзвонила хлопцю і розповідала, що я сексу йому не дам у грубій формі
5. топила бички в грейпфрутовому соку
6. дзвонила мамі і розповідала, що я твереза
7. ораза хлопцю подруги,щоб він приніс нам вина
8. написала повз унітаз,оскільки їх було 2-обрала я не той,а потім впала
9. випала з ванної
Я з ножем за чоловіком по квартирі 1 раз бігала, хоч випила шампанського пару келихів. Вранці страшно стало - а раптом заїхала б ножем, я ж просто вирішила палякати. І не знаю, з чого на мене такий заскок знайшов!
Постійно одне й те саме, як нап'юся йду в парк дрочерів шукати, щоб поприколюватися над ними, або ці теми на форумі створюю... вранці думаю, чи не дурна?
До мене якось хлопець клеївся, а я така нажрана була, що почала йому популярно пояснювати, мовляв нічого не обламається, тому що я "кривава Мері" (кд були) і мені ця клікуха так в душу запала, що я мало не кожному зустрічному кричала: "я - кривава Мері" благо, що це було не в моєму рідному місті
-*Дзвонила колишньому, висловлювала про нього все, що думаю
*дзвонила коханому, говорила, як я його хочу
* з'їла шаурму з фольгою
*обматерила хлопця, з яким у клубі познайомилася, хоча він мені допомагав, водичку приносив
*танцювала стриптиз у клубі у конкурсі, здобула друге місце
*кидалася льодом у незнайомого хлопця
*друг мого чоловіка на жовтому хамері їздить. таких у Кореї я більше ні в кого не бачила. коли ми всі разом поїхали гасити. я висунулася з машини свого чоловіка, і, показуючи на попереду хамер, що кричав, кричала, що ця машина мого друга
*вчора я була в клубі. коли їхала вранці додому на таксі, сказала водієві: [Переклад з корейського. дослівний!] "Їжа, яку я з'їла, сказала мені щойно, що хоче вийти назовні!"
* Ще в мене друг монгол є. я до нього на спину залізла, змусила його возити мене на спині і на всю вулицю репетувала, що він мій монгольський кінь
а взагалі ще багато всього було, так і не згадаєш...
Хлопцю написала о 4 годині ранку “на добраніч, дорогий“
Напилися з подругами коньяку, всі роз'їхалися по будинках, а я дура викликала таксі (було 3 години ночі) і поїхала до колишнього хлопця з яким розлучилася 4 місяці тому (залишилися друзями). ломитися до нього в двері і кричати "Одружися на мені марат", він бідний охренел.Затягнув мене додому і під душ холодний, напоїв міцним чаєм. Через години дві я відійшла, а потім коли зрозуміла що зробила мені мені було не соромно. помирилися і досі разом. Після цього більше келиха шампанського не п'ю.
...
02.10.2011, 22:09
Вбита йшла з хлопцем через сад і тут мені здалося, що він мене образив тим, що йшов на кілька кроків попереду мене, а не поруч. Я образилася і залізла на дерево, а він і не помітив ... він півтори години бігав по саду і шукав мене ..... а я спокійнісінько вирубалася на дереві і проспала до ранку. ...
...
Зате вранці відразу не змогла згадати, як я опинилась на гілках. Сті-иыыыи-дно те, як було після
;D;D;D
Інтерпретація
02.10.2011, 22:42
мені соромно за деякі вчинки, здійснені у тверезому вигляді:) Але це ближче до філософії. У нетверезому ... дуже багато було весело, соромно ... ні, соррі)))))
Олена Lotus
02.10.2011, 23:17
Вчора після того як я зрозуміла що мій телефон (ТОВ!!!) тепер не мій я напилася. сильно. секрету і то я подумаю кому можна розповісти ... Ресторан "Паберті" був знову шокований.
02.10.2011, 23:31
;D;D;D
Я завжди стверджував, що тіткам пити не можна! ВСІМ!;)
..;D;D;D істину говориш
02.10.2011, 23:36
П'яна жінка – як китайський пуховик: м'яка та лізе.
Галигін
Олена Lotus
02.10.2011, 23:41
а чоловікам типу можна так? Пити можна всім. і тіткам і дядькам ... тільки треба акуратненько так ... сьогодні голова не болить ... душа болить ... напиться? прямо замкнене коло;)
03.10.2011, 00:15
Пийте все і все, поки молоді і є здоров'я:)
03.10.2011, 01:48
вибірково сперто з жіночого форуму:)
Ледве не написала в кошик для білизни, при мамі ..... поплутала з унітазом .... упс ....
Вбита йшла з хлопцем через сад і тут мені здалося, що він мене образив тим, що йшов на кілька кроків попереду мене, а не поруч. Я образилася і залізла на дерево, а він і не помітив ... він півтори години бігав по саду і шукав мене ..... а я спокійнісінько вирубалася на дереві і проспала до ранку. ...
...
Зате вранці відразу не змогла згадати, як я опинилась на гілках. Сті-иыыыи-дно те, як було після
На балконі стояли тази і горщики, де мама вирощувала лук. вирішила твою зелень обссати! ну і ріс він далі, потім у салат пішов
03.10.2011, 05:35
Напилися з подругами коньяку, всі роз'їхалися по будинках, а я дура викликала таксі (було 3 години ночі) і поїхала до колишнього хлопця з яким розлучилася 4 місяці тому (залишилися друзями). ломитися до нього в двері і кричати "Одружися на мені марат", він бідний охренел.Затягнув мене додому і під душ холодний, напоїв міцним чаєм. Через години дві я відійшла, а потім коли зрозуміла що зробила мені мені було не соромно. помирилися і досі разом. Після цього більше келиха шампанського не п'ю.
Ось цілком реальна історія кохання.
І про користь пияцтва теж.
(Пийте, дівчатка, і прийде вам щастя.)
Схоже, PHP розробники рідко використовують паралельність. Говорити про простоту синхронного коду не буду, однопоточне програмування, звичайно, простіше і зрозуміліше, але іноді невелике використання паралельності може принести відчутне підвищення продуктивності.
У цій статті ми поглянемо на те, як багатопоточність може бути досягнута в PHP за допомогою розширення pthreads. Для цього буде потрібна встановлена ZTS (Zend Thread Safety) версія PHP 7.x, разом із встановленим розширенням pthreads v3. (На момент написання статті, в PHP 7.1 користувачам потрібно буде встановити з гілки master в репозиторії pthreads - див.
Невелике уточнення: pthreads v2 призначений для PHP 5.x і більше не підтримується, pthreads v3 - для PHP 7.х і активно розвивається.
Після такого відступу, давайте одразу перейдемо до справи!
Обробка разових завдань
Іноді ви хочете обробляти разові завдання багатопотоковим способом (наприклад, виконання певної задачі, зав'язаної на введення-виведення). У таких випадках можна використовувати клас Thread , щоб створити новий потік і запустити обробку в окремому потоці.
Наприклад:
$task = new class extends Thread ( private $response; public function run() ( $content = file_get_contents("http://google.com"); preg_match("~
Тут метод run це наша обробка, яка буде виконуватися всередині нового потоку. При виклик Thread::start , породжується новий потік і викликається метод run . Потім ми приєднуємо породжений потік назад до основного потоку, викликавши Thread::join , який буде заблокований доти, поки породжений потік не завершить своє виконання. Це гарантує, що завдання завершить виконання, перш ніж спробуємо вивести результат (який зберігається в $task->response).
Можливо, не бажано забруднювати клас додатковою відповідальністю, що з логікою потоку (зокрема обов'язок визначення методу run). Ми можемо виділити такі класи, успадкувавши їх від класу Threaded. Тоді вони можуть бути запущені всередині іншого потоку:
Class Task extends Threaded ( public $response; public function someWork() ( $content = file_get_contents("http://google.com"); preg_match("~ (.+) ~", $content, $matches); $ this->response = $matches; ) ) $task = new Task; $thread = new class($task) extends Thread ( private $task; public function __construct(Threaded $task) ( $this->task = $task; ) public function run() ( $this->task->someWork( ); )); $thread->start() && $thread->join(); var_dump($task->response);
Будь-який клас, який має бути запущений в окремому потоці, повиненуспадковуватися від класу Threaded. Це тому що він надає необхідні можливості для обробки в різних потоках, а також неявну безпеку і корисні інтерфейси (такі, як синхронізація ресурсів).
Погляньмо на ієрархію класів, пропоновану розширенням pthreads:
Threaded (implements Traversable, Collectable) Thread Worker Volatile Pool
Ми вже розглянули і дізналися основи класів Thread і Threaded, тепер поглянемо на інші три (Worker, Volatile, і Pool).
Перевикористання потоків
Запуск нового потоку для кожного завдання, яке потрібно розпаралелити, є досить витратним. Це тому, що архітектура "нічого-загального" повинна бути реалізована в pthreads, щоб домогтися багатопоточності всередині PHP. Що означає, що весь контекст виконання поточного екземпляра інтерпретатора PHP (у тому числі й кожен клас, інтерфейс, трейт та функція) має бути скопійований для кожного створеного потоку. Оскільки це тягне помітний вплив на продуктивність, потік завжди повинен бути повторно використаний, коли це можливо. Потоки можуть бути перевикористані двома способами: за допомогою Worker-ів або за допомогою Pool-ів.
Клас Worker використовується для виконання ряду завдань одночасно всередині іншого потоку. Це робиться шляхом створення нового екземпляра Worker -а (який створює новий потік), а потім внесенням завдань у стек цього окремого потоку (за допомогою Worker::stack).
Ось невеликий приклад:
Class Task extends Threaded ( private $value; public function __construct(int $i) ( $this->value = $i; ) public function run() ( usleep(250000); echo "Task: ($this->value) \n"; ) ) $worker = new Worker(); $worker->start(); for ($i = 0; $i stack(new Task($i)); ) while ($worker->collect()); $worker->shutdown();
У наведеному прикладі в стек заноситься 15 завдань для нового об'єкта $worker через метод Worker::stack , а потім вони обробляються в порядку їх внесення. Метод Worker::collect, як показано вище, використовується для очищення завдань, як тільки вони закінчать виконання. З його допомогою всередині циклу while, ми блокуємо основний потік, доки не будуть завершені всі завдання зі стеку і поки вони не будуть очищені - до того як ми викличемо Worker::shutdown. Завершення worker -а достроково (тобто доки є ще завдання, які мають бути виконані) буде, як і раніше, блокувати основний потік доти, доки всі завдання не завершать своє виконання, просто завдання не будуть почищені збирачем сміття (що тягне за собою витоку пам'яті).
Клас Worker надає кілька інших методів, що стосуються його стеку завдань, включаючи Worker::unstack для видалення останньої внесеної задачі та Worker::getStacked для отримання кількості завдань у стеку виконання. Стек worker -а містить лише завдання, які мають бути виконані. Як тільки завдання зі стеку було виконано, воно видаляється та розміщується в окремому (внутрішньому) стеку для складання сміття (за допомогою методу Worker::collect).
Ще один спосіб перевикористовувати потік при виконанні багатьох завдань – це використання пулу потоків (через клас Pool). Пул потоків використовує групу Worker -ів, щоб дати можливість виконувати завдання одночасно, В якому фактор паралельності (кількість потоків пулу, з якими він працює) задається при створенні пулу.
Давайте адаптуємо наведений вище приклад для використання пулу worker -ів:
Class Task extends Threaded ( private $value; public function __construct(int $i) ( $this->value = $i; ) public function run() ( usleep(250000); echo "Task: ($this->value) \n"; ) ) $pool = new Pool(4); for ($i = 0; $i submit(new Task($i)); ) while ($pool->collect()); $pool->shutdown();
Є кілька помітних відмінностей під час використання пулу, на відміну воркера. По-перше, пул не вимагає запуску вручну, він починає виконання завдань, як тільки вони стають доступними. По-друге, ми відправляємозавдання в пул, а не вкладаємо їх у стек. Крім того, клас Pool не успадковується від Threaded і тому він не може бути переданий в інші потоки (на відміну від Worker).
Як хороша практика, для воркерів і пулів слід завжди підчищати їхні завдання, як тільки вони завершилися, а потім вручну завершувати їх самих. Потоки, створені за допомогою класу Thread, також повинні бути приєднані до потоку, що породжує.
pthreads та (не)змінність
Останній клас, якого ми торкнемося, - Volatile, - новий додаток до pthreads v3. Поняття незмінності стало важливою концепцією в pthreads, оскільки без неї продуктивність суттєво знижується. Тому за умовчанням, властивості Threaded-класів, які самі є Threaded-об'єктами, зараз є незмінними, і тому вони не можуть бути перезаписані після їхнього первісного присвоєння. Явна змінність для таких властивостей зараз поки що краща, і все ще може бути досягнута за допомогою нового класу Volatile.
Погляньмо на приклад, який продемонструє нові обмеження незмінності:
Class Task extends Threaded // a Threaded class ( public function __construct() ( $this->data = new Threaded(); // $this->data is not overwritable, since it is a Threaded property of a Threaded class ) ) $task = New class(new Task()) extends Thread ( // a Threaded class, since Thread extends Threaded public function __construct($tm) ( $this->threadedMember = $tm; var_dump($this->threadedMember-> data); // object(Threaded)#3 (0) () $this->threadedMember = new StdClass(); // invalid, since the property is a Threaded member of a Threaded class)
Threaded -властивості у класів Volatile, з іншого боку, змінюються:
Class Task extends Volatile ( public function __construct() ( $this->data = new Threaded(); $this->data = new StdClass(); // valid, since we are in a volatile class ) ) $task = new class(new Task()) extends Thread ( public function __construct($vm) ( $this->volatileMember = $vm; var_dump($this->volatileMember->data); // object(stdClass)#4 (0) () // still invalid, since Volatile extends Threaded, so the property is still a Threaded member of Threaded class $this->volatileMember = new StdClass() ) );
Ми бачимо, що клас Volatile перевизначає незмінність, нав'язану батьківським класом Threaded , щоб надати можливість змінювати Threaded -властивості (а також unset() -ить).
Є ще один предмет обговорення, щоб розкрити тему змінності та класу Volatile - масиви. У pthreads масиви автоматично наводяться до Volatile-об'єктів при присвоєнні властивості класу Threaded . Це тому, що просто небезпечно маніпулювати масивом з кількох контекстів PHP.
Давайте знову поглянемо на приклад, щоб краще розуміти деякі речі:
$array = ; $task = New class($array) extends Thread ( private $data; public function __construct(array $array) ( $this->data = $array; ) public function run() ( $this->data = 4; $ this->data = 5; print_r($this->data); $task->start() && $task->join(); /* Висновок: Volatile Object ( => 1 => 2 => 3 => 4 => 5) */
Ми бачимо, що Volatile -об'єкти можуть бути оброблені так, ніби вони були масивами, тому що вони підтримують операції з масивами, такі як (як показано вище) оператор підмножин (). Однак, класи Volatile не підтримують базові функції з масивами, такі як array_pop та array_shift. Натомість клас Threaded надає нам подібні операції як вбудовані методи.
Як демонстрація:
$data = new class extends Volatile ( public $a = 1; public $b = 2; public $c = 3; ); var_dump($data); var_dump($data->pop()); var_dump($data->shift()); var_dump($data); /* Висновок: object(class@anonymous)#1 (3) ( [ "a"]=> int(1) ["b"]=> int(2) ["c"]=> int(3) ) int(3) int(1) object(class@anonymous)#1 (1) ( [ "b"]=> int(2) ) */
Інші підтримувані операції включають Threaded::chunk і Threaded::merge .
Синхронізація
В останньому розділі цієї статті ми розглянемо синхронізацію в pthreads. Синхронізація – це метод, який дозволяє контролювати доступ до загальних ресурсів.
Для прикладу, давайте реалізуємо найпростіший лічильник:
$counter = new class extends Thread ( public $i = 0; public function run() ( for ($i = 0; $i i; ) ) ); $counter->start(); for ($i = 0; $i i; ) $counter->join(); var_dump($counter->i); // виведе число від 10 до 20
Без використання синхронізації висновок не детермінований. Декілька потоків пишуть в одну змінну без контрольованого доступу, що означає, що оновлення будуть втрачені.
Давайте виправимо це так, що ми отримаємо правильний висновок 20 шляхом додавання синхронізації:
$counter = new class extends Thread ( public $i = 0; public function run() ( $this->synchronized(function () ( for ($i = 0; $i i; ) )); ) ); $counter->start(); $counter->synchronized(function ($counter) ( for ($i = 0; $i i; ) ), $counter); $counter->join(); var_dump($counter->i); // int(20)
Синхронізовані блоки коду можуть також взаємодіяти один з одним, використовуючи методи Threaded::wait та Threaded::notify (або Threaded::notifyAll).
Ось почерговий інкремент у двох синхронізованих циклах while:
$counter = New class extends Thread ( public $cond = 1; public function run() ( $this->synchronized(function () ( for ($i = 0; $i notify()); if ($this->cond === 1) ( $this->cond = 2; $this->wait(); ) ))); $counter->start(); $counter->synchronized(function ($counter) ( if ($counter->cond !== 2)) ( $counter->wait(); // wait for the start to first ) for ($i = 10; $i notify(); if ($counter->cond === 2) ( $counter->cond = 1; $counter->wait(); ) ) , $counter); $counter->join(); /* Висновок: int(0) int(10) int(1) int(11) int(2) int(12) int(3) int(13) int(4) int(14) int(5) int( 15) int(6) int(16) int(7) int(17) int(19) int(19) */
Ви можете помітити додаткові умови, які розміщені навколо звернення до Threaded::wait . Ці умови мають вирішальне значення, оскільки вони дозволяють синхронізованому колбеку відновити роботу, коли він отримав повідомлення і зазначена умова true . Це важливо, тому що повідомлення можуть надходити з інших місць, окрім Threaded::notify . Таким чином, якщо виклики методу Threaded::wait не були укладені в умовах, ми будемо виконувати помилкові виклики пробудження, які призведуть до непередбачуваної поведінки коду
Висновок
Ми розглянули п'ять класів пакету pthreads (Threaded, Thread, Worker, Volatile і Pool), а також як кожен із класів використовується. А ще ми глянули на нову концепцію незмінності в pthreads, зробили короткий огляд підтримуваних можливостей синхронізації. З цими основами, ми можемо тепер приступити до розгляду застосування pthreads у випадках реального світу! Це і буде темою нашого наступного посту.
Якщо вам цікавий переклад наступного посту, дайте знати: коментуйте у соц. мережах, плюсуйте та ділитесь постом з колегами та друзями.
У мережі гуляє багато рішень для емуляції багатопоточності в php. Найчастіше вони ґрунтуються на форках, але є й варіації на тему із використанням curl, proc_openі т.п.
Усі зустрінуті варіанти з тих чи інших причин мене не влаштували, і довелося написати своє рішення. Набір вимог у мене був наступний:
- використання форків;
- Синхронний режим із збереженням інтерфейсу за відсутності необхідних розширень;
- багаторазове використання дочірніх процесів;
- Повноцінний обмін даними між процесами. Тобто. запуск з аргументами та отримання результату після завершення;
- Можливість обміну подіями між дочірнім процесом-«потоком» та основним процесом під час роботи;
- Робота з пулом потоків із збереженням багаторазового використання, передачі аргументів та отримання результатів;
- Обробка помилок виконання;
- Таймаути виконання роботи, очікування роботи потоком, ініціалізацію;
- Максимум продуктивності.
В результаті вийшла бібліотека AzaThread (стара назва - CThread).
Опис
AzaThread надає простий інтерфейс створення класів-потоків. Які насправді використовують окремі процеси для асинхронної роботи, але це вас не повинно хвилювати. Ви можете посилати події з потоку, повертати результати, використовувати один потік безліч разів, передаючи йому аргументи запуску, або створити пул з 16 потоків, що розгрібають ваші завдання як гарячі пиріжки, не звертаючи уваги на те, що робота відбувається в різних процесах.
Крім цього, ви можете легко протестувати продуктивність бібліотеки в різних режимах, вибравши оптимальну кількість потоків і варіант передачі даних між процесами спеціально для вашої конфігурації.
Для повноцінної роботи потрібні такі розширення: libevent, posixі pcntl.
Бібліотека використовує LibEvent та парні сокети для спілкування між процесами. Підтримує 5 варіантів передачі даних (аргументів, результатів та даних подій)!
Варіанти наводжу одночасно з даними продуктивності. Тестувалося з пулом із восьми потоків на Intel Core i7 2600K 3.40 Ghz (Ubuntu 11.04 на віртуалці VMware). Наведено середні результати за 10 повторів тесту в jps (jobs per second — кількість завдань, що просто отримують аргументи і віддають дані в секунду).
Автоматично вибирається розширення для роботи із сокетами. Якщо доступно, то використовується розширення socketsщо дає поліпшення продуктивності. В іншому випадку задіюється stream.
У дочірньому процесі слухаються всі доступні сигнали. За замовчуванням на всі (крім SIGWINCH і SIGINFO) слід завершити роботу. Але це легко можна перевизначити, створивши в класі потоку метод з ім'ям сигналу. Наприклад sigWinch.
У батьківському процесі за умовчанням теж перехоплюються всі сигнали. Це можна змінити, виставивши у класу параметр listenMasterSignals false . У цьому випадку оброблятиметься лише SIGCHLD. Свої власні обробники можна легко додати, створивши статичний метод під назвою m<имя сигнала>. Наприклад, mSigTerm.
Якщо дочірній процес помре з якихось причин, клас автоматично форкнеться під час запуску нового завдання. Це відбувається непомітно і можна взагалі про це не замислюватись. Просто інстанс не потрібно перетворювати у разі будь-якої помилки.
Дочірній процес іноді перевіряє існування батьківського. Якщо він раптом помре, то автоматичний дочірній завершиться.
Усі ресурси, що використовуються потоком або пулом потоків, автоматично очищаються під час виклику деструктора. Але їх можна очистити примусово, якщо викликати метод cleanup. У цьому випадку потік/пул не можна використовувати.
При стандартних налаштуваннях потік ініціалізується заздалегідь, відразу під час створення класу. Якщо встановити параметр prefork у false , то форк відбуватиметься лише у момент запуску завдання.
Взагалі настроюваних параметрів досить багато. Зміна імені дочірнього процесу після форка (параметр pName конструктора), таймують на час виконання завдання (timeoutWork), таймують на максимальний час очікування задач дочірнім процесом (timeoutMaxWait), таймують на час преініціалізації (timeoutInit), розміри буферів для читання сокетів (pipeRead , pipeMasterReadSize). Можна вимкнути режим мультизадачності потоків (multitask). У цьому випадку щоразу після завершення завдання дочірній процес помиратиме і форкатиметься заново для наступного запуску. Це помітно зменшить продуктивність.
Код покритий тестами та детально документований, приклади використання можна подивитися та запустити у файлі example.php . Складніші приклади з обробкою помилок можна побачити в коді юніт-тесту.
Є режим налагодження, в якому виводиться дуже докладна інформація про те, що саме і де відбувається.
Приклади використання
Основна фіча – максимальна простота. Якщо ви хочете просто запустити що-небудь в окремому потоку достатньо наступного коду:
Class ExampleThread extends Thread ( protected function process() ( // Some work here ) ) $thread = new ExampleThread(); $thread->wait()->run();
Якщо є все необхідне для повноцінної роботи, завдання буде виконано асинхронно. Якщо ні, то все буде працювати, але в синхронному режимі.
З передачею параметра та отриманням результату код виглядатиме лише трохи складніше:
Class ExampleThread extends Thread ( protected function process() ( return $this->getParam(0); ) ) $thread = new ExampleThread(); $thread->wait()->run(123); $result = $thread->wait()->getResult();
Аналогічно, легким помахом руки, додаємо обробку подій з потоку:
Class ExampleThread extends Thread ( const EV_PROCESS = "process"; protected function process() ( $events = $this->getParam(0); for ($i = 0; $i trigger(self::EV_PROCESS, $event_data)); ) ) ) // Додатковий аргумент. $additionalArgument = 123; $thread->bind(ExampleThread::EV_PROCESS, function($event_name, $event_data, $additional_arg) ( // обробка події ), $additionalArgument); $events = 10; // кількість подій, що згенерує потік // Щоб не викликати вручну очікування потоку перед першим викликом, // можна перевизначити властивість preforkWait у TRUE у класі-спадкоємці $thread->wait(); $thread = New ExampleThread(); $thread->run($events)->wait();
І, нарешті, використання пулу з восьми потоків з обробкою помилок виконання:
$ Threads = 8 // Число потоків $ Pool = New ThreadPool ("ExampleThread", $ Threads); $num = 25; //Кількість завдань $left = $num; // Кількість завдань, що залишилися do ( // Якщо в кулі є вільні потоки // І у нас залишилися завдання для виконання while ($pool->hasWaiting() && $left > 0) ( // При запуску отримуємо id потоку $threadId = $pool->run(); $left-; Результат можна ідентифікувати // по id потоку ($threadId) $num--; таймують виконання завдання foreach ($failed as $threadId) ( $left++; ) ) ) while ($num > 0); // Завершуємо всі дочірні процеси. Очищаємо ресурси, що використовуються пулом. $pool->cleanup();
Результати тестування продуктивності
Тести запускав на двох машинах із Ubuntu 11.04.
Перша - Intel Core i3 540 3.07 Ghz.
Друга - Intel Core i7 2600K 3.40 Ghz (убунту стоїть на VMware віртуалці).
Результати наводжу просто, щоб можна було оцінити зростання продуктивності. Знову ж таки, це середні результати за серію з 10 повторів тесту в jps (jobs per second — кількість завдань за секунду).
Як завдання потоки виконують наступну фігню:
For ($i = 0; $i
Перший результат вказаний для синхронного режиму (без форків). 18 і 20 потоків на першій конфігурації я не пробував, тому що вже для 12 почалося падіння продуктивності.
| Число потоків | Перша конфігурація | Друга |
|---|---|---|
| 0 | 553 | 763 |
| 1 | 330 | 669 |
| 2 | 580 | 1254 |
| 4 | 1015 | 2188 |
| 8 | 1040 | 2618 |
| 10 | 1027 | 2719 |
| 12 | 970 | 2739 |
| 16 | 958 | 2904 |
| 18 | - | 2830 |
| 20 | - | 2730 |
Тобто продуктивність піднімається у 2-4 рази і більше залежно від процесора!
Код, який виконує серію тестів із потрібними параметрами, лежить у файлі examples/speed_test.php . Так що ви легко можете потестувати продуктивність і вибрати оптимальну кількість потоків у себе.
Буду дуже радий, якщо бібліотека комусь стане в нагоді. Будь-які фіч-реквести або виявлені баги можна залишати на гітхабі, оперативно фіксуватиму і покращуватиму бібліотеку.
НЕВРОЗ СЕРЦЯ (КАРДІОФОБІЯ). Особливою формою фобії поряд із синдромом паніки є кардіофобія, яку слід описати особливо внаслідок її характерної клініки та значної частоти. В основному вона зустрічається у молодих людей, частіше у чоловіків, а також у дітей.Приступообразные тривожні стану, у яких хворі побоюються припинення роботи серця та настання смерті, можуть і без наявності соматичного захворювання. На початку чергового нападу з'являються нудота, запаморочення, внутрішній занепокоєння, легкий стиск серця.
Однак у багатьох випадках без будь-яких провісників виникає тяжкий напад: сильне серцебиття, що відчувається по всьому тілу, деяке підвищення кров'яного тиску, тяжке почуття стиснення і стиснення в області серця, нестача повітря, пітливість, запаморочення і відчуття непритомності (але не втрати свідомості), тремтіння у всьому тілі та елементарний страх. Пацієнт вважає, що його серце через секунду зупиниться і він упаде мертвим. Це страх самознищення та смерті. При сильному збудженні хворі бігають і благають допомоги. Якщо напад страху виникає під час поїздки в машині, пацієнт змушений зупинитися і перепочити.
Після першого нападу відбувається фобічний розвиток. Хворі втрачають душевну рівновагу, живуть у постійному страху, очікуючи чергового нападу чи смерті, відчуваючи страх виникнення страху (страх очікування, фобія). При цьому їм не допомагають ні повідомлення терапевта про нормальні показники роботи серця, ні домовленості про те, що попередні напади не мали наслідків. Частота нападів та проміжки між ними нерегулярні. В інтервалах хворий небезпечно стежить за своїми серцевими функціями, контролює пульс та реєструє найменші його відхилення. Випадкові екстрасистоли він сприймає як незаперечні ознаки хвороби з безнадійним результатом.
З побоюванням хворі спостерігають за іншими вегетативними проявами, а також за легкими коливаннями свого самопочуття. Хворі бережуть себе, насилу наважуються ходити, прагнуть усунути всі навантаження, хвилювання і в першу чергу важкі ситуації, щоб запобігти нападу (поводження, що ухиляється). Замість страху смерті у багатьох випадках дедалі більше виступає страх перед страхом і перед викликають страх ситуаціями.
УМОВИ ВИНИКНЕННЯ. Приводом до першого кардіофобічного нападу частіше бувають гострий конфлікт і перенапруга, розлука та розчарування, ситуація самотності та покинутість, а також переживання у разі серцевої смерті когось із близьких.
Знання про те, що може статися серцева смерть, навіть у молодих і здорових стає тривожним чинником. Інтенсивне вживання кави та нікотину може викликати до життя цей процес. Початок часто виходить з дитинства. Переважно уражаються розпещені та несамостійні діти з вираженою залежністю від матері, багато в чому з амбівалентними установками: очікування кохання, з одного боку, та бажання самостійності з агресивними поривами – з іншого, з суперечливими фантазіями прихильності та розставання. Такі установки особливо небезпечні при обриві зв'язків, розлуки та розчарування. Кардіофобік часто живе у страху перед розлукою, перш ніж зрозуміє, що він її хоче і її боїться. Регулярно виникають спільні проблеми з батьками та конфлікти з партнерами.
ЛІКУВАННЯ
Якщо в гострому стані наявність лікаря та бесіда з ним не викликають покращення, показані транквілізатори або бета-блокатори. Як і інші хворі на неврози зі страхом, багато фобиків намагаються проводити самолікування алкоголем; але ефект його недостатній, а небезпека появи залежність від алкоголю велика. Фармакотерапія - це лише підсобний засіб, насамперед у гострих випадках, а також початковий ефективний засіб.
Вирішальною ж є психотерапія. Чим раніше її починають, тим краще. Вивчення причин та конфліктних ситуацій одразу після перших кардіофобічних нападів може призупинити подальший фобічний розвиток. Пізніше лікування проводити важче і потрібна тривала психотерапія.
При цих та інших тривожних розладах особливо показано поведінкову терапію (збудлива конфронтація, пізнавальна терапія, тренування впевненості у собі). Відмінна особливість тренінгу уникнення тривоги полягає в тому, що він працює на моделі десенсибілізації (до відповідних умов звичайних ситуацій), а тренінг управління тривогою - за допомогою форсованого занурення у фобічну ситуацію (повінь) та формування стратегій подолання. При тяжких тривожних розладах необхідно проводити клінічне лікування з використанням різних моделей психотерапії.