1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 try:
18
19 from collections import deque
20 assert deque
21 except ImportError:
22 deque = None
23 import os
24 import pickle
25
26 from zope.interface import implements, Interface
27
28
30 f = open(path, 'rb')
31 try:
32 return f.read()
33 finally:
34 f.close()
35
36
38 f = open(path, 'wb')
39 try:
40 f.write(buf)
41 finally:
42 f.close()
43
44
46 """Abstraction of a queue."""
48 """Adds an individual item to the end of the queue.
49
50 Returns an item if it was overflowed."""
51
53 """Adds a list of items as the oldest entries.
54
55 Normally called in case of failure to process the data, queue the data
56 back so it can be retrieved at a later time.
57
58 Returns a list of items if it was overflowed."""
59
61 """Pop many items at once. Defaults to self.maxItems()."""
62
64 """Save the queue to storage if implemented."""
65
67 """Returns items in the queue.
68
69 Warning: Can be extremely slow for queue on disk."""
70
72 """Returns the number of items in the queue."""
73
75 """Returns the maximum number of items this queue can hold."""
76
77
78
80 """Simple length bounded queue using list."""
81 implements(IQueue)
82
84 self._maxItems = maxItems
85 if self._maxItems is None:
86 self._maxItems = 10000
87 self._items = []
88
90 self._items.append(item)
91 if len(self._items) > self._maxItems:
92 return self._items.pop(0)
93
95 ret = None
96 excess = len(self._items) + len(chunk) - self._maxItems
97 if excess > 0:
98 ret = chunk[0:excess]
99 chunk = chunk[excess:]
100 self._items = chunk + self._items
101 return ret
102
113
116
119
121 return len(self._items)
122
124 return self._maxItems
125
126 if deque:
128 """Simple length bounded queue using deque.
129
130 list.pop(0) operation is O(n) so for a 10000 items list, it can start to
131 be real slow. On the contrary, deque.popleft() is O(1) most of the time.
132 See http://docs.python.org/library/collections.html for more
133 information.
134 """
135 implements(IQueue)
136
138 self._maxItems = maxItems
139 if self._maxItems is None:
140 self._maxItems = 10000
141 self._items = deque()
142
144 ret = None
145 if len(self._items) == self._maxItems:
146 ret = self._items.popleft()
147 self._items.append(item)
148 return ret
149
151 ret = None
152 excess = len(self._items) + len(chunk) - self._maxItems
153 if excess > 0:
154 ret = chunk[0:excess]
155 chunk = chunk[excess:]
156 self._items.extendleft(reversed(chunk))
157 return ret
158
160 if nbItems is None:
161 nbItems = self._maxItems
162 if nbItems > len(self._items):
163 items = list(self._items)
164 self._items = deque()
165 else:
166 items = []
167 for i in range(nbItems):
168 items.append(self._items.popleft())
169 return items
170
173
175 return list(self._items)
176
178 return len(self._items)
179
181 return self._maxItems
182
183 MemoryQueue = DequeMemoryQueue
184 else:
185 MemoryQueue = ListMemoryQueue
186
187
189 """Keeps a list of abstract items and serializes it to the disk.
190
191 Use pickle for serialization."""
192 implements(IQueue)
193
194 - def __init__(self, path, maxItems=None, pickleFn=pickle.dumps,
195 unpickleFn=pickle.loads):
196 """
197 @path: directory to save the items.
198 @maxItems: maximum number of items to keep on disk, flush the
199 older ones.
200 @pickleFn: function used to pack the items to disk.
201 @unpickleFn: function used to unpack items from disk.
202 """
203 self.path = path
204 self._maxItems = maxItems
205 if self._maxItems is None:
206 self._maxItems = 100000
207 if not os.path.isdir(self.path):
208 os.mkdir(self.path)
209 self.pickleFn = pickleFn
210 self.unpickleFn = unpickleFn
211
212
213 self._nbItems = 0
214
215 self.firstItemId = 0
216 self.lastItemId = 0
217 self._loadFromDisk()
218
220 ret = None
221 if self._nbItems == self._maxItems:
222 id = self._findNext(self.firstItemId)
223 path = os.path.join(self.path, str(id))
224 ret = self.unpickleFn(ReadFile(path))
225 os.remove(path)
226 self.firstItemId = id + 1
227 else:
228 self._nbItems += 1
229 self.lastItemId += 1
230 path = os.path.join(self.path, str(self.lastItemId))
231 if os.path.exists(path):
232 raise IOError('%s already exists.' % path)
233 WriteFile(path, self.pickleFn(item))
234 return ret
235
237 ret = None
238 excess = self._nbItems + len(chunk) - self._maxItems
239 if excess > 0:
240 ret = chunk[0:excess]
241 chunk = chunk[excess:]
242 for i in reversed(chunk):
243 self.firstItemId -= 1
244 path = os.path.join(self.path, str(self.firstItemId))
245 if os.path.exists(path):
246 raise IOError('%s already exists.' % path)
247 WriteFile(path, self.pickleFn(i))
248 self._nbItems += 1
249 return ret
250
252 if nbItems is None:
253 nbItems = self._maxItems
254 ret = []
255 for i in range(nbItems):
256 if self._nbItems == 0:
257 break
258 id = self._findNext(self.firstItemId)
259 path = os.path.join(self.path, str(id))
260 ret.append(self.unpickleFn(ReadFile(path)))
261 os.remove(path)
262 self._nbItems -= 1
263 self.firstItemId = id + 1
264 return ret
265
268
270 """Warning, very slow."""
271 ret = []
272 for id in range(self.firstItemId, self.lastItemId + 1):
273 path = os.path.join(self.path, str(id))
274 if os.path.exists(path):
275 ret.append(self.unpickleFn(ReadFile(path)))
276 return ret
277
280
282 return self._maxItems
283
284
285
287 while True:
288 path = os.path.join(self.path, str(id))
289 if os.path.isfile(path):
290 return id
291 id += 1
292 return None
293
295 """Loads the queue from disk upto self.maxMemoryItems items into
296 self.items.
297 """
298 def SafeInt(item):
299 try:
300 return int(item)
301 except ValueError:
302 return None
303
304 files = filter(None, [SafeInt(x) for x in os.listdir(self.path)])
305 files.sort()
306 self._nbItems = len(files)
307 if self._nbItems:
308 self.firstItemId = files[0]
309 self.lastItemId = files[-1]
310
311
313 """Keeps a list of abstract items and serializes it to the disk.
314
315 It has 2 layers of queue, normally an in-memory queue and an on-disk queue.
316 When the number of items in the primary queue gets too large, the new items
317 are automatically saved to the secondary queue. The older items are kept in
318 the primary queue.
319 """
320 implements(IQueue)
321
322 - def __init__(self, primaryQueue=None, secondaryQueue=None, path=None):
323 """
324 @primaryQueue: memory queue to use before buffering to disk.
325 @secondaryQueue: disk queue to use as permanent buffer.
326 @path: path is a shortcut when using default DiskQueue settings.
327 """
328 self.primaryQueue = primaryQueue
329 if self.primaryQueue is None:
330 self.primaryQueue = MemoryQueue()
331 self.secondaryQueue = secondaryQueue
332 if self.secondaryQueue is None:
333 self.secondaryQueue = DiskQueue(path)
334
335
336 if self.secondaryQueue.nbItems() < self.primaryQueue.maxItems():
337 self.primaryQueue.insertBackChunk(
338 self.secondaryQueue.popChunk(self.primaryQueue.maxItems()))
339
341
342
343
344 if (self.secondaryQueue.nbItems() or
345 self.primaryQueue.nbItems() == self.primaryQueue.maxItems()):
346 item = self.secondaryQueue.pushItem(item)
347 if item is None:
348 return item
349
350
351
352 return self.primaryQueue.pushItem(item)
353
355 ret = None
356
357 excess = self.nbItems() + len(chunk) - self.maxItems()
358 if excess > 0:
359 ret = chunk[0:excess]
360 chunk = chunk[excess:]
361
362 excess = (self.primaryQueue.nbItems() + len(chunk) -
363 self.primaryQueue.maxItems())
364 if excess > 0:
365 chunk2 = []
366 for i in range(excess):
367 chunk2.append(self.primaryQueue.items().pop())
368 chunk2.reverse()
369 x = self.primaryQueue.insertBackChunk(chunk)
370 assert x is None, "primaryQueue.insertBackChunk did not return None"
371 if excess > 0:
372 x = self.secondaryQueue.insertBackChunk(chunk2)
373 assert x is None, ("secondaryQueue.insertBackChunk did not return "
374 " None")
375 return ret
376
385
388
390 return self.primaryQueue.items() + self.secondaryQueue.items()
391
393 return self.primaryQueue.nbItems() + self.secondaryQueue.nbItems()
394
397
398
400 """Adds functionality to a IQueue object to track its usage.
401
402 Adds a new member function getIndex() and modify popChunk() and
403 insertBackChunk() to keep a virtual pointer to the queue's first entry
404 index."""
405 implements(IQueue)
406
408
409
410 assert IQueue.providedBy(queue)
411 def Filter(m):
412 return (m[0] != '_' and callable(getattr(queue, m))
413 and not hasattr(self, m))
414 for member in filter(Filter, dir(queue)):
415 setattr(self, member, getattr(queue, member))
416 self.queue = queue
417 self._index = 0
418
421
427
434
435
443
444
445