Package buildbot :: Package status :: Module persistent_queue
[frames] | no frames]

Source Code for Module buildbot.status.persistent_queue

  1  # This file is part of Buildbot.  Buildbot is free software: you can 
  2  # redistribute it and/or modify it under the terms of the GNU General Public 
  3  # License as published by the Free Software Foundation, version 2. 
  4  # 
  5  # This program is distributed in the hope that it will be useful, but WITHOUT 
  6  # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS 
  7  # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more 
  8  # details. 
  9  # 
 10  # You should have received a copy of the GNU General Public License along with 
 11  # this program; if not, write to the Free Software Foundation, Inc., 51 
 12  # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. 
 13  # 
 14  # Copyright Buildbot Team Members 
 15   
 16   
 17  try: 
 18      # Python 2.4+ 
 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   
29 -def ReadFile(path):
30 f = open(path, 'rb') 31 try: 32 return f.read() 33 finally: 34 f.close()
35 36
37 -def WriteFile(path, buf):
38 f = open(path, 'wb') 39 try: 40 f.write(buf) 41 finally: 42 f.close()
43 44
45 -class IQueue(Interface):
46 """Abstraction of a queue."""
47 - def pushItem(item):
48 """Adds an individual item to the end of the queue. 49 50 Returns an item if it was overflowed."""
51
52 - def insertBackChunk(items):
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
60 - def popChunk(nbItems=None):
61 """Pop many items at once. Defaults to self.maxItems()."""
62
63 - def save():
64 """Save the queue to storage if implemented."""
65
66 - def items():
67 """Returns items in the queue. 68 69 Warning: Can be extremely slow for queue on disk."""
70
71 - def nbItems():
72 """Returns the number of items in the queue."""
73
74 - def maxItems():
75 """Returns the maximum number of items this queue can hold."""
76 77 78 # Available for python 2.3 and earlier.
79 -class ListMemoryQueue(object):
80 """Simple length bounded queue using list.""" 81 implements(IQueue) 82
83 - def __init__(self, maxItems=None):
84 self._maxItems = maxItems 85 if self._maxItems is None: 86 self._maxItems = 10000 87 self._items = []
88
89 - def pushItem(self, item):
90 self._items.append(item) 91 if len(self._items) > self._maxItems: 92 return self._items.pop(0)
93
94 - def insertBackChunk(self, chunk):
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
103 - def popChunk(self, nbItems=None):
104 if nbItems is None: 105 nbItems = self._maxItems 106 if nbItems > len(self._items): 107 items = self._items 108 self._items = [] 109 else: 110 items = self._items[0:nbItems] 111 del self._items[0:nbItems] 112 return items
113
114 - def save(self):
115 pass
116
117 - def items(self):
118 return self._items
119
120 - def nbItems(self):
121 return len(self._items)
122
123 - def maxItems(self):
124 return self._maxItems
125 126 if deque:
127 - class DequeMemoryQueue(object):
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
137 - def __init__(self, maxItems=None):
138 self._maxItems = maxItems 139 if self._maxItems is None: 140 self._maxItems = 10000 141 self._items = deque()
142
143 - def pushItem(self, item):
144 ret = None 145 if len(self._items) == self._maxItems: 146 ret = self._items.popleft() 147 self._items.append(item) 148 return ret
149
150 - def insertBackChunk(self, chunk):
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
159 - def popChunk(self, nbItems=None):
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
171 - def save(self):
172 pass
173
174 - def items(self):
175 return list(self._items)
176
177 - def nbItems(self):
178 return len(self._items)
179
180 - def maxItems(self):
181 return self._maxItems
182 183 MemoryQueue = DequeMemoryQueue 184 else: 185 MemoryQueue = ListMemoryQueue 186 187
188 -class DiskQueue(object):
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 # Total number of items. 213 self._nbItems = 0 214 # The actual items id start at one. 215 self.firstItemId = 0 216 self.lastItemId = 0 217 self._loadFromDisk()
218
219 - def pushItem(self, item):
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
236 - def insertBackChunk(self, chunk):
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
251 - def popChunk(self, nbItems=None):
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
266 - def save(self):
267 pass
268
269 - def items(self):
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
278 - def nbItems(self):
279 return self._nbItems
280
281 - def maxItems(self):
282 return self._maxItems
283 284 #### Protected functions 285
286 - def _findNext(self, id):
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
294 - def _loadFromDisk(self):
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
312 -class PersistentQueue(object):
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 # Preload data from the secondary queue only if we know we won't start 335 # using the secondary queue right away. 336 if self.secondaryQueue.nbItems() < self.primaryQueue.maxItems(): 337 self.primaryQueue.insertBackChunk( 338 self.secondaryQueue.popChunk(self.primaryQueue.maxItems()))
339
340 - def pushItem(self, item):
341 # If there is already items in secondaryQueue, we'd need to pop them 342 # all to start inserting them into primaryQueue so don't bother and 343 # just push it in secondaryQueue. 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 # If item is not None, secondaryQueue overflowed. We need to push it 350 # back to primaryQueue so the oldest item is dumped. 351 # Or everything fit in the primaryQueue. 352 return self.primaryQueue.pushItem(item)
353
354 - def insertBackChunk(self, chunk):
355 ret = None 356 # Overall excess 357 excess = self.nbItems() + len(chunk) - self.maxItems() 358 if excess > 0: 359 ret = chunk[0:excess] 360 chunk = chunk[excess:] 361 # Memory excess 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
377 - def popChunk(self, nbItems=None):
378 if nbItems is None: 379 nbItems = self.primaryQueue.maxItems() 380 ret = self.primaryQueue.popChunk(nbItems) 381 nbItems -= len(ret) 382 if nbItems and self.secondaryQueue.nbItems(): 383 ret.extend(self.secondaryQueue.popChunk(nbItems)) 384 return ret
385
386 - def save(self):
387 self.secondaryQueue.insertBackChunk(self.primaryQueue.popChunk())
388
389 - def items(self):
390 return self.primaryQueue.items() + self.secondaryQueue.items()
391
392 - def nbItems(self):
393 return self.primaryQueue.nbItems() + self.secondaryQueue.nbItems()
394
395 - def maxItems(self):
396 return self.primaryQueue.maxItems() + self.secondaryQueue.maxItems()
397 398
399 -class IndexedQueue(object):
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
407 - def __init__(self, queue):
408 # Copy all the member functions from the other object that this class 409 # doesn't already define. 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
419 - def getIndex(self):
420 return self._index
421
422 - def popChunk(self, *args, **kwargs):
423 items = self.queue.popChunk(*args, **kwargs) 424 if items: 425 self._index += len(items) 426 return items
427
428 - def insertBackChunk(self, items):
429 self._index -= len(items) 430 ret = self.queue.insertBackChunk(items) 431 if ret: 432 self._index += len(ret) 433 return ret
434 435
436 -def ToIndexedQueue(queue):
437 """If the IQueue wasn't already a IndexedQueue, makes it an IndexedQueue.""" 438 if not IQueue.providedBy(queue): 439 raise TypeError("queue doesn't implement IQueue", queue) 440 if isinstance(queue, IndexedQueue): 441 return queue 442 return IndexedQueue(queue)
443 444 # vim: set ts=4 sts=4 sw=4 et: 445