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/OStatus/OStatusPlugin.php | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) (limited to 'plugins/OStatus') 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; } -- 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/OStatus') 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/OStatus') 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