From 5f4c6ec626d3d641f0712b276deb32b218b7a330 Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Thu, 3 Jun 2010 16:58:45 -0700 Subject: Skip enqueueing to outgoing bridges on incoming remote messages. Twitter, Facebook, RSSCloud, and OStatus checks were enqueued on these when they'd never do anything but churn the queue servers. Notice::isLocal() can replace a number of manual checks for $notice->is_local being LOCAL_PUBLIC or LOCAL_NONPUBLIC. --- plugins/Facebook/FacebookPlugin.php | 2 +- plugins/OStatus/OStatusPlugin.php | 6 ++++-- plugins/RSSCloud/RSSCloudPlugin.php | 18 +++--------------- plugins/TwitterBridge/TwitterBridgePlugin.php | 2 +- 4 files changed, 9 insertions(+), 19 deletions(-) (limited to 'plugins') diff --git a/plugins/Facebook/FacebookPlugin.php b/plugins/Facebook/FacebookPlugin.php index 5dba73a5d..19989a952 100644 --- a/plugins/Facebook/FacebookPlugin.php +++ b/plugins/Facebook/FacebookPlugin.php @@ -585,7 +585,7 @@ class FacebookPlugin extends Plugin function onStartEnqueueNotice($notice, &$transports) { - if (self::hasKeys()) { + if (self::hasKeys() && $notice->isLocal()) { array_push($transports, 'facebook'); } return true; diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php index 5b153216e..5a657c83d 100644 --- a/plugins/OStatus/OStatusPlugin.php +++ b/plugins/OStatus/OStatusPlugin.php @@ -102,8 +102,10 @@ class OStatusPlugin extends Plugin */ function onStartEnqueueNotice($notice, &$transports) { - // put our transport first, in case there's any conflict (like OMB) - array_unshift($transports, 'ostatus'); + if ($notice->isLocal()) { + // put our transport first, in case there's any conflict (like OMB) + array_unshift($transports, 'ostatus'); + } return true; } diff --git a/plugins/RSSCloud/RSSCloudPlugin.php b/plugins/RSSCloud/RSSCloudPlugin.php index 661c32141..c1951cdbf 100644 --- a/plugins/RSSCloud/RSSCloudPlugin.php +++ b/plugins/RSSCloud/RSSCloudPlugin.php @@ -192,24 +192,12 @@ class RSSCloudPlugin extends Plugin function onStartEnqueueNotice($notice, &$transports) { - array_push($transports, 'rsscloud'); + if ($notice->isLocal()) { + array_push($transports, 'rsscloud'); + } return true; } - /** - * Determine whether the notice was locally created - * - * @param Notice $notice the notice in question - * - * @return boolean locality - */ - - function _isLocal($notice) - { - return ($notice->is_local == Notice::LOCAL_PUBLIC || - $notice->is_local == Notice::LOCAL_NONPUBLIC); - } - /** * Create the rsscloud_subscription table if it's not * already in the DB diff --git a/plugins/TwitterBridge/TwitterBridgePlugin.php b/plugins/TwitterBridge/TwitterBridgePlugin.php index 1a0a69682..65b3a6b38 100644 --- a/plugins/TwitterBridge/TwitterBridgePlugin.php +++ b/plugins/TwitterBridge/TwitterBridgePlugin.php @@ -221,7 +221,7 @@ class TwitterBridgePlugin extends Plugin */ function onStartEnqueueNotice($notice, &$transports) { - if (self::hasKeys()) { + if (self::hasKeys() && $notice->isLocal()) { // Avoid a possible loop if ($notice->source != 'twitter') { array_push($transports, 'twitter'); -- cgit v1.2.3-54-g00ecf From a75095fa1a3926d1fcc18c3d7285141fa3bef344 Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Thu, 3 Jun 2010 17:41:26 -0700 Subject: Meteor realtime plugin: use persistent connections by default when pushing updates from our queue threads --- plugins/Meteor/MeteorPlugin.php | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) (limited to 'plugins') diff --git a/plugins/Meteor/MeteorPlugin.php b/plugins/Meteor/MeteorPlugin.php index 5600d5fcc..ec8c9e217 100644 --- a/plugins/Meteor/MeteorPlugin.php +++ b/plugins/Meteor/MeteorPlugin.php @@ -50,6 +50,7 @@ class MeteorPlugin extends RealtimePlugin public $controlport = null; public $controlserver = null; public $channelbase = null; + public $persistent = true; protected $_socket = null; function __construct($webserver=null, $webport=4670, $controlport=4671, $controlserver=null, $channelbase='') @@ -102,8 +103,14 @@ class MeteorPlugin extends RealtimePlugin function _connect() { $controlserver = (empty($this->controlserver)) ? $this->webserver : $this->controlserver; + + $errno = $errstr = null; + $timeout = 5; + $flags = STREAM_CLIENT_CONNECT; + if ($this->persistent) $flags |= STREAM_CLIENT_PERSISTENT; + // May throw an exception. - $this->_socket = stream_socket_client("tcp://{$controlserver}:{$this->controlport}"); + $this->_socket = stream_socket_client("tcp://{$controlserver}:{$this->controlport}", $errno, $errstr, $timeout, $flags); if (!$this->_socket) { throw new Exception("Couldn't connect to {$controlserver} on {$this->controlport}"); } @@ -124,8 +131,10 @@ class MeteorPlugin extends RealtimePlugin function _disconnect() { - $cnt = fwrite($this->_socket, "QUIT\n"); - @fclose($this->_socket); + if (!$this->persistent) { + $cnt = fwrite($this->_socket, "QUIT\n"); + @fclose($this->_socket); + } } // Meteord flips out with default '/' separator -- cgit v1.2.3-54-g00ecf From 8b9436e8ae1ebcc7ef10752bb9666939200e26aa Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Thu, 3 Jun 2010 17:49:20 -0700 Subject: Option to divert PuSH items directly to the target site's queue when local --- classes/Status_network.php | 29 ++++++++++++++++++++--------- lib/stompqueuemanager.php | 9 +++++---- plugins/OStatus/classes/HubSub.php | 31 +++++++++++++++++++++++++++++++ 3 files changed, 56 insertions(+), 13 deletions(-) (limited to 'plugins') diff --git a/classes/Status_network.php b/classes/Status_network.php index a452c32ce..4a1f2c374 100644 --- a/classes/Status_network.php +++ b/classes/Status_network.php @@ -149,21 +149,15 @@ class Status_network extends Safe_DataObject $this->decache(); # while we still have the values! return parent::delete(); } - + /** * @param string $servername hostname - * @param string $pathname URL base path * @param string $wildcard hostname suffix to match wildcard config + * @return mixed Status_network or null */ - static function setupSite($servername, $pathname, $wildcard) + static function getFromHostname($servername, $wildcard) { - global $config; - $sn = null; - - // XXX I18N, probably not crucial for hostnames - // XXX This probably needs a tune up - if (0 == strncasecmp(strrev($wildcard), strrev($servername), strlen($wildcard))) { // special case for exact match if (0 == strcasecmp($servername, $wildcard)) { @@ -182,6 +176,23 @@ class Status_network extends Safe_DataObject } } } + return $sn; + } + + /** + * @param string $servername hostname + * @param string $pathname URL base path + * @param string $wildcard hostname suffix to match wildcard config + */ + static function setupSite($servername, $pathname, $wildcard) + { + global $config; + + $sn = null; + + // XXX I18N, probably not crucial for hostnames + // XXX This probably needs a tune up + $sn = self::getFromHostname($servername, $wildcard); if (!empty($sn)) { diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php index de4ba7f01..91faa8c36 100644 --- a/lib/stompqueuemanager.php +++ b/lib/stompqueuemanager.php @@ -115,11 +115,12 @@ class StompQueueManager extends QueueManager * * @param mixed $object * @param string $queue + * @param string $siteNickname optional override to drop into another site's queue * * @return boolean true on success * @throws StompException on connection or send error */ - public function enqueue($object, $queue) + public function enqueue($object, $queue, $siteNickname=null) { $this->_connect(); if (common_config('queue', 'stomp_enqueue_on')) { @@ -134,7 +135,7 @@ class StompQueueManager extends QueueManager } else { $idx = $this->defaultIdx; } - return $this->_doEnqueue($object, $queue, $idx); + return $this->_doEnqueue($object, $queue, $idx, $siteNickname); } /** @@ -144,10 +145,10 @@ class StompQueueManager extends QueueManager * @return boolean true on success * @throws StompException on connection or send error */ - protected function _doEnqueue($object, $queue, $idx) + protected function _doEnqueue($object, $queue, $idx, $siteNickname=null) { $rep = $this->logrep($object); - $envelope = array('site' => common_config('site', 'nickname'), + $envelope = array('site' => $siteNickname ? $siteNickname : common_config('site', 'nickname'), 'handler' => $queue, 'payload' => $this->encode($object)); $msg = serialize($envelope); diff --git a/plugins/OStatus/classes/HubSub.php b/plugins/OStatus/classes/HubSub.php index cdace3c1f..9748b4a56 100644 --- a/plugins/OStatus/classes/HubSub.php +++ b/plugins/OStatus/classes/HubSub.php @@ -260,6 +260,37 @@ class HubSub extends Memcached_DataObject $retries = intval(common_config('ostatus', 'hub_retries')); } + if (common_config('ostatus', 'local_push_bypass')) { + // If target is a local site, bypass the web server and drop the + // item directly into the target's input queue. + $url = parse_url($this->callback); + $wildcard = common_config('ostatus', 'local_wildcard'); + $site = Status_network::getFromHostname($url['host'], $wildcard); + + if ($site) { + if ($this->secret) { + $hmac = 'sha1=' . hash_hmac('sha1', $atom, $this->secret); + } else { + $hmac = ''; + } + + // Hack: at the moment we stick the subscription ID in the callback + // URL so we don't have to look inside the Atom to route the subscription. + // For now this means we need to extract that from the target URL + // so we can include it in the data. + $parts = explode('/', $url['path']); + $subId = intval(array_pop($parts)); + + $data = array('feedsub_id' => $subId, + 'post' => $atom, + 'hmac' => $hmac); + common_log(LOG_DEBUG, "Cross-site PuSH bypass enqueueing straight to $site->nickname feed $subId"); + $qm = QueueManager::get(); + $qm->enqueue($data, 'pushin', $site->nickname); + return; + } + } + // We dare not clone() as when the clone is discarded it'll // destroy the result data for the parent query. // @fixme use clone() again when it's safe to copy an -- cgit v1.2.3-54-g00ecf From 41e9dba7297d43b7de0cb7665901869910d1047a Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Fri, 4 Jun 2010 11:48:54 -0700 Subject: OStatus plugin: Rolling batch queueing for PuSH output to >50 subscribing sites. Keeps latency down for other things enqueued while we work... --- plugins/OStatus/OStatusPlugin.php | 2 ++ plugins/OStatus/classes/HubSub.php | 20 +++++++++++++++++++ plugins/OStatus/lib/ostatusqueuehandler.php | 31 ++++++++++++++++++++++++++++- 3 files changed, 52 insertions(+), 1 deletion(-) (limited to 'plugins') diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php index 5a657c83d..c61e2cc5f 100644 --- a/plugins/OStatus/OStatusPlugin.php +++ b/plugins/OStatus/OStatusPlugin.php @@ -87,6 +87,8 @@ class OStatusPlugin extends Plugin // Outgoing from our internal PuSH hub $qm->connect('hubconf', 'HubConfQueueHandler'); + $qm->connect('hubprep', 'HubPrepQueueHandler'); + $qm->connect('hubout', 'HubOutQueueHandler'); // Outgoing Salmon replies (when we don't need a return value) diff --git a/plugins/OStatus/classes/HubSub.php b/plugins/OStatus/classes/HubSub.php index 9748b4a56..7db528a4e 100644 --- a/plugins/OStatus/classes/HubSub.php +++ b/plugins/OStatus/classes/HubSub.php @@ -304,6 +304,26 @@ class HubSub extends Memcached_DataObject $qm->enqueue($data, 'hubout'); } + /** + * Queue up a large batch of pushes to multiple subscribers + * for this same topic update. + * + * If queues are disabled, this will run immediately. + * + * @param string $atom well-formed Atom feed + * @param array $pushCallbacks list of callback URLs + */ + function bulkDistribute($atom, $pushCallbacks) + { + $data = array('atom' => $atom, + 'topic' => $this->topic, + 'pushCallbacks' => $pushCallbacks); + common_log(LOG_INFO, "Queuing PuSH batch: $this->topic to " . + count($pushCallbacks) . " sites"); + $qm = QueueManager::get(); + $qm->enqueue($data, 'hubprep'); + } + /** * Send a 'fat ping' to the subscriber's callback endpoint * containing the given Atom feed chunk. diff --git a/plugins/OStatus/lib/ostatusqueuehandler.php b/plugins/OStatus/lib/ostatusqueuehandler.php index d1e58f1d6..8905d2e21 100644 --- a/plugins/OStatus/lib/ostatusqueuehandler.php +++ b/plugins/OStatus/lib/ostatusqueuehandler.php @@ -25,6 +25,18 @@ */ class OStatusQueueHandler extends QueueHandler { + // If we have more than this many subscribing sites on a single feed, + // break up the PuSH distribution into smaller batches which will be + // rolled into the queue progressively. This reduces disruption to + // other, shorter activities being enqueued while we work. + const MAX_UNBATCHED = 50; + + // Each batch (a 'hubprep' entry) will have this many items. + // Selected to provide a balance between queue packet size + // and number of batches that will end up getting processed. + // For 20,000 target sites, 1000 should work acceptably. + const BATCH_SIZE = 1000; + function transport() { return 'ostatus'; @@ -147,14 +159,31 @@ class OStatusQueueHandler extends QueueHandler /** * Queue up direct feed update pushes to subscribers on our internal hub. + * If there are a large number of subscriber sites, intermediate bulk + * distribution triggers may be queued. + * * @param string $atom update feed, containing only new/changed items * @param HubSub $sub open query of subscribers */ function pushFeedInternal($atom, $sub) { common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic"); + $n = 0; + $batch = array(); while ($sub->fetch()) { - $sub->distribute($atom); + $n++; + if ($n < self::MAX_UNBATCHED) { + $sub->distribute($atom); + } else { + $batch[] = $sub->callback; + if (count($batch) >= self::BATCH_SIZE) { + $sub->bulkDistribute($atom, $batch); + $batch = array(); + } + } + } + if (count($batch) >= 0) { + $sub->bulkDistribute($atom, $batch); } } -- cgit v1.2.3-54-g00ecf