🖼️ 👐🏼 👨‍👨‍👧‍👦 PHP में Coroutines और गैर-अवरुद्ध कार्यों के साथ काम करना 🈷️ 🎅🏾 🤜🏻

PHP 5.5 में सबसे बड़ी नवाचारों में से एक जनरेटर और कोरआउट (कोराउटाइन) के लिए समर्थन होगा। जनरेटर पहले से ही दस्तावेज में और कई अन्य पदों पर पर्याप्त रूप से कवर किए गए हैं (उदाहरण के लिए, यह या यह )। Coroutines को बहुत कम ध्यान मिला। यह बहुत अधिक शक्तिशाली है, लेकिन टूल को समझने और समझाने में भी अधिक कठिन है।

इस लेख में मैं दिखाऊंगा कि कोरटाइन का उपयोग करके एक कार्य अनुसूचक को कैसे लागू किया जाए, ताकि आप समझ सकें कि आप उनके साथ क्या कर सकते हैं और उन्हें कैसे लागू कर सकते हैं। चलिए कुछ परिचयात्मक शब्दों के साथ शुरू करते हैं। यदि आपको लगता है कि आप पहले से ही अच्छी तरह से जानते हैं कि जनरेटर और कोरटाइन कैसे काम करते हैं, तो आप तुरंत "संयुक्त मल्टीटास्किंग" अनुभाग पर जा सकते हैं।

जनरेटर

जनरेटर का सार यह है कि यह एक ऐसा फ़ंक्शन है जो न केवल एक मान लौटाता है, बल्कि मूल्यों का एक क्रम है , जहां प्रत्येक मान को एक दूसरे से फेंक दिया जाता है। या, दूसरे शब्दों में, जनरेटर आपको अतिरिक्त कोड के बिना, एक पुनरावृत्ति को लागू करने की अनुमति देता है।

एक बहुत ही सरल उदाहरण xrange() फ़ंक्शन है:

 function xrange($start, $end, $step = 1) { for ($i = $start; $i <= $end; $i += $step) { yield $i; } } foreach (xrange(1, 1000000) as $num) { echo $num, "\n"; }

यह फ़ंक्शन अंतर्निहित PHP, range() । अंतर केवल इतना है कि वह range() आपको एक मिलियन संख्याओं की एक सरणी xrange() , और xrange() एक xrange() लौटाएगा जो उन संख्याओं को दूर फेंक देगा, लेकिन उन सभी के साथ एक सरणी कभी नहीं बनाएगा।

इस दृष्टिकोण के लाभ स्पष्ट होना चाहिए। इस तरह आप उन सभी को मेमोरी में लोड किए बिना बहुत बड़े डेटा सेट के साथ काम कर सकते हैं। तुम भी अंतहीन डेटा धाराओं के साथ काम कर सकते हैं।

यह सब जेनरेटर के बिना भी किया जा सकता है Iterator इंटरफ़ेस को लागू करने वाला एक वर्ग बनाकर। जनरेटर केवल इसे (बहुत) सरल अभ्यास करते हैं क्योंकि आपको प्रत्येक पुनरावृत्ति के लिए पांच अलग-अलग तरीकों को लागू करने की आवश्यकता नहीं है।

जनरेटरों में रुकावट के रूप में कार्य

जनरेटर से कोरटाइन में स्थानांतरित करने के लिए, यह समझना महत्वपूर्ण है कि वे अंदर से कैसे बनाए जाते हैं: जनरेटर अवरोधक कार्य हैं जिसमें yield कीवर्ड का उपयोग रुकावटों के लिए किया जाता है।

पिछले उदाहरण पर xrange(1, 1000000) , जब आप xrange(1, 1000000) , तो xrange फ़ंक्शन के शरीर से वास्तव में कुछ भी नहीं कहा जाएगा। इसके बजाय, PHP बस एक Generator ऑब्जेक्ट लौटाएगा जो Iterator इंटरफ़ेस को लागू करता है:

 <?php $range = xrange(1, 1000000); var_dump($range); // object(Generator)#1 var_dump($range instanceof Iterator); // bool(true)

कोड केवल तभी चलेगा जब आप कुछ पुनरावृत्त तरीके चलाते हैं। उदाहरण के लिए, यदि आप $range->rewind() तो पहली yield से पहले xrange() फ़ंक्शन $range->rewind() कोड को निष्पादित किया जाएगा। इस मामले में, इसका मतलब है कि $i = $start को पहले निष्पादित किया जाएगा और फिर yield $i किया जाएगा। हम जो कुछ भी yield करने के लिए पास करते हैं, वह $range->current() प्राप्त किया जा सकता है।

कोड निष्पादित करने के लिए, आपको $range->next() को कॉल करना होगा। यह जनरेटर को अगली yield तक कोड निष्पादित करने के लिए मजबूर करता yield । इस प्रकार, next() और current() लिए लगातार कॉल का उपयोग करके, आप जनरेटर से सभी मान प्राप्त कर सकते हैं जब तक कि यह उस बिंदु तक नहीं पहुंचता जहां कोड बस समाप्त होता है। xrange() के मामले में xrange() यह तब होगा जब $i $end तक पहुंच $end । इस मामले में, निष्पादन का धागा अधिक कोड छोड़ने के बिना फ़ंक्शन के अंत तक पहुंचता है। ऐसा होने के बाद, valid() false वापस आ जाएगा और चलना बंद हो जाएगा।

coroutine

उपरोक्त कार्यशीलता में मुख्य चीज़ जो कोरटाइन जोड़ती है, वह जनरेटर को मान भेजने की क्षमता है। यह जनरेटर के सामान्य एकालाप को पूर्ण संवाद बनाता है, जहां कॉलर जनरेटर को भी कुछ बता सकता है।

next() बजाय कॉल send() द्वारा मानों को कोरटाइन पास किया जाता है। यह कैसे काम करता है, इसका एक उदाहरण है:

 function logger($fileName) { $fileHandle = fopen($fileName, 'a'); while (true) { fwrite($fileHandle, yield . "\n"); } } $logger = logger(__DIR__ . '/log'); $logger->send('Foo'); $logger->send('Bar');

जैसा कि आप देख सकते हैं, यहां yield एक बयान (जैसे return या echo ) के रूप में नहीं है, लेकिन एक अभिव्यक्ति के रूप में, यह एक मूल्य है। यह वह send() जो भेजे send() माध्यम से भेजा गया था। इस उदाहरण में, yield "Foo" और फिर "Bar" वापस आ yield ।

इस उदाहरण में, यह प्रस्तुत किया गया कि yield एक साधारण प्राप्तकर्ता के रूप में कैसे कार्य कर सकती है। लेकिन आप दोनों प्रकार के उपयोग को भी जोड़ सकते हैं, इसलिए आप मान भेज और प्राप्त कर सकते हैं। यह कैसे काम करता है इसका एक उदाहरण है:

 function gen() { $ret = (yield 'yield1'); var_dump($ret); $ret = (yield 'yield2'); var_dump($ret); } $gen = gen(); var_dump($gen->current()); // string(6) "yield1" var_dump($gen->send('ret1')); // string(4) "ret1" (the first var_dump in gen) // string(6) "yield2" (the var_dump of the ->send() return value) var_dump($gen->send('ret2')); // string(4) "ret2" (again from within gen) // NULL (the return value of ->send())

सटीक आउटपुट ऑर्डर पहली नज़र में समझने में थोड़ा मुश्किल लग सकता है, इसलिए कोड को फिर से पढ़ें और यह जानने के लिए खुद को चलाने की कोशिश करें कि उस क्रम में सब कुछ क्यों होता है। यहां दो चीजें हैं जो मैं आपका ध्यान आकर्षित करना चाहता हूं: पहला, yield आसपास कोष्ठक का उपयोग कोई दुर्घटना नहीं है। हमें तकनीकी कारणों से इन कोष्ठकों की आवश्यकता है (मैंने प्रत्यक्ष असाइनमेंट में एक अपवाद फेंक को जोड़ने के बारे में भी सोचा था)। दूसरे, आपने देखा होगा कि current() का उपयोग rewind() बिना कॉल किया गया था। rewind() , वास्तव में, तब अंतर्निहित रूप से लागू किया जाता है।

सहयोगी मल्टीटास्किंग

यदि आप उदाहरण logger() फ़ंक्शन पढ़ते हैं, तो आपने सोचा, “मैं इसके लिए कोरटाइन का उपयोग क्यों करूंगा? इसके लिए एक साधारण वर्ग क्यों नहीं बनाया गया? ”, तो आप बिल्कुल सही थे। यह उदाहरण केवल यह दर्शाता है कि इसका उपयोग कैसे किया जा सकता है, लेकिन कोरटाइन का उपयोग करने का कोई कारण नहीं है। जैसा कि ऊपर उल्लेख किया गया है, परिचय में, कोरटाइन एक बहुत ही शक्तिशाली चीज है, लेकिन उनका उपयोग बहुत ही दुर्लभ और अक्सर बहुत जटिल होता है, जो सरल और दूर-दराज के उदाहरणों के साथ आने का कार्य काफी कठिन बना देता है।

मैंने आपको कोरटाइन का उपयोग करके सहयोगी मल्टीटास्किंग के कार्यान्वयन को दिखाने का निर्णय लिया। लब्बोलुआब यह है कि हमारे पास कई कार्य हैं जिन्हें समानांतर में चलाने की आवश्यकता है। लेकिन एक प्रोसेसर (लगभग लेन। गोलाकार और वैक्यूम में) एक समय में केवल एक ही कार्य कर सकता है। इस प्रकार, प्रोसेसर को विभिन्न कार्यों के बीच स्विच करने और प्रत्येक "थोड़ा काम" करने की आवश्यकता होती है।

"जॉइंट" यह मल्टीटास्किंग है क्योंकि इसमें शेड्यूलर को निष्पादन का नियंत्रण स्वेच्छा से स्थानांतरित करना शामिल है ताकि वह एक और कार्य शुरू कर सके। प्रीमेप्टिव मल्टीटास्किंग भी है, जहाँ शेड्यूलर स्वयं कार्य को बाधित कर सकता है। सहयोगात्मक मल्टीटास्किंग का उपयोग विंडोज के पुराने संस्करणों (Win95 से पहले) और मैक ओएस में किया गया था, लेकिन फिर उन्होंने प्रीमेप्टिव पर स्विच कर दिया। कारण स्पष्ट है - यदि आप नियंत्रण प्रवाह को स्वेच्छा से छोड़ने के लिए किसी भी कार्यक्रम पर भरोसा करते हैं, तो कोई भी कार्यक्रम केवल संपूर्ण सीपीयू पर कब्जा कर सकता है।

अब आपको पहले से ही कोरटाइन और टास्क शेड्यूलर के बीच संबंध देखना चाहिए: yield आपको शेड्यूलर नियंत्रण प्रवाह देने के लिए कार्य को स्वयं बाधित करने की अनुमति देता है और यह एक और कार्य शुरू कर सकता है। इसके अतिरिक्त, शेड्यूलर के साथ कार्यों को संप्रेषित करने के लिए yield का उपयोग किया जा सकता है।

हमारे मामले में, कार्य जनरेटर फ़ंक्शन के आसपास एक पतला आवरण होगा:

 class Task { protected $taskId; protected $coroutine; protected $sendValue = null; protected $beforeFirstYield = true; public function __construct($taskId, Generator $coroutine) { $this->taskId = $taskId; $this->coroutine = $coroutine; } public function getTaskId() { return $this->taskId; } public function setSendValue($sendValue) { $this->sendValue = $sendValue; } public function run() { if ($this->beforeFirstYield) { $this->beforeFirstYield = false; return $this->coroutine->current(); } else { $retval = $this->coroutine->send($this->sendValue); $this->sendValue = null; return $retval; } } public function isFinished() { return !$this->coroutine->valid(); } }

प्रत्येक कार्य का अपना पहचानकर्ता ( taskId ) होगा। setSendValue() विधि का उपयोग करके, आप अगले रन पर कार्य को भेजने के लिए क्या मूल्य निर्दिष्ट कर सकते हैं (आपको बाद में पता लगाने की आवश्यकता क्यों है)। run() विधि वास्तव में सिर्फ कॉरटाइन्स send() विधि कहती है।
यह समझने के लिए कि हमें पहले beforeFirstYield फ़्लैग की आवश्यकता क्यों है, निम्न कोड को देखें:

 function gen() { yield 'foo'; yield 'bar'; } $gen = gen(); var_dump($gen->send('something')); //   send(),   yield    rewind() //       : $gen->rewind(); var_dump($gen->send('something')); // rewind()    yield (   ), send() //    yield ( var_dump-  ). //

beforeFirstYield साथ beforeFirstYield हमें पता चल जाएगा कि क्या पहली उपज पहले ही वापस आ गई है।

शेड्यूलर को अब सभी कार्यों से थोड़ा अधिक चलना होगा:

 class Scheduler { protected $maxTaskId = 0; protected $taskMap = []; // taskId => task protected $taskQueue; public function __construct() { $this->taskQueue = new SplQueue(); } public function newTask(Generator $coroutine) { $tid = ++$this->maxTaskId; $task = new Task($tid, $coroutine); $this->taskMap[$tid] = $task; $this->schedule($task); return $tid; } public function schedule(Task $task) { $this->taskQueue->enqueue($task); } public function run() { while (!$this->taskQueue->isEmpty()) { $task = $this->taskQueue->dequeue(); $task->run(); if ($task->isFinished()) { unset($this->taskMap[$task->getTaskId()]); } else { $this->schedule($task); } } } }

newTask() विधि एक नया कार्य बनाती है और इसे taskMap में taskMap । इसके अलावा, वह इसे taskQueue जोड़ता है। run() विधि तब इस कतार से गुजरती है और कार्यों को शुरू करती है। यदि कार्य पूरा हो गया है, तो इसे हटा दिया जाता है, अन्यथा इसे कतार के अंत में जोड़ा जाता है।

आइए एक अनुसूचक को दो सरल (और अर्थहीन) कार्यों के साथ आज़माएँ:

 function task1() { for ($i = 1; $i <= 10; ++$i) { echo "This is task 1 iteration $i.\n"; yield; } } function task2() { for ($i = 1; $i <= 5; ++$i) { echo "This is task 2 iteration $i.\n"; yield; } } $scheduler = new Scheduler; $scheduler->newTask(task1()); $scheduler->newTask(task2()); $scheduler->run();

दोनों कार्य केवल एक संदेश प्रदर्शित करते हैं, और कंट्रोलर को वापस शेड्यूलर को देते हैं। यहाँ आउटपुट क्या होगा:

  यह कार्य 1 पुनरावृत्ति 1 है।
 यह कार्य 2 पुनरावृत्ति 1 है।
 यह कार्य 1 पुनरावृत्ति 2 है।
 यह कार्य 2 पुनरावृत्ति 2 है।
 यह कार्य 1 पुनरावृत्ति 3 है।
 यह कार्य 2 पुनरावृत्ति 3 है।
 यह कार्य 1 पुनरावृत्ति 4 है।
 यह कार्य 2 पुनरावृत्ति 4 है।
 यह कार्य 1 पुनरावृत्ति 5 है।
 यह कार्य 2 पुनरावृत्ति 5 है।
 यह कार्य 1 पुनरावृत्ति 6 है।
 यह कार्य 1 पुनरावृत्ति 7 है।
 यह कार्य 1 पुनरावृत्ति 8 है।
 यह कार्य 1 पुनरावृत्ति 9 है।
 यह कार्य 1 पुनरावृत्ति 10 है।

शेड्यूलर इंटरैक्शन

इसलिए, हमारा शेड्यूलर काम करता है और हम एजेंडे पर अगले आइटम पर जा सकते हैं: कार्यों और अनुसूचक के बीच बातचीत। हम उसी विधि का उपयोग करेंगे जो प्रक्रियाएं ऑपरेटिंग सिस्टम के साथ संचार करने के लिए उपयोग करती हैं: सिस्टम कॉल (syscall, siskol)। हमें syskols की आवश्यकता है, क्योंकि ऑपरेटिंग सिस्टम स्वयं की प्रक्रियाओं की तुलना में विशेषाधिकारों के एक अलग स्तर पर है। इसलिए, कुछ कार्यों को करने के लिए जिन्हें अधिक विशेषाधिकारों की आवश्यकता होती है (उदाहरण के लिए, अन्य प्रक्रियाओं को मारना), कर्नेल को नियंत्रण देने का एक तरीका होना चाहिए ताकि यह उन कार्यों को कर सके। आंतरिक रूप से, यह रुकावट निर्देशों का उपयोग करके लागू किया जाता है।

हमारा शेड्यूलर इस आर्किटेक्चर का पालन करेगा: शेड्यूलर को कार्य के पास करने के बजाय (जो इसके साथ कुछ भी कर सकता है), हम yield माध्यम से पारित पासवर्ड की मदद से बातचीत करेंगे। इस स्थिति में, yield एक ब्रेकर के रूप में और अनुसूचक को सूचना प्रसारित करने (और प्राप्त करने) के रूप में कार्य करेगा।

सिस्कोल का प्रतिनिधित्व करने के लिए, मैं कॉल करने योग्य पर एक छोटे आवरण का उपयोग करूंगा:

 class SystemCall { protected $callback; public function __construct(callable $callback) { $this->callback = $callback; } public function __invoke(Task $task, Scheduler $scheduler) { $callback = $this->callback; // Can't call it directly in PHP :/ return $callback($task, $scheduler); } }

यह किसी भी अन्य callable योग्य व्यवहार करेगा, लेकिन कार्य और अनुसूचक को तर्कों में ले जाएगा। इसके साथ काम करने के लिए, हमें शेड्यूलर की run() विधि को थोड़ा संशोधित करना होगा:

 public function run() { while (!$this->taskQueue->isEmpty()) { $task = $this->taskQueue->dequeue(); $retval = $task->run(); if ($retval instanceof SystemCall) { $retval($task, $this); continue; } if ($task->isFinished()) { unset($this->taskMap[$task->getTaskId()]); } else { $this->schedule($task); } } }

हमारा पहला syskol बस कार्य पहचानकर्ता को लौटाएगा:

 function getTaskId() { return new SystemCall(function(Task $task, Scheduler $scheduler) { $task->setSendValue($task->getTaskId()); $scheduler->schedule($task); }); }

ऐसा इसलिए होता है क्योंकि हम भेजने के लिए मान सेट करते हैं और कार्य को शेड्यूलर पर वापस सेट करते हैं। Syskol के लिए, अनुसूचक स्वचालित रूप से कतार में कार्य नहीं करता है, हमें इसे मैन्युअल रूप से करने की आवश्यकता है (आपको पता चल जाएगा कि थोड़ी देर बाद क्यों)।
इस नए सिस्कोल का उपयोग करके, हम पिछले उदाहरण को फिर से लिख सकते हैं:

 function task($max) { $tid = (yield getTaskId()); // <-- here's the syscall! for ($i = 1; $i <= $max; ++$i) { echo "This is task $tid iteration $i.\n"; yield; } } $scheduler = new Scheduler; $scheduler->newTask(task(10)); $scheduler->newTask(task(5)); $scheduler->run();

यह कोड हमें पिछले उदाहरण के समान आउटपुट देगा। नए और मार कार्यों को बनाने के लिए एक जोड़ी siskols:

 function newTask(Generator $coroutine) { return new SystemCall( function(Task $task, Scheduler $scheduler) use ($coroutine) { $task->setSendValue($scheduler->newTask($coroutine)); $scheduler->schedule($task); } ); } function killTask($tid) { return new SystemCall( function(Task $task, Scheduler $scheduler) use ($tid) { $task->setSendValue($scheduler->killTask($tid)); $scheduler->schedule($task); } ); }

killTask फ़ंक्शन के लिए killTask हमें शेड्यूलर में एक अतिरिक्त विधि की आवश्यकता है:

 public function killTask($tid) { if (!isset($this->taskMap[$tid])) { return false; } unset($this->taskMap[$tid]); //        , //      ,       foreach ($this->taskQueue as $i => $task) { if ($task->getTaskId() === $tid) { unset($this->taskQueue[$i]); break; } } return true; }

इस कार्यक्षमता का परीक्षण करने के लिए एक छोटी स्क्रिप्ट:

 function childTask() { $tid = (yield getTaskId()); while (true) { echo "Child task $tid still alive!\n"; yield; } } function task() { $tid = (yield getTaskId()); $childTid = (yield newTask(childTask())); for ($i = 1; $i <= 6; ++$i) { echo "Parent task $tid iteration $i.\n"; yield; if ($i == 3) yield killTask($childTid); } } $scheduler = new Scheduler; $scheduler->newTask(task()); $scheduler->run();

निष्कर्ष इस प्रकार होगा:

  मूल कार्य 1 पुनरावृत्ति 1।
 बाल कार्य 2 अभी भी जीवित है!
 मूल कार्य 1 पुनरावृत्ति 2।
 बाल कार्य 2 अभी भी जीवित है!
 मूल कार्य 1 पुनरावृत्ति 3।
 बाल कार्य 2 अभी भी जीवित है!
 मूल कार्य 1 पुनरावृत्ति 4।
 मूल कार्य 1 पुनरावृत्ति 5।
 मूल कार्य 1 पुनरावृत्ति 6।

बाल कार्य तीन पुनरावृत्तियों के बाद मारा जाता है, फिर "बाल कार्य अभी भी जीवित है!" संदेश समाप्त होता है। यह ध्यान देने योग्य है कि यह एक वास्तविक माता-पिता और बच्चा नहीं है, क्योंकि बच्चे के कार्य को तब भी निष्पादित किया जा सकता है जब माता-पिता पहले ही पूरा कर चुके हों। या बेटी माता-पिता को भी मार सकती है। हम माता-पिता और बच्चे के बीच एक सामान्य संबंध प्राप्त करने के लिए अनुसूचक को संशोधित कर सकते हैं, लेकिन इस लेख में नहीं।

कुछ अलग-अलग प्रकार की कॉल हैं जिन्हें आप कार्यान्वित कर सकते हैं, जैसे कि wait (कार्य पूरी तरह से समाप्त होने तक प्रतीक्षा करें), exec (जो निर्धारित करता है कि आपको अब किस कार्य को पूरा करने की आवश्यकता है) और fork (कार्य का क्लोन बनाना)। क्लोनिंग एक बहुत अच्छी सुविधा है और आप इसे कोरटाइन के साथ लागू कर सकते हैं, क्योंकि वे क्लोनिंग का समर्थन करते हैं।

गैर-अवरुद्ध बातचीत

एक वेब सर्वर - हमारे अनुसूचक का एक बहुत अच्छा अनुप्रयोग स्पष्ट है। नए कनेक्शन के लिए सॉकेट पर एक कार्य सुनना हो सकता है, और हर बार एक नया कनेक्शन बनाया जाता है, इस कनेक्शन को संसाधित करने के लिए एक नया कार्य बनाया जाएगा।

कठिनाई इस तथ्य में निहित है कि PHP में डेटा पढ़ने जैसे सॉकेट्स पर संचालन अवरुद्ध हो रहा है, अर्थात, जब तक क्लाइंट डेटा भेजने का काम पूरा नहीं करेगा, तब तक PHP प्रतीक्षा करेगा। एक वेब सर्वर के लिए, यह स्पष्ट रूप से अच्छा नहीं है: इसका मतलब है कि यह केवल एक समय में एक अनुरोध को संसाधित कर सकता है।

समाधान के रूप में, हमें सॉकेट से पूछना होगा कि क्या यह डेटा पढ़ने या लिखने से पहले "तैयार" है। यह पता लगाने के लिए कि कौन से सॉकेट डेटा भेजने या प्राप्त करने के लिए तैयार हैं, हम stream_select() फ़ंक्शन का उपयोग करेंगे।

आरंभ करने के लिए, आइए कुछ नए सिस्कोल्स जोड़ते हैं जो एक विशिष्ट सॉकेट भेजेंगे जो या तो पढ़ने या लिखने के लिए इंतजार कर रहा है:

 function waitForRead($socket) { return new SystemCall( function(Task $task, Scheduler $scheduler) use ($socket) { $scheduler->waitForRead($socket, $task); } ); } function waitForWrite($socket) { return new SystemCall( function(Task $task, Scheduler $scheduler) use ($socket) { $scheduler->waitForWrite($socket, $task); } ); }

ये सिस्कोल्स इसी अनुसूचक विधियों के लिए समीपवर्ती हैं:

 <?php // resourceID => [socket, tasks] protected $waitingForRead = []; protected $waitingForWrite = []; public function waitForRead($socket, Task $task) { if (isset($this->waitingForRead[(int) $socket])) { $this->waitingForRead[(int) $socket][1][] = $task; } else { $this->waitingForRead[(int) $socket] = [$socket, [$task]]; } } public function waitForWrite($socket, Task $task) { if (isset($this->waitingForWrite[(int) $socket])) { $this->waitingForWrite[(int) $socket][1][] = $task; } else { $this->waitingForWrite[(int) $socket] = [$socket, [$task]]; } }

waitingForRead और waitingForWrite बस इंतज़ार कर रहे सॉकेट्स और संबंधित कार्य हैं। सबसे दिलचस्प हिस्सा यह विधि है, जो यह जांचती है कि क्या सॉकेट तैयार हैं और उनके कार्यों को फिर से शुरू करता है:

 protected function ioPoll($timeout) { $rSocks = []; foreach ($this->waitingForRead as list($socket)) { $rSocks[] = $socket; } $wSocks = []; foreach ($this->waitingForWrite as list($socket)) { $wSocks[] = $socket; } $eSocks = []; // dummy if (!stream_select($rSocks, $wSocks, $eSocks, $timeout)) { return; } foreach ($rSocks as $socket) { list(, $tasks) = $this->waitingForRead[(int) $socket]; unset($this->waitingForRead[(int) $socket]); foreach ($tasks as $task) { $this->schedule($task); } } foreach ($wSocks as $socket) { list(, $tasks) = $this->waitingForWrite[(int) $socket]; unset($this->waitingForWrite[(int) $socket]); foreach ($tasks as $task) { $this->schedule($task); } } }

stream_select फ़ंक्शन सॉकेट के इनपुट एरे में लेता है जो पढ़ने, लिखे जाने और फेंके जाने की प्रतीक्षा में है (बाद वाले को अनदेखा करें)। एरे को संदर्भ द्वारा पारित किया जाता है और यह फ़ंक्शन केवल उन तत्वों को छोड़ देगा जिनकी स्थिति बदल गई है। उसके बाद हम इन सभी सरणियों के माध्यम से जा सकते हैं और इन सॉकेट्स से जुड़े सभी कार्यों को पुनर्निर्धारित कर सकते हैं।

इन सभी कार्यों को करने के लिए, हम निम्नलिखित विधि को अनुसूचक में जोड़ देंगे:

 protected function ioPollTask() { while (true) { if ($this->taskQueue->isEmpty()) { $this->ioPoll(null); } else { $this->ioPoll(0); } yield; } }

इस कार्य को कुछ बिंदु पर पंजीकृत करने की आवश्यकता है, उदाहरण के लिए आप $this->newTask($this->ioPollTask()) जोड़ सकते $this->newTask($this->ioPollTask()) run() के शीर्ष run() विधि के लिए। फिर यह किसी भी अन्य कार्य की तरह काम करेगा, कार्यों के बीच प्रत्येक स्विचिंग के लिए उपलब्ध तैयार सॉकेट्स की जांच करना। ioPollTask पद्धति शून्य टाइमआउट के साथ ioPoll को कॉल ioPoll , इसलिए stream_select बिना प्रतीक्षा किए तुरंत परिणाम वापस कर देगी।

केवल अगर कार्य कतार खाली है, तो हम समय-सीमा के रूप में null उपयोग करते हैं, जिस स्थिति में stream_select इंतजार करेंगे जब तक कि सॉकेट्स में से कोई भी तैयार न हो। यदि हमने ऐसा नहीं किया, तो हम पूरा सीपीयू (कम से कम कोर) खाएंगे, क्योंकि यह कार्य एक लूप में बार-बार किया जाएगा जब तक कि कोई जुड़ा नहीं।

सर्वर स्वयं बहुत सरल दिखता है:

 function server($port) { echo "Starting server at port $port...\n"; $socket = @stream_socket_server("tcp://localhost:$port", $errNo, $errStr); if (!$socket) throw new Exception($errStr, $errNo); stream_set_blocking($socket, 0); while (true) { yield waitForRead($socket); $clientSocket = stream_socket_accept($socket, 0); yield newTask(handleClient($clientSocket)); } } function handleClient($socket) { yield waitForRead($socket); $data = fread($socket, 8192); $msg = "Received following request:\n\n$data"; $msgLength = strlen($msg); $response = <<<RES HTTP/1.1 200 OK\r Content-Type: text/plain\r Content-Length: $msgLength\r Connection: close\r \r $msg RES; yield waitForWrite($socket); fwrite($socket, $response); fclose($socket); } $scheduler = new Scheduler; $scheduler->newTask(server(8000)); $scheduler->run();

यह पोर्ट 8000 पर कनेक्शन स्वीकार करेगा और केवल प्रतिक्रिया में अनुरोध की सामग्री भेजेगा। कुछ "वास्तविक" करना सही होगा (HTTP अनुरोधों का सही हैंडलिंग इस लेख का विषय नहीं है)।

अब आप इस सर्वर को ab -n 10000 -c 100 localhost:8000/ जैसी किसी चीज के साथ टेस्ट कर सकते हैं। तो हम 10000 अनुरोध भेजेंगे, जिनमें से 100 उसी समय भेजे जाएंगे। इस बेंचमार्क का उपयोग करते हुए, मुझे 10 मिलीसेकंड की औसत प्रतिक्रिया मिली। लेकिन कुछ अनुरोधों के साथ एक समस्या थी जिसे संसाधित करने में बहुत समय लगा (5 सेकंड में जिला), इसलिए कुल थ्रूपुट प्रति सेकंड केवल 2000 अनुरोध हैं। उच्च प्रतियोगिता (उदाहरण के लिए, -c 500 ) के साथ, स्क्रिप्ट भी अच्छी तरह से काम करती है, लेकिन कुछ कनेक्शन "सहकर्मी द्वारा रीसेट रीसेट" त्रुटि फेंकते हैं।

दूरस्थ कोरटाइन

यदि आप हमारे अनुसूचक का उपयोग करके एक बड़ी प्रणाली बनाना शुरू करते हैं, तो आप जल्द ही एक समस्या में आ जाएंगे: हम अक्सर कुछ टुकड़ों को अलग-अलग कार्यों में लगाकर और उन्हें कॉल करके कोड को विभाजित करते हैं। लेकिन coroutines के साथ यह संभव नहीं है। इस कोड की कल्पना करें:

 function echoTimes($msg, $max) { for ($i = 1; $i <= $max; ++$i) { echo "$msg iteration $i\n"; yield; } } function task() { echoTimes('foo', 10); // print foo ten times echo "---\n"; echoTimes('bar', 5); // print bar five times yield; // force it to be a coroutine } $scheduler = new Scheduler; $scheduler->newTask(task()); $scheduler->run();

इस कोड में हमने task() से कोरटाइन निकालने और उसे निष्पादित करने का प्रयास किया। लेकिन वह काम नहीं करेगा।जैसा कि लेख की शुरुआत में कहा गया था, जनरेटर को कॉल करने से इसमें कोई कोड निष्पादित नहीं होगा, इसके बजाय, यह केवल एक क्लास ऑब्जेक्ट वापस करेगा Generator। हमारे मामले में, वास्तव में यही होता है, कॉल echoTimes()कुछ भी नहीं करेंगे, वे केवल ऑब्जेक्ट वापस करेंगे।

इसके लिए बाहर काम करने के लिए, हमें कोरआउट के लिए एक छोटा आवरण लिखना होगा। इस तरह से हम उप-कॉरआउट को कॉल कर सकते हैं:

 $retval = (yield someCoroutine($foo, $bar));

उप-कोरआउट्स का उपयोग करके एक मान भी लौटाया जा सकता है yield:

 yield retval("I'm return value!");

फ़ंक्शन retval()मान पर एक आवरण को छोड़कर कुछ भी नहीं करता है, जो हमें बताता है कि यह वापसी मूल्य है:

 class CoroutineReturnValue { protected $value; public function __construct($value) { $this->value = $value; } public function getValue() { return $this->value; } } function retval($value) { return new CoroutineReturnValue($value); }

एक नियमित कोरआउट से एक मिश्रित कोरआउट (उप-कोरआउट के साथ) बनाने के लिए, हमें एक और कार्य लिखना होगा (जो स्पष्ट रूप से एक और कोरआउट है):

 function stackedCoroutine(Generator $gen) { $stack = new SplStack; for (;;) { $value = $gen->current(); if ($value instanceof Generator) { $stack->push($gen); $gen = $value; continue; } $isReturnValue = $value instanceof CoroutineReturnValue; if (!$gen->valid() || $isReturnValue) { if ($stack->isEmpty()) { return; } $gen = $stack->pop(); $gen->send($isReturnValue ? $value->getValue() : NULL); continue; } $gen->send(yield $gen->key() => $value); } }

यह फ़ंक्शन कॉलर और चल रहे उप-कोरआउट के बीच एक सरल प्रॉक्सी के रूप में कार्य करता है। यह भी जांचता है कि क्या वापसी मूल्य भी एक जनरेटर है और यदि हां, तो इसे शुरू करता है। जब वह वस्तु प्राप्त करती है CoroutineReturnValue, तो वह माता-पिता को कोरटाइन लेगी और इसका निष्पादन जारी रखेगी।

ताकि यौगिक कोरटाइन को कार्यों में उपयोग किया जा सके, $this->coroutine = $coroutine;वर्ग निर्माणकर्ता की पंक्ति को इसके Taskसाथ बदलना होगा$this->coroutine = stackedCoroutine($coroutine); ।

अब हम पढ़ने, लिखने और एक नया कनेक्शन प्राप्त करने के कार्यों के साथ प्रतीक्षा कार्यों को समूहीकृत करके अपने वेब सर्वर को थोड़ा सुधार सकते हैं। ऐसा करने के लिए, इस वर्ग का उपयोग करें:

 class CoSocket { protected $socket; public function __construct($socket) { $this->socket = $socket; } public function accept() { yield waitForRead($this->socket); yield retval(new CoSocket(stream_socket_accept($this->socket, 0))); } public function read($size) { yield waitForRead($this->socket); yield retval(fread($this->socket, $size)); } public function write($string) { yield waitForWrite($this->socket); fwrite($this->socket, $string); } public function close() { @fclose($this->socket); } }

अब सर्वर को फिर से लिखा जा सकता है:

 function server($port) { echo "Starting server at port $port...\n"; $socket = @stream_socket_server("tcp://localhost:$port", $errNo, $errStr); if (!$socket) throw new Exception($errStr, $errNo); stream_set_blocking($socket, 0); $socket = new CoSocket($socket); while (true) { yield newTask( handleClient(yield $socket->accept()) ); } } function handleClient($socket) { $data = (yield $socket->read(8192)); $msg = "Received following request:\n\n$data"; $msgLength = strlen($msg); $response = <<<RES HTTP/1.1 200 OK\r Content-Type: text/plain\r Content-Length: $msgLength\r Connection: close\r \r $msg RES; yield $socket->write($response); yield $socket->close(); }

हैंडलिंग में त्रुटि

एक अच्छे प्रोग्रामर के रूप में, आपने स्पष्ट रूप से देखा है कि उपरोक्त कोड उदाहरणों में त्रुटि हैंडलिंग की कमी है। सॉकेट्स के साथ कोई भी ऑपरेशन हमेशा त्रुटि में पड़ सकता है। मैंने ऐसा इसलिए किया क्योंकि उदाहरणों के लिए त्रुटि से निपटना बहुत थकाऊ होता है (विशेषकर सॉकेट के मामले में) और यह कई बार हमारे कोड को बढ़ाता है।

लेकिन, फिर भी, मैं सामान्य रूप से coroutines के लिए त्रुटि से निपटने के बारे में बात करना चाहूंगा: coroutines आपको विधि का उपयोग करके उनके अंदर अपवाद फेंकने की अनुमति देता है throw()।

विधि throw()पहले तर्क में एक अपवाद लेती है और इसे उस स्थान पर फेंकती है जहां वर्तमान उपज है (जिसका मूल्य विधि का उपयोग करके प्राप्त किया जा सकता है current()):

 function gen() { echo "Foo\n"; try { yield; } catch (Exception $e) { echo "Exception: {$e->getMessage()}\n"; } echo "Bar\n"; } $gen = gen(); $gen->rewind(); // echos "Foo" $gen->throw(new Exception('Test')); // echos "Exception: Test" // and "Bar"

यह एक बहुत अच्छी बात है, क्योंकि उदाहरण के लिए, हम अपवादों को फेंकने के लिए सिस्कोल और उप-कॉरटाइन के माध्यम से अवसर दे सकते हैं। सिस्कोल के लिए, विधि Scheduler::run()को थोड़ा संशोधित करने की आवश्यकता है:

 if ($retval instanceof SystemCall) { try { $retval($task, $this); } catch (Exception $e) { $task->setException($e); $this->schedule($task); } continue; }

और कक्षा Taskको कॉल संभालनी चाहिए throw():

 class Task { // ... protected $exception = null; public function setException($exception) { $this->exception = $exception; } public function run() { if ($this->beforeFirstYield) { $this->beforeFirstYield = false; return $this->coroutine->current(); } elseif ($this->exception) { $retval = $this->coroutine->throw($this->exception); $this->exception = null; return $retval; } else { $retval = $this->coroutine->send($this->sendValue); $this->sendValue = null; return $retval; } } // ... }

अब हम सिस्कोल से अपवाद फेंकना शुरू कर सकते हैं! उदाहरण के लिए, killTaskयदि एक ट्रांसमिटेड टास्क आइडेंटिफ़ायर अमान्य है, तो इसका अपवाद छोड़ दें:

 function killTask($tid) { return new SystemCall( function(Task $task, Scheduler $scheduler) use ($tid) { if ($scheduler->killTask($tid)) { $scheduler->schedule($task); } else { throw new InvalidArgumentException('Invalid task ID!'); } } ); }

अब चलो कोशिश करते हैं:

 function task() { try { yield killTask(500); } catch (Exception $e) { echo 'Tried to kill task 500 but failed: ', $e->getMessage(), "\n"; } }

अब तक यह काम नहीं करेगा, जैसा कि फ़ंक्शन stackedCoroutineअपवादों को नहीं संभालता है। इसे ठीक करने के लिए, इसे थोड़ा संशोधित करें:

 function stackedCoroutine(Generator $gen) { $stack = new SplStack; $exception = null; for (;;) { try { if ($exception) { $gen->throw($exception); $exception = null; continue; } $value = $gen->current(); if ($value instanceof Generator) { $stack->push($gen); $gen = $value; continue; } $isReturnValue = $value instanceof CoroutineReturnValue; if (!$gen->valid() || $isReturnValue) { if ($stack->isEmpty()) { return; } $gen = $stack->pop(); $gen->send($isReturnValue ? $value->getValue() : NULL); continue; } try { $sendValue = (yield $gen->key() => $value); } catch (Exception $e) { $gen->throw($e); continue; } $gen->send($sendValue); } catch (Exception $e) { if ($stack->isEmpty()) { throw $e; } $gen = $stack->pop(); $exception = $e; } } }

संक्षेप में देना

इस लेख में, हमने संयुक्त मल्टीटास्किंग, संदेशों को भेजने की क्षमता, गैर-अवरुद्ध संचालन और त्रुटि से निपटने के साथ काम करते हुए एक कार्य अनुसूचक का निर्माण किया। इस सब के बारे में सबसे अच्छी बात यह है कि अंतिम कोड पूरी तरह से सिंक्रोनस दिखता है, भले ही यह कई अतुल्यकालिक संचालन करता है। यदि आपको सॉकेट से डेटा पढ़ने की आवश्यकता है, तो आपको कॉलबैक भेजने या श्रोता को पंजीकृत करने की आवश्यकता नहीं है। इसके बजाय, आप लिखते हैं yield $socket->read()।

जब मैंने पहली बार इसके बारे में सुना, तो मुझे एहसास हुआ कि यह कितना अच्छा है, और इसने मुझे उन्हें PHP में लागू करने के लिए प्रेरित किया। एक ही समय में, मुझे लगता है कि वास्तव में डरावने हैं। शांत कोड और खौफनाक गंदगी के बीच एक महीन रेखा है, और मुझे लगता है कि उस लाइन पर कोरटाइन हैं। मेरे लिए यह कहना कठिन है कि क्या इस तरह से कोरटाइन का उपयोग करना, जैसा कि मैंने ऊपर वर्णित किया है, किसी तरह से फायदेमंद है।

किसी भी मामले में, मुझे लगता है कि यह एक बहुत ही दिलचस्प विषय है, और मुझे आशा है कि आप भी :)

अनुवादक से पुनश्च: लेख बहुत बड़ा है और समझना मुश्किल है, इसलिए इसमें त्रुटियां हो सकती हैं, जिनमें शामिल हैं अर्थ। अगर आपने यह देखा - कृपया भेजें।

PHP में Coroutines और गैर-अवरुद्ध कार्यों के साथ काम करना