Revolutionizingenterprise web development

\

Drupal Queue API:Built In Procrastination

Overview

• What is a queue?• When do you want to queue?• Drupal Queue 7 Classes and Methods• Batch API• Implementing Batch• Summary

What is a Queue?

• Queue: A line or sequence of people or vehicles awaiting their turn to be attended to or to proceed.

What does this mean to developers?• Queue: A sequence of items awaiting

their turn to be processed.

Presentation Problem

• Client Wants a Drupal site where user interactions need to be processed.– User updates a title -> gets put into a queue

to be processed.– User changes a field -> gets put into a queue

to be processed.– User publishes a node -> gets put into a

queue to be processed.• Requests are programmatically

moderated. • Moderating AI can be slow, likes to take

lunch breaks, has a mind of it’s own.

Very Simple Queue Example

User Requests: A sequence* of requests from a user.

* These items must be done in order. FIFO

item_id data

1 {'uid' => 100, 'request' => 'Publish My Blog Post'}

2 {'uid' => 100, 'request' => 'Update My Blog Title'}

3 {'uid' => 100, 'request' => 'Publish Latest Blogs To Drupal Planet'}

FIFO

What is a Queue (again) ?

• Queue: A sequence of items awaiting their turn to be processed.

• Do items always need to be processed in a sequence? – No.

• Queue: A set of items awaiting their turn to be processed.

Very Simple Queue Example 2

User Requests: A set* of requests from a user.

* These items don’t need to be done in order.

item_id data

1 {'uid' => 100, 'request' => 'Up vote Blog-A'}

2 {'uid' => 200, 'request' => 'Down vote Blog-A'}

3 {'uid' => 300, 'request' => 'Up vote Blog-B'}

Set

Multiple Workers

When do you want a queue?

Queues work best when you have a lot of items which eventually need to be processed.

Some examples:

• Mass Email

• Content Indexing

• Order Processing

• Third Party Service Requests (Saleforce, MailChimp, etc.)

When do you want a queue?

// When you are doing way too much$query = selectQuery('user_requests', 'ur');

$res = $query->fields('ur')->execute();while ($req = $res->fetchObject()) { if (user_requests_process_request($req)) {

user_requests_delete_request($req); }}

Queue Basics• Create a Queue

o Gotta start somewhere

• Add item to the Queueo Put something into the queue for later.

• Claim an item from the Queueo Get me the next Item that needs to be

done.

• Release the item from the Queueo Leave the item back in the queue because

it couldn't be processed yet.

• Remove the item from the Queue

• Count how many items are left in the Queue.

DrupalQueueInterface::

createQueue() - Create a queuecreateItem() - Add an item to the queueclaimItem() - Claim next queue item. Lock.releaseItem() - Leave the item in the queuedeleteItem() - Delete the item from the

queue.numberOffItems() - Retrieve the number of

items in the queue.

More@ http://dgo.to/a/DrupalQueueInterface

Drupal Core Queues

Storage: Database

• SystemQueue

• BatchQueue extends SystemQueue

Storage: Memory

• MemoryQueue

• BatchMemoryQueue extends MemoryQueue

Adding Items to SystemQueue$queue = DrupalQueue::get('user_requests');$queue->createQueue();$requests = array();$requests[]= array('uid' => 100, 'request' => 'upvote node 1');$requests[]= array('uid' => 200, 'request' => 'downvote node 2');$requests[]= array('uid' => 300, 'request' => 'upvote node 1');

foreach ($requests as $request) { // Adds the item as a serialized string to the queue table $queue->createItem($request);}

SystemQueue Queue Table

Work is lined up, now what?

Data processing strategy:

• Does the work have to be done in order it was queued?

• Do we have a limit of items we're allowed to process per day?

• Do we have the resources to run multiple processing workers in parallel?

Simple Queue processor

function user_requests_processor_1() { $queue = DrupalQueue::get('user_requests'); while($queue->numberOfItems()) { $request = $queue->claimItem(); user_requests_process_request($request); $queue->deleteItem($item); // Done, remove it from queue }}

Feedback on Worker

Worker: user_requests_processor_1

• Still times out

• Potential headaches depending on the success rate of user_request_process_request()

• Potential headaches if running multiple workers.

• Better memory footprint.

• Overall a positive change

Working with a Queuefunction user_requests_processor_2() { $queue = DrupalQueue::get('user_requests'); $ops = array('%count' => $queue->numberOfItems()); drupal_set_message("Processing user requests. %count Items in

queue.", $ops); while ($request = $queue->claimItem()) { if (user_requests_process_request($request)) { $queue->deleteItem($item); // Done, remove it from queue } else { // Couldn't process this. Leave it in the queue $queue->releaseItem($item); } }}

Feedback on Worker

Worker: user_requests_processor_2

• Both will time out... but....

• Can run in scaled up or down as needed. Big win.

Need a better Queue?

• Store items in a different location?• Want to notify somebody when you

create a Queue?• Don’t want to allow locking of items?

Subclass SystemQueueclass BatchQueue extends SystemQueue { public function claimItem($lease_time = 0) { // Ignores lease time. Forces sequential processing. $item = db_query_range('SELECT data, item_id FROM {queue} q WHERE name = :name ORDER BY item_id

ASC', 0, 1, array(':name' => $this->name))->fetchObject(); if ($item) { $item->data = unserialize($item->data); return $item; } return FALSE; }

public function getAllItems() { $result = array(); $items = db_query('SELECT data FROM {queue} q WHERE name = :name ORDER BY item_id ASC',

array(':name' => $this->name))->fetchAll(); foreach ($items as $item) { $result[] = unserialize($item->data); } return $result; }}

Subclass SystemQueueclass AwesomeQueue extends SystemQueue { public function releaseMultipleItems($items) { // Release multiple items }}

Extending SystemQueue

DrupalQueue::get('user_requests');new SystemQueue('user_requests');

// But we need an AwesomeQueuenew AwesomeQueue('user_requests');

Need a better Queue?Changing our Queue Classvariable_set('queue_class_user_request', 'AwesomeQueue');

Change Queue Class by Queuevariable_get('queue_class_' . $name, NULL);

Additionally…

Change Default Queue Classvariable_get('queue_default_class', 'SystemQueue');

What about MemoryQueue?

• MemoryQueue implements the same base interface as SystemQueue.

• Items you put into a MemoryQueue need to be processed within the same request.

• Use it for small queues you know you can take care off in one page request. Otherwise, stick to the SystemQueue.

Helpful Links/Debugging

Interactive Queuedhttp://tinyurl.com/cyk9qch

Views Queuehttp://dgo.to/views_queue

Queue UIhttp://dgo.to/queue_ui

Simple Queue process in chunks

function user_requests_processor_3() { $queue = DrupalQueue::get('user_requests'); while($queue->numberOfItems()) { $limit = 1000; $chunk = array(); while ($limit && $request = $queue->claimItem()) { $limit--; $chunk[] = $request; $queue->deleteItem($item); // Done, remove it from queue } user_requests_process_requests($chunk); }}

Batch API

Built In Procrastination

What is Batch?

• Another tool for getting a lot of work done.

• Built to help fight off issues with hitting max_execution_time

When to use Batch

• When you have a lot of tasks that need to be done.

• When you want inform the user of progress on the work.

You've seen Batch at work

Familiar Cases:Drupal InstallRebuild Node AccessDrupal UpdatePretty much whenever you see a progress

bar

How does batch fight off timeout?

Batch Form Submit

Batch Engine

How much work is there to do? Where are we at?Batch

EngineDo SOME of the work.

Are we finishe

d?No

New HTTP RequestReset the

clock! yey!

YesBatch Engine

Complete

Lets get to it!function user_requests_batch_form($form, &$form_state) { $form['submit'] = array( '#type' => 'submit', '#value' => t('Process User Requests'), ); return $form;}function user_requests_batch_form_submit($form, &$form_state) { $batch = array( 'operations' => array( array(user_requests_process_batch_requests', array()), ), ); batch_set($batch);}

Multiple Operationsfunction user_requests_batch_form_submit($form, &$form_state) { $batch = array( 'operations' => array( array( 'user_requests_preprocess_requests', array() ), array( 'user_requests_process_requests', array('argument_1', 'argument2') ), ), ); batch_set($batch);}

Batch variables $batch = array( 'init_message' => t('Getting ready to Process User Requests'), 'title' => t('Processing User Requests'), 'progress_message' => t('Processed @current of @total User

Requests'), 'error_message' => t('An Error occurred while processing User

Request'), 'finished' => 'user_requests_finished_batch_work', 'operations' => array( array( 'user_requests_preprocess_requests',array()), ), ), );

$batch['progess_message']

@current - current operation@remaining - remaining operations@total - total operations@percentage - percentage complete@elapsed - Time elapsed

Defaults to t('Completed @current of @total.').

Batch Operation Callbacks

Operation callbacks need to accept $context variable.

function user_requests_preprocess_requests(&$context);

function user_requests_process_requests($arg1, $arg2, &$context);

Batch $context

// Operations can tell Batch what they have done$context['results'][];// The message to display on the progress page$context['message'];// Sandbox is primary means of communication between

Iterations$context['sandbox']; // A float between 0 and 1 the tells the batch engine is the

batch work is done. Defaults to 1.// Moves the progress bar forward.$context['finished'] = 0;

Back to our user requestsuser_requests_process_batch_requests(&$context) { if (empty($context['sandbox'])) { $context['sandbox']['max'] = count_db_items(); $context['sandbox']['current'] = 0; } // Fetch next item $context['sandbox']['current']++; user_requests_process_request($request); $context['message'] = 'Processed request ' . $request-

>id $max = $context['sandbox']['max']; $context['finished'] =

$context['sandbox']['current']/$max;}

User Requests Demo

Bringing it together

• DrupalQueue and Batch work great as Complementary Systems.– DrupalQueue set tasks up.– Batch knocks them down.

DrupalQueue in Batch Operation #1

user_requests_process_batch_requests(&$context) { $queue = DrupalQueue::get('user_requests'); if (empty($context['sandbox'])) { $context['sandbox']['max']=$queue->numberOfItems(); $context['sandbox']['current']=0; } $request=$queue->claimItem(); $context['sandbox']['current']++; user_requests_process_request($request); $queue->deleteItem($request); $max = $context['sandbox']['max']; $context['finished'] = $context['sandbox']['current']/$max;

}

DrupalQueue in Batch Operation #2user_requests_process_batch_requests(&$context) { $queue = DrupalQueue::get('user_requests'); if (empty($context['sandbox'])) { $context['sandbox']['max'] = $queue->numberOfItems(); $context['sandbox']['current'] = 0; } for ($i = 1; $i < 20; $i++) { // Process 20 items on each request. if ($request = $queue->claimItem()){ $context['sandbox']['current']++; user_requests_process_request($request); $queue->deleteItem($request); } } $max = $context['sandbox']['max']; $context['finished'] = $context['sandbox']['current']/$max;}

Further Reading

Batch Operations Pagehttp://tinyurl.com/7asc4p8

Batch Example Module in Examples projecthttp://dgo.to/examples

Summary

DrupalQueue - SystemQueue• Great for saving work you can do later.• Simple tool with big implications.

– Great way to bring distributed processing of data to you Drupal projects.

DrupalQueue - MemoryQueue• Use it for lining up tasks that can be

handled in one request.

Summary

Batch API• Drupal's goto solution for getting past

max_execution_time limits• Great for processing a large amount of in

chunks at a time.• Really shines when users need to be

notified of progress on the work being completed.

• Compliments DrupalQueue very well.

Thank YouDagoberto Aceves

www.achieveinternet.comdago.aceves