+ self._AppendFeedback(timestamp, log_type, log_msg)
+
+ def CheckCancel(self):
+ """Check whether job has been cancelled.
+
+ """
+ assert self._op.status in (constants.OP_STATUS_WAITING,
+ constants.OP_STATUS_CANCELING)
+
+ # Cancel here if we were asked to
+ self._CheckCancel()
+
+ def SubmitManyJobs(self, jobs):
+ """Submits jobs for processing.
+
+ See L{JobQueue.SubmitManyJobs}.
+
+ """
+ # Locking is done in job queue
+ return self._queue.SubmitManyJobs(jobs)
+
+
+class _JobChangesChecker(object):
+ def __init__(self, fields, prev_job_info, prev_log_serial):
+ """Initializes this class.
+
+ @type fields: list of strings
+ @param fields: Fields requested by LUXI client
+ @type prev_job_info: string
+ @param prev_job_info: previous job info, as passed by the LUXI client
+ @type prev_log_serial: string
+ @param prev_log_serial: previous job serial, as passed by the LUXI client
+
+ """
+ self._fields = fields
+ self._prev_job_info = prev_job_info
+ self._prev_log_serial = prev_log_serial
+
+ def __call__(self, job):
+ """Checks whether job has changed.
+
+ @type job: L{_QueuedJob}
+ @param job: Job object
+
+ """
+ assert not job.writable, "Expected read-only job"
+
+ status = job.CalcStatus()
+ job_info = job.GetInfo(self._fields)
+ log_entries = job.GetLogEntries(self._prev_log_serial)
+
+ # Serializing and deserializing data can cause type changes (e.g. from
+ # tuple to list) or precision loss. We're doing it here so that we get
+ # the same modifications as the data received from the client. Without
+ # this, the comparison afterwards might fail without the data being
+ # significantly different.
+ # TODO: we just deserialized from disk, investigate how to make sure that
+ # the job info and log entries are compatible to avoid this further step.
+ # TODO: Doing something like in testutils.py:UnifyValueType might be more
+ # efficient, though floats will be tricky
+ job_info = serializer.LoadJson(serializer.DumpJson(job_info))
+ log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
+
+ # Don't even try to wait if the job is no longer running, there will be
+ # no changes.
+ if (status not in (constants.JOB_STATUS_QUEUED,
+ constants.JOB_STATUS_RUNNING,
+ constants.JOB_STATUS_WAITING) or
+ job_info != self._prev_job_info or
+ (log_entries and self._prev_log_serial != log_entries[0][0])):
+ logging.debug("Job %s changed", job.id)
+ return (job_info, log_entries)
+
+ return None
+
+
+class _JobFileChangesWaiter(object):
+ def __init__(self, filename):
+ """Initializes this class.
+
+ @type filename: string
+ @param filename: Path to job file
+ @raises errors.InotifyError: if the notifier cannot be setup
+
+ """
+ self._wm = pyinotify.WatchManager()
+ self._inotify_handler = \
+ asyncnotifier.SingleFileEventHandler(self._wm, self._OnInotify, filename)
+ self._notifier = \
+ pyinotify.Notifier(self._wm, default_proc_fun=self._inotify_handler)
+ try:
+ self._inotify_handler.enable()
+ except Exception:
+ # pyinotify doesn't close file descriptors automatically
+ self._notifier.stop()
+ raise
+
+ def _OnInotify(self, notifier_enabled):
+ """Callback for inotify.
+
+ """
+ if not notifier_enabled:
+ self._inotify_handler.enable()
+
+ def Wait(self, timeout):
+ """Waits for the job file to change.
+
+ @type timeout: float
+ @param timeout: Timeout in seconds
+ @return: Whether there have been events
+
+ """
+ assert timeout >= 0
+ have_events = self._notifier.check_events(timeout * 1000)
+ if have_events:
+ self._notifier.read_events()
+ self._notifier.process_events()
+ return have_events
+
+ def Close(self):
+ """Closes underlying notifier and its file descriptor.
+
+ """
+ self._notifier.stop()
+
+
+class _JobChangesWaiter(object):
+ def __init__(self, filename):
+ """Initializes this class.
+
+ @type filename: string
+ @param filename: Path to job file
+
+ """
+ self._filewaiter = None
+ self._filename = filename
+
+ def Wait(self, timeout):
+ """Waits for a job to change.
+
+ @type timeout: float
+ @param timeout: Timeout in seconds
+ @return: Whether there have been events
+
+ """
+ if self._filewaiter:
+ return self._filewaiter.Wait(timeout)
+
+ # Lazy setup: Avoid inotify setup cost when job file has already changed.
+ # If this point is reached, return immediately and let caller check the job
+ # file again in case there were changes since the last check. This avoids a
+ # race condition.
+ self._filewaiter = _JobFileChangesWaiter(self._filename)
+
+ return True
+
+ def Close(self):
+ """Closes underlying waiter.
+
+ """
+ if self._filewaiter:
+ self._filewaiter.Close()
+
+
+class _WaitForJobChangesHelper(object):
+ """Helper class using inotify to wait for changes in a job file.
+
+ This class takes a previous job status and serial, and alerts the client when
+ the current job status has changed.
+
+ """
+ @staticmethod
+ def _CheckForChanges(counter, job_load_fn, check_fn):
+ if counter.next() > 0:
+ # If this isn't the first check the job is given some more time to change
+ # again. This gives better performance for jobs generating many
+ # changes/messages.
+ time.sleep(0.1)
+
+ job = job_load_fn()
+ if not job:
+ raise errors.JobLost()
+
+ result = check_fn(job)
+ if result is None:
+ raise utils.RetryAgain()
+
+ return result
+
+ def __call__(self, filename, job_load_fn,
+ fields, prev_job_info, prev_log_serial, timeout):
+ """Waits for changes on a job.
+
+ @type filename: string
+ @param filename: File on which to wait for changes
+ @type job_load_fn: callable
+ @param job_load_fn: Function to load job
+ @type fields: list of strings
+ @param fields: Which fields to check for changes
+ @type prev_job_info: list or None
+ @param prev_job_info: Last job information returned
+ @type prev_log_serial: int
+ @param prev_log_serial: Last job message serial number
+ @type timeout: float
+ @param timeout: maximum time to wait in seconds
+
+ """
+ counter = itertools.count()
+ try:
+ check_fn = _JobChangesChecker(fields, prev_job_info, prev_log_serial)
+ waiter = _JobChangesWaiter(filename)
+ try:
+ return utils.Retry(compat.partial(self._CheckForChanges,
+ counter, job_load_fn, check_fn),
+ utils.RETRY_REMAINING_TIME, timeout,
+ wait_fn=waiter.Wait)
+ finally:
+ waiter.Close()
+ except (errors.InotifyError, errors.JobLost):
+ return None
+ except utils.RetryTimeout:
+ return constants.JOB_NOTCHANGED
+
+
+def _EncodeOpError(err):
+ """Encodes an error which occurred while processing an opcode.
+
+ """
+ if isinstance(err, errors.GenericError):
+ to_encode = err
+ else:
+ to_encode = errors.OpExecError(str(err))
+
+ return errors.EncodeException(to_encode)
+
+
+class _TimeoutStrategyWrapper:
+ def __init__(self, fn):
+ """Initializes this class.
+
+ """
+ self._fn = fn
+ self._next = None
+
+ def _Advance(self):
+ """Gets the next timeout if necessary.
+
+ """
+ if self._next is None:
+ self._next = self._fn()
+
+ def Peek(self):
+ """Returns the next timeout.
+
+ """
+ self._Advance()
+ return self._next
+
+ def Next(self):
+ """Returns the current timeout and advances the internal state.
+
+ """
+ self._Advance()
+ result = self._next
+ self._next = None
+ return result
+
+
+class _OpExecContext:
+ def __init__(self, op, index, log_prefix, timeout_strategy_factory):
+ """Initializes this class.
+
+ """
+ self.op = op
+ self.index = index
+ self.log_prefix = log_prefix
+ self.summary = op.input.Summary()
+
+ # Create local copy to modify
+ if getattr(op.input, opcodes.DEPEND_ATTR, None):
+ self.jobdeps = op.input.depends[:]
+ else:
+ self.jobdeps = None
+
+ self._timeout_strategy_factory = timeout_strategy_factory
+ self._ResetTimeoutStrategy()
+
+ def _ResetTimeoutStrategy(self):
+ """Creates a new timeout strategy.
+
+ """
+ self._timeout_strategy = \
+ _TimeoutStrategyWrapper(self._timeout_strategy_factory().NextAttempt)
+
+ def CheckPriorityIncrease(self):
+ """Checks whether priority can and should be increased.
+
+ Called when locks couldn't be acquired.
+
+ """
+ op = self.op
+
+ # Exhausted all retries and next round should not use blocking acquire
+ # for locks?
+ if (self._timeout_strategy.Peek() is None and
+ op.priority > constants.OP_PRIO_HIGHEST):
+ logging.debug("Increasing priority")
+ op.priority -= 1
+ self._ResetTimeoutStrategy()
+ return True
+
+ return False
+
+ def GetNextLockTimeout(self):
+ """Returns the next lock acquire timeout.
+
+ """
+ return self._timeout_strategy.Next()
+
+
+class _JobProcessor(object):
+ (DEFER,
+ WAITDEP,
+ FINISHED) = range(1, 4)
+
+ def __init__(self, queue, opexec_fn, job,
+ _timeout_strategy_factory=mcpu.LockAttemptTimeoutStrategy):
+ """Initializes this class.
+
+ """
+ self.queue = queue
+ self.opexec_fn = opexec_fn
+ self.job = job
+ self._timeout_strategy_factory = _timeout_strategy_factory
+
+ @staticmethod
+ def _FindNextOpcode(job, timeout_strategy_factory):
+ """Locates the next opcode to run.
+
+ @type job: L{_QueuedJob}
+ @param job: Job object
+ @param timeout_strategy_factory: Callable to create new timeout strategy
+
+ """
+ # Create some sort of a cache to speed up locating next opcode for future
+ # lookups
+ # TODO: Consider splitting _QueuedJob.ops into two separate lists, one for
+ # pending and one for processed ops.
+ if job.ops_iter is None:
+ job.ops_iter = enumerate(job.ops)
+
+ # Find next opcode to run
+ while True:
+ try:
+ (idx, op) = job.ops_iter.next()
+ except StopIteration:
+ raise errors.ProgrammerError("Called for a finished job")
+
+ if op.status == constants.OP_STATUS_RUNNING:
+ # Found an opcode already marked as running
+ raise errors.ProgrammerError("Called for job marked as running")
+
+ opctx = _OpExecContext(op, idx, "Op %s/%s" % (idx + 1, len(job.ops)),
+ timeout_strategy_factory)
+
+ if op.status not in constants.OPS_FINALIZED:
+ return opctx
+
+ # This is a job that was partially completed before master daemon
+ # shutdown, so it can be expected that some opcodes are already
+ # completed successfully (if any did error out, then the whole job
+ # should have been aborted and not resubmitted for processing).
+ logging.info("%s: opcode %s already processed, skipping",
+ opctx.log_prefix, opctx.summary)
+
+ @staticmethod
+ def _MarkWaitlock(job, op):
+ """Marks an opcode as waiting for locks.
+
+ The job's start timestamp is also set if necessary.
+
+ @type job: L{_QueuedJob}
+ @param job: Job object
+ @type op: L{_QueuedOpCode}
+ @param op: Opcode object
+
+ """
+ assert op in job.ops
+ assert op.status in (constants.OP_STATUS_QUEUED,
+ constants.OP_STATUS_WAITING)
+
+ update = False
+
+ op.result = None
+
+ if op.status == constants.OP_STATUS_QUEUED:
+ op.status = constants.OP_STATUS_WAITING
+ update = True
+
+ if op.start_timestamp is None:
+ op.start_timestamp = TimeStampNow()
+ update = True
+
+ if job.start_timestamp is None:
+ job.start_timestamp = op.start_timestamp
+ update = True
+
+ assert op.status == constants.OP_STATUS_WAITING
+
+ return update
+
+ @staticmethod
+ def _CheckDependencies(queue, job, opctx):
+ """Checks if an opcode has dependencies and if so, processes them.
+
+ @type queue: L{JobQueue}
+ @param queue: Queue object
+ @type job: L{_QueuedJob}
+ @param job: Job object
+ @type opctx: L{_OpExecContext}
+ @param opctx: Opcode execution context
+ @rtype: bool
+ @return: Whether opcode will be re-scheduled by dependency tracker
+
+ """
+ op = opctx.op
+
+ result = False
+
+ while opctx.jobdeps:
+ (dep_job_id, dep_status) = opctx.jobdeps[0]
+
+ (depresult, depmsg) = queue.depmgr.CheckAndRegister(job, dep_job_id,
+ dep_status)
+ assert ht.TNonEmptyString(depmsg), "No dependency message"
+
+ logging.info("%s: %s", opctx.log_prefix, depmsg)
+
+ if depresult == _JobDependencyManager.CONTINUE:
+ # Remove dependency and continue
+ opctx.jobdeps.pop(0)
+
+ elif depresult == _JobDependencyManager.WAIT:
+ # Need to wait for notification, dependency tracker will re-add job
+ # to workerpool
+ result = True
+ break
+
+ elif depresult == _JobDependencyManager.CANCEL:
+ # Job was cancelled, cancel this job as well
+ job.Cancel()
+ assert op.status == constants.OP_STATUS_CANCELING
+ break
+
+ elif depresult in (_JobDependencyManager.WRONGSTATUS,
+ _JobDependencyManager.ERROR):
+ # Job failed or there was an error, this job must fail
+ op.status = constants.OP_STATUS_ERROR
+ op.result = _EncodeOpError(errors.OpExecError(depmsg))
+ break
+
+ else:
+ raise errors.ProgrammerError("Unknown dependency result '%s'" %
+ depresult)
+
+ return result
+
+ def _ExecOpCodeUnlocked(self, opctx):
+ """Processes one opcode and returns the result.
+
+ """
+ op = opctx.op
+
+ assert op.status == constants.OP_STATUS_WAITING
+
+ timeout = opctx.GetNextLockTimeout()