From 2a2dcac4cf445bd3988b42d476307a93adb5c827 Mon Sep 17 00:00:00 2001 From: Binaek Sarkar Date: Sat, 12 Dec 2015 01:51:23 +0530 Subject: [PATCH 1/8] Adding Throttling in individual Procedure registrations --- src/Thruway/Common/LeakyBucket.php | 71 ++++++++++++++++++++++++++++++ src/Thruway/Procedure.php | 29 +++++++++++- src/Thruway/Registration.php | 62 +++++++++++++++++++++++--- tests/Unit/RegistrationTest.php | 26 +++++++++++ 4 files changed, 181 insertions(+), 7 deletions(-) create mode 100644 src/Thruway/Common/LeakyBucket.php diff --git a/src/Thruway/Common/LeakyBucket.php b/src/Thruway/Common/LeakyBucket.php new file mode 100644 index 00000000..24ed27a7 --- /dev/null +++ b/src/Thruway/Common/LeakyBucket.php @@ -0,0 +1,71 @@ +maxRate = -1; + $this->objectQueue = new SplQueue(); + $this->lastSchedAction = time(); + $this->setMaxRate($maxRatePerSecond); + } + + public function enqueue($anyObject) + { + $this->objectQueue->enqueue($anyObject); + } + + public function count() + { + return $this->objectQueue->count(); + } + + public function setMaxRate($maxRatePerSecond) + { + if ($maxRatePerSecond > 0.0) { + $this->maxRate = $maxRatePerSecond; + $this->minTime = (int) (1000.0 / $maxRatePerSecond); + } + } + + public function consume() + { + if ($this->maxRate > 0) { + //we are rate limited + $curTime = time(); + //calculate when can we send back + $timeLeft = $this->lastSchedAction + $this->minTime - $curTime; + if ($timeLeft > 0) { + $this->lastSchedAction += $this->minTime; + //we need to sleep for sometime + echo "We are sleeping"; + sleep($timeLeft); + } else { + $this->lastSchedAction = $this->curTime; + } + } + //lets go back + return $this->objectQueue->dequeue(); + } + +} diff --git a/src/Thruway/Procedure.php b/src/Thruway/Procedure.php index d739de2b..babe740b 100644 --- a/src/Thruway/Procedure.php +++ b/src/Thruway/Procedure.php @@ -317,8 +317,33 @@ private function getNextRandomRegistration() //just return this so that we don't have to run mt_rand return $this->registrations[0]; } - //mt_rand is apparently faster than array_rand(which uses the libc generator) - return $this->registrations[mt_rand(0, count($this->registrations) - 1)]; + + //getting registrations with 0 unprocessed calls + $possibleRegistrations = array_filter($this->registrations, function($theRegistration, $theIndex) { + return ($theRegistration->getStatistics()['invokeQueueCount'] === 0); + }, ARRAY_FILTER_USE_BOTH); + + if (count($possibleRegistrations) > 0) { + //return a random from this set + return $possibleRegistrations[mt_rand(0, count($possibleRegistrations) - 1)]; + } else { + //create a copy to maintain the original indexing + $possibleRegistrations = array_merge([], $this->registrations); + //sort ascending by number of unprocessed Invocations + usort($possibleRegistrations, function($registrationA, $registrationB) { + $unprocessedA = $registrationA->getStatistics()['invokeQueueCount']; + $unprocessedB = $registrationB->getStatistics()['invokeQueueCount']; + if ($unprocessedA == $unprocessedB) { + return 0; + } else if ($unprocessedA < $unprocessedB) { + return -1; + } else { + return 1; + } + }); + //return the first one + return $possibleRegistrations[0]; + } } private function getNextRoundRobinRegistration() diff --git a/src/Thruway/Registration.php b/src/Thruway/Registration.php index 1629fed0..1cebb544 100644 --- a/src/Thruway/Registration.php +++ b/src/Thruway/Registration.php @@ -3,6 +3,7 @@ namespace Thruway; use Thruway\Common\Utils; +use Thruway\Common\LeakyBucket; use Thruway\Message\ErrorMessage; use Thruway\Message\RegisterMessage; @@ -19,6 +20,16 @@ class Registration */ private $id; + /** + * @var int + */ + private $limit; + + /** + * @var LeakyBucket + */ + private $leakyQueue; + /** * @var \Thruway\Session */ @@ -97,11 +108,11 @@ class Registration private $completedCallTimeTotal; const SINGLE_REGISTRATION = 'single'; - const THRUWAY_REGISTRATION = '_thruway'; const ROUNDROBIN_REGISTRATION = 'roundrobin'; const RANDOM_REGISTRATION = 'random'; const FIRST_REGISTRATION = 'first'; const LAST_REGISTRATION = 'last'; + const THRUWAY_REGISTRATION = '_thruway'; /** * Constructor @@ -127,6 +138,9 @@ public function __construct(Session $session, $procedureName) $this->lastIdledAt = $this->registeredAt; $this->busyStart = null; $this->completedCallTimeTotal = 0; + + $this->limit = -1; + $this->leakyQueue = new LeakyBucket(); //no throtling by default } /** @@ -154,6 +168,11 @@ public static function createRegistrationFromRegisterMessage(Session $session, R $registration->setInvokeType(Registration::SINGLE_REGISTRATION); } } + if (isset($options->_limit) && settype($options->_limit, "integer")) { + $registration->setLimit($options->_limit); + } else { + $registration->setLimit(-1); //setting to UNLIMITED + } return $registration; } @@ -210,14 +229,31 @@ public function setInvokeType($type) if ($type !== Registration::SINGLE_REGISTRATION) { $this->invokeType = $type; $this->setAllowMultipleRegistrations(true); - } - else { + } else { $this->invokeType = Registration::SINGLE_REGISTRATION; $this->setAllowMultipleRegistrations(false); } } } + /** + * @param int $limit The number of calls allowed per second + */ + public function setLimit($limit) + { + $this->limit = $limit; + $this->leakyQueue->setMaxRate($limit); + } + + /** + * Get the Limit per second on this registrations + * @return int + */ + public function getLimit() + { + return $this->limit; + } + /** * Process call * @@ -245,7 +281,22 @@ public function processCall(Call $call) $this->invocationCount++; $this->lastCallStartedAt = new \DateTime(); - $this->getSession()->sendMessage($call->getInvocationMessage()); + $this->leakyQueue->enqueue($call->getInvocationMessage()); + + $this->processInvocationQueue(); + } + + /** + * Process Invocation Queue + * + * @param none + */ + private function processInvocationQueue() + { + while ($this->leakyQueue->count() > 0) { + //this will sleep till we can actually make the call + $this->getSession()->sendMessage($this->leakyQueue->consume()); + } } /** @@ -386,7 +437,8 @@ public function getStatistics() 'busyStart' => $this->busyStart, 'lastIdledAt' => $this->lastIdledAt, 'lastCallStartedAt' => $this->lastCallStartedAt, - 'completedCallTimeTotal' => $this->completedCallTimeTotal + 'completedCallTimeTotal' => $this->completedCallTimeTotal, + 'invokeQueueCount' => $this->leakyQueue->count() ]; } diff --git a/tests/Unit/RegistrationTest.php b/tests/Unit/RegistrationTest.php index 626067f6..93476570 100644 --- a/tests/Unit/RegistrationTest.php +++ b/tests/Unit/RegistrationTest.php @@ -1,4 +1,5 @@ 0], 'rate.limit.procedure' + ); + $procedure->processRegister($calleeSession, $registerMsg); + + $this->assertEquals(1, count($procedure->getRegistrations())); + + for ($i = 5; $i-- > 0;) { + //send an invocation + $callMessage = new \Thruway\Message\CallMessage( + \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' + ); + $call = new \Thruway\Call($callerSession, $callMessage, $procedure); + $procedure->processCall($callerSession, $call); + } + $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['invokeQueueCount'])); + } /** * xdepends testAddCall From 0d04041c0b3efa3eb8e27c6d6935db1b9aceea11 Mon Sep 17 00:00:00 2001 From: Binaek Date: Tue, 15 Dec 2015 18:07:14 +0530 Subject: [PATCH 2/8] --- src/Thruway/Common/LeakyBucket.php | 43 +++++++---------- src/Thruway/Procedure.php | 6 +-- src/Thruway/Registration.php | 75 +++++++++++++++++++----------- tests/Unit/RegistrationTest.php | 34 +++++++------- 4 files changed, 85 insertions(+), 73 deletions(-) diff --git a/src/Thruway/Common/LeakyBucket.php b/src/Thruway/Common/LeakyBucket.php index 24ed27a7..d6282c02 100644 --- a/src/Thruway/Common/LeakyBucket.php +++ b/src/Thruway/Common/LeakyBucket.php @@ -16,30 +16,17 @@ class LeakyBucket protected $minTime; //holds time of last action (past or future!) protected $lastSchedAction; - - /** - * @var SplQueue The Object Storage - */ + protected $eventLoop; + protected $timer; protected $objectQueue; public function __construct($maxRatePerSecond = -1) { $this->maxRate = -1; - $this->objectQueue = new SplQueue(); $this->lastSchedAction = time(); $this->setMaxRate($maxRatePerSecond); } - public function enqueue($anyObject) - { - $this->objectQueue->enqueue($anyObject); - } - - public function count() - { - return $this->objectQueue->count(); - } - public function setMaxRate($maxRatePerSecond) { if ($maxRatePerSecond > 0.0) { @@ -48,24 +35,28 @@ public function setMaxRate($maxRatePerSecond) } } - public function consume() + public function canConsume() { + return ($this->timeLeft() <= 0); + } + + public function getTimeLeft() + { + $timeLeft = 0; if ($this->maxRate > 0) { //we are rate limited $curTime = time(); //calculate when can we send back $timeLeft = $this->lastSchedAction + $this->minTime - $curTime; - if ($timeLeft > 0) { - $this->lastSchedAction += $this->minTime; - //we need to sleep for sometime - echo "We are sleeping"; - sleep($timeLeft); - } else { - $this->lastSchedAction = $this->curTime; - } } - //lets go back - return $this->objectQueue->dequeue(); + return $timeLeft; + } + + public function consume() + { + if ($this->canConsume()) { + $this->lastSchedAction = time(); + } } } diff --git a/src/Thruway/Procedure.php b/src/Thruway/Procedure.php index babe740b..a064dc6c 100644 --- a/src/Thruway/Procedure.php +++ b/src/Thruway/Procedure.php @@ -320,7 +320,7 @@ private function getNextRandomRegistration() //getting registrations with 0 unprocessed calls $possibleRegistrations = array_filter($this->registrations, function($theRegistration, $theIndex) { - return ($theRegistration->getStatistics()['invokeQueueCount'] === 0); + return ($theRegistration->getStatistics()['pendingInvokeCount'] === 0); }, ARRAY_FILTER_USE_BOTH); if (count($possibleRegistrations) > 0) { @@ -331,8 +331,8 @@ private function getNextRandomRegistration() $possibleRegistrations = array_merge([], $this->registrations); //sort ascending by number of unprocessed Invocations usort($possibleRegistrations, function($registrationA, $registrationB) { - $unprocessedA = $registrationA->getStatistics()['invokeQueueCount']; - $unprocessedB = $registrationB->getStatistics()['invokeQueueCount']; + $unprocessedA = $registrationA->getStatistics()['pendingInvokeCount']; + $unprocessedB = $registrationB->getStatistics()['pendingInvokeCount']; if ($unprocessedA == $unprocessedB) { return 0; } else if ($unprocessedA < $unprocessedB) { diff --git a/src/Thruway/Registration.php b/src/Thruway/Registration.php index 1cebb544..a95ee703 100644 --- a/src/Thruway/Registration.php +++ b/src/Thruway/Registration.php @@ -3,7 +3,6 @@ namespace Thruway; use Thruway\Common\Utils; -use Thruway\Common\LeakyBucket; use Thruway\Message\ErrorMessage; use Thruway\Message\RegisterMessage; @@ -20,16 +19,6 @@ class Registration */ private $id; - /** - * @var int - */ - private $limit; - - /** - * @var LeakyBucket - */ - private $leakyQueue; - /** * @var \Thruway\Session */ @@ -106,6 +95,16 @@ class Registration * @var float */ private $completedCallTimeTotal; +//for ratelimiting + + /** + * @var int + */ + private $limit; + private $eventLoop; + private $leakyBucket; + private $invokeQueue; + private $rateLimited; const SINGLE_REGISTRATION = 'single'; const ROUNDROBIN_REGISTRATION = 'roundrobin'; @@ -140,7 +139,7 @@ public function __construct(Session $session, $procedureName) $this->completedCallTimeTotal = 0; $this->limit = -1; - $this->leakyQueue = new LeakyBucket(); //no throtling by default + $this->rateLimited = false; } /** @@ -242,7 +241,17 @@ public function setInvokeType($type) public function setLimit($limit) { $this->limit = $limit; - $this->leakyQueue->setMaxRate($limit); + if ($limit > 0) { + $this->rateLimited = true; + $this->invokeQueue = new \SplQueue(); + $this->leakyBucket = new Common\LeakyBucket($this->limit); + $this->eventLoop = \React\EventLoop\Factory::create(); + $this->eventLoop->addPeriodicTimer(30, function() { + //just to keep the event loop running + //no idea if this is required + }); + $this->eventLoop->run(); + } } /** @@ -280,22 +289,32 @@ public function processCall(Call $call) } $this->invocationCount++; $this->lastCallStartedAt = new \DateTime(); + if ($this->rateLimited) { + if ($this->leakyBucket->canConsume()) { + $this->leakyBucket->consume(); + $this->getSession()->sendMessage($call->getInvocationMessage()); + } else { + $this->invokeQueue->enqueue($call->getInvocationMessage()); + if ($this->invokeQueue->count() === 1) { + //start the timer if I am the first addition to the queue + $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); + } + } + } else { + $this->getSession()->sendMessage($call->getInvocationMessage()); + } + } - $this->leakyQueue->enqueue($call->getInvocationMessage()); - - $this->processInvocationQueue(); + public function isRateLimited() + { + return $this->rateLimited; } - /** - * Process Invocation Queue - * - * @param none - */ - private function processInvocationQueue() + private function invokeNextMessage() { - while ($this->leakyQueue->count() > 0) { - //this will sleep till we can actually make the call - $this->getSession()->sendMessage($this->leakyQueue->consume()); + $this->getSession()->sendMessage($this->invokeQueue->dequeue()); + if ($this->invokeQueue->count() > 0) { + $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); } } @@ -331,10 +350,10 @@ public function removeCall($callToRemove) $this->session->decPendingCallCount(); $callEnd = microtime(true); - // average call time +// average call time $callsInAverage = $this->invocationCount - count($this->calls) - 1; - // add this call time into the total +// add this call time into the total $this->completedCallTimeTotal += $callEnd - $call->getCallStart(); $callsInAverage++; $this->invocationAverageTime = ((float) $this->completedCallTimeTotal) / $callsInAverage; @@ -438,7 +457,7 @@ public function getStatistics() 'lastIdledAt' => $this->lastIdledAt, 'lastCallStartedAt' => $this->lastCallStartedAt, 'completedCallTimeTotal' => $this->completedCallTimeTotal, - 'invokeQueueCount' => $this->leakyQueue->count() + 'pendingInvokeCount' => ($this->rateLimited ? $this->invokeQueue->count() : 0) ]; } diff --git a/tests/Unit/RegistrationTest.php b/tests/Unit/RegistrationTest.php index 93476570..c2716de7 100644 --- a/tests/Unit/RegistrationTest.php +++ b/tests/Unit/RegistrationTest.php @@ -1,17 +1,20 @@ _calleeSession = new \Thruway\Session(new \Thruway\Transport\DummyTransport()); $this->_callerSession = new \Thruway\Session(new \Thruway\Transport\DummyTransport()); - $this->_registration = new \Thruway\Registration($this->_calleeSession, 'test_procedure'); + $this->_registration = new \Thruway\Registration($this->_calleeSession, 'test_procedure'); } public function testMakingCallIncrementsCallCount() @@ -31,9 +34,7 @@ public function testMakingCallIncrementsCallCount() $this->assertEquals(0, $this->_registration->getCurrentCallCount()); $callMsg = new \Thruway\Message\CallMessage( - \Thruway\Common\Utils::getUniqueId(), - new \stdClass(), - 'test_procedure' + \Thruway\Common\Utils::getUniqueId(), new \stdClass(), 'test_procedure' ); @@ -43,10 +44,8 @@ public function testMakingCallIncrementsCallCount() $this->_registration->processCall($call); $this->assertEquals(1, $this->_registration->getCurrentCallCount()); - - } - + public function testRateLimitedRegistration() { $procedure = new \Thruway\Procedure('rate.limit.procedure'); @@ -55,21 +54,23 @@ public function testRateLimitedRegistration() $calleeSession = new \Thruway\Session(new Thruway\Transport\DummyTransport()); $registerMsg = new \Thruway\Message\RegisterMessage( - \Thruway\Common\Utils::getUniqueId(), [ "_limit" => 0], 'rate.limit.procedure' + \Thruway\Common\Utils::getUniqueId(), [ "_limit" => 1], 'rate.limit.procedure' ); $procedure->processRegister($calleeSession, $registerMsg); $this->assertEquals(1, count($procedure->getRegistrations())); + + $this->assertTrue($procedure->getRegistrations()[0]->isRateLimited()); for ($i = 5; $i-- > 0;) { //send an invocation - $callMessage = new \Thruway\Message\CallMessage( - \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' - ); - $call = new \Thruway\Call($callerSession, $callMessage, $procedure); - $procedure->processCall($callerSession, $call); +// $callMessage = new \Thruway\Message\CallMessage( +// \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' +// ); +// $call = new \Thruway\Call($callerSession, $callMessage, $procedure); +// $procedure->processCall($callerSession, $call); } - $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['invokeQueueCount'])); +// $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['pendingInvokeCount'])); } /** @@ -79,4 +80,5 @@ public function testRemoveCall() { $this->assertTrue(true); } -} \ No newline at end of file + +} From 8b1390c2dae1cace970305c2945b25b217506681 Mon Sep 17 00:00:00 2001 From: Binaek Date: Thu, 31 Dec 2015 12:52:20 +0530 Subject: [PATCH 3/8] Adding Rate Limiting --- src/Thruway/Common/LeakyBucket.php | 8 ++-- src/Thruway/Registration.php | 60 +++++++++++++++++++----------- tests/Unit/RegistrationTest.php | 28 +++++++------- 3 files changed, 57 insertions(+), 39 deletions(-) diff --git a/src/Thruway/Common/LeakyBucket.php b/src/Thruway/Common/LeakyBucket.php index d6282c02..9de04c33 100644 --- a/src/Thruway/Common/LeakyBucket.php +++ b/src/Thruway/Common/LeakyBucket.php @@ -16,28 +16,26 @@ class LeakyBucket protected $minTime; //holds time of last action (past or future!) protected $lastSchedAction; - protected $eventLoop; - protected $timer; - protected $objectQueue; public function __construct($maxRatePerSecond = -1) { $this->maxRate = -1; - $this->lastSchedAction = time(); $this->setMaxRate($maxRatePerSecond); + $this->lastSchedAction = time() - $this->minTime; } public function setMaxRate($maxRatePerSecond) { if ($maxRatePerSecond > 0.0) { $this->maxRate = $maxRatePerSecond; + //milliseconds between successive calls $this->minTime = (int) (1000.0 / $maxRatePerSecond); } } public function canConsume() { - return ($this->timeLeft() <= 0); + return ($this->getTimeLeft() <= 0); } public function getTimeLeft() diff --git a/src/Thruway/Registration.php b/src/Thruway/Registration.php index a95ee703..22b1f415 100644 --- a/src/Thruway/Registration.php +++ b/src/Thruway/Registration.php @@ -3,6 +3,7 @@ namespace Thruway; use Thruway\Common\Utils; +use Thruway\Common\LeakyBucket; use Thruway\Message\ErrorMessage; use Thruway\Message\RegisterMessage; @@ -19,6 +20,21 @@ class Registration */ private $id; + /** + * @var int + */ + private $limit; + + /** + * @var LeakyBucket + */ + private $leakyBucket; + + /** + * @var \SplQueue + */ + private $invokeQueue; + /** * @var \Thruway\Session */ @@ -95,15 +111,10 @@ class Registration * @var float */ private $completedCallTimeTotal; -//for ratelimiting /** - * @var int + * @bool */ - private $limit; - private $eventLoop; - private $leakyBucket; - private $invokeQueue; private $rateLimited; const SINGLE_REGISTRATION = 'single'; @@ -138,7 +149,9 @@ public function __construct(Session $session, $procedureName) $this->busyStart = null; $this->completedCallTimeTotal = 0; + //no throtling by default $this->limit = -1; + $this->leakyBucket = new LeakyBucket(); $this->rateLimited = false; } @@ -245,12 +258,6 @@ public function setLimit($limit) $this->rateLimited = true; $this->invokeQueue = new \SplQueue(); $this->leakyBucket = new Common\LeakyBucket($this->limit); - $this->eventLoop = \React\EventLoop\Factory::create(); - $this->eventLoop->addPeriodicTimer(30, function() { - //just to keep the event loop running - //no idea if this is required - }); - $this->eventLoop->run(); } } @@ -297,24 +304,35 @@ public function processCall(Call $call) $this->invokeQueue->enqueue($call->getInvocationMessage()); if ($this->invokeQueue->count() === 1) { //start the timer if I am the first addition to the queue - $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); + $this->session->getLoop()->addTimer($this->leakyBucket->getTimeLeft() / 1000, $this); } } } else { $this->getSession()->sendMessage($call->getInvocationMessage()); } } - - public function isRateLimited() - { + + /** + * Get whether rate limited + * + * @return bool + */ + public function isRateLimited(){ return $this->rateLimited; } - private function invokeNextMessage() + /** + * Process Invocation Queue + * + * Using __invoke magic method + * + * @param none + */ + public function __invoke() { $this->getSession()->sendMessage($this->invokeQueue->dequeue()); if ($this->invokeQueue->count() > 0) { - $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); + $this->session->getLoop()->addTimer($this->leakyBucket->getTimeLeft() / 1000, $this); } } @@ -350,10 +368,10 @@ public function removeCall($callToRemove) $this->session->decPendingCallCount(); $callEnd = microtime(true); -// average call time + // average call time $callsInAverage = $this->invocationCount - count($this->calls) - 1; -// add this call time into the total + // add this call time into the total $this->completedCallTimeTotal += $callEnd - $call->getCallStart(); $callsInAverage++; $this->invocationAverageTime = ((float) $this->completedCallTimeTotal) / $callsInAverage; @@ -457,7 +475,7 @@ public function getStatistics() 'lastIdledAt' => $this->lastIdledAt, 'lastCallStartedAt' => $this->lastCallStartedAt, 'completedCallTimeTotal' => $this->completedCallTimeTotal, - 'pendingInvokeCount' => ($this->rateLimited ? $this->invokeQueue->count() : 0) + 'invokeQueueCount' => $this->rateLimited ? $this->invokeQueue->count() : 0 ]; } diff --git a/tests/Unit/RegistrationTest.php b/tests/Unit/RegistrationTest.php index c2716de7..c9364470 100644 --- a/tests/Unit/RegistrationTest.php +++ b/tests/Unit/RegistrationTest.php @@ -2,7 +2,7 @@ require_once __DIR__ . '/../bootstrap.php'; -class RegistrationTest extends PHPUnit_Framework_TestCase +class RegistrationTest extends Thruway\TestCase { /** @@ -52,25 +52,27 @@ public function testRateLimitedRegistration() $callerSession = new \Thruway\Session(new Thruway\Transport\DummyTransport()); $calleeSession = new \Thruway\Session(new Thruway\Transport\DummyTransport()); + + $calleeSession->setLoop(\React\EventLoop\Factory::create()); - $registerMsg = new \Thruway\Message\RegisterMessage( + $throttledRegisterMsg = new \Thruway\Message\RegisterMessage( \Thruway\Common\Utils::getUniqueId(), [ "_limit" => 1], 'rate.limit.procedure' ); - $procedure->processRegister($calleeSession, $registerMsg); + + $procedure->processRegister($calleeSession, $throttledRegisterMsg); $this->assertEquals(1, count($procedure->getRegistrations())); - + $this->assertTrue($procedure->getRegistrations()[0]->isRateLimited()); - for ($i = 5; $i-- > 0;) { - //send an invocation -// $callMessage = new \Thruway\Message\CallMessage( -// \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' -// ); -// $call = new \Thruway\Call($callerSession, $callMessage, $procedure); -// $procedure->processCall($callerSession, $call); - } -// $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['pendingInvokeCount'])); + $callMsg = new \Thruway\Message\CallMessage( + \Thruway\Common\Utils::getUniqueId(), new \stdClass(), 'rate.limit.procedure' + ); + $call = new \Thruway\Call($callerSession, $callMsg, $procedure); + + $procedure->getRegistrations()[0]->processCall($call); + + $this->assertEquals(1, $procedure->getRegistrations()[0]->getStatistics()['invokeQueueCount']); } /** From 109626a519abf5cb1a17cbc20f259b9fcc07a753 Mon Sep 17 00:00:00 2001 From: Binaek Date: Thu, 31 Dec 2015 13:02:10 +0530 Subject: [PATCH 4/8] Fixing incorrect key in Registration lookup for Random registration --- src/Thruway/Procedure.php | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/Thruway/Procedure.php b/src/Thruway/Procedure.php index a064dc6c..0d249c74 100644 --- a/src/Thruway/Procedure.php +++ b/src/Thruway/Procedure.php @@ -320,19 +320,21 @@ private function getNextRandomRegistration() //getting registrations with 0 unprocessed calls $possibleRegistrations = array_filter($this->registrations, function($theRegistration, $theIndex) { - return ($theRegistration->getStatistics()['pendingInvokeCount'] === 0); + return ($theRegistration->getStatistics()['invokeQueueCount'] === 0); }, ARRAY_FILTER_USE_BOTH); if (count($possibleRegistrations) > 0) { //return a random from this set return $possibleRegistrations[mt_rand(0, count($possibleRegistrations) - 1)]; + } else if (count($possibleRegistrations) === 1) { + return $possibleRegistrations[0]; } else { //create a copy to maintain the original indexing $possibleRegistrations = array_merge([], $this->registrations); //sort ascending by number of unprocessed Invocations usort($possibleRegistrations, function($registrationA, $registrationB) { - $unprocessedA = $registrationA->getStatistics()['pendingInvokeCount']; - $unprocessedB = $registrationB->getStatistics()['pendingInvokeCount']; + $unprocessedA = $registrationA->getStatistics()['invokeQueueCount']; + $unprocessedB = $registrationB->getStatistics()['invokeQueueCount']; if ($unprocessedA == $unprocessedB) { return 0; } else if ($unprocessedA < $unprocessedB) { From ccff14e52e6ce13f3768fdd7848498f3cc7f4f1a Mon Sep 17 00:00:00 2001 From: Binaek Sarkar Date: Sat, 12 Dec 2015 01:51:23 +0530 Subject: [PATCH 5/8] Adding Throttling in individual Procedure registrations --- src/Thruway/Common/LeakyBucket.php | 71 ++++++++++++++++++++++++++++++ src/Thruway/Procedure.php | 29 +++++++++++- src/Thruway/Registration.php | 62 +++++++++++++++++++++++--- tests/Unit/RegistrationTest.php | 26 +++++++++++ 4 files changed, 181 insertions(+), 7 deletions(-) create mode 100644 src/Thruway/Common/LeakyBucket.php diff --git a/src/Thruway/Common/LeakyBucket.php b/src/Thruway/Common/LeakyBucket.php new file mode 100644 index 00000000..24ed27a7 --- /dev/null +++ b/src/Thruway/Common/LeakyBucket.php @@ -0,0 +1,71 @@ +maxRate = -1; + $this->objectQueue = new SplQueue(); + $this->lastSchedAction = time(); + $this->setMaxRate($maxRatePerSecond); + } + + public function enqueue($anyObject) + { + $this->objectQueue->enqueue($anyObject); + } + + public function count() + { + return $this->objectQueue->count(); + } + + public function setMaxRate($maxRatePerSecond) + { + if ($maxRatePerSecond > 0.0) { + $this->maxRate = $maxRatePerSecond; + $this->minTime = (int) (1000.0 / $maxRatePerSecond); + } + } + + public function consume() + { + if ($this->maxRate > 0) { + //we are rate limited + $curTime = time(); + //calculate when can we send back + $timeLeft = $this->lastSchedAction + $this->minTime - $curTime; + if ($timeLeft > 0) { + $this->lastSchedAction += $this->minTime; + //we need to sleep for sometime + echo "We are sleeping"; + sleep($timeLeft); + } else { + $this->lastSchedAction = $this->curTime; + } + } + //lets go back + return $this->objectQueue->dequeue(); + } + +} diff --git a/src/Thruway/Procedure.php b/src/Thruway/Procedure.php index d739de2b..babe740b 100644 --- a/src/Thruway/Procedure.php +++ b/src/Thruway/Procedure.php @@ -317,8 +317,33 @@ private function getNextRandomRegistration() //just return this so that we don't have to run mt_rand return $this->registrations[0]; } - //mt_rand is apparently faster than array_rand(which uses the libc generator) - return $this->registrations[mt_rand(0, count($this->registrations) - 1)]; + + //getting registrations with 0 unprocessed calls + $possibleRegistrations = array_filter($this->registrations, function($theRegistration, $theIndex) { + return ($theRegistration->getStatistics()['invokeQueueCount'] === 0); + }, ARRAY_FILTER_USE_BOTH); + + if (count($possibleRegistrations) > 0) { + //return a random from this set + return $possibleRegistrations[mt_rand(0, count($possibleRegistrations) - 1)]; + } else { + //create a copy to maintain the original indexing + $possibleRegistrations = array_merge([], $this->registrations); + //sort ascending by number of unprocessed Invocations + usort($possibleRegistrations, function($registrationA, $registrationB) { + $unprocessedA = $registrationA->getStatistics()['invokeQueueCount']; + $unprocessedB = $registrationB->getStatistics()['invokeQueueCount']; + if ($unprocessedA == $unprocessedB) { + return 0; + } else if ($unprocessedA < $unprocessedB) { + return -1; + } else { + return 1; + } + }); + //return the first one + return $possibleRegistrations[0]; + } } private function getNextRoundRobinRegistration() diff --git a/src/Thruway/Registration.php b/src/Thruway/Registration.php index 1629fed0..1cebb544 100644 --- a/src/Thruway/Registration.php +++ b/src/Thruway/Registration.php @@ -3,6 +3,7 @@ namespace Thruway; use Thruway\Common\Utils; +use Thruway\Common\LeakyBucket; use Thruway\Message\ErrorMessage; use Thruway\Message\RegisterMessage; @@ -19,6 +20,16 @@ class Registration */ private $id; + /** + * @var int + */ + private $limit; + + /** + * @var LeakyBucket + */ + private $leakyQueue; + /** * @var \Thruway\Session */ @@ -97,11 +108,11 @@ class Registration private $completedCallTimeTotal; const SINGLE_REGISTRATION = 'single'; - const THRUWAY_REGISTRATION = '_thruway'; const ROUNDROBIN_REGISTRATION = 'roundrobin'; const RANDOM_REGISTRATION = 'random'; const FIRST_REGISTRATION = 'first'; const LAST_REGISTRATION = 'last'; + const THRUWAY_REGISTRATION = '_thruway'; /** * Constructor @@ -127,6 +138,9 @@ public function __construct(Session $session, $procedureName) $this->lastIdledAt = $this->registeredAt; $this->busyStart = null; $this->completedCallTimeTotal = 0; + + $this->limit = -1; + $this->leakyQueue = new LeakyBucket(); //no throtling by default } /** @@ -154,6 +168,11 @@ public static function createRegistrationFromRegisterMessage(Session $session, R $registration->setInvokeType(Registration::SINGLE_REGISTRATION); } } + if (isset($options->_limit) && settype($options->_limit, "integer")) { + $registration->setLimit($options->_limit); + } else { + $registration->setLimit(-1); //setting to UNLIMITED + } return $registration; } @@ -210,14 +229,31 @@ public function setInvokeType($type) if ($type !== Registration::SINGLE_REGISTRATION) { $this->invokeType = $type; $this->setAllowMultipleRegistrations(true); - } - else { + } else { $this->invokeType = Registration::SINGLE_REGISTRATION; $this->setAllowMultipleRegistrations(false); } } } + /** + * @param int $limit The number of calls allowed per second + */ + public function setLimit($limit) + { + $this->limit = $limit; + $this->leakyQueue->setMaxRate($limit); + } + + /** + * Get the Limit per second on this registrations + * @return int + */ + public function getLimit() + { + return $this->limit; + } + /** * Process call * @@ -245,7 +281,22 @@ public function processCall(Call $call) $this->invocationCount++; $this->lastCallStartedAt = new \DateTime(); - $this->getSession()->sendMessage($call->getInvocationMessage()); + $this->leakyQueue->enqueue($call->getInvocationMessage()); + + $this->processInvocationQueue(); + } + + /** + * Process Invocation Queue + * + * @param none + */ + private function processInvocationQueue() + { + while ($this->leakyQueue->count() > 0) { + //this will sleep till we can actually make the call + $this->getSession()->sendMessage($this->leakyQueue->consume()); + } } /** @@ -386,7 +437,8 @@ public function getStatistics() 'busyStart' => $this->busyStart, 'lastIdledAt' => $this->lastIdledAt, 'lastCallStartedAt' => $this->lastCallStartedAt, - 'completedCallTimeTotal' => $this->completedCallTimeTotal + 'completedCallTimeTotal' => $this->completedCallTimeTotal, + 'invokeQueueCount' => $this->leakyQueue->count() ]; } diff --git a/tests/Unit/RegistrationTest.php b/tests/Unit/RegistrationTest.php index 626067f6..93476570 100644 --- a/tests/Unit/RegistrationTest.php +++ b/tests/Unit/RegistrationTest.php @@ -1,4 +1,5 @@ 0], 'rate.limit.procedure' + ); + $procedure->processRegister($calleeSession, $registerMsg); + + $this->assertEquals(1, count($procedure->getRegistrations())); + + for ($i = 5; $i-- > 0;) { + //send an invocation + $callMessage = new \Thruway\Message\CallMessage( + \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' + ); + $call = new \Thruway\Call($callerSession, $callMessage, $procedure); + $procedure->processCall($callerSession, $call); + } + $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['invokeQueueCount'])); + } /** * xdepends testAddCall From 4aa27bb067e825e0764d900f8b99b0959669d21e Mon Sep 17 00:00:00 2001 From: Binaek Date: Tue, 15 Dec 2015 18:07:14 +0530 Subject: [PATCH 6/8] --- src/Thruway/Common/LeakyBucket.php | 43 +++++++---------- src/Thruway/Procedure.php | 6 +-- src/Thruway/Registration.php | 75 +++++++++++++++++++----------- tests/Unit/RegistrationTest.php | 34 +++++++------- 4 files changed, 85 insertions(+), 73 deletions(-) diff --git a/src/Thruway/Common/LeakyBucket.php b/src/Thruway/Common/LeakyBucket.php index 24ed27a7..d6282c02 100644 --- a/src/Thruway/Common/LeakyBucket.php +++ b/src/Thruway/Common/LeakyBucket.php @@ -16,30 +16,17 @@ class LeakyBucket protected $minTime; //holds time of last action (past or future!) protected $lastSchedAction; - - /** - * @var SplQueue The Object Storage - */ + protected $eventLoop; + protected $timer; protected $objectQueue; public function __construct($maxRatePerSecond = -1) { $this->maxRate = -1; - $this->objectQueue = new SplQueue(); $this->lastSchedAction = time(); $this->setMaxRate($maxRatePerSecond); } - public function enqueue($anyObject) - { - $this->objectQueue->enqueue($anyObject); - } - - public function count() - { - return $this->objectQueue->count(); - } - public function setMaxRate($maxRatePerSecond) { if ($maxRatePerSecond > 0.0) { @@ -48,24 +35,28 @@ public function setMaxRate($maxRatePerSecond) } } - public function consume() + public function canConsume() { + return ($this->timeLeft() <= 0); + } + + public function getTimeLeft() + { + $timeLeft = 0; if ($this->maxRate > 0) { //we are rate limited $curTime = time(); //calculate when can we send back $timeLeft = $this->lastSchedAction + $this->minTime - $curTime; - if ($timeLeft > 0) { - $this->lastSchedAction += $this->minTime; - //we need to sleep for sometime - echo "We are sleeping"; - sleep($timeLeft); - } else { - $this->lastSchedAction = $this->curTime; - } } - //lets go back - return $this->objectQueue->dequeue(); + return $timeLeft; + } + + public function consume() + { + if ($this->canConsume()) { + $this->lastSchedAction = time(); + } } } diff --git a/src/Thruway/Procedure.php b/src/Thruway/Procedure.php index babe740b..a064dc6c 100644 --- a/src/Thruway/Procedure.php +++ b/src/Thruway/Procedure.php @@ -320,7 +320,7 @@ private function getNextRandomRegistration() //getting registrations with 0 unprocessed calls $possibleRegistrations = array_filter($this->registrations, function($theRegistration, $theIndex) { - return ($theRegistration->getStatistics()['invokeQueueCount'] === 0); + return ($theRegistration->getStatistics()['pendingInvokeCount'] === 0); }, ARRAY_FILTER_USE_BOTH); if (count($possibleRegistrations) > 0) { @@ -331,8 +331,8 @@ private function getNextRandomRegistration() $possibleRegistrations = array_merge([], $this->registrations); //sort ascending by number of unprocessed Invocations usort($possibleRegistrations, function($registrationA, $registrationB) { - $unprocessedA = $registrationA->getStatistics()['invokeQueueCount']; - $unprocessedB = $registrationB->getStatistics()['invokeQueueCount']; + $unprocessedA = $registrationA->getStatistics()['pendingInvokeCount']; + $unprocessedB = $registrationB->getStatistics()['pendingInvokeCount']; if ($unprocessedA == $unprocessedB) { return 0; } else if ($unprocessedA < $unprocessedB) { diff --git a/src/Thruway/Registration.php b/src/Thruway/Registration.php index 1cebb544..a95ee703 100644 --- a/src/Thruway/Registration.php +++ b/src/Thruway/Registration.php @@ -3,7 +3,6 @@ namespace Thruway; use Thruway\Common\Utils; -use Thruway\Common\LeakyBucket; use Thruway\Message\ErrorMessage; use Thruway\Message\RegisterMessage; @@ -20,16 +19,6 @@ class Registration */ private $id; - /** - * @var int - */ - private $limit; - - /** - * @var LeakyBucket - */ - private $leakyQueue; - /** * @var \Thruway\Session */ @@ -106,6 +95,16 @@ class Registration * @var float */ private $completedCallTimeTotal; +//for ratelimiting + + /** + * @var int + */ + private $limit; + private $eventLoop; + private $leakyBucket; + private $invokeQueue; + private $rateLimited; const SINGLE_REGISTRATION = 'single'; const ROUNDROBIN_REGISTRATION = 'roundrobin'; @@ -140,7 +139,7 @@ public function __construct(Session $session, $procedureName) $this->completedCallTimeTotal = 0; $this->limit = -1; - $this->leakyQueue = new LeakyBucket(); //no throtling by default + $this->rateLimited = false; } /** @@ -242,7 +241,17 @@ public function setInvokeType($type) public function setLimit($limit) { $this->limit = $limit; - $this->leakyQueue->setMaxRate($limit); + if ($limit > 0) { + $this->rateLimited = true; + $this->invokeQueue = new \SplQueue(); + $this->leakyBucket = new Common\LeakyBucket($this->limit); + $this->eventLoop = \React\EventLoop\Factory::create(); + $this->eventLoop->addPeriodicTimer(30, function() { + //just to keep the event loop running + //no idea if this is required + }); + $this->eventLoop->run(); + } } /** @@ -280,22 +289,32 @@ public function processCall(Call $call) } $this->invocationCount++; $this->lastCallStartedAt = new \DateTime(); + if ($this->rateLimited) { + if ($this->leakyBucket->canConsume()) { + $this->leakyBucket->consume(); + $this->getSession()->sendMessage($call->getInvocationMessage()); + } else { + $this->invokeQueue->enqueue($call->getInvocationMessage()); + if ($this->invokeQueue->count() === 1) { + //start the timer if I am the first addition to the queue + $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); + } + } + } else { + $this->getSession()->sendMessage($call->getInvocationMessage()); + } + } - $this->leakyQueue->enqueue($call->getInvocationMessage()); - - $this->processInvocationQueue(); + public function isRateLimited() + { + return $this->rateLimited; } - /** - * Process Invocation Queue - * - * @param none - */ - private function processInvocationQueue() + private function invokeNextMessage() { - while ($this->leakyQueue->count() > 0) { - //this will sleep till we can actually make the call - $this->getSession()->sendMessage($this->leakyQueue->consume()); + $this->getSession()->sendMessage($this->invokeQueue->dequeue()); + if ($this->invokeQueue->count() > 0) { + $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); } } @@ -331,10 +350,10 @@ public function removeCall($callToRemove) $this->session->decPendingCallCount(); $callEnd = microtime(true); - // average call time +// average call time $callsInAverage = $this->invocationCount - count($this->calls) - 1; - // add this call time into the total +// add this call time into the total $this->completedCallTimeTotal += $callEnd - $call->getCallStart(); $callsInAverage++; $this->invocationAverageTime = ((float) $this->completedCallTimeTotal) / $callsInAverage; @@ -438,7 +457,7 @@ public function getStatistics() 'lastIdledAt' => $this->lastIdledAt, 'lastCallStartedAt' => $this->lastCallStartedAt, 'completedCallTimeTotal' => $this->completedCallTimeTotal, - 'invokeQueueCount' => $this->leakyQueue->count() + 'pendingInvokeCount' => ($this->rateLimited ? $this->invokeQueue->count() : 0) ]; } diff --git a/tests/Unit/RegistrationTest.php b/tests/Unit/RegistrationTest.php index 93476570..c2716de7 100644 --- a/tests/Unit/RegistrationTest.php +++ b/tests/Unit/RegistrationTest.php @@ -1,17 +1,20 @@ _calleeSession = new \Thruway\Session(new \Thruway\Transport\DummyTransport()); $this->_callerSession = new \Thruway\Session(new \Thruway\Transport\DummyTransport()); - $this->_registration = new \Thruway\Registration($this->_calleeSession, 'test_procedure'); + $this->_registration = new \Thruway\Registration($this->_calleeSession, 'test_procedure'); } public function testMakingCallIncrementsCallCount() @@ -31,9 +34,7 @@ public function testMakingCallIncrementsCallCount() $this->assertEquals(0, $this->_registration->getCurrentCallCount()); $callMsg = new \Thruway\Message\CallMessage( - \Thruway\Common\Utils::getUniqueId(), - new \stdClass(), - 'test_procedure' + \Thruway\Common\Utils::getUniqueId(), new \stdClass(), 'test_procedure' ); @@ -43,10 +44,8 @@ public function testMakingCallIncrementsCallCount() $this->_registration->processCall($call); $this->assertEquals(1, $this->_registration->getCurrentCallCount()); - - } - + public function testRateLimitedRegistration() { $procedure = new \Thruway\Procedure('rate.limit.procedure'); @@ -55,21 +54,23 @@ public function testRateLimitedRegistration() $calleeSession = new \Thruway\Session(new Thruway\Transport\DummyTransport()); $registerMsg = new \Thruway\Message\RegisterMessage( - \Thruway\Common\Utils::getUniqueId(), [ "_limit" => 0], 'rate.limit.procedure' + \Thruway\Common\Utils::getUniqueId(), [ "_limit" => 1], 'rate.limit.procedure' ); $procedure->processRegister($calleeSession, $registerMsg); $this->assertEquals(1, count($procedure->getRegistrations())); + + $this->assertTrue($procedure->getRegistrations()[0]->isRateLimited()); for ($i = 5; $i-- > 0;) { //send an invocation - $callMessage = new \Thruway\Message\CallMessage( - \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' - ); - $call = new \Thruway\Call($callerSession, $callMessage, $procedure); - $procedure->processCall($callerSession, $call); +// $callMessage = new \Thruway\Message\CallMessage( +// \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' +// ); +// $call = new \Thruway\Call($callerSession, $callMessage, $procedure); +// $procedure->processCall($callerSession, $call); } - $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['invokeQueueCount'])); +// $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['pendingInvokeCount'])); } /** @@ -79,4 +80,5 @@ public function testRemoveCall() { $this->assertTrue(true); } -} \ No newline at end of file + +} From 78bc037b2d049a77d52a0b1bad3c9ac4014f3644 Mon Sep 17 00:00:00 2001 From: Binaek Date: Thu, 31 Dec 2015 12:52:20 +0530 Subject: [PATCH 7/8] Adding Rate Limiting --- src/Thruway/Common/LeakyBucket.php | 8 ++-- src/Thruway/Registration.php | 60 +++++++++++++++++++----------- tests/Unit/RegistrationTest.php | 28 +++++++------- 3 files changed, 57 insertions(+), 39 deletions(-) diff --git a/src/Thruway/Common/LeakyBucket.php b/src/Thruway/Common/LeakyBucket.php index d6282c02..9de04c33 100644 --- a/src/Thruway/Common/LeakyBucket.php +++ b/src/Thruway/Common/LeakyBucket.php @@ -16,28 +16,26 @@ class LeakyBucket protected $minTime; //holds time of last action (past or future!) protected $lastSchedAction; - protected $eventLoop; - protected $timer; - protected $objectQueue; public function __construct($maxRatePerSecond = -1) { $this->maxRate = -1; - $this->lastSchedAction = time(); $this->setMaxRate($maxRatePerSecond); + $this->lastSchedAction = time() - $this->minTime; } public function setMaxRate($maxRatePerSecond) { if ($maxRatePerSecond > 0.0) { $this->maxRate = $maxRatePerSecond; + //milliseconds between successive calls $this->minTime = (int) (1000.0 / $maxRatePerSecond); } } public function canConsume() { - return ($this->timeLeft() <= 0); + return ($this->getTimeLeft() <= 0); } public function getTimeLeft() diff --git a/src/Thruway/Registration.php b/src/Thruway/Registration.php index a95ee703..22b1f415 100644 --- a/src/Thruway/Registration.php +++ b/src/Thruway/Registration.php @@ -3,6 +3,7 @@ namespace Thruway; use Thruway\Common\Utils; +use Thruway\Common\LeakyBucket; use Thruway\Message\ErrorMessage; use Thruway\Message\RegisterMessage; @@ -19,6 +20,21 @@ class Registration */ private $id; + /** + * @var int + */ + private $limit; + + /** + * @var LeakyBucket + */ + private $leakyBucket; + + /** + * @var \SplQueue + */ + private $invokeQueue; + /** * @var \Thruway\Session */ @@ -95,15 +111,10 @@ class Registration * @var float */ private $completedCallTimeTotal; -//for ratelimiting /** - * @var int + * @bool */ - private $limit; - private $eventLoop; - private $leakyBucket; - private $invokeQueue; private $rateLimited; const SINGLE_REGISTRATION = 'single'; @@ -138,7 +149,9 @@ public function __construct(Session $session, $procedureName) $this->busyStart = null; $this->completedCallTimeTotal = 0; + //no throtling by default $this->limit = -1; + $this->leakyBucket = new LeakyBucket(); $this->rateLimited = false; } @@ -245,12 +258,6 @@ public function setLimit($limit) $this->rateLimited = true; $this->invokeQueue = new \SplQueue(); $this->leakyBucket = new Common\LeakyBucket($this->limit); - $this->eventLoop = \React\EventLoop\Factory::create(); - $this->eventLoop->addPeriodicTimer(30, function() { - //just to keep the event loop running - //no idea if this is required - }); - $this->eventLoop->run(); } } @@ -297,24 +304,35 @@ public function processCall(Call $call) $this->invokeQueue->enqueue($call->getInvocationMessage()); if ($this->invokeQueue->count() === 1) { //start the timer if I am the first addition to the queue - $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); + $this->session->getLoop()->addTimer($this->leakyBucket->getTimeLeft() / 1000, $this); } } } else { $this->getSession()->sendMessage($call->getInvocationMessage()); } } - - public function isRateLimited() - { + + /** + * Get whether rate limited + * + * @return bool + */ + public function isRateLimited(){ return $this->rateLimited; } - private function invokeNextMessage() + /** + * Process Invocation Queue + * + * Using __invoke magic method + * + * @param none + */ + public function __invoke() { $this->getSession()->sendMessage($this->invokeQueue->dequeue()); if ($this->invokeQueue->count() > 0) { - $this->eventLoop->addTimer($this->leakyBucket->getTimeLeft() / 1000, 'invokeNextMessage'); + $this->session->getLoop()->addTimer($this->leakyBucket->getTimeLeft() / 1000, $this); } } @@ -350,10 +368,10 @@ public function removeCall($callToRemove) $this->session->decPendingCallCount(); $callEnd = microtime(true); -// average call time + // average call time $callsInAverage = $this->invocationCount - count($this->calls) - 1; -// add this call time into the total + // add this call time into the total $this->completedCallTimeTotal += $callEnd - $call->getCallStart(); $callsInAverage++; $this->invocationAverageTime = ((float) $this->completedCallTimeTotal) / $callsInAverage; @@ -457,7 +475,7 @@ public function getStatistics() 'lastIdledAt' => $this->lastIdledAt, 'lastCallStartedAt' => $this->lastCallStartedAt, 'completedCallTimeTotal' => $this->completedCallTimeTotal, - 'pendingInvokeCount' => ($this->rateLimited ? $this->invokeQueue->count() : 0) + 'invokeQueueCount' => $this->rateLimited ? $this->invokeQueue->count() : 0 ]; } diff --git a/tests/Unit/RegistrationTest.php b/tests/Unit/RegistrationTest.php index c2716de7..c9364470 100644 --- a/tests/Unit/RegistrationTest.php +++ b/tests/Unit/RegistrationTest.php @@ -2,7 +2,7 @@ require_once __DIR__ . '/../bootstrap.php'; -class RegistrationTest extends PHPUnit_Framework_TestCase +class RegistrationTest extends Thruway\TestCase { /** @@ -52,25 +52,27 @@ public function testRateLimitedRegistration() $callerSession = new \Thruway\Session(new Thruway\Transport\DummyTransport()); $calleeSession = new \Thruway\Session(new Thruway\Transport\DummyTransport()); + + $calleeSession->setLoop(\React\EventLoop\Factory::create()); - $registerMsg = new \Thruway\Message\RegisterMessage( + $throttledRegisterMsg = new \Thruway\Message\RegisterMessage( \Thruway\Common\Utils::getUniqueId(), [ "_limit" => 1], 'rate.limit.procedure' ); - $procedure->processRegister($calleeSession, $registerMsg); + + $procedure->processRegister($calleeSession, $throttledRegisterMsg); $this->assertEquals(1, count($procedure->getRegistrations())); - + $this->assertTrue($procedure->getRegistrations()[0]->isRateLimited()); - for ($i = 5; $i-- > 0;) { - //send an invocation -// $callMessage = new \Thruway\Message\CallMessage( -// \Thruway\Common\Utils::getUniqueId(), [], 'rate.limit.procedure' -// ); -// $call = new \Thruway\Call($callerSession, $callMessage, $procedure); -// $procedure->processCall($callerSession, $call); - } -// $this->assertEquals(1, count($procedure->getRegistrations()[0]->getStatistics()['pendingInvokeCount'])); + $callMsg = new \Thruway\Message\CallMessage( + \Thruway\Common\Utils::getUniqueId(), new \stdClass(), 'rate.limit.procedure' + ); + $call = new \Thruway\Call($callerSession, $callMsg, $procedure); + + $procedure->getRegistrations()[0]->processCall($call); + + $this->assertEquals(1, $procedure->getRegistrations()[0]->getStatistics()['invokeQueueCount']); } /** From 0154289b77dabd3ce221af3f13e773fd32fa0ee3 Mon Sep 17 00:00:00 2001 From: Binaek Date: Thu, 31 Dec 2015 13:02:10 +0530 Subject: [PATCH 8/8] Fixing incorrect key in Registration lookup for Random registration --- src/Thruway/Procedure.php | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/Thruway/Procedure.php b/src/Thruway/Procedure.php index a064dc6c..0d249c74 100644 --- a/src/Thruway/Procedure.php +++ b/src/Thruway/Procedure.php @@ -320,19 +320,21 @@ private function getNextRandomRegistration() //getting registrations with 0 unprocessed calls $possibleRegistrations = array_filter($this->registrations, function($theRegistration, $theIndex) { - return ($theRegistration->getStatistics()['pendingInvokeCount'] === 0); + return ($theRegistration->getStatistics()['invokeQueueCount'] === 0); }, ARRAY_FILTER_USE_BOTH); if (count($possibleRegistrations) > 0) { //return a random from this set return $possibleRegistrations[mt_rand(0, count($possibleRegistrations) - 1)]; + } else if (count($possibleRegistrations) === 1) { + return $possibleRegistrations[0]; } else { //create a copy to maintain the original indexing $possibleRegistrations = array_merge([], $this->registrations); //sort ascending by number of unprocessed Invocations usort($possibleRegistrations, function($registrationA, $registrationB) { - $unprocessedA = $registrationA->getStatistics()['pendingInvokeCount']; - $unprocessedB = $registrationB->getStatistics()['pendingInvokeCount']; + $unprocessedA = $registrationA->getStatistics()['invokeQueueCount']; + $unprocessedB = $registrationB->getStatistics()['invokeQueueCount']; if ($unprocessedA == $unprocessedB) { return 0; } else if ($unprocessedA < $unprocessedB) {