From 427ac3a3a6eafa9a82006e9f0a9c2ab19a09fb4b Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Wed, 27 Jan 2010 20:51:04 -0800 Subject: debug log line for control channel sub --- lib/stompqueuemanager.php | 1 + 1 file changed, 1 insertion(+) (limited to 'lib') diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php index 19e8c49b5..da70d9ae3 100644 --- a/lib/stompqueuemanager.php +++ b/lib/stompqueuemanager.php @@ -242,6 +242,7 @@ class StompQueueManager extends QueueManager parent::start($master); $this->_connect(); + common_log(LOG_INFO, "Subscribing to $this->control"); $this->con->subscribe($this->control); if ($this->sites) { foreach ($this->sites as $server) { -- cgit v1.2.3-54-g00ecf From 1ba8045a9b318fd19fb55d2c469bd23a3976c2e8 Mon Sep 17 00:00:00 2001 From: Evan Prodromou Date: Thu, 28 Jan 2010 01:24:00 -0500 Subject: set session cookie correctly --- lib/util.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/util.php b/lib/util.php index 4312f9876..c6dc4b43a 100644 --- a/lib/util.php +++ b/lib/util.php @@ -178,7 +178,7 @@ function common_ensure_session() } if (isset($id)) { session_id($id); - setcookie(session_name(), $id); + setcookie(session_name(), $id, 0, common_config('site', 'path')); } @session_start(); if (!isset($_SESSION['started'])) { -- cgit v1.2.3-54-g00ecf From 65c4cff01c42a5c432db968cf6731104b8040134 Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Wed, 27 Jan 2010 23:14:49 -0800 Subject: append '/' on cookie path for now (may still need some refactoring) --- lib/util.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/util.php b/lib/util.php index c6dc4b43a..26a12871b 100644 --- a/lib/util.php +++ b/lib/util.php @@ -178,7 +178,7 @@ function common_ensure_session() } if (isset($id)) { session_id($id); - setcookie(session_name(), $id, 0, common_config('site', 'path')); + setcookie(session_name(), $id, 0, common_config('site', 'path') . '/'); } @session_start(); if (!isset($_SESSION['started'])) { -- cgit v1.2.3-54-g00ecf From 78fe76b058157670bf6c0ab5f3454733d465684e Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Wed, 27 Jan 2010 23:16:06 -0800 Subject: dropping the setcookie() call from common_ensure_session() since we're pretty sure it's unnecessary --- lib/util.php | 1 - 1 file changed, 1 deletion(-) (limited to 'lib') diff --git a/lib/util.php b/lib/util.php index 26a12871b..dd8189a58 100644 --- a/lib/util.php +++ b/lib/util.php @@ -178,7 +178,6 @@ function common_ensure_session() } if (isset($id)) { session_id($id); - setcookie(session_name(), $id, 0, common_config('site', 'path') . '/'); } @session_start(); if (!isset($_SESSION['started'])) { -- cgit v1.2.3-54-g00ecf From a868a523a5ab042e75d333298a75aaa369d445cc Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Thu, 28 Jan 2010 09:52:35 -0800 Subject: Can now set $config['queue']['stomp_persistent'] = false; to explicitly disable persistence when we queue items --- lib/default.php | 1 + lib/stompqueuemanager.php | 9 +++++---- 2 files changed, 6 insertions(+), 4 deletions(-) (limited to 'lib') diff --git a/lib/default.php b/lib/default.php index c729193b5..8de8b1097 100644 --- a/lib/default.php +++ b/lib/default.php @@ -84,6 +84,7 @@ $default = 'control_channel' => '/topic/statusnet-control', // broadcasts to all queue daemons 'stomp_username' => null, 'stomp_password' => null, + 'stomp_persistent' => true, // keep items across queue server restart, if persistence is enabled 'monitor' => null, // URL to monitor ping endpoint (work in progress) 'softlimit' => '90%', // total size or % of memory_limit at which to restart queue threads gracefully 'debug_memory' => false, // true to spit memory usage to log diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php index da70d9ae3..4e2b58602 100644 --- a/lib/stompqueuemanager.php +++ b/lib/stompqueuemanager.php @@ -174,12 +174,13 @@ class StompQueueManager extends QueueManager $this->_connect(); - // XXX: serialize and send entire notice - + $props = array('created' => common_sql_now()); + if (common_config('queue', 'stomp_persistent')) { + $props['persistent'] = 'true'; + } $result = $this->con->send($this->queueName($queue), $msg, // BODY of the message - array ('created' => common_sql_now(), - 'persistent' => 'true')); + $props); if (!$result) { common_log(LOG_ERR, "Error sending $rep to $queue queue"); -- cgit v1.2.3-54-g00ecf From 155a5d446f96651abf3eb62f9b5748e4bdfa0a76 Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Thu, 28 Jan 2010 16:49:32 -0800 Subject: Manual failover for stomp queues. If an array of multiple servers is put in $config['queue']['stomp_server'], enqueues will pick a random server to send to (failing over automatically if any are down). Queue handling daemons connect all servers so they get events no matter where they were delivered. In case of disconnection, daemons should now handle it gracefully and attempt to reconnect every 60 seconds or so, automatically resubscribing to all queues once it's back up. Can put to 'native' failover for reads as well by disabling $config['stomp']['manual_failover'] = false; but this is untested and may explode in addition to requiring that your ActiveMQ cluster actually be set up to handle its own data distribution. Additionally, can choose which queues to mark as persistent by setting $config['stomp']['persistent'] to an array of queue names. --- lib/default.php | 1 + lib/liberalstomp.php | 24 +++- lib/stompqueuemanager.php | 354 ++++++++++++++++++++++++++++++++++++---------- 3 files changed, 299 insertions(+), 80 deletions(-) (limited to 'lib') diff --git a/lib/default.php b/lib/default.php index 8de8b1097..c01508695 100644 --- a/lib/default.php +++ b/lib/default.php @@ -85,6 +85,7 @@ $default = 'stomp_username' => null, 'stomp_password' => null, 'stomp_persistent' => true, // keep items across queue server restart, if persistence is enabled + 'stomp_manual_failover' => true, // if multiple servers are listed, treat them as separate (enqueue on one randomly, listen on all) 'monitor' => null, // URL to monitor ping endpoint (work in progress) 'softlimit' => '90%', // total size or % of memory_limit at which to restart queue threads gracefully 'debug_memory' => false, // true to spit memory usage to log diff --git a/lib/liberalstomp.php b/lib/liberalstomp.php index c9233843a..3d38953fd 100644 --- a/lib/liberalstomp.php +++ b/lib/liberalstomp.php @@ -33,6 +33,22 @@ class LiberalStomp extends Stomp return $this->_socket; } + /** + * Return the host we're currently connected to. + * + * @return string + */ + function getServer() + { + $idx = $this->_currentHost; + if ($idx >= 0) { + $host = $this->_hosts[$idx]; + return "$host[0]:$host[1]"; + } else { + return '[unconnected]'; + } + } + /** * Make socket connection to the server * We also set the stream to non-blocking mode, since we'll be @@ -71,10 +87,12 @@ class LiberalStomp extends Stomp // @fixme this sometimes hangs in blocking mode... // shouldn't we have been idle until we found there's more data? $read = fread($this->_socket, $rb); - if ($read === false) { - $this->_reconnect(); + if ($read === false || ($read === '' && feof($this->_socket))) { + // @fixme possibly attempt an auto reconnect as old code? + throw new StompException("Error reading"); + //$this->_reconnect(); // @fixme this will lose prior items - return $this->readFrames(); + //return $this->readFrames(); } $data .= $read; if (strpos($data, "\x00") !== false) { diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php index 4e2b58602..ec150bbb6 100644 --- a/lib/stompqueuemanager.php +++ b/lib/stompqueuemanager.php @@ -29,28 +29,37 @@ */ require_once 'Stomp.php'; +require_once 'Stomp/Exception.php'; class StompQueueManager extends QueueManager { - var $server = null; - var $username = null; - var $password = null; - var $base = null; - var $con = null; + protected $servers; + protected $username; + protected $password; + protected $base; protected $control; + + protected $useTransactions = true; protected $sites = array(); protected $subscriptions = array(); - protected $useTransactions = true; - protected $transaction = null; - protected $transactionCount = 0; + protected $cons = array(); // all open connections + protected $disconnect = array(); + protected $transaction = array(); + protected $transactionCount = array(); + protected $defaultIdx = 0; function __construct() { parent::__construct(); - $this->server = common_config('queue', 'stomp_server'); + $server = common_config('queue', 'stomp_server'); + if (is_array($server)) { + $this->servers = $server; + } else { + $this->servers = array($server); + } $this->username = common_config('queue', 'stomp_username'); $this->password = common_config('queue', 'stomp_password'); $this->base = common_config('queue', 'queue_basename'); @@ -99,9 +108,9 @@ class StompQueueManager extends QueueManager $message .= ':' . $param; } $this->_connect(); - $result = $this->con->send($this->control, - $message, - array ('created' => common_sql_now())); + $result = $this->_send($this->control, + $message, + array ('created' => common_sql_now())); if ($result) { $this->_log(LOG_INFO, "Sent control ping to queue daemons: $message"); return true; @@ -166,29 +175,59 @@ class StompQueueManager extends QueueManager /** * Saves a notice object reference into the queue item table. * @return boolean true on success + * @throws StompException on connection or send error */ public function enqueue($object, $queue) + { + $this->_connect(); + return $this->_doEnqueue($object, $queue, $this->defaultIdx); + } + + /** + * Saves a notice object reference into the queue item table + * on the given connection. + * + * @return boolean true on success + * @throws StompException on connection or send error + */ + protected function _doEnqueue($object, $queue, $idx) { $msg = $this->encode($object); $rep = $this->logrep($object); - $this->_connect(); - $props = array('created' => common_sql_now()); - if (common_config('queue', 'stomp_persistent')) { + if ($this->isPersistent($queue)) { $props['persistent'] = 'true'; } - $result = $this->con->send($this->queueName($queue), - $msg, // BODY of the message - $props); + + $con = $this->cons[$idx]; + $host = $con->getServer(); + $result = $con->send($this->queueName($queue), $msg, $props); if (!$result) { - common_log(LOG_ERR, "Error sending $rep to $queue queue"); + common_log(LOG_ERR, "Error sending $rep to $queue queue on $host"); return false; } - common_log(LOG_DEBUG, "complete remote queueing $rep for $queue"); + common_log(LOG_DEBUG, "complete remote queueing $rep for $queue on $host"); $this->stats('enqueued', $queue); + return true; + } + + /** + * Determine whether messages to this queue should be marked as persistent. + * Actual persistent storage depends on the queue server's configuration. + * @param string $queue + * @return bool + */ + protected function isPersistent($queue) + { + $mode = common_config('queue', 'stomp_persistent'); + if (is_array($mode)) { + return in_array($queue, $mode); + } else { + return (bool)$mode; + } } /** @@ -199,7 +238,29 @@ class StompQueueManager extends QueueManager */ public function getSockets() { - return array($this->con->getSocket()); + $sockets = array(); + foreach ($this->cons as $con) { + if ($con) { + $sockets[] = $con->getSocket(); + } + } + return $sockets; + } + + /** + * Get the Stomp connection object associated with the given socket. + * @param resource $socket + * @return int index into connections list + * @throws Exception + */ + protected function connectionFromSocket($socket) + { + foreach ($this->cons as $i => $con) { + if ($con && $con->getSocket() === $socket) { + return $i; + } + } + throw new Exception(__CLASS__ . " asked to read from unrecognized socket"); } /** @@ -211,27 +272,56 @@ class StompQueueManager extends QueueManager */ public function handleInput($socket) { - assert($socket === $this->con->getSocket()); + $idx = $this->connectionFromSocket($socket); + $con = $this->cons[$idx]; + $host = $con->getServer(); + $ok = true; - $frames = $this->con->readFrames(); + try { + $frames = $con->readFrames(); + } catch (StompException $e) { + common_log(LOG_ERR, "Lost connection to $host: " . $e->getMessage()); + $this->cons[$idx] = null; + $this->transaction[$idx] = null; + $this->disconnect[$idx] = time(); + return false; + } foreach ($frames as $frame) { $dest = $frame->headers['destination']; if ($dest == $this->control) { - if (!$this->handleControlSignal($frame)) { + if (!$this->handleControlSignal($idx, $frame)) { // We got a control event that requests a shutdown; // close out and stop handling anything else! break; } } else { - $ok = $ok && $this->handleItem($frame); + $ok = $ok && $this->handleItem($idx, $frame); } } return $ok; } + /** + * Attempt to reconnect in background if we lost a connection. + */ + function idle() + { + $now = time(); + foreach ($this->cons as $idx => $con) { + if (empty($con)) { + $age = $now - $this->disconnect[$idx]; + if ($age >= 60) { + $this->_reconnect($idx); + } + } + } + return true; + } + /** * Initialize our connection and subscribe to all the queues - * we're going to need to handle... + * we're going to need to handle... If multiple queue servers + * are configured for failover, we'll listen to all of them. * * Side effects: in multi-site mode, may reset site configuration. * @@ -241,10 +331,14 @@ class StompQueueManager extends QueueManager public function start($master) { parent::start($master); - $this->_connect(); + $this->_connectAll(); common_log(LOG_INFO, "Subscribing to $this->control"); - $this->con->subscribe($this->control); + foreach ($this->cons as $con) { + if ($con) { + $con->subscribe($this->control); + } + } if ($this->sites) { foreach ($this->sites as $server) { StatusNet::init($server); @@ -253,7 +347,11 @@ class StompQueueManager extends QueueManager } else { $this->doSubscribe(); } - $this->begin(); + foreach ($this->cons as $i => $con) { + if ($con) { + $this->begin($i); + } + } return true; } @@ -268,8 +366,12 @@ class StompQueueManager extends QueueManager { // If there are any outstanding delivered messages we haven't processed, // free them for another thread to take. - $this->rollback(); - $this->con->unsubscribe($this->control); + foreach ($this->cons as $i => $con) { + if ($con) { + $this->rollback($i); + $con->unsubscribe($this->control); + } + } if ($this->sites) { foreach ($this->sites as $server) { StatusNet::init($server); @@ -291,23 +393,106 @@ class StompQueueManager extends QueueManager } /** - * Lazy open connection to Stomp queue server. + * Lazy open a single connection to Stomp queue server. + * If multiple servers are configured, we let the Stomp client library + * worry about finding a working connection among them. */ protected function _connect() { - if (empty($this->con)) { - $this->_log(LOG_INFO, "Connecting to '$this->server' as '$this->username'..."); - $this->con = new LiberalStomp($this->server); - - if ($this->con->connect($this->username, $this->password)) { - $this->_log(LOG_INFO, "Connected."); + if (empty($this->cons)) { + $list = $this->servers; + if (count($list) > 1) { + shuffle($list); // Randomize to spread load + $url = 'failover://(' . implode(',', $list) . ')'; } else { - $this->_log(LOG_ERR, 'Failed to connect to queue server'); - throw new ServerException('Failed to connect to queue server'); + $url = $list[0]; } + $con = $this->_doConnect($url); + $this->cons = array($con); + $this->transactionCount = array(0); + $this->transaction = array(null); + $this->disconnect = array(null); } } + /** + * Lazy open connections to all Stomp servers, if in manual failover + * mode. This means the queue servers don't speak to each other, so + * we have to listen to all of them to make sure we get all events. + */ + protected function _connectAll() + { + if (!common_config('queue', 'stomp_manual_failover')) { + return $this->_connect(); + } + if (empty($this->cons)) { + $this->cons = array(); + $this->transactionCount = array(); + $this->transaction = array(); + foreach ($this->servers as $idx => $server) { + try { + $this->cons[] = $this->_doConnect($server); + $this->disconnect[] = null; + } catch (Exception $e) { + // s'okay, we'll live + $this->cons[] = null; + $this->disconnect[] = time(); + } + $this->transactionCount[] = 0; + $this->transaction[] = null; + } + if (empty($this->cons)) { + throw new ServerException("No queue servers reachable..."); + return false; + } + } + } + + protected function _reconnect($idx) + { + try { + $con = $this->_doConnect($this->servers[$idx]); + } catch (Exception $e) { + $this->_log(LOG_ERR, $e->getMessage()); + $con = null; + } + if ($con) { + $this->cons[$idx] = $con; + $this->disconnect[$idx] = null; + + // now we have to listen to everything... + // @fixme refactor this nicer. :P + $host = $con->getServer(); + $this->_log(LOG_INFO, "Resubscribing to $this->control on $host"); + $con->subscribe($this->control); + foreach ($this->subscriptions as $site => $queues) { + foreach ($queues as $queue) { + $this->_log(LOG_INFO, "Resubscribing to $queue on $host"); + $con->subscribe($queue); + } + } + $this->begin($idx); + } else { + // Try again later... + $this->disconnect[$idx] = time(); + } + } + + protected function _doConnect($server) + { + $this->_log(LOG_INFO, "Connecting to '$server' as '$this->username'..."); + $con = new LiberalStomp($server); + + if ($con->connect($this->username, $this->password)) { + $this->_log(LOG_INFO, "Connected."); + } else { + $this->_log(LOG_ERR, 'Failed to connect to queue server'); + throw new ServerException('Failed to connect to queue server'); + } + + return $con; + } + /** * Subscribe to all enabled notice queues for the current site. */ @@ -319,7 +504,11 @@ class StompQueueManager extends QueueManager $rawqueue = $this->queueName($queue); $this->subscriptions[$site][$queue] = $rawqueue; $this->_log(LOG_INFO, "Subscribing to $rawqueue"); - $this->con->subscribe($rawqueue); + foreach ($this->cons as $con) { + if ($con) { + $con->subscribe($rawqueue); + } + } } } @@ -333,7 +522,11 @@ class StompQueueManager extends QueueManager if (!empty($this->subscriptions[$site])) { foreach ($this->subscriptions[$site] as $queue => $rawqueue) { $this->_log(LOG_INFO, "Unsubscribing from $rawqueue"); - $this->con->unsubscribe($rawqueue); + foreach ($this->cons as $con) { + if ($con) { + $con->unsubscribe($rawqueue); + } + } unset($this->subscriptions[$site][$queue]); } } @@ -348,27 +541,31 @@ class StompQueueManager extends QueueManager * Side effects: in multi-site mode, may reset site configuration to * match the site that queued the event. * + * @param int $idx connection index * @param StompFrame $frame * @return bool */ - protected function handleItem($frame) + protected function handleItem($idx, $frame) { + $this->defaultIdx = $idx; + list($site, $queue) = $this->parseDestination($frame->headers['destination']); if ($site != $this->currentSite()) { $this->stats('switch'); StatusNet::init($site); } + $host = $this->cons[$idx]->getServer(); if (is_numeric($frame->body)) { $id = intval($frame->body); - $info = "notice $id posted at {$frame->headers['created']} in queue $queue"; + $info = "notice $id posted at {$frame->headers['created']} in queue $queue from $host"; $notice = Notice::staticGet('id', $id); if (empty($notice)) { $this->_log(LOG_WARNING, "Skipping missing $info"); - $this->ack($frame); - $this->commit(); - $this->begin(); + $this->ack($idx, $frame); + $this->commit($idx); + $this->begin($idx); $this->stats('badnotice', $queue); return false; } @@ -376,16 +573,16 @@ class StompQueueManager extends QueueManager $item = $notice; } else { // @fixme should we serialize, or json, or what here? - $info = "string posted at {$frame->headers['created']} in queue $queue"; + $info = "string posted at {$frame->headers['created']} in queue $queue from $host"; $item = $frame->body; } $handler = $this->getHandler($queue); if (!$handler) { $this->_log(LOG_ERR, "Missing handler class; skipping $info"); - $this->ack($frame); - $this->commit(); - $this->begin(); + $this->ack($idx, $frame); + $this->commit($idx); + $this->begin($idx); $this->stats('badhandler', $queue); return false; } @@ -397,18 +594,18 @@ class StompQueueManager extends QueueManager // FIXME we probably shouldn't have to do // this kind of queue management ourselves; // if we don't ack, it should resend... - $this->ack($frame); + $this->ack($idx, $frame); $this->enqueue($item, $queue); - $this->commit(); - $this->begin(); + $this->commit($idx); + $this->begin($idx); $this->stats('requeued', $queue); return false; } $this->_log(LOG_INFO, "Successfully handled $info"); - $this->ack($frame); - $this->commit(); - $this->begin(); + $this->ack($idx, $frame); + $this->commit($idx); + $this->begin($idx); $this->stats('handled', $queue); return true; } @@ -416,10 +613,11 @@ class StompQueueManager extends QueueManager /** * Process a control signal broadcast. * + * @param int $idx connection index * @param array $frame Stomp frame * @return bool true to continue; false to stop further processing. */ - protected function handleControlSignal($frame) + protected function handleControlSignal($idx, $frame) { $message = trim($frame->body); if (strpos($message, ':') !== false) { @@ -443,9 +641,9 @@ class StompQueueManager extends QueueManager $this->_log(LOG_ERR, "Ignoring unrecognized control message: $message"); } - $this->ack($frame); - $this->commit(); - $this->begin(); + $this->ack($idx, $frame); + $this->commit($idx); + $this->begin($idx); return $shutdown; } @@ -522,47 +720,49 @@ class StompQueueManager extends QueueManager common_log($level, 'StompQueueManager: '.$msg); } - protected function begin() + protected function begin($idx) { if ($this->useTransactions) { - if ($this->transaction) { + if (!empty($this->transaction[$idx])) { throw new Exception("Tried to start transaction in the middle of a transaction"); } - $this->transactionCount++; - $this->transaction = $this->master->id . '-' . $this->transactionCount . '-' . time(); - $this->con->begin($this->transaction); + $this->transactionCount[$idx]++; + $this->transaction[$idx] = $this->master->id . '-' . $this->transactionCount[$idx] . '-' . time(); + $this->cons[$idx]->begin($this->transaction[$idx]); } } - protected function ack($frame) + protected function ack($idx, $frame) { if ($this->useTransactions) { - if (!$this->transaction) { + if (empty($this->transaction[$idx])) { throw new Exception("Tried to ack but not in a transaction"); } + $this->cons[$idx]->ack($frame, $this->transaction[$idx]); + } else { + $this->cons[$idx]->ack($frame); } - $this->con->ack($frame, $this->transaction); } - protected function commit() + protected function commit($idx) { if ($this->useTransactions) { - if (!$this->transaction) { + if (empty($this->transaction[$idx])) { throw new Exception("Tried to commit but not in a transaction"); } - $this->con->commit($this->transaction); - $this->transaction = null; + $this->cons[$idx]->commit($this->transaction[$idx]); + $this->transaction[$idx] = null; } } - protected function rollback() + protected function rollback($idx) { if ($this->useTransactions) { - if (!$this->transaction) { + if (empty($this->transaction[$idx])) { throw new Exception("Tried to rollback but not in a transaction"); } - $this->con->commit($this->transaction); - $this->transaction = null; + $this->cons[$idx]->commit($this->transaction[$idx]); + $this->transaction[$idx] = null; } } } -- cgit v1.2.3-54-g00ecf