code.grnet.gr Git - ganeti-local/blob - lib/jqueue.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Module implementing the job queue handling.
  23
  24 Locking: there's a single, large lock in the L{JobQueue} class. It's
  25 used by all other classes in this module.
  26
  27 @var JOBQUEUE_THREADS: the number of worker threads we start for
  28     processing jobs
  29
  30 """
  31
  32 import logging
  33 import errno
  34 import time
  35 import weakref
  36 import threading
  37 import itertools
  38
  39 try:
  40   # pylint: disable=E0611
  41   from pyinotify import pyinotify
  42 except ImportError:
  43   import pyinotify
  44
  45 from ganeti import asyncnotifier
  46 from ganeti import constants
  47 from ganeti import serializer
  48 from ganeti import workerpool
  49 from ganeti import locking
  50 from ganeti import opcodes
  51 from ganeti import errors
  52 from ganeti import mcpu
  53 from ganeti import utils
  54 from ganeti import jstore
  55 from ganeti import rpc
  56 from ganeti import runtime
  57 from ganeti import netutils
  58 from ganeti import compat
  59 from ganeti import ht
  60
  61
  62 JOBQUEUE_THREADS = 25
  63 JOBS_PER_ARCHIVE_DIRECTORY = 10000
  64
  65 # member lock names to be passed to @ssynchronized decorator
  66 _LOCK = "_lock"
  67 _QUEUE = "_queue"
  68
  69
  70 class CancelJob(Exception):
  71   """Special exception to cancel a job.
  72
  73   """
  74
  75
  76 def TimeStampNow():
  77   """Returns the current timestamp.
  78
  79   @rtype: tuple
  80   @return: the current time in the (seconds, microseconds) format
  81
  82   """
  83   return utils.SplitTime(time.time())
  84
  85
  86 class _QueuedOpCode(object):
  87   """Encapsulates an opcode object.
  88
  89   @ivar log: holds the execution log and consists of tuples
  90   of the form C{(log_serial, timestamp, level, message)}
  91   @ivar input: the OpCode we encapsulate
  92   @ivar status: the current status
  93   @ivar result: the result of the LU execution
  94   @ivar start_timestamp: timestamp for the start of the execution
  95   @ivar exec_timestamp: timestamp for the actual LU Exec() function invocation
  96   @ivar stop_timestamp: timestamp for the end of the execution
  97
  98   """
  99   __slots__ = ["input", "status", "result", "log", "priority",
 100                "start_timestamp", "exec_timestamp", "end_timestamp",
 101                "__weakref__"]
 102
 103   def __init__(self, op):
 104     """Constructor for the _QuededOpCode.
 105
 106     @type op: L{opcodes.OpCode}
 107     @param op: the opcode we encapsulate
 108
 109     """
 110     self.input = op
 111     self.status = constants.OP_STATUS_QUEUED
 112     self.result = None
 113     self.log = []
 114     self.start_timestamp = None
 115     self.exec_timestamp = None
 116     self.end_timestamp = None
 117
 118     # Get initial priority (it might change during the lifetime of this opcode)
 119     self.priority = getattr(op, "priority", constants.OP_PRIO_DEFAULT)
 120
 121   @classmethod
 122   def Restore(cls, state):
 123     """Restore the _QueuedOpCode from the serialized form.
 124
 125     @type state: dict
 126     @param state: the serialized state
 127     @rtype: _QueuedOpCode
 128     @return: a new _QueuedOpCode instance
 129
 130     """
 131     obj = _QueuedOpCode.__new__(cls)
 132     obj.input = opcodes.OpCode.LoadOpCode(state["input"])
 133     obj.status = state["status"]
 134     obj.result = state["result"]
 135     obj.log = state["log"]
 136     obj.start_timestamp = state.get("start_timestamp", None)
 137     obj.exec_timestamp = state.get("exec_timestamp", None)
 138     obj.end_timestamp = state.get("end_timestamp", None)
 139     obj.priority = state.get("priority", constants.OP_PRIO_DEFAULT)
 140     return obj
 141
 142   def Serialize(self):
 143     """Serializes this _QueuedOpCode.
 144
 145     @rtype: dict
 146     @return: the dictionary holding the serialized state
 147
 148     """
 149     return {
 150       "input": self.input.__getstate__(),
 151       "status": self.status,
 152       "result": self.result,
 153       "log": self.log,
 154       "start_timestamp": self.start_timestamp,
 155       "exec_timestamp": self.exec_timestamp,
 156       "end_timestamp": self.end_timestamp,
 157       "priority": self.priority,
 158       }
 159
 160
 161 class _QueuedJob(object):
 162   """In-memory job representation.
 163
 164   This is what we use to track the user-submitted jobs. Locking must
 165   be taken care of by users of this class.
 166
 167   @type queue: L{JobQueue}
 168   @ivar queue: the parent queue
 169   @ivar id: the job ID
 170   @type ops: list
 171   @ivar ops: the list of _QueuedOpCode that constitute the job
 172   @type log_serial: int
 173   @ivar log_serial: holds the index for the next log entry
 174   @ivar received_timestamp: the timestamp for when the job was received
 175   @ivar start_timestmap: the timestamp for start of execution
 176   @ivar end_timestamp: the timestamp for end of execution
 177   @ivar writable: Whether the job is allowed to be modified
 178
 179   """
 180   # pylint: disable=W0212
 181   __slots__ = ["queue", "id", "ops", "log_serial", "ops_iter", "cur_opctx",
 182                "received_timestamp", "start_timestamp", "end_timestamp",
 183                "__weakref__", "processor_lock", "writable"]
 184
 185   def __init__(self, queue, job_id, ops, writable):
 186     """Constructor for the _QueuedJob.
 187
 188     @type queue: L{JobQueue}
 189     @param queue: our parent queue
 190     @type job_id: job_id
 191     @param job_id: our job id
 192     @type ops: list
 193     @param ops: the list of opcodes we hold, which will be encapsulated
 194         in _QueuedOpCodes
 195     @type writable: bool
 196     @param writable: Whether job can be modified
 197
 198     """
 199     if not ops:
 200       raise errors.GenericError("A job needs at least one opcode")
 201
 202     self.queue = queue
 203     self.id = job_id
 204     self.ops = [_QueuedOpCode(op) for op in ops]
 205     self.log_serial = 0
 206     self.received_timestamp = TimeStampNow()
 207     self.start_timestamp = None
 208     self.end_timestamp = None
 209
 210     self._InitInMemory(self, writable)
 211
 212   @staticmethod
 213   def _InitInMemory(obj, writable):
 214     """Initializes in-memory variables.
 215
 216     """
 217     obj.writable = writable
 218     obj.ops_iter = None
 219     obj.cur_opctx = None
 220
 221     # Read-only jobs are not processed and therefore don't need a lock
 222     if writable:
 223       obj.processor_lock = threading.Lock()
 224     else:
 225       obj.processor_lock = None
 226
 227   def __repr__(self):
 228     status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
 229               "id=%s" % self.id,
 230               "ops=%s" % ",".join([op.input.Summary() for op in self.ops])]
 231
 232     return "<%s at %#x>" % (" ".join(status), id(self))
 233
 234   @classmethod
 235   def Restore(cls, queue, state, writable):
 236     """Restore a _QueuedJob from serialized state:
 237
 238     @type queue: L{JobQueue}
 239     @param queue: to which queue the restored job belongs
 240     @type state: dict
 241     @param state: the serialized state
 242     @type writable: bool
 243     @param writable: Whether job can be modified
 244     @rtype: _JobQueue
 245     @return: the restored _JobQueue instance
 246
 247     """
 248     obj = _QueuedJob.__new__(cls)
 249     obj.queue = queue
 250     obj.id = state["id"]
 251     obj.received_timestamp = state.get("received_timestamp", None)
 252     obj.start_timestamp = state.get("start_timestamp", None)
 253     obj.end_timestamp = state.get("end_timestamp", None)
 254
 255     obj.ops = []
 256     obj.log_serial = 0
 257     for op_state in state["ops"]:
 258       op = _QueuedOpCode.Restore(op_state)
 259       for log_entry in op.log:
 260         obj.log_serial = max(obj.log_serial, log_entry[0])
 261       obj.ops.append(op)
 262
 263     cls._InitInMemory(obj, writable)
 264
 265     return obj
 266
 267   def Serialize(self):
 268     """Serialize the _JobQueue instance.
 269
 270     @rtype: dict
 271     @return: the serialized state
 272
 273     """
 274     return {
 275       "id": self.id,
 276       "ops": [op.Serialize() for op in self.ops],
 277       "start_timestamp": self.start_timestamp,
 278       "end_timestamp": self.end_timestamp,
 279       "received_timestamp": self.received_timestamp,
 280       }
 281
 282   def CalcStatus(self):
 283     """Compute the status of this job.
 284
 285     This function iterates over all the _QueuedOpCodes in the job and
 286     based on their status, computes the job status.
 287
 288     The algorithm is:
 289       - if we find a cancelled, or finished with error, the job
 290         status will be the same
 291       - otherwise, the last opcode with the status one of:
 292           - waitlock
 293           - canceling
 294           - running
 295
 296         will determine the job status
 297
 298       - otherwise, it means either all opcodes are queued, or success,
 299         and the job status will be the same
 300
 301     @return: the job status
 302
 303     """
 304     status = constants.JOB_STATUS_QUEUED
 305
 306     all_success = True
 307     for op in self.ops:
 308       if op.status == constants.OP_STATUS_SUCCESS:
 309         continue
 310
 311       all_success = False
 312
 313       if op.status == constants.OP_STATUS_QUEUED:
 314         pass
 315       elif op.status == constants.OP_STATUS_WAITING:
 316         status = constants.JOB_STATUS_WAITING
 317       elif op.status == constants.OP_STATUS_RUNNING:
 318         status = constants.JOB_STATUS_RUNNING
 319       elif op.status == constants.OP_STATUS_CANCELING:
 320         status = constants.JOB_STATUS_CANCELING
 321         break
 322       elif op.status == constants.OP_STATUS_ERROR:
 323         status = constants.JOB_STATUS_ERROR
 324         # The whole job fails if one opcode failed
 325         break
 326       elif op.status == constants.OP_STATUS_CANCELED:
 327         status = constants.OP_STATUS_CANCELED
 328         break
 329
 330     if all_success:
 331       status = constants.JOB_STATUS_SUCCESS
 332
 333     return status
 334
 335   def CalcPriority(self):
 336     """Gets the current priority for this job.
 337
 338     Only unfinished opcodes are considered. When all are done, the default
 339     priority is used.
 340
 341     @rtype: int
 342
 343     """
 344     priorities = [op.priority for op in self.ops
 345                   if op.status not in constants.OPS_FINALIZED]
 346
 347     if not priorities:
 348       # All opcodes are done, assume default priority
 349       return constants.OP_PRIO_DEFAULT
 350
 351     return min(priorities)
 352
 353   def GetLogEntries(self, newer_than):
 354     """Selectively returns the log entries.
 355
 356     @type newer_than: None or int
 357     @param newer_than: if this is None, return all log entries,
 358         otherwise return only the log entries with serial higher
 359         than this value
 360     @rtype: list
 361     @return: the list of the log entries selected
 362
 363     """
 364     if newer_than is None:
 365       serial = -1
 366     else:
 367       serial = newer_than
 368
 369     entries = []
 370     for op in self.ops:
 371       entries.extend(filter(lambda entry: entry[0] > serial, op.log))
 372
 373     return entries
 374
 375   def GetInfo(self, fields):
 376     """Returns information about a job.
 377
 378     @type fields: list
 379     @param fields: names of fields to return
 380     @rtype: list
 381     @return: list with one element for each field
 382     @raise errors.OpExecError: when an invalid field
 383         has been passed
 384
 385     """
 386     row = []
 387     for fname in fields:
 388       if fname == "id":
 389         row.append(self.id)
 390       elif fname == "status":
 391         row.append(self.CalcStatus())
 392       elif fname == "priority":
 393         row.append(self.CalcPriority())
 394       elif fname == "ops":
 395         row.append([op.input.__getstate__() for op in self.ops])
 396       elif fname == "opresult":
 397         row.append([op.result for op in self.ops])
 398       elif fname == "opstatus":
 399         row.append([op.status for op in self.ops])
 400       elif fname == "oplog":
 401         row.append([op.log for op in self.ops])
 402       elif fname == "opstart":
 403         row.append([op.start_timestamp for op in self.ops])
 404       elif fname == "opexec":
 405         row.append([op.exec_timestamp for op in self.ops])
 406       elif fname == "opend":
 407         row.append([op.end_timestamp for op in self.ops])
 408       elif fname == "oppriority":
 409         row.append([op.priority for op in self.ops])
 410       elif fname == "received_ts":
 411         row.append(self.received_timestamp)
 412       elif fname == "start_ts":
 413         row.append(self.start_timestamp)
 414       elif fname == "end_ts":
 415         row.append(self.end_timestamp)
 416       elif fname == "summary":
 417         row.append([op.input.Summary() for op in self.ops])
 418       else:
 419         raise errors.OpExecError("Invalid self query field '%s'" % fname)
 420     return row
 421
 422   def MarkUnfinishedOps(self, status, result):
 423     """Mark unfinished opcodes with a given status and result.
 424
 425     This is an utility function for marking all running or waiting to
 426     be run opcodes with a given status. Opcodes which are already
 427     finalised are not changed.
 428
 429     @param status: a given opcode status
 430     @param result: the opcode result
 431
 432     """
 433     not_marked = True
 434     for op in self.ops:
 435       if op.status in constants.OPS_FINALIZED:
 436         assert not_marked, "Finalized opcodes found after non-finalized ones"
 437         continue
 438       op.status = status
 439       op.result = result
 440       not_marked = False
 441
 442   def Finalize(self):
 443     """Marks the job as finalized.
 444
 445     """
 446     self.end_timestamp = TimeStampNow()
 447
 448   def Cancel(self):
 449     """Marks job as canceled/-ing if possible.
 450
 451     @rtype: tuple; (bool, string)
 452     @return: Boolean describing whether job was successfully canceled or marked
 453       as canceling and a text message
 454
 455     """
 456     status = self.CalcStatus()
 457
 458     if status == constants.JOB_STATUS_QUEUED:
 459       self.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
 460                              "Job canceled by request")
 461       self.Finalize()
 462       return (True, "Job %s canceled" % self.id)
 463
 464     elif status == constants.JOB_STATUS_WAITING:
 465       # The worker will notice the new status and cancel the job
 466       self.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
 467       return (True, "Job %s will be canceled" % self.id)
 468
 469     else:
 470       logging.debug("Job %s is no longer waiting in the queue", self.id)
 471       return (False, "Job %s is no longer waiting in the queue" % self.id)
 472
 473
 474 class _OpExecCallbacks(mcpu.OpExecCbBase):
 475   def __init__(self, queue, job, op):
 476     """Initializes this class.
 477
 478     @type queue: L{JobQueue}
 479     @param queue: Job queue
 480     @type job: L{_QueuedJob}
 481     @param job: Job object
 482     @type op: L{_QueuedOpCode}
 483     @param op: OpCode
 484
 485     """
 486     assert queue, "Queue is missing"
 487     assert job, "Job is missing"
 488     assert op, "Opcode is missing"
 489
 490     self._queue = queue
 491     self._job = job
 492     self._op = op
 493
 494   def _CheckCancel(self):
 495     """Raises an exception to cancel the job if asked to.
 496
 497     """
 498     # Cancel here if we were asked to
 499     if self._op.status == constants.OP_STATUS_CANCELING:
 500       logging.debug("Canceling opcode")
 501       raise CancelJob()
 502
 503   @locking.ssynchronized(_QUEUE, shared=1)
 504   def NotifyStart(self):
 505     """Mark the opcode as running, not lock-waiting.
 506
 507     This is called from the mcpu code as a notifier function, when the LU is
 508     finally about to start the Exec() method. Of course, to have end-user
 509     visible results, the opcode must be initially (before calling into
 510     Processor.ExecOpCode) set to OP_STATUS_WAITING.
 511
 512     """
 513     assert self._op in self._job.ops
 514     assert self._op.status in (constants.OP_STATUS_WAITING,
 515                                constants.OP_STATUS_CANCELING)
 516
 517     # Cancel here if we were asked to
 518     self._CheckCancel()
 519
 520     logging.debug("Opcode is now running")
 521
 522     self._op.status = constants.OP_STATUS_RUNNING
 523     self._op.exec_timestamp = TimeStampNow()
 524
 525     # And finally replicate the job status
 526     self._queue.UpdateJobUnlocked(self._job)
 527
 528   @locking.ssynchronized(_QUEUE, shared=1)
 529   def _AppendFeedback(self, timestamp, log_type, log_msg):
 530     """Internal feedback append function, with locks
 531
 532     """
 533     self._job.log_serial += 1
 534     self._op.log.append((self._job.log_serial, timestamp, log_type, log_msg))
 535     self._queue.UpdateJobUnlocked(self._job, replicate=False)
 536
 537   def Feedback(self, *args):
 538     """Append a log entry.
 539
 540     """
 541     assert len(args) < 3
 542
 543     if len(args) == 1:
 544       log_type = constants.ELOG_MESSAGE
 545       log_msg = args[0]
 546     else:
 547       (log_type, log_msg) = args
 548
 549     # The time is split to make serialization easier and not lose
 550     # precision.
 551     timestamp = utils.SplitTime(time.time())
 552     self._AppendFeedback(timestamp, log_type, log_msg)
 553
 554   def CheckCancel(self):
 555     """Check whether job has been cancelled.
 556
 557     """
 558     assert self._op.status in (constants.OP_STATUS_WAITING,
 559                                constants.OP_STATUS_CANCELING)
 560
 561     # Cancel here if we were asked to
 562     self._CheckCancel()
 563
 564   def SubmitManyJobs(self, jobs):
 565     """Submits jobs for processing.
 566
 567     See L{JobQueue.SubmitManyJobs}.
 568
 569     """
 570     # Locking is done in job queue
 571     return self._queue.SubmitManyJobs(jobs)
 572
 573
 574 class _JobChangesChecker(object):
 575   def __init__(self, fields, prev_job_info, prev_log_serial):
 576     """Initializes this class.
 577
 578     @type fields: list of strings
 579     @param fields: Fields requested by LUXI client
 580     @type prev_job_info: string
 581     @param prev_job_info: previous job info, as passed by the LUXI client
 582     @type prev_log_serial: string
 583     @param prev_log_serial: previous job serial, as passed by the LUXI client
 584
 585     """
 586     self._fields = fields
 587     self._prev_job_info = prev_job_info
 588     self._prev_log_serial = prev_log_serial
 589
 590   def __call__(self, job):
 591     """Checks whether job has changed.
 592
 593     @type job: L{_QueuedJob}
 594     @param job: Job object
 595
 596     """
 597     assert not job.writable, "Expected read-only job"
 598
 599     status = job.CalcStatus()
 600     job_info = job.GetInfo(self._fields)
 601     log_entries = job.GetLogEntries(self._prev_log_serial)
 602
 603     # Serializing and deserializing data can cause type changes (e.g. from
 604     # tuple to list) or precision loss. We're doing it here so that we get
 605     # the same modifications as the data received from the client. Without
 606     # this, the comparison afterwards might fail without the data being
 607     # significantly different.
 608     # TODO: we just deserialized from disk, investigate how to make sure that
 609     # the job info and log entries are compatible to avoid this further step.
 610     # TODO: Doing something like in testutils.py:UnifyValueType might be more
 611     # efficient, though floats will be tricky
 612     job_info = serializer.LoadJson(serializer.DumpJson(job_info))
 613     log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
 614
 615     # Don't even try to wait if the job is no longer running, there will be
 616     # no changes.
 617     if (status not in (constants.JOB_STATUS_QUEUED,
 618                        constants.JOB_STATUS_RUNNING,
 619                        constants.JOB_STATUS_WAITING) or
 620         job_info != self._prev_job_info or
 621         (log_entries and self._prev_log_serial != log_entries[0][0])):
 622       logging.debug("Job %s changed", job.id)
 623       return (job_info, log_entries)
 624
 625     return None
 626
 627
 628 class _JobFileChangesWaiter(object):
 629   def __init__(self, filename):
 630     """Initializes this class.
 631
 632     @type filename: string
 633     @param filename: Path to job file
 634     @raises errors.InotifyError: if the notifier cannot be setup
 635
 636     """
 637     self._wm = pyinotify.WatchManager()
 638     self._inotify_handler = \
 639       asyncnotifier.SingleFileEventHandler(self._wm, self._OnInotify, filename)
 640     self._notifier = \
 641       pyinotify.Notifier(self._wm, default_proc_fun=self._inotify_handler)
 642     try:
 643       self._inotify_handler.enable()
 644     except Exception:
 645       # pyinotify doesn't close file descriptors automatically
 646       self._notifier.stop()
 647       raise
 648
 649   def _OnInotify(self, notifier_enabled):
 650     """Callback for inotify.
 651
 652     """
 653     if not notifier_enabled:
 654       self._inotify_handler.enable()
 655
 656   def Wait(self, timeout):
 657     """Waits for the job file to change.
 658
 659     @type timeout: float
 660     @param timeout: Timeout in seconds
 661     @return: Whether there have been events
 662
 663     """
 664     assert timeout >= 0
 665     have_events = self._notifier.check_events(timeout * 1000)
 666     if have_events:
 667       self._notifier.read_events()
 668     self._notifier.process_events()
 669     return have_events
 670
 671   def Close(self):
 672     """Closes underlying notifier and its file descriptor.
 673
 674     """
 675     self._notifier.stop()
 676
 677
 678 class _JobChangesWaiter(object):
 679   def __init__(self, filename):
 680     """Initializes this class.
 681
 682     @type filename: string
 683     @param filename: Path to job file
 684
 685     """
 686     self._filewaiter = None
 687     self._filename = filename
 688
 689   def Wait(self, timeout):
 690     """Waits for a job to change.
 691
 692     @type timeout: float
 693     @param timeout: Timeout in seconds
 694     @return: Whether there have been events
 695
 696     """
 697     if self._filewaiter:
 698       return self._filewaiter.Wait(timeout)
 699
 700     # Lazy setup: Avoid inotify setup cost when job file has already changed.
 701     # If this point is reached, return immediately and let caller check the job
 702     # file again in case there were changes since the last check. This avoids a
 703     # race condition.
 704     self._filewaiter = _JobFileChangesWaiter(self._filename)
 705
 706     return True
 707
 708   def Close(self):
 709     """Closes underlying waiter.
 710
 711     """
 712     if self._filewaiter:
 713       self._filewaiter.Close()
 714
 715
 716 class _WaitForJobChangesHelper(object):
 717   """Helper class using inotify to wait for changes in a job file.
 718
 719   This class takes a previous job status and serial, and alerts the client when
 720   the current job status has changed.
 721
 722   """
 723   @staticmethod
 724   def _CheckForChanges(counter, job_load_fn, check_fn):
 725     if counter.next() > 0:
 726       # If this isn't the first check the job is given some more time to change
 727       # again. This gives better performance for jobs generating many
 728       # changes/messages.
 729       time.sleep(0.1)
 730
 731     job = job_load_fn()
 732     if not job:
 733       raise errors.JobLost()
 734
 735     result = check_fn(job)
 736     if result is None:
 737       raise utils.RetryAgain()
 738
 739     return result
 740
 741   def __call__(self, filename, job_load_fn,
 742                fields, prev_job_info, prev_log_serial, timeout):
 743     """Waits for changes on a job.
 744
 745     @type filename: string
 746     @param filename: File on which to wait for changes
 747     @type job_load_fn: callable
 748     @param job_load_fn: Function to load job
 749     @type fields: list of strings
 750     @param fields: Which fields to check for changes
 751     @type prev_job_info: list or None
 752     @param prev_job_info: Last job information returned
 753     @type prev_log_serial: int
 754     @param prev_log_serial: Last job message serial number
 755     @type timeout: float
 756     @param timeout: maximum time to wait in seconds
 757
 758     """
 759     counter = itertools.count()
 760     try:
 761       check_fn = _JobChangesChecker(fields, prev_job_info, prev_log_serial)
 762       waiter = _JobChangesWaiter(filename)
 763       try:
 764         return utils.Retry(compat.partial(self._CheckForChanges,
 765                                           counter, job_load_fn, check_fn),
 766                            utils.RETRY_REMAINING_TIME, timeout,
 767                            wait_fn=waiter.Wait)
 768       finally:
 769         waiter.Close()
 770     except (errors.InotifyError, errors.JobLost):
 771       return None
 772     except utils.RetryTimeout:
 773       return constants.JOB_NOTCHANGED
 774
 775
 776 def _EncodeOpError(err):
 777   """Encodes an error which occurred while processing an opcode.
 778
 779   """
 780   if isinstance(err, errors.GenericError):
 781     to_encode = err
 782   else:
 783     to_encode = errors.OpExecError(str(err))
 784
 785   return errors.EncodeException(to_encode)
 786
 787
 788 class _TimeoutStrategyWrapper:
 789   def __init__(self, fn):
 790     """Initializes this class.
 791
 792     """
 793     self._fn = fn
 794     self._next = None
 795
 796   def _Advance(self):
 797     """Gets the next timeout if necessary.
 798
 799     """
 800     if self._next is None:
 801       self._next = self._fn()
 802
 803   def Peek(self):
 804     """Returns the next timeout.
 805
 806     """
 807     self._Advance()
 808     return self._next
 809
 810   def Next(self):
 811     """Returns the current timeout and advances the internal state.
 812
 813     """
 814     self._Advance()
 815     result = self._next
 816     self._next = None
 817     return result
 818
 819
 820 class _OpExecContext:
 821   def __init__(self, op, index, log_prefix, timeout_strategy_factory):
 822     """Initializes this class.
 823
 824     """
 825     self.op = op
 826     self.index = index
 827     self.log_prefix = log_prefix
 828     self.summary = op.input.Summary()
 829
 830     # Create local copy to modify
 831     if getattr(op.input, opcodes.DEPEND_ATTR, None):
 832       self.jobdeps = op.input.depends[:]
 833     else:
 834       self.jobdeps = None
 835
 836     self._timeout_strategy_factory = timeout_strategy_factory
 837     self._ResetTimeoutStrategy()
 838
 839   def _ResetTimeoutStrategy(self):
 840     """Creates a new timeout strategy.
 841
 842     """
 843     self._timeout_strategy = \
 844       _TimeoutStrategyWrapper(self._timeout_strategy_factory().NextAttempt)
 845
 846   def CheckPriorityIncrease(self):
 847     """Checks whether priority can and should be increased.
 848
 849     Called when locks couldn't be acquired.
 850
 851     """
 852     op = self.op
 853
 854     # Exhausted all retries and next round should not use blocking acquire
 855     # for locks?
 856     if (self._timeout_strategy.Peek() is None and
 857         op.priority > constants.OP_PRIO_HIGHEST):
 858       logging.debug("Increasing priority")
 859       op.priority -= 1
 860       self._ResetTimeoutStrategy()
 861       return True
 862
 863     return False
 864
 865   def GetNextLockTimeout(self):
 866     """Returns the next lock acquire timeout.
 867
 868     """
 869     return self._timeout_strategy.Next()
 870
 871
 872 class _JobProcessor(object):
 873   (DEFER,
 874    WAITDEP,
 875    FINISHED) = range(1, 4)
 876
 877   def __init__(self, queue, opexec_fn, job,
 878                _timeout_strategy_factory=mcpu.LockAttemptTimeoutStrategy):
 879     """Initializes this class.
 880
 881     """
 882     self.queue = queue
 883     self.opexec_fn = opexec_fn
 884     self.job = job
 885     self._timeout_strategy_factory = _timeout_strategy_factory
 886
 887   @staticmethod
 888   def _FindNextOpcode(job, timeout_strategy_factory):
 889     """Locates the next opcode to run.
 890
 891     @type job: L{_QueuedJob}
 892     @param job: Job object
 893     @param timeout_strategy_factory: Callable to create new timeout strategy
 894
 895     """
 896     # Create some sort of a cache to speed up locating next opcode for future
 897     # lookups
 898     # TODO: Consider splitting _QueuedJob.ops into two separate lists, one for
 899     # pending and one for processed ops.
 900     if job.ops_iter is None:
 901       job.ops_iter = enumerate(job.ops)
 902
 903     # Find next opcode to run
 904     while True:
 905       try:
 906         (idx, op) = job.ops_iter.next()
 907       except StopIteration:
 908         raise errors.ProgrammerError("Called for a finished job")
 909
 910       if op.status == constants.OP_STATUS_RUNNING:
 911         # Found an opcode already marked as running
 912         raise errors.ProgrammerError("Called for job marked as running")
 913
 914       opctx = _OpExecContext(op, idx, "Op %s/%s" % (idx + 1, len(job.ops)),
 915                              timeout_strategy_factory)
 916
 917       if op.status not in constants.OPS_FINALIZED:
 918         return opctx
 919
 920       # This is a job that was partially completed before master daemon
 921       # shutdown, so it can be expected that some opcodes are already
 922       # completed successfully (if any did error out, then the whole job
 923       # should have been aborted and not resubmitted for processing).
 924       logging.info("%s: opcode %s already processed, skipping",
 925                    opctx.log_prefix, opctx.summary)
 926
 927   @staticmethod
 928   def _MarkWaitlock(job, op):
 929     """Marks an opcode as waiting for locks.
 930
 931     The job's start timestamp is also set if necessary.
 932
 933     @type job: L{_QueuedJob}
 934     @param job: Job object
 935     @type op: L{_QueuedOpCode}
 936     @param op: Opcode object
 937
 938     """
 939     assert op in job.ops
 940     assert op.status in (constants.OP_STATUS_QUEUED,
 941                          constants.OP_STATUS_WAITING)
 942
 943     update = False
 944
 945     op.result = None
 946
 947     if op.status == constants.OP_STATUS_QUEUED:
 948       op.status = constants.OP_STATUS_WAITING
 949       update = True
 950
 951     if op.start_timestamp is None:
 952       op.start_timestamp = TimeStampNow()
 953       update = True
 954
 955     if job.start_timestamp is None:
 956       job.start_timestamp = op.start_timestamp
 957       update = True
 958
 959     assert op.status == constants.OP_STATUS_WAITING
 960
 961     return update
 962
 963   @staticmethod
 964   def _CheckDependencies(queue, job, opctx):
 965     """Checks if an opcode has dependencies and if so, processes them.
 966
 967     @type queue: L{JobQueue}
 968     @param queue: Queue object
 969     @type job: L{_QueuedJob}
 970     @param job: Job object
 971     @type opctx: L{_OpExecContext}
 972     @param opctx: Opcode execution context
 973     @rtype: bool
 974     @return: Whether opcode will be re-scheduled by dependency tracker
 975
 976     """
 977     op = opctx.op
 978
 979     result = False
 980
 981     while opctx.jobdeps:
 982       (dep_job_id, dep_status) = opctx.jobdeps[0]
 983
 984       (depresult, depmsg) = queue.depmgr.CheckAndRegister(job, dep_job_id,
 985                                                           dep_status)
 986       assert ht.TNonEmptyString(depmsg), "No dependency message"
 987
 988       logging.info("%s: %s", opctx.log_prefix, depmsg)
 989
 990       if depresult == _JobDependencyManager.CONTINUE:
 991         # Remove dependency and continue
 992         opctx.jobdeps.pop(0)
 993
 994       elif depresult == _JobDependencyManager.WAIT:
 995         # Need to wait for notification, dependency tracker will re-add job
 996         # to workerpool
 997         result = True
 998         break
 999
1000       elif depresult == _JobDependencyManager.CANCEL:
1001         # Job was cancelled, cancel this job as well
1002         job.Cancel()
1003         assert op.status == constants.OP_STATUS_CANCELING
1004         break
1005
1006       elif depresult in (_JobDependencyManager.WRONGSTATUS,
1007                          _JobDependencyManager.ERROR):
1008         # Job failed or there was an error, this job must fail
1009         op.status = constants.OP_STATUS_ERROR
1010         op.result = _EncodeOpError(errors.OpExecError(depmsg))
1011         break
1012
1013       else:
1014         raise errors.ProgrammerError("Unknown dependency result '%s'" %
1015                                      depresult)
1016
1017     return result
1018
1019   def _ExecOpCodeUnlocked(self, opctx):
1020     """Processes one opcode and returns the result.
1021
1022     """
1023     op = opctx.op
1024
1025     assert op.status == constants.OP_STATUS_WAITING
1026
1027     timeout = opctx.GetNextLockTimeout()
1028
1029     try:
1030       # Make sure not to hold queue lock while calling ExecOpCode
1031       result = self.opexec_fn(op.input,
1032                               _OpExecCallbacks(self.queue, self.job, op),
1033                               timeout=timeout, priority=op.priority)
1034     except mcpu.LockAcquireTimeout:
1035       assert timeout is not None, "Received timeout for blocking acquire"
1036       logging.debug("Couldn't acquire locks in %0.6fs", timeout)
1037
1038       assert op.status in (constants.OP_STATUS_WAITING,
1039                            constants.OP_STATUS_CANCELING)
1040
1041       # Was job cancelled while we were waiting for the lock?
1042       if op.status == constants.OP_STATUS_CANCELING:
1043         return (constants.OP_STATUS_CANCELING, None)
1044
1045       # Stay in waitlock while trying to re-acquire lock
1046       return (constants.OP_STATUS_WAITING, None)
1047     except CancelJob:
1048       logging.exception("%s: Canceling job", opctx.log_prefix)
1049       assert op.status == constants.OP_STATUS_CANCELING
1050       return (constants.OP_STATUS_CANCELING, None)
1051     except Exception, err: # pylint: disable=W0703
1052       logging.exception("%s: Caught exception in %s",
1053                         opctx.log_prefix, opctx.summary)
1054       return (constants.OP_STATUS_ERROR, _EncodeOpError(err))
1055     else:
1056       logging.debug("%s: %s successful",
1057                     opctx.log_prefix, opctx.summary)
1058       return (constants.OP_STATUS_SUCCESS, result)
1059
1060   def __call__(self, _nextop_fn=None):
1061     """Continues execution of a job.
1062
1063     @param _nextop_fn: Callback function for tests
1064     @return: C{FINISHED} if job is fully processed, C{DEFER} if the job should
1065       be deferred and C{WAITDEP} if the dependency manager
1066       (L{_JobDependencyManager}) will re-schedule the job when appropriate
1067
1068     """
1069     queue = self.queue
1070     job = self.job
1071
1072     logging.debug("Processing job %s", job.id)
1073
1074     queue.acquire(shared=1)
1075     try:
1076       opcount = len(job.ops)
1077
1078       assert job.writable, "Expected writable job"
1079
1080       # Don't do anything for finalized jobs
1081       if job.CalcStatus() in constants.JOBS_FINALIZED:
1082         return self.FINISHED
1083
1084       # Is a previous opcode still pending?
1085       if job.cur_opctx:
1086         opctx = job.cur_opctx
1087         job.cur_opctx = None
1088       else:
1089         if __debug__ and _nextop_fn:
1090           _nextop_fn()
1091         opctx = self._FindNextOpcode(job, self._timeout_strategy_factory)
1092
1093       op = opctx.op
1094
1095       # Consistency check
1096       assert compat.all(i.status in (constants.OP_STATUS_QUEUED,
1097                                      constants.OP_STATUS_CANCELING)
1098                         for i in job.ops[opctx.index + 1:])
1099
1100       assert op.status in (constants.OP_STATUS_QUEUED,
1101                            constants.OP_STATUS_WAITING,
1102                            constants.OP_STATUS_CANCELING)
1103
1104       assert (op.priority <= constants.OP_PRIO_LOWEST and
1105               op.priority >= constants.OP_PRIO_HIGHEST)
1106
1107       waitjob = None
1108
1109       if op.status != constants.OP_STATUS_CANCELING:
1110         assert op.status in (constants.OP_STATUS_QUEUED,
1111                              constants.OP_STATUS_WAITING)
1112
1113         # Prepare to start opcode
1114         if self._MarkWaitlock(job, op):
1115           # Write to disk
1116           queue.UpdateJobUnlocked(job)
1117
1118         assert op.status == constants.OP_STATUS_WAITING
1119         assert job.CalcStatus() == constants.JOB_STATUS_WAITING
1120         assert job.start_timestamp and op.start_timestamp
1121         assert waitjob is None
1122
1123         # Check if waiting for a job is necessary
1124         waitjob = self._CheckDependencies(queue, job, opctx)
1125
1126         assert op.status in (constants.OP_STATUS_WAITING,
1127                              constants.OP_STATUS_CANCELING,
1128                              constants.OP_STATUS_ERROR)
1129
1130         if not (waitjob or op.status in (constants.OP_STATUS_CANCELING,
1131                                          constants.OP_STATUS_ERROR)):
1132           logging.info("%s: opcode %s waiting for locks",
1133                        opctx.log_prefix, opctx.summary)
1134
1135           assert not opctx.jobdeps, "Not all dependencies were removed"
1136
1137           queue.release()
1138           try:
1139             (op_status, op_result) = self._ExecOpCodeUnlocked(opctx)
1140           finally:
1141             queue.acquire(shared=1)
1142
1143           op.status = op_status
1144           op.result = op_result
1145
1146           assert not waitjob
1147
1148         if op.status == constants.OP_STATUS_WAITING:
1149           # Couldn't get locks in time
1150           assert not op.end_timestamp
1151         else:
1152           # Finalize opcode
1153           op.end_timestamp = TimeStampNow()
1154
1155           if op.status == constants.OP_STATUS_CANCELING:
1156             assert not compat.any(i.status != constants.OP_STATUS_CANCELING
1157                                   for i in job.ops[opctx.index:])
1158           else:
1159             assert op.status in constants.OPS_FINALIZED
1160
1161       if op.status == constants.OP_STATUS_WAITING or waitjob:
1162         finalize = False
1163
1164         if not waitjob and opctx.CheckPriorityIncrease():
1165           # Priority was changed, need to update on-disk file
1166           queue.UpdateJobUnlocked(job)
1167
1168         # Keep around for another round
1169         job.cur_opctx = opctx
1170
1171         assert (op.priority <= constants.OP_PRIO_LOWEST and
1172                 op.priority >= constants.OP_PRIO_HIGHEST)
1173
1174         # In no case must the status be finalized here
1175         assert job.CalcStatus() == constants.JOB_STATUS_WAITING
1176
1177       else:
1178         # Ensure all opcodes so far have been successful
1179         assert (opctx.index == 0 or
1180                 compat.all(i.status == constants.OP_STATUS_SUCCESS
1181                            for i in job.ops[:opctx.index]))
1182
1183         # Reset context
1184         job.cur_opctx = None
1185
1186         if op.status == constants.OP_STATUS_SUCCESS:
1187           finalize = False
1188
1189         elif op.status == constants.OP_STATUS_ERROR:
1190           # Ensure failed opcode has an exception as its result
1191           assert errors.GetEncodedError(job.ops[opctx.index].result)
1192
1193           to_encode = errors.OpExecError("Preceding opcode failed")
1194           job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,
1195                                 _EncodeOpError(to_encode))
1196           finalize = True
1197
1198           # Consistency check
1199           assert compat.all(i.status == constants.OP_STATUS_ERROR and
1200                             errors.GetEncodedError(i.result)
1201                             for i in job.ops[opctx.index:])
1202
1203         elif op.status == constants.OP_STATUS_CANCELING:
1204           job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
1205                                 "Job canceled by request")
1206           finalize = True
1207
1208         else:
1209           raise errors.ProgrammerError("Unknown status '%s'" % op.status)
1210
1211         if opctx.index == (opcount - 1):
1212           # Finalize on last opcode
1213           finalize = True
1214
1215         if finalize:
1216           # All opcodes have been run, finalize job
1217           job.Finalize()
1218
1219         # Write to disk. If the job status is final, this is the final write
1220         # allowed. Once the file has been written, it can be archived anytime.
1221         queue.UpdateJobUnlocked(job)
1222
1223         assert not waitjob
1224
1225         if finalize:
1226           logging.info("Finished job %s, status = %s", job.id, job.CalcStatus())
1227           return self.FINISHED
1228
1229       assert not waitjob or queue.depmgr.JobWaiting(job)
1230
1231       if waitjob:
1232         return self.WAITDEP
1233       else:
1234         return self.DEFER
1235     finally:
1236       assert job.writable, "Job became read-only while being processed"
1237       queue.release()
1238
1239
1240 def _EvaluateJobProcessorResult(depmgr, job, result):
1241   """Looks at a result from L{_JobProcessor} for a job.
1242
1243   To be used in a L{_JobQueueWorker}.
1244
1245   """
1246   if result == _JobProcessor.FINISHED:
1247     # Notify waiting jobs
1248     depmgr.NotifyWaiters(job.id)
1249
1250   elif result == _JobProcessor.DEFER:
1251     # Schedule again
1252     raise workerpool.DeferTask(priority=job.CalcPriority())
1253
1254   elif result == _JobProcessor.WAITDEP:
1255     # No-op, dependency manager will re-schedule
1256     pass
1257
1258   else:
1259     raise errors.ProgrammerError("Job processor returned unknown status %s" %
1260                                  (result, ))
1261
1262
1263 class _JobQueueWorker(workerpool.BaseWorker):
1264   """The actual job workers.
1265
1266   """
1267   def RunTask(self, job): # pylint: disable=W0221
1268     """Job executor.
1269
1270     @type job: L{_QueuedJob}
1271     @param job: the job to be processed
1272
1273     """
1274     assert job.writable, "Expected writable job"
1275
1276     # Ensure only one worker is active on a single job. If a job registers for
1277     # a dependency job, and the other job notifies before the first worker is
1278     # done, the job can end up in the tasklist more than once.
1279     job.processor_lock.acquire()
1280     try:
1281       return self._RunTaskInner(job)
1282     finally:
1283       job.processor_lock.release()
1284
1285   def _RunTaskInner(self, job):
1286     """Executes a job.
1287
1288     Must be called with per-job lock acquired.
1289
1290     """
1291     queue = job.queue
1292     assert queue == self.pool.queue
1293
1294     setname_fn = lambda op: self.SetTaskName(self._GetWorkerName(job, op))
1295     setname_fn(None)
1296
1297     proc = mcpu.Processor(queue.context, job.id)
1298
1299     # Create wrapper for setting thread name
1300     wrap_execop_fn = compat.partial(self._WrapExecOpCode, setname_fn,
1301                                     proc.ExecOpCode)
1302
1303     _EvaluateJobProcessorResult(queue.depmgr, job,
1304                                 _JobProcessor(queue, wrap_execop_fn, job)())
1305
1306   @staticmethod
1307   def _WrapExecOpCode(setname_fn, execop_fn, op, *args, **kwargs):
1308     """Updates the worker thread name to include a short summary of the opcode.
1309
1310     @param setname_fn: Callable setting worker thread name
1311     @param execop_fn: Callable for executing opcode (usually
1312                       L{mcpu.Processor.ExecOpCode})
1313
1314     """
1315     setname_fn(op)
1316     try:
1317       return execop_fn(op, *args, **kwargs)
1318     finally:
1319       setname_fn(None)
1320
1321   @staticmethod
1322   def _GetWorkerName(job, op):
1323     """Sets the worker thread name.
1324
1325     @type job: L{_QueuedJob}
1326     @type op: L{opcodes.OpCode}
1327
1328     """
1329     parts = ["Job%s" % job.id]
1330
1331     if op:
1332       parts.append(op.TinySummary())
1333
1334     return "/".join(parts)
1335
1336
1337 class _JobQueueWorkerPool(workerpool.WorkerPool):
1338   """Simple class implementing a job-processing workerpool.
1339
1340   """
1341   def __init__(self, queue):
1342     super(_JobQueueWorkerPool, self).__init__("Jq",
1343                                               JOBQUEUE_THREADS,
1344                                               _JobQueueWorker)
1345     self.queue = queue
1346
1347
1348 class _JobDependencyManager:
1349   """Keeps track of job dependencies.
1350
1351   """
1352   (WAIT,
1353    ERROR,
1354    CANCEL,
1355    CONTINUE,
1356    WRONGSTATUS) = range(1, 6)
1357
1358   def __init__(self, getstatus_fn, enqueue_fn):
1359     """Initializes this class.
1360
1361     """
1362     self._getstatus_fn = getstatus_fn
1363     self._enqueue_fn = enqueue_fn
1364
1365     self._waiters = {}
1366     self._lock = locking.SharedLock("JobDepMgr")
1367
1368   @locking.ssynchronized(_LOCK, shared=1)
1369   def GetLockInfo(self, requested): # pylint: disable=W0613
1370     """Retrieves information about waiting jobs.
1371
1372     @type requested: set
1373     @param requested: Requested information, see C{query.LQ_*}
1374
1375     """
1376     # No need to sort here, that's being done by the lock manager and query
1377     # library. There are no priorities for notifying jobs, hence all show up as
1378     # one item under "pending".
1379     return [("job/%s" % job_id, None, None,
1380              [("job", [job.id for job in waiters])])
1381             for job_id, waiters in self._waiters.items()
1382             if waiters]
1383
1384   @locking.ssynchronized(_LOCK, shared=1)
1385   def JobWaiting(self, job):
1386     """Checks if a job is waiting.
1387
1388     """
1389     return compat.any(job in jobs
1390                       for jobs in self._waiters.values())
1391
1392   @locking.ssynchronized(_LOCK)
1393   def CheckAndRegister(self, job, dep_job_id, dep_status):
1394     """Checks if a dependency job has the requested status.
1395
1396     If the other job is not yet in a finalized status, the calling job will be
1397     notified (re-added to the workerpool) at a later point.
1398
1399     @type job: L{_QueuedJob}
1400     @param job: Job object
1401     @type dep_job_id: string
1402     @param dep_job_id: ID of dependency job
1403     @type dep_status: list
1404     @param dep_status: Required status
1405
1406     """
1407     assert ht.TString(job.id)
1408     assert ht.TString(dep_job_id)
1409     assert ht.TListOf(ht.TElemOf(constants.JOBS_FINALIZED))(dep_status)
1410
1411     if job.id == dep_job_id:
1412       return (self.ERROR, "Job can't depend on itself")
1413
1414     # Get status of dependency job
1415     try:
1416       status = self._getstatus_fn(dep_job_id)
1417     except errors.JobLost, err:
1418       return (self.ERROR, "Dependency error: %s" % err)
1419
1420     assert status in constants.JOB_STATUS_ALL
1421
1422     job_id_waiters = self._waiters.setdefault(dep_job_id, set())
1423
1424     if status not in constants.JOBS_FINALIZED:
1425       # Register for notification and wait for job to finish
1426       job_id_waiters.add(job)
1427       return (self.WAIT,
1428               "Need to wait for job %s, wanted status '%s'" %
1429               (dep_job_id, dep_status))
1430
1431     # Remove from waiters list
1432     if job in job_id_waiters:
1433       job_id_waiters.remove(job)
1434
1435     if (status == constants.JOB_STATUS_CANCELED and
1436         constants.JOB_STATUS_CANCELED not in dep_status):
1437       return (self.CANCEL, "Dependency job %s was cancelled" % dep_job_id)
1438
1439     elif not dep_status or status in dep_status:
1440       return (self.CONTINUE,
1441               "Dependency job %s finished with status '%s'" %
1442               (dep_job_id, status))
1443
1444     else:
1445       return (self.WRONGSTATUS,
1446               "Dependency job %s finished with status '%s',"
1447               " not one of '%s' as required" %
1448               (dep_job_id, status, utils.CommaJoin(dep_status)))
1449
1450   def _RemoveEmptyWaitersUnlocked(self):
1451     """Remove all jobs without actual waiters.
1452
1453     """
1454     for job_id in [job_id for (job_id, waiters) in self._waiters.items()
1455                    if not waiters]:
1456       del self._waiters[job_id]
1457
1458   def NotifyWaiters(self, job_id):
1459     """Notifies all jobs waiting for a certain job ID.
1460
1461     @attention: Do not call until L{CheckAndRegister} returned a status other
1462       than C{WAITDEP} for C{job_id}, or behaviour is undefined
1463     @type job_id: string
1464     @param job_id: Job ID
1465
1466     """
1467     assert ht.TString(job_id)
1468
1469     self._lock.acquire()
1470     try:
1471       self._RemoveEmptyWaitersUnlocked()
1472
1473       jobs = self._waiters.pop(job_id, None)
1474     finally:
1475       self._lock.release()
1476
1477     if jobs:
1478       # Re-add jobs to workerpool
1479       logging.debug("Re-adding %s jobs which were waiting for job %s",
1480                     len(jobs), job_id)
1481       self._enqueue_fn(jobs)
1482
1483
1484 def _RequireOpenQueue(fn):
1485   """Decorator for "public" functions.
1486
1487   This function should be used for all 'public' functions. That is,
1488   functions usually called from other classes. Note that this should
1489   be applied only to methods (not plain functions), since it expects
1490   that the decorated function is called with a first argument that has
1491   a '_queue_filelock' argument.
1492
1493   @warning: Use this decorator only after locking.ssynchronized
1494
1495   Example::
1496     @locking.ssynchronized(_LOCK)
1497     @_RequireOpenQueue
1498     def Example(self):
1499       pass
1500
1501   """
1502   def wrapper(self, *args, **kwargs):
1503     # pylint: disable=W0212
1504     assert self._queue_filelock is not None, "Queue should be open"
1505     return fn(self, *args, **kwargs)
1506   return wrapper
1507
1508
1509 class JobQueue(object):
1510   """Queue used to manage the jobs.
1511
1512   """
1513   def __init__(self, context):
1514     """Constructor for JobQueue.
1515
1516     The constructor will initialize the job queue object and then
1517     start loading the current jobs from disk, either for starting them
1518     (if they were queue) or for aborting them (if they were already
1519     running).
1520
1521     @type context: GanetiContext
1522     @param context: the context object for access to the configuration
1523         data and other ganeti objects
1524
1525     """
1526     self.context = context
1527     self._memcache = weakref.WeakValueDictionary()
1528     self._my_hostname = netutils.Hostname.GetSysName()
1529
1530     # The Big JobQueue lock. If a code block or method acquires it in shared
1531     # mode safe it must guarantee concurrency with all the code acquiring it in
1532     # shared mode, including itself. In order not to acquire it at all
1533     # concurrency must be guaranteed with all code acquiring it in shared mode
1534     # and all code acquiring it exclusively.
1535     self._lock = locking.SharedLock("JobQueue")
1536
1537     self.acquire = self._lock.acquire
1538     self.release = self._lock.release
1539
1540     # Initialize the queue, and acquire the filelock.
1541     # This ensures no other process is working on the job queue.
1542     self._queue_filelock = jstore.InitAndVerifyQueue(must_lock=True)
1543
1544     # Read serial file
1545     self._last_serial = jstore.ReadSerial()
1546     assert self._last_serial is not None, ("Serial file was modified between"
1547                                            " check in jstore and here")
1548
1549     # Get initial list of nodes
1550     self._nodes = dict((n.name, n.primary_ip)
1551                        for n in self.context.cfg.GetAllNodesInfo().values()
1552                        if n.master_candidate)
1553
1554     # Remove master node
1555     self._nodes.pop(self._my_hostname, None)
1556
1557     # TODO: Check consistency across nodes
1558
1559     self._queue_size = 0
1560     self._UpdateQueueSizeUnlocked()
1561     self._drained = jstore.CheckDrainFlag()
1562
1563     # Job dependencies
1564     self.depmgr = _JobDependencyManager(self._GetJobStatusForDependencies,
1565                                         self._EnqueueJobs)
1566     self.context.glm.AddToLockMonitor(self.depmgr)
1567
1568     # Setup worker pool
1569     self._wpool = _JobQueueWorkerPool(self)
1570     try:
1571       self._InspectQueue()
1572     except:
1573       self._wpool.TerminateWorkers()
1574       raise
1575
1576   @locking.ssynchronized(_LOCK)
1577   @_RequireOpenQueue
1578   def _InspectQueue(self):
1579     """Loads the whole job queue and resumes unfinished jobs.
1580
1581     This function needs the lock here because WorkerPool.AddTask() may start a
1582     job while we're still doing our work.
1583
1584     """
1585     logging.info("Inspecting job queue")
1586
1587     restartjobs = []
1588
1589     all_job_ids = self._GetJobIDsUnlocked()
1590     jobs_count = len(all_job_ids)
1591     lastinfo = time.time()
1592     for idx, job_id in enumerate(all_job_ids):
1593       # Give an update every 1000 jobs or 10 seconds
1594       if (idx % 1000 == 0 or time.time() >= (lastinfo + 10.0) or
1595           idx == (jobs_count - 1)):
1596         logging.info("Job queue inspection: %d/%d (%0.1f %%)",
1597                      idx, jobs_count - 1, 100.0 * (idx + 1) / jobs_count)
1598         lastinfo = time.time()
1599
1600       job = self._LoadJobUnlocked(job_id)
1601
1602       # a failure in loading the job can cause 'None' to be returned
1603       if job is None:
1604         continue
1605
1606       status = job.CalcStatus()
1607
1608       if status == constants.JOB_STATUS_QUEUED:
1609         restartjobs.append(job)
1610
1611       elif status in (constants.JOB_STATUS_RUNNING,
1612                       constants.JOB_STATUS_WAITING,
1613                       constants.JOB_STATUS_CANCELING):
1614         logging.warning("Unfinished job %s found: %s", job.id, job)
1615
1616         if status == constants.JOB_STATUS_WAITING:
1617           # Restart job
1618           job.MarkUnfinishedOps(constants.OP_STATUS_QUEUED, None)
1619           restartjobs.append(job)
1620         else:
1621           job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,
1622                                 "Unclean master daemon shutdown")
1623           job.Finalize()
1624
1625         self.UpdateJobUnlocked(job)
1626
1627     if restartjobs:
1628       logging.info("Restarting %s jobs", len(restartjobs))
1629       self._EnqueueJobsUnlocked(restartjobs)
1630
1631     logging.info("Job queue inspection finished")
1632
1633   @locking.ssynchronized(_LOCK)
1634   @_RequireOpenQueue
1635   def AddNode(self, node):
1636     """Register a new node with the queue.
1637
1638     @type node: L{objects.Node}
1639     @param node: the node object to be added
1640
1641     """
1642     node_name = node.name
1643     assert node_name != self._my_hostname
1644
1645     # Clean queue directory on added node
1646     result = rpc.RpcRunner.call_jobqueue_purge(node_name)
1647     msg = result.fail_msg
1648     if msg:
1649       logging.warning("Cannot cleanup queue directory on node %s: %s",
1650                       node_name, msg)
1651
1652     if not node.master_candidate:
1653       # remove if existing, ignoring errors
1654       self._nodes.pop(node_name, None)
1655       # and skip the replication of the job ids
1656       return
1657
1658     # Upload the whole queue excluding archived jobs
1659     files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
1660
1661     # Upload current serial file
1662     files.append(constants.JOB_QUEUE_SERIAL_FILE)
1663
1664     for file_name in files:
1665       # Read file content
1666       content = utils.ReadFile(file_name)
1667
1668       result = rpc.RpcRunner.call_jobqueue_update([node_name],
1669                                                   [node.primary_ip],
1670                                                   file_name, content)
1671       msg = result[node_name].fail_msg
1672       if msg:
1673         logging.error("Failed to upload file %s to node %s: %s",
1674                       file_name, node_name, msg)
1675
1676     self._nodes[node_name] = node.primary_ip
1677
1678   @locking.ssynchronized(_LOCK)
1679   @_RequireOpenQueue
1680   def RemoveNode(self, node_name):
1681     """Callback called when removing nodes from the cluster.
1682
1683     @type node_name: str
1684     @param node_name: the name of the node to remove
1685
1686     """
1687     self._nodes.pop(node_name, None)
1688
1689   @staticmethod
1690   def _CheckRpcResult(result, nodes, failmsg):
1691     """Verifies the status of an RPC call.
1692
1693     Since we aim to keep consistency should this node (the current
1694     master) fail, we will log errors if our rpc fail, and especially
1695     log the case when more than half of the nodes fails.
1696
1697     @param result: the data as returned from the rpc call
1698     @type nodes: list
1699     @param nodes: the list of nodes we made the call to
1700     @type failmsg: str
1701     @param failmsg: the identifier to be used for logging
1702
1703     """
1704     failed = []
1705     success = []
1706
1707     for node in nodes:
1708       msg = result[node].fail_msg
1709       if msg:
1710         failed.append(node)
1711         logging.error("RPC call %s (%s) failed on node %s: %s",
1712                       result[node].call, failmsg, node, msg)
1713       else:
1714         success.append(node)
1715
1716     # +1 for the master node
1717     if (len(success) + 1) < len(failed):
1718       # TODO: Handle failing nodes
1719       logging.error("More than half of the nodes failed")
1720
1721   def _GetNodeIp(self):
1722     """Helper for returning the node name/ip list.
1723
1724     @rtype: (list, list)
1725     @return: a tuple of two lists, the first one with the node
1726         names and the second one with the node addresses
1727
1728     """
1729     # TODO: Change to "tuple(map(list, zip(*self._nodes.items())))"?
1730     name_list = self._nodes.keys()
1731     addr_list = [self._nodes[name] for name in name_list]
1732     return name_list, addr_list
1733
1734   def _UpdateJobQueueFile(self, file_name, data, replicate):
1735     """Writes a file locally and then replicates it to all nodes.
1736
1737     This function will replace the contents of a file on the local
1738     node and then replicate it to all the other nodes we have.
1739
1740     @type file_name: str
1741     @param file_name: the path of the file to be replicated
1742     @type data: str
1743     @param data: the new contents of the file
1744     @type replicate: boolean
1745     @param replicate: whether to spread the changes to the remote nodes
1746
1747     """
1748     getents = runtime.GetEnts()
1749     utils.WriteFile(file_name, data=data, uid=getents.masterd_uid,
1750                     gid=getents.masterd_gid)
1751
1752     if replicate:
1753       names, addrs = self._GetNodeIp()
1754       result = rpc.RpcRunner.call_jobqueue_update(names, addrs, file_name, data)
1755       self._CheckRpcResult(result, self._nodes, "Updating %s" % file_name)
1756
1757   def _RenameFilesUnlocked(self, rename):
1758     """Renames a file locally and then replicate the change.
1759
1760     This function will rename a file in the local queue directory
1761     and then replicate this rename to all the other nodes we have.
1762
1763     @type rename: list of (old, new)
1764     @param rename: List containing tuples mapping old to new names
1765
1766     """
1767     # Rename them locally
1768     for old, new in rename:
1769       utils.RenameFile(old, new, mkdir=True)
1770
1771     # ... and on all nodes
1772     names, addrs = self._GetNodeIp()
1773     result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
1774     self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
1775
1776   @staticmethod
1777   def _FormatJobID(job_id):
1778     """Convert a job ID to string format.
1779
1780     Currently this just does C{str(job_id)} after performing some
1781     checks, but if we want to change the job id format this will
1782     abstract this change.
1783
1784     @type job_id: int or long
1785     @param job_id: the numeric job id
1786     @rtype: str
1787     @return: the formatted job id
1788
1789     """
1790     if not isinstance(job_id, (int, long)):
1791       raise errors.ProgrammerError("Job ID '%s' not numeric" % job_id)
1792     if job_id < 0:
1793       raise errors.ProgrammerError("Job ID %s is negative" % job_id)
1794
1795     return str(job_id)
1796
1797   @classmethod
1798   def _GetArchiveDirectory(cls, job_id):
1799     """Returns the archive directory for a job.
1800
1801     @type job_id: str
1802     @param job_id: Job identifier
1803     @rtype: str
1804     @return: Directory name
1805
1806     """
1807     return str(int(job_id) / JOBS_PER_ARCHIVE_DIRECTORY)
1808
1809   def _NewSerialsUnlocked(self, count):
1810     """Generates a new job identifier.
1811
1812     Job identifiers are unique during the lifetime of a cluster.
1813
1814     @type count: integer
1815     @param count: how many serials to return
1816     @rtype: str
1817     @return: a string representing the job identifier.
1818
1819     """
1820     assert ht.TPositiveInt(count)
1821
1822     # New number
1823     serial = self._last_serial + count
1824
1825     # Write to file
1826     self._UpdateJobQueueFile(constants.JOB_QUEUE_SERIAL_FILE,
1827                              "%s\n" % serial, True)
1828
1829     result = [self._FormatJobID(v)
1830               for v in range(self._last_serial + 1, serial + 1)]
1831
1832     # Keep it only if we were able to write the file
1833     self._last_serial = serial
1834
1835     assert len(result) == count
1836
1837     return result
1838
1839   @staticmethod
1840   def _GetJobPath(job_id):
1841     """Returns the job file for a given job id.
1842
1843     @type job_id: str
1844     @param job_id: the job identifier
1845     @rtype: str
1846     @return: the path to the job file
1847
1848     """
1849     return utils.PathJoin(constants.QUEUE_DIR, "job-%s" % job_id)
1850
1851   @classmethod
1852   def _GetArchivedJobPath(cls, job_id):
1853     """Returns the archived job file for a give job id.
1854
1855     @type job_id: str
1856     @param job_id: the job identifier
1857     @rtype: str
1858     @return: the path to the archived job file
1859
1860     """
1861     return utils.PathJoin(constants.JOB_QUEUE_ARCHIVE_DIR,
1862                           cls._GetArchiveDirectory(job_id), "job-%s" % job_id)
1863
1864   @staticmethod
1865   def _GetJobIDsUnlocked(sort=True):
1866     """Return all known job IDs.
1867
1868     The method only looks at disk because it's a requirement that all
1869     jobs are present on disk (so in the _memcache we don't have any
1870     extra IDs).
1871
1872     @type sort: boolean
1873     @param sort: perform sorting on the returned job ids
1874     @rtype: list
1875     @return: the list of job IDs
1876
1877     """
1878     jlist = []
1879     for filename in utils.ListVisibleFiles(constants.QUEUE_DIR):
1880       m = constants.JOB_FILE_RE.match(filename)
1881       if m:
1882         jlist.append(m.group(1))
1883     if sort:
1884       jlist = utils.NiceSort(jlist)
1885     return jlist
1886
1887   def _LoadJobUnlocked(self, job_id):
1888     """Loads a job from the disk or memory.
1889
1890     Given a job id, this will return the cached job object if
1891     existing, or try to load the job from the disk. If loading from
1892     disk, it will also add the job to the cache.
1893
1894     @param job_id: the job id
1895     @rtype: L{_QueuedJob} or None
1896     @return: either None or the job object
1897
1898     """
1899     job = self._memcache.get(job_id, None)
1900     if job:
1901       logging.debug("Found job %s in memcache", job_id)
1902       assert job.writable, "Found read-only job in memcache"
1903       return job
1904
1905     try:
1906       job = self._LoadJobFromDisk(job_id, False)
1907       if job is None:
1908         return job
1909     except errors.JobFileCorrupted:
1910       old_path = self._GetJobPath(job_id)
1911       new_path = self._GetArchivedJobPath(job_id)
1912       if old_path == new_path:
1913         # job already archived (future case)
1914         logging.exception("Can't parse job %s", job_id)
1915       else:
1916         # non-archived case
1917         logging.exception("Can't parse job %s, will archive.", job_id)
1918         self._RenameFilesUnlocked([(old_path, new_path)])
1919       return None
1920
1921     assert job.writable, "Job just loaded is not writable"
1922
1923     self._memcache[job_id] = job
1924     logging.debug("Added job %s to the cache", job_id)
1925     return job
1926
1927   def _LoadJobFromDisk(self, job_id, try_archived, writable=None):
1928     """Load the given job file from disk.
1929
1930     Given a job file, read, load and restore it in a _QueuedJob format.
1931
1932     @type job_id: string
1933     @param job_id: job identifier
1934     @type try_archived: bool
1935     @param try_archived: Whether to try loading an archived job
1936     @rtype: L{_QueuedJob} or None
1937     @return: either None or the job object
1938
1939     """
1940     path_functions = [(self._GetJobPath, True)]
1941
1942     if try_archived:
1943       path_functions.append((self._GetArchivedJobPath, False))
1944
1945     raw_data = None
1946     writable_default = None
1947
1948     for (fn, writable_default) in path_functions:
1949       filepath = fn(job_id)
1950       logging.debug("Loading job from %s", filepath)
1951       try:
1952         raw_data = utils.ReadFile(filepath)
1953       except EnvironmentError, err:
1954         if err.errno != errno.ENOENT:
1955           raise
1956       else:
1957         break
1958
1959     if not raw_data:
1960       return None
1961
1962     if writable is None:
1963       writable = writable_default
1964
1965     try:
1966       data = serializer.LoadJson(raw_data)
1967       job = _QueuedJob.Restore(self, data, writable)
1968     except Exception, err: # pylint: disable=W0703
1969       raise errors.JobFileCorrupted(err)
1970
1971     return job
1972
1973   def SafeLoadJobFromDisk(self, job_id, try_archived, writable=None):
1974     """Load the given job file from disk.
1975
1976     Given a job file, read, load and restore it in a _QueuedJob format.
1977     In case of error reading the job, it gets returned as None, and the
1978     exception is logged.
1979
1980     @type job_id: string
1981     @param job_id: job identifier
1982     @type try_archived: bool
1983     @param try_archived: Whether to try loading an archived job
1984     @rtype: L{_QueuedJob} or None
1985     @return: either None or the job object
1986
1987     """
1988     try:
1989       return self._LoadJobFromDisk(job_id, try_archived, writable=writable)
1990     except (errors.JobFileCorrupted, EnvironmentError):
1991       logging.exception("Can't load/parse job %s", job_id)
1992       return None
1993
1994   def _UpdateQueueSizeUnlocked(self):
1995     """Update the queue size.
1996
1997     """
1998     self._queue_size = len(self._GetJobIDsUnlocked(sort=False))
1999
2000   @locking.ssynchronized(_LOCK)
2001   @_RequireOpenQueue
2002   def SetDrainFlag(self, drain_flag):
2003     """Sets the drain flag for the queue.
2004
2005     @type drain_flag: boolean
2006     @param drain_flag: Whether to set or unset the drain flag
2007
2008     """
2009     jstore.SetDrainFlag(drain_flag)
2010
2011     self._drained = drain_flag
2012
2013     return True
2014
2015   @_RequireOpenQueue
2016   def _SubmitJobUnlocked(self, job_id, ops):
2017     """Create and store a new job.
2018
2019     This enters the job into our job queue and also puts it on the new
2020     queue, in order for it to be picked up by the queue processors.
2021
2022     @type job_id: job ID
2023     @param job_id: the job ID for the new job
2024     @type ops: list
2025     @param ops: The list of OpCodes that will become the new job.
2026     @rtype: L{_QueuedJob}
2027     @return: the job object to be queued
2028     @raise errors.JobQueueDrainError: if the job queue is marked for draining
2029     @raise errors.JobQueueFull: if the job queue has too many jobs in it
2030     @raise errors.GenericError: If an opcode is not valid
2031
2032     """
2033     # Ok when sharing the big job queue lock, as the drain file is created when
2034     # the lock is exclusive.
2035     if self._drained:
2036       raise errors.JobQueueDrainError("Job queue is drained, refusing job")
2037
2038     if self._queue_size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
2039       raise errors.JobQueueFull()
2040
2041     job = _QueuedJob(self, job_id, ops, True)
2042
2043     # Check priority
2044     for idx, op in enumerate(job.ops):
2045       if op.priority not in constants.OP_PRIO_SUBMIT_VALID:
2046         allowed = utils.CommaJoin(constants.OP_PRIO_SUBMIT_VALID)
2047         raise errors.GenericError("Opcode %s has invalid priority %s, allowed"
2048                                   " are %s" % (idx, op.priority, allowed))
2049
2050       dependencies = getattr(op.input, opcodes.DEPEND_ATTR, None)
2051       if not opcodes.TNoRelativeJobDependencies(dependencies):
2052         raise errors.GenericError("Opcode %s has invalid dependencies, must"
2053                                   " match %s: %s" %
2054                                   (idx, opcodes.TNoRelativeJobDependencies,
2055                                    dependencies))
2056
2057     # Write to disk
2058     self.UpdateJobUnlocked(job)
2059
2060     self._queue_size += 1
2061
2062     logging.debug("Adding new job %s to the cache", job_id)
2063     self._memcache[job_id] = job
2064
2065     return job
2066
2067   @locking.ssynchronized(_LOCK)
2068   @_RequireOpenQueue
2069   def SubmitJob(self, ops):
2070     """Create and store a new job.
2071
2072     @see: L{_SubmitJobUnlocked}
2073
2074     """
2075     (job_id, ) = self._NewSerialsUnlocked(1)
2076     self._EnqueueJobsUnlocked([self._SubmitJobUnlocked(job_id, ops)])
2077     return job_id
2078
2079   @locking.ssynchronized(_LOCK)
2080   @_RequireOpenQueue
2081   def SubmitManyJobs(self, jobs):
2082     """Create and store multiple jobs.
2083
2084     @see: L{_SubmitJobUnlocked}
2085
2086     """
2087     all_job_ids = self._NewSerialsUnlocked(len(jobs))
2088
2089     (results, added_jobs) = \
2090       self._SubmitManyJobsUnlocked(jobs, all_job_ids, [])
2091
2092     self._EnqueueJobsUnlocked(added_jobs)
2093
2094     return results
2095
2096   @staticmethod
2097   def _FormatSubmitError(msg, ops):
2098     """Formats errors which occurred while submitting a job.
2099
2100     """
2101     return ("%s; opcodes %s" %
2102             (msg, utils.CommaJoin(op.Summary() for op in ops)))
2103
2104   @staticmethod
2105   def _ResolveJobDependencies(resolve_fn, deps):
2106     """Resolves relative job IDs in dependencies.
2107
2108     @type resolve_fn: callable
2109     @param resolve_fn: Function to resolve a relative job ID
2110     @type deps: list
2111     @param deps: Dependencies
2112     @rtype: list
2113     @return: Resolved dependencies
2114
2115     """
2116     result = []
2117
2118     for (dep_job_id, dep_status) in deps:
2119       if ht.TRelativeJobId(dep_job_id):
2120         assert ht.TInt(dep_job_id) and dep_job_id < 0
2121         try:
2122           job_id = resolve_fn(dep_job_id)
2123         except IndexError:
2124           # Abort
2125           return (False, "Unable to resolve relative job ID %s" % dep_job_id)
2126       else:
2127         job_id = dep_job_id
2128
2129       result.append((job_id, dep_status))
2130
2131     return (True, result)
2132
2133   def _SubmitManyJobsUnlocked(self, jobs, job_ids, previous_job_ids):
2134     """Create and store multiple jobs.
2135
2136     @see: L{_SubmitJobUnlocked}
2137
2138     """
2139     results = []
2140     added_jobs = []
2141
2142     def resolve_fn(job_idx, reljobid):
2143       assert reljobid < 0
2144       return (previous_job_ids + job_ids[:job_idx])[reljobid]
2145
2146     for (idx, (job_id, ops)) in enumerate(zip(job_ids, jobs)):
2147       for op in ops:
2148         if getattr(op, opcodes.DEPEND_ATTR, None):
2149           (status, data) = \
2150             self._ResolveJobDependencies(compat.partial(resolve_fn, idx),
2151                                          op.depends)
2152           if not status:
2153             # Abort resolving dependencies
2154             assert ht.TNonEmptyString(data), "No error message"
2155             break
2156           # Use resolved dependencies
2157           op.depends = data
2158       else:
2159         try:
2160           job = self._SubmitJobUnlocked(job_id, ops)
2161         except errors.GenericError, err:
2162           status = False
2163           data = self._FormatSubmitError(str(err), ops)
2164         else:
2165           status = True
2166           data = job_id
2167           added_jobs.append(job)
2168
2169       results.append((status, data))
2170
2171     return (results, added_jobs)
2172
2173   @locking.ssynchronized(_LOCK)
2174   def _EnqueueJobs(self, jobs):
2175     """Helper function to add jobs to worker pool's queue.
2176
2177     @type jobs: list
2178     @param jobs: List of all jobs
2179
2180     """
2181     return self._EnqueueJobsUnlocked(jobs)
2182
2183   def _EnqueueJobsUnlocked(self, jobs):
2184     """Helper function to add jobs to worker pool's queue.
2185
2186     @type jobs: list
2187     @param jobs: List of all jobs
2188
2189     """
2190     assert self._lock.is_owned(shared=0), "Must own lock in exclusive mode"
2191     self._wpool.AddManyTasks([(job, ) for job in jobs],
2192                              priority=[job.CalcPriority() for job in jobs])
2193
2194   def _GetJobStatusForDependencies(self, job_id):
2195     """Gets the status of a job for dependencies.
2196
2197     @type job_id: string
2198     @param job_id: Job ID
2199     @raise errors.JobLost: If job can't be found
2200
2201     """
2202     if not isinstance(job_id, basestring):
2203       job_id = self._FormatJobID(job_id)
2204
2205     # Not using in-memory cache as doing so would require an exclusive lock
2206
2207     # Try to load from disk
2208     job = self.SafeLoadJobFromDisk(job_id, True, writable=False)
2209
2210     assert not job.writable, "Got writable job" # pylint: disable=E1101
2211
2212     if job:
2213       return job.CalcStatus()
2214
2215     raise errors.JobLost("Job %s not found" % job_id)
2216
2217   @_RequireOpenQueue
2218   def UpdateJobUnlocked(self, job, replicate=True):
2219     """Update a job's on disk storage.
2220
2221     After a job has been modified, this function needs to be called in
2222     order to write the changes to disk and replicate them to the other
2223     nodes.
2224
2225     @type job: L{_QueuedJob}
2226     @param job: the changed job
2227     @type replicate: boolean
2228     @param replicate: whether to replicate the change to remote nodes
2229
2230     """
2231     if __debug__:
2232       finalized = job.CalcStatus() in constants.JOBS_FINALIZED
2233       assert (finalized ^ (job.end_timestamp is None))
2234       assert job.writable, "Can't update read-only job"
2235
2236     filename = self._GetJobPath(job.id)
2237     data = serializer.DumpJson(job.Serialize(), indent=False)
2238     logging.debug("Writing job %s to %s", job.id, filename)
2239     self._UpdateJobQueueFile(filename, data, replicate)
2240
2241   def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial,
2242                         timeout):
2243     """Waits for changes in a job.
2244
2245     @type job_id: string
2246     @param job_id: Job identifier
2247     @type fields: list of strings
2248     @param fields: Which fields to check for changes
2249     @type prev_job_info: list or None
2250     @param prev_job_info: Last job information returned
2251     @type prev_log_serial: int
2252     @param prev_log_serial: Last job message serial number
2253     @type timeout: float
2254     @param timeout: maximum time to wait in seconds
2255     @rtype: tuple (job info, log entries)
2256     @return: a tuple of the job information as required via
2257         the fields parameter, and the log entries as a list
2258
2259         if the job has not changed and the timeout has expired,
2260         we instead return a special value,
2261         L{constants.JOB_NOTCHANGED}, which should be interpreted
2262         as such by the clients
2263
2264     """
2265     load_fn = compat.partial(self.SafeLoadJobFromDisk, job_id, False,
2266                              writable=False)
2267
2268     helper = _WaitForJobChangesHelper()
2269
2270     return helper(self._GetJobPath(job_id), load_fn,
2271                   fields, prev_job_info, prev_log_serial, timeout)
2272
2273   @locking.ssynchronized(_LOCK)
2274   @_RequireOpenQueue
2275   def CancelJob(self, job_id):
2276     """Cancels a job.
2277
2278     This will only succeed if the job has not started yet.
2279
2280     @type job_id: string
2281     @param job_id: job ID of job to be cancelled.
2282
2283     """
2284     logging.info("Cancelling job %s", job_id)
2285
2286     job = self._LoadJobUnlocked(job_id)
2287     if not job:
2288       logging.debug("Job %s not found", job_id)
2289       return (False, "Job %s not found" % job_id)
2290
2291     assert job.writable, "Can't cancel read-only job"
2292
2293     (success, msg) = job.Cancel()
2294
2295     if success:
2296       # If the job was finalized (e.g. cancelled), this is the final write
2297       # allowed. The job can be archived anytime.
2298       self.UpdateJobUnlocked(job)
2299
2300     return (success, msg)
2301
2302   @_RequireOpenQueue
2303   def _ArchiveJobsUnlocked(self, jobs):
2304     """Archives jobs.
2305
2306     @type jobs: list of L{_QueuedJob}
2307     @param jobs: Job objects
2308     @rtype: int
2309     @return: Number of archived jobs
2310
2311     """
2312     archive_jobs = []
2313     rename_files = []
2314     for job in jobs:
2315       assert job.writable, "Can't archive read-only job"
2316
2317       if job.CalcStatus() not in constants.JOBS_FINALIZED:
2318         logging.debug("Job %s is not yet done", job.id)
2319         continue
2320
2321       archive_jobs.append(job)
2322
2323       old = self._GetJobPath(job.id)
2324       new = self._GetArchivedJobPath(job.id)
2325       rename_files.append((old, new))
2326
2327     # TODO: What if 1..n files fail to rename?
2328     self._RenameFilesUnlocked(rename_files)
2329
2330     logging.debug("Successfully archived job(s) %s",
2331                   utils.CommaJoin(job.id for job in archive_jobs))
2332
2333     # Since we haven't quite checked, above, if we succeeded or failed renaming
2334     # the files, we update the cached queue size from the filesystem. When we
2335     # get around to fix the TODO: above, we can use the number of actually
2336     # archived jobs to fix this.
2337     self._UpdateQueueSizeUnlocked()
2338     return len(archive_jobs)
2339
2340   @locking.ssynchronized(_LOCK)
2341   @_RequireOpenQueue
2342   def ArchiveJob(self, job_id):
2343     """Archives a job.
2344
2345     This is just a wrapper over L{_ArchiveJobsUnlocked}.
2346
2347     @type job_id: string
2348     @param job_id: Job ID of job to be archived.
2349     @rtype: bool
2350     @return: Whether job was archived
2351
2352     """
2353     logging.info("Archiving job %s", job_id)
2354
2355     job = self._LoadJobUnlocked(job_id)
2356     if not job:
2357       logging.debug("Job %s not found", job_id)
2358       return False
2359
2360     return self._ArchiveJobsUnlocked([job]) == 1
2361
2362   @locking.ssynchronized(_LOCK)
2363   @_RequireOpenQueue
2364   def AutoArchiveJobs(self, age, timeout):
2365     """Archives all jobs based on age.
2366
2367     The method will archive all jobs which are older than the age
2368     parameter. For jobs that don't have an end timestamp, the start
2369     timestamp will be considered. The special '-1' age will cause
2370     archival of all jobs (that are not running or queued).
2371
2372     @type age: int
2373     @param age: the minimum age in seconds
2374
2375     """
2376     logging.info("Archiving jobs with age more than %s seconds", age)
2377
2378     now = time.time()
2379     end_time = now + timeout
2380     archived_count = 0
2381     last_touched = 0
2382
2383     all_job_ids = self._GetJobIDsUnlocked()
2384     pending = []
2385     for idx, job_id in enumerate(all_job_ids):
2386       last_touched = idx + 1
2387
2388       # Not optimal because jobs could be pending
2389       # TODO: Measure average duration for job archival and take number of
2390       # pending jobs into account.
2391       if time.time() > end_time:
2392         break
2393
2394       # Returns None if the job failed to load
2395       job = self._LoadJobUnlocked(job_id)
2396       if job:
2397         if job.end_timestamp is None:
2398           if job.start_timestamp is None:
2399             job_age = job.received_timestamp
2400           else:
2401             job_age = job.start_timestamp
2402         else:
2403           job_age = job.end_timestamp
2404
2405         if age == -1 or now - job_age[0] > age:
2406           pending.append(job)
2407
2408           # Archive 10 jobs at a time
2409           if len(pending) >= 10:
2410             archived_count += self._ArchiveJobsUnlocked(pending)
2411             pending = []
2412
2413     if pending:
2414       archived_count += self._ArchiveJobsUnlocked(pending)
2415
2416     return (archived_count, len(all_job_ids) - last_touched)
2417
2418   def QueryJobs(self, job_ids, fields):
2419     """Returns a list of jobs in queue.
2420
2421     @type job_ids: list
2422     @param job_ids: sequence of job identifiers or None for all
2423     @type fields: list
2424     @param fields: names of fields to return
2425     @rtype: list
2426     @return: list one element per job, each element being list with
2427         the requested fields
2428
2429     """
2430     jobs = []
2431     list_all = False
2432     if not job_ids:
2433       # Since files are added to/removed from the queue atomically, there's no
2434       # risk of getting the job ids in an inconsistent state.
2435       job_ids = self._GetJobIDsUnlocked()
2436       list_all = True
2437
2438     for job_id in job_ids:
2439       job = self.SafeLoadJobFromDisk(job_id, True)
2440       if job is not None:
2441         jobs.append(job.GetInfo(fields))
2442       elif not list_all:
2443         jobs.append(None)
2444
2445     return jobs
2446
2447   @locking.ssynchronized(_LOCK)
2448   @_RequireOpenQueue
2449   def Shutdown(self):
2450     """Stops the job queue.
2451
2452     This shutdowns all the worker threads an closes the queue.
2453
2454     """
2455     self._wpool.TerminateWorkers()
2456
2457     self._queue_filelock.Close()
2458     self._queue_filelock = None