corpus_parole: comparison cms/drupal/modules/system/system.queue.inc

equal deleted inserted replaced

-:07239de796bb
+:e756a8c72c3d
+<?php
+/**
+* @file
+* Queue functionality.
+*/
+/**
+* @defgroup queue Queue operations
+* @{
+* Queue items to allow later processing.
+*
+* The queue system allows placing items in a queue and processing them later.
+* The system tries to ensure that only one consumer can process an item.
+*
+* Before a queue can be used it needs to be created by
+* DrupalQueueInterface::createQueue().
+*
+* Items can be added to the queue by passing an arbitrary data object to
+* DrupalQueueInterface::createItem().
+*
+* To process an item, call DrupalQueueInterface::claimItem() and specify how
+* long you want to have a lease for working on that item. When finished
+* processing, the item needs to be deleted by calling
+* DrupalQueueInterface::deleteItem(). If the consumer dies, the item will be
+* made available again by the DrupalQueueInterface implementation once the
+* lease expires. Another consumer will then be able to receive it when calling
+* DrupalQueueInterface::claimItem(). Due to this, the processing code should
+* be aware that an item might be handed over for processing more than once.
+*
+* The $item object used by the DrupalQueueInterface can contain arbitrary
+* metadata depending on the implementation. Systems using the interface should
+* only rely on the data property which will contain the information passed to
+* DrupalQueueInterface::createItem(). The full queue item returned by
+* DrupalQueueInterface::claimItem() needs to be passed to
+* DrupalQueueInterface::deleteItem() once processing is completed.
+*
+* There are two kinds of queue backends available: reliable, which preserves
+* the order of messages and guarantees that every item will be executed at
+* least once. The non-reliable kind only does a best effort to preserve order
+* in messages and to execute them at least once but there is a small chance
+* that some items get lost. For example, some distributed back-ends like
+* Amazon SQS will be managing jobs for a large set of producers and consumers
+* where a strict FIFO ordering will likely not be preserved. Another example
+* would be an in-memory queue backend which might lose items if it crashes.
+* However, such a backend would be able to deal with significantly more writes
+* than a reliable queue and for many tasks this is more important. See
+* aggregator_cron() for an example of how to effectively utilize a
+* non-reliable queue. Another example is doing Twitter statistics -- the small
+* possibility of losing a few items is insignificant next to power of the
+* queue being able to keep up with writes. As described in the processing
+* section, regardless of the queue being reliable or not, the processing code
+* should be aware that an item might be handed over for processing more than
+* once (because the processing code might time out before it finishes).
+*/
+/**
+* Factory class for interacting with queues.
+*/
+class DrupalQueue {
+/**
+* Returns the queue object for a given name.
+*
+* The following variables can be set by variable_set or $conf overrides:
+* - queue_class_$name: the class to be used for the queue $name.
+* - queue_default_class: the class to use when queue_class_$name is not
+*   defined. Defaults to SystemQueue, a reliable backend using SQL.
+* - queue_default_reliable_class: the class to use when queue_class_$name is
+*   not defined and the queue_default_class is not reliable. Defaults to
+*   SystemQueue.
+*
+* @param $name
+*   Arbitrary string. The name of the queue to work with.
+* @param $reliable
+*   TRUE if the ordering of items and guaranteeing every item executes at
+*   least once is important, FALSE if scalability is the main concern.
+*
+* @return
+*   The queue object for a given name.
+*/
+public static function get($name, $reliable = FALSE) {
+static $queues;
+if (!isset($queues[$name])) {
+$class = variable_get('queue_class_' . $name, NULL);
+if (!$class) {
+$class = variable_get('queue_default_class', 'SystemQueue');
+}
+$object = new $class($name);
+if ($reliable && !$object instanceof DrupalReliableQueueInterface) {
+$class = variable_get('queue_default_reliable_class', 'SystemQueue');
+$object = new $class($name);
+}
+$queues[$name] = $object;
+}
+return $queues[$name];
+}
+}
+interface DrupalQueueInterface {
+/**
+* Add a queue item and store it directly to the queue.
+*
+* @param $data
+*   Arbitrary data to be associated with the new task in the queue.
+* @return
+*   TRUE if the item was successfully created and was (best effort) added
+*   to the queue, otherwise FALSE. We don't guarantee the item was
+*   committed to disk etc, but as far as we know, the item is now in the
+*   queue.
+*/
+public function createItem($data);
+/**
+* Retrieve the number of items in the queue.
+*
+* This is intended to provide a "best guess" count of the number of items in
+* the queue. Depending on the implementation and the setup, the accuracy of
+* the results of this function may vary.
+*
+* e.g. On a busy system with a large number of consumers and items, the
+* result might only be valid for a fraction of a second and not provide an
+* accurate representation.
+*
+* @return
+*   An integer estimate of the number of items in the queue.
+*/
+public function numberOfItems();
+/**
+* Claim an item in the queue for processing.
+*
+* @param $lease_time
+*   How long the processing is expected to take in seconds, defaults to an
+*   hour. After this lease expires, the item will be reset and another
+*   consumer can claim the item. For idempotent tasks (which can be run
+*   multiple times without side effects), shorter lease times would result
+*   in lower latency in case a consumer fails. For tasks that should not be
+*   run more than once (non-idempotent), a larger lease time will make it
+*   more rare for a given task to run multiple times in cases of failure,
+*   at the cost of higher latency.
+* @return
+*   On success we return an item object. If the queue is unable to claim an
+*   item it returns false. This implies a best effort to retrieve an item
+*   and either the queue is empty or there is some other non-recoverable
+*   problem.
+*/
+public function claimItem($lease_time = 3600);
+/**
+* Delete a finished item from the queue.
+*
+* @param $item
+*   The item returned by DrupalQueueInterface::claimItem().
+*/
+public function deleteItem($item);
+/**
+* Release an item that the worker could not process, so another
+* worker can come in and process it before the timeout expires.
+*
+* @param $item
+* @return boolean
+*/
+public function releaseItem($item);
+/**
+* Create a queue.
+*
+* Called during installation and should be used to perform any necessary
+* initialization operations. This should not be confused with the
+* constructor for these objects, which is called every time an object is
+* instantiated to operate on a queue. This operation is only needed the
+* first time a given queue is going to be initialized (for example, to make
+* a new database table or directory to hold tasks for the queue -- it
+* depends on the queue implementation if this is necessary at all).
+*/
+public function createQueue();
+/**
+* Delete a queue and every item in the queue.
+*/
+public function deleteQueue();
+}
+/**
+* Reliable queue interface.
+*
+* Classes implementing this interface preserve the order of messages and
+* guarantee that every item will be executed at least once.
+*/
+interface DrupalReliableQueueInterface extends DrupalQueueInterface {
+}
+/**
+* Default queue implementation.
+*/
+class SystemQueue implements DrupalReliableQueueInterface {
+/**
+* The name of the queue this instance is working with.
+*
+* @var string
+*/
+protected $name;
+public function __construct($name) {
+$this->name = $name;
+}
+public function createItem($data) {
+// During a Drupal 6.x to 7.x update, drupal_get_schema() does not contain
+// the queue table yet, so we cannot rely on drupal_write_record().
+$query = db_insert('queue')
+->fields(array(
+'name' => $this->name,
+'data' => serialize($data),
+// We cannot rely on REQUEST_TIME because many items might be created
+// by a single request which takes longer than 1 second.
+'created' => time(),
+));
+return (bool) $query->execute();
+}
+public function numberOfItems() {
+return db_query('SELECT COUNT(item_id) FROM {queue} WHERE name = :name', array(':name' => $this->name))->fetchField();
+}
+public function claimItem($lease_time = 30) {
+// Claim an item by updating its expire fields. If claim is not successful
+// another thread may have claimed the item in the meantime. Therefore loop
+// until an item is successfully claimed or we are reasonably sure there
+// are no unclaimed items left.
+while (TRUE) {
+$item = db_query_range('SELECT data, item_id FROM {queue} q WHERE expire = 0 AND name = :name ORDER BY created, item_id ASC', 0, 1, array(':name' => $this->name))->fetchObject();
+if ($item) {
+// Try to update the item. Only one thread can succeed in UPDATEing the
+// same row. We cannot rely on REQUEST_TIME because items might be
+// claimed by a single consumer which runs longer than 1 second. If we
+// continue to use REQUEST_TIME instead of the current time(), we steal
+// time from the lease, and will tend to reset items before the lease
+// should really expire.
+$update = db_update('queue')
+->fields(array(
+'expire' => time() + $lease_time,
+))
+->condition('item_id', $item->item_id)
+->condition('expire', 0);
+// If there are affected rows, this update succeeded.
+if ($update->execute()) {
+$item->data = unserialize($item->data);
+return $item;
+}
+}
+else {
+// No items currently available to claim.
+return FALSE;
+}
+}
+}
+public function releaseItem($item) {
+$update = db_update('queue')
+->fields(array(
+'expire' => 0,
+))
+->condition('item_id', $item->item_id);
+return $update->execute();
+}
+public function deleteItem($item) {
+db_delete('queue')
+->condition('item_id', $item->item_id)
+->execute();
+}
+public function createQueue() {
+// All tasks are stored in a single database table (which is created when
+// Drupal is first installed) so there is nothing we need to do to create
+// a new queue.
+}
+public function deleteQueue() {
+db_delete('queue')
+->condition('name', $this->name)
+->execute();
+}
+}
+/**
+* Static queue implementation.
+*
+* This allows "undelayed" variants of processes relying on the Queue
+* interface. The queue data resides in memory. It should only be used for
+* items that will be queued and dequeued within a given page request.
+*/
+class MemoryQueue implements DrupalQueueInterface {
+/**
+* The queue data.
+*
+* @var array
+*/
+protected $queue;
+/**
+* Counter for item ids.
+*
+* @var int
+*/
+protected $id_sequence;
+/**
+* Start working with a queue.
+*
+* @param $name
+*   Arbitrary string. The name of the queue to work with.
+*/
+public function __construct($name) {
+$this->queue = array();
+$this->id_sequence = 0;
+}
+public function createItem($data) {
+$item = new stdClass();
+$item->item_id = $this->id_sequence++;
+$item->data = $data;
+$item->created = time();
+$item->expire = 0;
+$this->queue[$item->item_id] = $item;
+return TRUE;
+}
+public function numberOfItems() {
+return count($this->queue);
+}
+public function claimItem($lease_time = 30) {
+foreach ($this->queue as $key => $item) {
+if ($item->expire == 0) {
+$item->expire = time() + $lease_time;
+$this->queue[$key] = $item;
+return $item;
+}
+}
+return FALSE;
+}
+public function deleteItem($item) {
+unset($this->queue[$item->item_id]);
+}
+public function releaseItem($item) {
+if (isset($this->queue[$item->item_id]) && $this->queue[$item->item_id]->expire != 0) {
+$this->queue[$item->item_id]->expire = 0;
+return TRUE;
+}
+return FALSE;
+}
+public function createQueue() {
+// Nothing needed here.
+}
+public function deleteQueue() {
+$this->queue = array();
+$this->id_sequence = 0;
+}
+}
+/**
+* @} End of "defgroup queue".
+*/