Revision 5685c1a5 lib/jqueue.py
b/lib/jqueue.py | ||
---|---|---|
1 | 1 |
# |
2 | 2 |
# |
3 | 3 |
|
4 |
# Copyright (C) 2006, 2007 Google Inc. |
|
4 |
# Copyright (C) 2006, 2007, 2008 Google Inc.
|
|
5 | 5 |
# |
6 | 6 |
# This program is free software; you can redistribute it and/or modify |
7 | 7 |
# it under the terms of the GNU General Public License as published by |
... | ... | |
33 | 33 |
import errno |
34 | 34 |
import re |
35 | 35 |
import time |
36 |
import weakref |
|
36 | 37 |
|
37 | 38 |
from ganeti import constants |
38 | 39 |
from ganeti import serializer |
... | ... | |
312 | 313 |
|
313 | 314 |
def __init__(self, context): |
314 | 315 |
self.context = context |
315 |
self._memcache = {}
|
|
316 |
self._memcache = weakref.WeakValueDictionary()
|
|
316 | 317 |
self._my_hostname = utils.HostInfo().name |
317 | 318 |
|
318 | 319 |
# Locking |
... | ... | |
489 | 490 |
if self._RE_JOB_FILE.match(name)] |
490 | 491 |
|
491 | 492 |
def _LoadJobUnlocked(self, job_id): |
492 |
if job_id in self._memcache: |
|
493 |
job = self._memcache.get(job_id, None) |
|
494 |
if job: |
|
493 | 495 |
logging.debug("Found job %s in memcache", job_id) |
494 |
return self._memcache[job_id]
|
|
496 |
return job
|
|
495 | 497 |
|
496 | 498 |
filepath = self._GetJobPath(job_id) |
497 | 499 |
logging.debug("Loading job from %s", filepath) |
... | ... | |
536 | 538 |
# Write to disk |
537 | 539 |
self.UpdateJobUnlocked(job) |
538 | 540 |
|
539 |
logging.debug("Added new job %s to the cache", job_id)
|
|
541 |
logging.debug("Adding new job %s to the cache", job_id)
|
|
540 | 542 |
self._memcache[job_id] = job |
541 | 543 |
|
542 | 544 |
# Add to worker pool |
... | ... | |
550 | 552 |
data = serializer.DumpJson(job.Serialize(), indent=False) |
551 | 553 |
logging.debug("Writing job %s to %s", job.id, filename) |
552 | 554 |
self._WriteAndReplicateFileUnlocked(filename, data) |
553 |
self._CleanCacheUnlocked([job.id]) |
|
554 | 555 |
|
555 | 556 |
# Notify waiters about potential changes |
556 | 557 |
job.change.notifyAll() |
557 | 558 |
|
558 |
def _CleanCacheUnlocked(self, exclude): |
|
559 |
"""Clean the memory cache. |
|
560 |
|
|
561 |
The exceptions argument contains job IDs that should not be |
|
562 |
cleaned. |
|
563 |
|
|
564 |
""" |
|
565 |
assert isinstance(exclude, list) |
|
566 |
|
|
567 |
for job in self._memcache.values(): |
|
568 |
if job.id in exclude: |
|
569 |
continue |
|
570 |
if job.CalcStatus() not in (constants.JOB_STATUS_QUEUED, |
|
571 |
constants.JOB_STATUS_RUNNING): |
|
572 |
logging.debug("Cleaning job %s from the cache", job.id) |
|
573 |
try: |
|
574 |
del self._memcache[job.id] |
|
575 |
except KeyError: |
|
576 |
pass |
|
577 |
|
|
578 | 559 |
@utils.LockedMethod |
579 | 560 |
@_RequireOpenQueue |
580 | 561 |
def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial): |
... | ... | |
679 | 660 |
logging.debug("Job %s is not yet done", job.id) |
680 | 661 |
return |
681 | 662 |
|
682 |
try: |
|
683 |
old = self._GetJobPath(job.id) |
|
684 |
new = self._GetArchivedJobPath(job.id) |
|
663 |
old = self._GetJobPath(job.id) |
|
664 |
new = self._GetArchivedJobPath(job.id) |
|
685 | 665 |
|
686 |
self._RenameFileUnlocked(old, new)
|
|
666 |
self._RenameFileUnlocked(old, new) |
|
687 | 667 |
|
688 |
logging.debug("Successfully archived job %s", job.id) |
|
689 |
finally: |
|
690 |
# Cleaning the cache because we don't know what os.rename actually did |
|
691 |
# and to be on the safe side. |
|
692 |
self._CleanCacheUnlocked([]) |
|
668 |
logging.debug("Successfully archived job %s", job.id) |
|
693 | 669 |
|
694 | 670 |
def _GetJobInfoUnlocked(self, job, fields): |
695 | 671 |
row = [] |
Also available in: Unified diff