Library: queues

From Yombo
Jump to: navigation, search
queues
Commonly Used No
Status Active

Summary
FIFO queues allow you to add items to a queue and will call a function for each item to be processed. Prevents system overload and handled surges of requests gracefully.

Introduction

Work queues can be used as a method to throttle workload or to save items for processing later. Note: Currently, jobs in the queue are not saved when the system exists.

See also

Hooks

No hooks registered.

Usage

 1 from txrdq.rdq import ResizableDispatchQueue
 2 # Create a queue that will allow 2 simultaneous underway jobs.
 3 my_queue = self._Queue.new('module.mymodule.my_queue', self.sendSMS, 2)
 4 
 5 # Later... send off some SMS messages.
 6 my_queue.put(('2127399921', 'Hello...'), priority=5)
 7 
 8 # queue a higher priority message and tell the system to
 9 # call self.important_sms_sent(results) when it's done.
10 d1 = rdq.put((5052929919, 'Test...'), priority=0)
11 d1.addCallback(self.important_sms_sent)
12 
13 d2 = my_queue.put(('2127399921', 'Hello...'), priority=10)
14 # Cancel the second job
15 d2.cancel()
16 
17 # Widen the outgoing pipeline to 10 simultaneous jobs. Now it's 12 total.
18 my_queue.width = 10
19 
20 # We're dispatching jobs too fast, turn it down a little.
21 my_queue.width = 7
22 
23 # Get a copy of the list of pending jobs.
24 jobs = my_queue.pending()  # returns a tuple - (underway_jobs, queued_jobs)
25 
26 # Cancel one of the pending jobs from the jobs list.
27 jobs[1][2].cancel()
28 
29 # Reprioritize one of the pending jobs from the jobs list.
30 my_queue.reprioritize(jobs[1][0], -1)
31 
32 # In an hour from now, set the width to 20 jobs at a time.
33 reactor.callLater(3600, rdq.setWidth, 20)
34 
35 # Pause processing.
36 rdq.pause()
37 
38 # Resume processing, with a new width of 8.
39 rdq.resume(8)
40 
41 # Shutdown. Wait for any underway jobs to complete, and save
42 # the list of jobs not yet processed.
43 
44 def saveJobs(jobs):
45     pickle.dump(jobs, ...)
46 
47 d = rdq.stop()
48 d.addCallback(saveJobs)


This page was last edited on 9 December 2017.