diff --git a/cron/hosting_cron.module b/cron/hosting_cron.module
index 241eaf7..4f019ba 100644
--- a/cron/hosting_cron.module
+++ b/cron/hosting_cron.module
@@ -6,10 +6,12 @@
/**
* Implements hook_hosting_queues().
+ *
+ * @todo: In Hosting 4.x change the type to HOSTING_QUEUE_TYPE_SPREAD.
*/
function hosting_cron_hosting_queues() {
$items['cron'] = array(
- 'type' => 'batch',
+ 'type' => HOSTING_QUEUE_TYPE_BATCH,
'name' => t('Cron queue'),
'description' => t('Run cron on hosted sites.'),
'total_items' => hosting_site_count(),
diff --git a/hosting.api.php b/hosting.api.php
index 138b71a..e589a6f 100644
--- a/hosting.api.php
+++ b/hosting.api.php
@@ -132,13 +132,57 @@ function hook_hosting_feature() {
/**
* Define hosting queues.
*
- * @see hosting_get_queues()
+ * @see hosting_get_queues
*/
function hook_hosting_queues() {
}
/**
+ * Alter module defined queue definitions before they are processed.
+ *
+ * This hook is invoked before hosting calculates the number of items to
+ * process when processing queues, so, for example, you could alter the number of apparent
+ * items in the queue.
+ *
+ * @param array $queues
+ * The array of queue definitions of queues provided by modules.
+ *
+ * @see hosting_get_queues
+ * @see hook_hosting_queues
+ * @see hook_hosting_processed_queues_alter
+ */
+function hook_hosting_queues_alter(&$queues) {
+ if (isset($queues['cron'])) {
+ // Do not execute the cron queue at weekends.
+ if (date('N', REQUEST_TIME) > 5) {
+ $queues['cron']['total_items'] = 0;
+ }
+ }
+}
+
+/**
+ * Alter module defined queue definitions after they are processed.
+ *
+ * This hook is invoked after hosting module calculates the number of items to
+ * process when processing queues, and after the configurable information has
+ * been merged in.
+ *
+ * @param array $queues
+ * The processed array of queue definitions of queues provided by modules.
+ *
+ * @see hosting_get_queues
+ * @see hook_hosting_queues
+ * @see hook_hosting_queues_alter
+ */
+function hook_hosting_processed_queues_alter(&$queues) {
+ if (isset($queues['cron'])) {
+ // Force the cron queue to always be disabled.
+ $queues['cron']['enabled'] = FALSE;
+ }
+}
+
+/**
* Add or change context options before a hosting task runs.
*
* This hook is invoked just before any task that has the 'provision_save' flag
diff --git a/hosting.module b/hosting.module
index 7140cba..a5fcf63 100644
--- a/hosting.module
+++ b/hosting.module
@@ -14,6 +14,53 @@
define('HOSTING_MAX_ALIAS_LENGTH', 235);
/**
+ * A serial queue type.
+ *
+ * This queue will be processed at a specified interval. When the interval has
+ * elapsed a configurable number of the items will be processed.
+ */
+define('HOSTING_QUEUE_TYPE_SERIAL', 'serial');
+
+/**
+ * A batch queue type,
+ *
+ * This queue type aims to process repeatable tasks spread out over a
+ * configurable interval.
+ *
+ * See the processing code in hosting_get_queues() for the exact
+ * implementation.
+ *
+ * This queue type will:
+ * - Process items in the queue at some small number of specific points, evenly spaced throughout
+ * the configurable interval.
+ * - Process as many items at those intervals so that all the items are
+ * processed at least once over the configurable interval.
+ *
+ * @see HOSTING_QUEUE_TYPE_PARALLEL
+ */
+define('HOSTING_QUEUE_TYPE_BATCH', 'batch');
+
+/**
+ * A parallel queue type,
+ *
+ * This queue type aims to process repeatable tasks spread out over a
+ * configurable interval.
+ *
+ * See the processing code in hosting_get_queues() for the exact
+ * implementation.
+ *
+ * This queue type will attempt to:
+ * - Process items at as many points as possible evenly spread over the
+ * configurable interval.
+ * - Points will be at least 1 minute apart to allow for the default dispatcher to execute them.
+ * - At each point process the minimum number of items needed to result in all
+ * items being processed at least once over the configurable interval.
+ *
+ * @see HOSTING_QUEUE_TYPE_BATCH
+ */
+define('HOSTING_QUEUE_TYPE_SPREAD', 'spread');
+
+/**
* Not split for performance reasons. Just to keep code together.
*/
include_once dirname(__FILE__) . '/hosting.inc';
@@ -859,7 +906,17 @@ function hosting_queues_configure($form, &$form_state) {
$form[$queue]['frequency']['#prefix'] = "
";
$form[$queue]['frequency']['#suffix'] = "
";
- if ($info['type'] == 'batch') {
+ if ($info['type'] == HOSTING_QUEUE_TYPE_BATCH) {
+ $form[$queue]['frequency']['items'] = array(
+ '#markup' => t('%count %items every', array(
+ "%count" => $info['total_items'],
+ "%items" => format_plural($info['total_items'], $info['singular'], $info['plural']),
+ )),
+ '#prefix' => "",
+ '#suffix' => "
",
+ );
+ }
+ elseif ($info['type'] == HOSTING_QUEUE_TYPE_SPREAD) {
$form[$queue]['frequency']['items'] = array(
'#markup' => t('%count %items every', array(
"%count" => $info['total_items'],
@@ -982,7 +1039,7 @@ function hosting_queues_configure_submit($form, &$form_state) {
if ($form_state['values'][$queue]) {
variable_set("hosting_queue_" . $queue . "_enabled", $form_state['values'][$queue]['enabled']);
variable_set("hosting_queue_" . $queue . "_frequency", $form_state['values'][$queue]['frequency']['ticks'] * $form_state['values'][$queue]['frequency']['unit']);
- if ($info['type'] == 'serial') {
+ if ($info['type'] == HOSTING_QUEUE_TYPE_SERIAL) {
variable_set("hosting_queue_" . $queue . "_items", $form_state['values'][$queue]['frequency']['items']);
}
}
diff --git a/hosting.queues.inc b/hosting.queues.inc
index 7d7546d..7acb857 100644
--- a/hosting.queues.inc
+++ b/hosting.queues.inc
@@ -59,7 +59,7 @@ function hosting_get_queues($refresh = FALSE) {
if (is_null($cache) || $refresh) {
$cache = array();
$defaults = array(
- 'type' => 'serial',
+ 'type' => HOSTING_QUEUE_TYPE_SERIAL,
'max_threads' => 6,
'threshold' => '100',
'min_threads' => 1,
@@ -70,7 +70,11 @@ function hosting_get_queues($refresh = FALSE) {
'singular' => t('item'),
'plural' => t('items'),
);
- $queues = module_invoke_all("hosting_queues");
+ $queues = module_invoke_all('hosting_queues');
+ if (!is_array($queues)) {
+ $queues = array();
+ }
+ drupal_alter('hosting_queues', $queues);
foreach ($queues as $key => $queue) {
$queue = array_merge($defaults, $queue);
@@ -84,7 +88,7 @@ function hosting_get_queues($refresh = FALSE) {
);
$queue = array_merge($queue, $configured);
- if ($queue['type'] == 'batch') {
+ if ($queue['type'] == HOSTING_QUEUE_TYPE_BATCH) {
$threads = $queue['total_items'] / $queue['threshold'];
if ($threads <= $queue['min_threads']) {
$threads = $queue['min_threads'];
@@ -96,6 +100,24 @@ function hosting_get_queues($refresh = FALSE) {
$queue['calc_frequency'] = ceil($queue['frequency'] / $threads);
$queue['calc_items'] = ceil($queue['total_items'] / $threads);
}
+ elseif ($queue['type'] == HOSTING_QUEUE_TYPE_SPREAD) {
+ // If there are 0 items in the queue avoid a division by 0.
+ if ($queue['total_items'] > 0) {
+ // Process the queue as often as is needed to get through all the
+ // items once per frequency, but at most once a minute since the
+ // default dispatcher runs that quickly.
+ $queue['calc_frequency'] = max(60, $queue['frequency'] / $queue['total_items']);
+ // Compute the number of items to process per invocation. Usually this is will be 1.
+ $queue['calc_items'] = ceil($queue['calc_frequency'] / $queue['frequency'] * $queue['total_items']);
+ }
+ else {
+ // There are 0 items in the queue, process 0 tasks once a day anyway.
+ $queue['calc_frequency'] = 86400;
+ $queue['calc_items'] = 0;
+ }
+ // Pretend like we do not have any previously run items still processing.
+ $queue['running_items'] = 0;
+ }
else {
$queue['calc_frequency'] = $queue['frequency'];
$queue['calc_items'] = $queue['items'];
@@ -104,6 +126,8 @@ function hosting_get_queues($refresh = FALSE) {
$queue['last'] = variable_get('hosting_queue_' . $key . '_last_run', 0);
$queues[$key] = $queue;
}
+ // Allow altering the processed queue data.
+ drupal_alter('hosting_processed_queues', $queues);
$cache = $queues;
}
diff --git a/task_gc/hosting_task_gc.module b/task_gc/hosting_task_gc.module
index 4fb6273..a9b3fab 100644
--- a/task_gc/hosting_task_gc.module
+++ b/task_gc/hosting_task_gc.module
@@ -10,7 +10,7 @@ function hosting_task_gc_hosting_queues() {
$queue['task_gc'] = array(
'name' => t('Task GC'),
'description' => t('Process the garbage collection of task logs.'),
- 'type' => 'batch', # run queue sequentially. always with the same parameters.
+ 'type' => HOSTING_QUEUE_TYPE_BATCH, # run queue sequentially. always with the same parameters.
'frequency' => strtotime("1 hour", 0),
'items' => 20,
'total_items' => $count,