Toto provides build a built in task queue for performing work in the background while limiting the number of active jobs. The task queue is designed primarily for shorter, lightweight jobs. For CPU intensive tasks or tasks that are expected to run for a long time, look at Toto’s worker functionality instead.
- class toto.tasks.TaskQueue(thread_count=1, idle_timeout=60, name='TaskQueue')¶
Instances will run up to thread_count tasks at a time whenever there are tasks in the queue. Theads will be killed if unused for longer than idle_timeout seconds.
- classmethod TaskQueue.instance(name, thread_count=1, idle_timeout=60)¶
A convenience method for accessing shared instances of TaskQueue. If name references an existing instance created with this method, that instance will be returned. Otherwise, a new TaskQueue will be instantiated with thread_count threads and the specified idle_timeout then stored under name.
All arguments after name are ignored after the queue is created.
- TaskQueue.add_task(fn, *args, **kwargs)¶
Add the function fn to the queue to be invoked with args and kwargs as arguments. If the TaskQueue is not currently running, it will be started now.
- TaskQueue.yield_task(fn, *args, **kwargs)¶
Like add_task but will call the function as a coroutine, allowing you to yield the return value from within a function decorated with @tornado.gen.coroutine or @tornado.gen.engine.
def add(arg1, arg2): return arg1 + arg2 @tornado.gen.engine def caller(): value = yield TaskQueue.instance('myqueue').yield_task(add, 1, 2) print value #prints 3
Start processing jobs in the queue. You should not need to call this as add_task automatically starts the queue. Processing threads will stop when there are no jobs available in the queue for at least idle_timeout seconds.
Returns the number of active threads plus the number of queued tasks/
- class toto.tasks.AwaitableInstance(instance, queue_name='pool')¶
Use AwaitableInstance(instance) to create a wrapper around instance that may be used to automatically execute any of instance methods in a TaskQueue. Use queue_name to specify the name of the TaskQueue to use.
class MyClass(object): def do_something(self, arg): ... return result awaitable = AwaitableInstance(MyClass()) result = yield awaitable.do_something(args)
The above will call do_something() in a TaskQueue and await the result.
- class toto.tasks.InstancePool(instances, default_queue='pool')¶
Create a pool wrapping one or more instances of a class. The pool has convenience methods for asynchronous execution and transactions. May be used e.g. to share non-threadsafe classes among various threads.
Invoking a method other than those documented below will remove an instance from the pool, invoke the method, then return it to the pool.
instances should be either a single instance or a collection of instances.
default_queue is the name of the TaskQueue that will be used for all asynchronous operations.
Create an AwaitableInstance wrapping this queue.
For use with the with operator. Allows multiple methods to be called on a single instance instance in the pool and returns it when complete.
with pool.instance() as i: i.operation1() i.operation2()
Will remove an instance from the pool, call two methods on it, then return it to the queue.
Calls function with an instance from the pool as the single argument. Returns the result of function(instance)
- InstancePool.async_transaction(function, queue_name=None)¶
Like transaction() but runs the operation in a TaskQueue. queue_name may be specified to override the default.
- InstancePool.await_transaction(function, queue_name=None)¶
Like transaction() but runs the operation in a TaskQueue and returns a Future that may be used to retrieve the result. Must be called from a Tornado IOLoop. queue_name may be specified to override the default.