Framework for building threaded asynchronous job processors.
task_tempest
basically lets you build glorified loops, reading messages from a queue and dispatching them to classes to handle. In short, it’s just another background job processor for Ruby.
task_tempest
is based on the idea of a thread pool; each background job is executed on a thread. This can achieve high throughput, but your code will need to be threadsafe.
task_tempest
can also run in fibered mode in conjunction with EventMachine. This can achieve high(er?) throughput, but your code will need to be fiber aware.
task_tempest
is queue agnostic; you can use whatever queue you like, as long as it supports two operations: enqueue, dequeue.
task_tempest
is used in production at Onespot and processes over 1 million jobs a day.
Run the code below and both MathTempest.log and MathTempest.task.log will be created in your current working directory. To stop the tempest, send the process an Interrupt or SIGTERM.
require "task_tempest" class MemoryQueue def initialize; @array = []; end def enqueue(message); @array.push(message); end def dequeue; @array.shift; end end class MathTempest < TaskTempest::Engine configure do threads 2 queue MemoryQueue.new end end class Adder extend TaskTempest::Task def process(a, b) c = a + b task_logger.info "#{a} + #{b} = #{c}" end end 10.times do a, b = rand(10), rand(10) MathTempest.submit(Adder, a, b) end MathTempest.run
The queue object given in the TaskTempest::Engine’s configure
block must define 2 methods: enqueue
and dequeue
.
task_tempest
doesn’t care what kind of queue you use or what you actually store in it, as long as dequeue
returns a message as described here. You are responsible for any serialization and deserialization that may be required.
A message is an array with the following format:
[task_id, task_class_name, arg1, arg2, ...]
An example message is:
[nil, "Adder", 2, 5]
The queue’s enqueue
method will receive a message as its first argument, plus any additional arguments given to submit
.
The queue’s dequeue
method must return a message.
If the task_id
of a message is nil
, then task_tempest
will automatically assign an id.
Messages are dispatched to a task class to handle. A task class must extend TaskTempest::Task and define a process
method. The process
method will be called with the arguments from the message being dispatched.
class Averager extend TaskTempest::Task configure_task do ... end def process(*args) avg = args.inject(0){ |memo, n| memo += n; memo } / args.length task_logger.info "The average of #{args.inspect} is #{avg}" end end MathTempest.submit(Averager, 2, 4) # Will produce a message like ["9ac4f", "Averager", 2, 4] # And eventually Averager.process(2, 4) will be called. MathTempest.submit(Averager, 2, 4, 6) # Will produce a message like ["b733c", "Averager", 2, 4, 6] # And eventually Averager.process(2, 4, 6) will be called.
See TaskTempest::Task::Configuration for what options can be set in a task’s configure
block.
task_tempest
logs to a main log and a task log.
The main log shows a high level view of all the tasks that are run (when they started, when the finished, if they failed, if they timed out, etc).
The task log shows detailed information about each task. Each line in the task log is prefixed with a task id. This is so that if you see a task failed in the main log, you can grep for its id in the task log.
Any logging done in a task using task_logger
will be written to the task log.
Each task class can define callbacks for success
, failure
, and timeout
of a task.
class Averager extend TaskTempest::Task configure_task do timeout 1.5 after_failure proc { |exception| HoptoadNotifier.notify(exception) } after_timeout proc { task_logger.warn "Crap, I took longer than 1.5 seconds" } end ... end
See TaskTempest::Task::Configuration for the names of each callback and arguments yielded.
The fibers
option puts task_tempest
into fibered mode and also defines the size of the fiber pool.
class FiberedTempest << TaskTempest::Engine configure do fibers 5 ... end ... end
Now when FiberedTempest.run
is called, The EventMachine reactor will be started and each task will be dispatched on a fiber.
If you run in fibered mode, you need to install fiber_storm
.
There is no code in task_tempest
to run as a daemon, that is left to you. It’s easy with the Daemons gem though.
Assuming your tempest is defined in my_tempest.rb
, just put the following code at the bottom of the file.
if $0 == __FILE__ require "daemons" Daemons.run_proc(MyTempest.conf.name, :log_output => true) do MyTempest.run end end
Now you can run it as a daemon from the command line.
ruby my_tempest.rb start ruby my_tempest.rb stop ruby my_tempest.rb run # Run in foreground
See the Daemons rdoc for more info.
When task_tempest
is used as an asynchronous task processor for a Rails app, you probably want to load the Rails environment so you can have access to models, configuration, etc.
You can load the Rails environment however you like, but if you choose to load it in one of task_tempest
‘s initialization callbacks, then there are some caveats to be aware of.
class MathTempest < TaskTempest::Engine configure do root{ Rails.root } after_initialize proc { require "config/environment" # Load Rails. } end end
That won’t work and will result in an exception saying Rails
isn’t defined, because the root
configuration option is used during the initialization process before the after_initialized
callback is called. To fix the error, change after_initialized
to before_initialize
.
See the examples
directory.
Copyright © 2010 Christopher J. Bottaro. See LICENSE for details.