diff options
author | Brion Vibber <brion@pobox.com> | 2010-06-11 12:04:03 -0700 |
---|---|---|
committer | Brion Vibber <brion@pobox.com> | 2010-06-11 12:04:03 -0700 |
commit | 47665e845ae74e6ee5b9a39565fb45dd9a93f921 (patch) | |
tree | a24680e60e250f085c7290c8fa0ecbc3ec4c6ea8 /plugins/OStatus/lib/ostatusqueuehandler.php | |
parent | ec155464765db36591bbb183b335abbc1ec8638c (diff) | |
parent | 054ac909bf6222254cf9b2c3a4eda824f4f4eb8f (diff) |
Merge branch 'testing' of gitorious.org:statusnet/mainline into testing
Diffstat (limited to 'plugins/OStatus/lib/ostatusqueuehandler.php')
-rw-r--r-- | plugins/OStatus/lib/ostatusqueuehandler.php | 31 |
1 files changed, 30 insertions, 1 deletions
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); } } |