import logging
import errno
-import re
import time
import weakref
import threading
+import itertools
try:
- # pylint: disable-msg=E0611
+ # pylint: disable=E0611
from pyinotify import pyinotify
except ImportError:
import pyinotify
@ivar received_timestamp: the timestamp for when the job was received
@ivar start_timestmap: the timestamp for start of execution
@ivar end_timestamp: the timestamp for end of execution
+ @ivar writable: Whether the job is allowed to be modified
"""
- # pylint: disable-msg=W0212
+ # pylint: disable=W0212
__slots__ = ["queue", "id", "ops", "log_serial", "ops_iter", "cur_opctx",
"received_timestamp", "start_timestamp", "end_timestamp",
- "__weakref__", "processor_lock"]
+ "__weakref__", "processor_lock", "writable"]
- def __init__(self, queue, job_id, ops):
+ def __init__(self, queue, job_id, ops, writable):
"""Constructor for the _QueuedJob.
@type queue: L{JobQueue}
@type ops: list
@param ops: the list of opcodes we hold, which will be encapsulated
in _QueuedOpCodes
+ @type writable: bool
+ @param writable: Whether job can be modified
"""
if not ops:
self.start_timestamp = None
self.end_timestamp = None
- self._InitInMemory(self)
+ self._InitInMemory(self, writable)
@staticmethod
- def _InitInMemory(obj):
+ def _InitInMemory(obj, writable):
"""Initializes in-memory variables.
"""
+ obj.writable = writable
obj.ops_iter = None
obj.cur_opctx = None
- obj.processor_lock = threading.Lock()
+
+ # Read-only jobs are not processed and therefore don't need a lock
+ if writable:
+ obj.processor_lock = threading.Lock()
+ else:
+ obj.processor_lock = None
def __repr__(self):
status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
return "<%s at %#x>" % (" ".join(status), id(self))
@classmethod
- def Restore(cls, queue, state):
+ def Restore(cls, queue, state, writable):
"""Restore a _QueuedJob from serialized state:
@type queue: L{JobQueue}
@param queue: to which queue the restored job belongs
@type state: dict
@param state: the serialized state
+ @type writable: bool
+ @param writable: Whether job can be modified
@rtype: _JobQueue
@return: the restored _JobQueue instance
obj.log_serial = max(obj.log_serial, log_entry[0])
obj.ops.append(op)
- cls._InitInMemory(obj)
+ cls._InitInMemory(obj, writable)
return obj
if op.status == constants.OP_STATUS_QUEUED:
pass
- elif op.status == constants.OP_STATUS_WAITLOCK:
- status = constants.JOB_STATUS_WAITLOCK
+ elif op.status == constants.OP_STATUS_WAITING:
+ status = constants.JOB_STATUS_WAITING
elif op.status == constants.OP_STATUS_RUNNING:
status = constants.JOB_STATUS_RUNNING
elif op.status == constants.OP_STATUS_CANCELING:
self.Finalize()
return (True, "Job %s canceled" % self.id)
- elif status == constants.JOB_STATUS_WAITLOCK:
+ elif status == constants.JOB_STATUS_WAITING:
# The worker will notice the new status and cancel the job
self.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
return (True, "Job %s will be canceled" % self.id)
This is called from the mcpu code as a notifier function, when the LU is
finally about to start the Exec() method. Of course, to have end-user
visible results, the opcode must be initially (before calling into
- Processor.ExecOpCode) set to OP_STATUS_WAITLOCK.
+ Processor.ExecOpCode) set to OP_STATUS_WAITING.
"""
assert self._op in self._job.ops
- assert self._op.status in (constants.OP_STATUS_WAITLOCK,
+ assert self._op.status in (constants.OP_STATUS_WAITING,
constants.OP_STATUS_CANCELING)
# Cancel here if we were asked to
"""Check whether job has been cancelled.
"""
- assert self._op.status in (constants.OP_STATUS_WAITLOCK,
+ assert self._op.status in (constants.OP_STATUS_WAITING,
constants.OP_STATUS_CANCELING)
# Cancel here if we were asked to
@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)
# no changes.
if (status not in (constants.JOB_STATUS_QUEUED,
constants.JOB_STATUS_RUNNING,
- constants.JOB_STATUS_WAITLOCK) or
+ 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)
"""
@staticmethod
- def _CheckForChanges(job_load_fn, check_fn):
+ 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()
@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,
- job_load_fn, check_fn),
+ counter, job_load_fn, check_fn),
utils.RETRY_REMAINING_TIME, timeout,
wait_fn=waiter.Wait)
finally:
class _JobProcessor(object):
+ (DEFER,
+ WAITDEP,
+ FINISHED) = range(1, 4)
+
def __init__(self, queue, opexec_fn, job,
_timeout_strategy_factory=mcpu.LockAttemptTimeoutStrategy):
"""Initializes this class.
"""
assert op in job.ops
assert op.status in (constants.OP_STATUS_QUEUED,
- constants.OP_STATUS_WAITLOCK)
+ constants.OP_STATUS_WAITING)
update = False
op.result = None
if op.status == constants.OP_STATUS_QUEUED:
- op.status = constants.OP_STATUS_WAITLOCK
+ op.status = constants.OP_STATUS_WAITING
update = True
if op.start_timestamp is None:
job.start_timestamp = op.start_timestamp
update = True
- assert op.status == constants.OP_STATUS_WAITLOCK
+ assert op.status == constants.OP_STATUS_WAITING
return update
"""
op = opctx.op
- assert op.status == constants.OP_STATUS_WAITLOCK
+ assert op.status == constants.OP_STATUS_WAITING
timeout = opctx.GetNextLockTimeout()
assert timeout is not None, "Received timeout for blocking acquire"
logging.debug("Couldn't acquire locks in %0.6fs", timeout)
- assert op.status in (constants.OP_STATUS_WAITLOCK,
+ assert op.status in (constants.OP_STATUS_WAITING,
constants.OP_STATUS_CANCELING)
# Was job cancelled while we were waiting for the lock?
return (constants.OP_STATUS_CANCELING, None)
# Stay in waitlock while trying to re-acquire lock
- return (constants.OP_STATUS_WAITLOCK, None)
+ return (constants.OP_STATUS_WAITING, None)
except CancelJob:
logging.exception("%s: Canceling job", opctx.log_prefix)
assert op.status == constants.OP_STATUS_CANCELING
return (constants.OP_STATUS_CANCELING, None)
- except Exception, err: # pylint: disable-msg=W0703
+ except Exception, err: # pylint: disable=W0703
logging.exception("%s: Caught exception in %s",
opctx.log_prefix, opctx.summary)
return (constants.OP_STATUS_ERROR, _EncodeOpError(err))
"""Continues execution of a job.
@param _nextop_fn: Callback function for tests
- @rtype: bool
- @return: True if job is finished, False if processor needs to be called
- again
+ @return: C{FINISHED} if job is fully processed, C{DEFER} if the job should
+ be deferred and C{WAITDEP} if the dependency manager
+ (L{_JobDependencyManager}) will re-schedule the job when appropriate
"""
queue = self.queue
try:
opcount = len(job.ops)
+ assert job.writable, "Expected writable job"
+
# Don't do anything for finalized jobs
if job.CalcStatus() in constants.JOBS_FINALIZED:
- return True
+ return self.FINISHED
# Is a previous opcode still pending?
if job.cur_opctx:
for i in job.ops[opctx.index + 1:])
assert op.status in (constants.OP_STATUS_QUEUED,
- constants.OP_STATUS_WAITLOCK,
+ constants.OP_STATUS_WAITING,
constants.OP_STATUS_CANCELING)
assert (op.priority <= constants.OP_PRIO_LOWEST and
if op.status != constants.OP_STATUS_CANCELING:
assert op.status in (constants.OP_STATUS_QUEUED,
- constants.OP_STATUS_WAITLOCK)
+ constants.OP_STATUS_WAITING)
# Prepare to start opcode
if self._MarkWaitlock(job, op):
# Write to disk
queue.UpdateJobUnlocked(job)
- assert op.status == constants.OP_STATUS_WAITLOCK
- assert job.CalcStatus() == constants.JOB_STATUS_WAITLOCK
+ assert op.status == constants.OP_STATUS_WAITING
+ assert job.CalcStatus() == constants.JOB_STATUS_WAITING
assert job.start_timestamp and op.start_timestamp
assert waitjob is None
# Check if waiting for a job is necessary
waitjob = self._CheckDependencies(queue, job, opctx)
- assert op.status in (constants.OP_STATUS_WAITLOCK,
+ assert op.status in (constants.OP_STATUS_WAITING,
constants.OP_STATUS_CANCELING,
constants.OP_STATUS_ERROR)
assert not waitjob
- if op.status == constants.OP_STATUS_WAITLOCK:
+ if op.status == constants.OP_STATUS_WAITING:
# Couldn't get locks in time
assert not op.end_timestamp
else:
else:
assert op.status in constants.OPS_FINALIZED
- if op.status == constants.OP_STATUS_WAITLOCK or waitjob:
+ if op.status == constants.OP_STATUS_WAITING or waitjob:
finalize = False
if not waitjob and opctx.CheckPriorityIncrease():
op.priority >= constants.OP_PRIO_HIGHEST)
# In no case must the status be finalized here
- assert job.CalcStatus() == constants.JOB_STATUS_WAITLOCK
+ assert job.CalcStatus() == constants.JOB_STATUS_WAITING
else:
# Ensure all opcodes so far have been successful
if finalize:
logging.info("Finished job %s, status = %s", job.id, job.CalcStatus())
- # TODO: Check locking
- queue.depmgr.NotifyWaiters(job.id)
- return True
+ return self.FINISHED
assert not waitjob or queue.depmgr.JobWaiting(job)
- return bool(waitjob)
+ if waitjob:
+ return self.WAITDEP
+ else:
+ return self.DEFER
finally:
+ assert job.writable, "Job became read-only while being processed"
queue.release()
"""The actual job workers.
"""
- def RunTask(self, job): # pylint: disable-msg=W0221
+ def RunTask(self, job): # pylint: disable=W0221
"""Job executor.
@type job: L{_QueuedJob}
@param job: the job to be processed
"""
+ assert job.writable, "Expected writable job"
+
# Ensure only one worker is active on a single job. If a job registers for
# a dependency job, and the other job notifies before the first worker is
# done, the job can end up in the tasklist more than once.
wrap_execop_fn = compat.partial(self._WrapExecOpCode, setname_fn,
proc.ExecOpCode)
- if not _JobProcessor(queue, wrap_execop_fn, job)():
+ result = _JobProcessor(queue, wrap_execop_fn, job)()
+
+ if result == _JobProcessor.FINISHED:
+ # Notify waiting jobs
+ queue.depmgr.NotifyWaiters(job.id)
+
+ elif result == _JobProcessor.DEFER:
# Schedule again
raise workerpool.DeferTask(priority=job.CalcPriority())
+ elif result == _JobProcessor.WAITDEP:
+ # No-op, dependency manager will re-schedule
+ pass
+
+ else:
+ raise errors.ProgrammerError("Job processor returned unknown status %s" %
+ (result, ))
+
@staticmethod
def _WrapExecOpCode(setname_fn, execop_fn, op, *args, **kwargs):
"""Updates the worker thread name to include a short summary of the opcode.
CONTINUE,
WRONGSTATUS) = range(1, 6)
- # TODO: Export waiter information to lock monitor
-
def __init__(self, getstatus_fn, enqueue_fn):
"""Initializes this class.
self._lock = locking.SharedLock("JobDepMgr")
@locking.ssynchronized(_LOCK, shared=1)
+ def GetLockInfo(self, requested): # pylint: disable=W0613
+ """Retrieves information about waiting jobs.
+
+ @type requested: set
+ @param requested: Requested information, see C{query.LQ_*}
+
+ """
+ # No need to sort here, that's being done by the lock manager and query
+ # library. There are no priorities for notifying jobs, hence all show up as
+ # one item under "pending".
+ return [("job/%s" % job_id, None, None,
+ [("job", [job.id for job in waiters])])
+ for job_id, waiters in self._waiters.items()
+ if waiters]
+
+ @locking.ssynchronized(_LOCK, shared=1)
def JobWaiting(self, job):
"""Checks if a job is waiting.
"""
def wrapper(self, *args, **kwargs):
- # pylint: disable-msg=W0212
+ # pylint: disable=W0212
assert self._queue_filelock is not None, "Queue should be open"
return fn(self, *args, **kwargs)
return wrapper
class JobQueue(object):
"""Queue used to manage the jobs.
- @cvar _RE_JOB_FILE: regex matching the valid job file names
-
"""
- _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
-
def __init__(self, context):
"""Constructor for JobQueue.
# Job dependencies
self.depmgr = _JobDependencyManager(self._GetJobStatusForDependencies,
self._EnqueueJobs)
+ self.context.glm.AddToLockMonitor(self.depmgr)
# Setup worker pool
self._wpool = _JobQueueWorkerPool(self)
restartjobs.append(job)
elif status in (constants.JOB_STATUS_RUNNING,
- constants.JOB_STATUS_WAITLOCK,
+ constants.JOB_STATUS_WAITING,
constants.JOB_STATUS_CANCELING):
logging.warning("Unfinished job %s found: %s", job.id, job)
- if status == constants.JOB_STATUS_WAITLOCK:
+ if status == constants.JOB_STATUS_WAITING:
# Restart job
job.MarkUnfinishedOps(constants.OP_STATUS_QUEUED, None)
restartjobs.append(job)
if restartjobs:
logging.info("Restarting %s jobs", len(restartjobs))
- self._EnqueueJobs(restartjobs)
+ self._EnqueueJobsUnlocked(restartjobs)
logging.info("Job queue inspection finished")
@return: a string representing the job identifier.
"""
- assert count > 0
+ assert ht.TPositiveInt(count)
+
# New number
serial = self._last_serial + count
return utils.PathJoin(constants.JOB_QUEUE_ARCHIVE_DIR,
cls._GetArchiveDirectory(job_id), "job-%s" % job_id)
- def _GetJobIDsUnlocked(self, sort=True):
+ @staticmethod
+ def _GetJobIDsUnlocked(sort=True):
"""Return all known job IDs.
The method only looks at disk because it's a requirement that all
"""
jlist = []
for filename in utils.ListVisibleFiles(constants.QUEUE_DIR):
- m = self._RE_JOB_FILE.match(filename)
+ m = constants.JOB_FILE_RE.match(filename)
if m:
jlist.append(m.group(1))
if sort:
job = self._memcache.get(job_id, None)
if job:
logging.debug("Found job %s in memcache", job_id)
+ assert job.writable, "Found read-only job in memcache"
return job
try:
self._RenameFilesUnlocked([(old_path, new_path)])
return None
+ assert job.writable, "Job just loaded is not writable"
+
self._memcache[job_id] = job
logging.debug("Added job %s to the cache", job_id)
return job
- def _LoadJobFromDisk(self, job_id, try_archived):
+ def _LoadJobFromDisk(self, job_id, try_archived, writable=None):
"""Load the given job file from disk.
Given a job file, read, load and restore it in a _QueuedJob format.
@return: either None or the job object
"""
- path_functions = [self._GetJobPath]
+ path_functions = [(self._GetJobPath, True)]
if try_archived:
- path_functions.append(self._GetArchivedJobPath)
+ path_functions.append((self._GetArchivedJobPath, False))
raw_data = None
+ writable_default = None
- for fn in path_functions:
+ for (fn, writable_default) in path_functions:
filepath = fn(job_id)
logging.debug("Loading job from %s", filepath)
try:
if not raw_data:
return None
+ if writable is None:
+ writable = writable_default
+
try:
data = serializer.LoadJson(raw_data)
- job = _QueuedJob.Restore(self, data)
- except Exception, err: # pylint: disable-msg=W0703
+ job = _QueuedJob.Restore(self, data, writable)
+ except Exception, err: # pylint: disable=W0703
raise errors.JobFileCorrupted(err)
return job
- def SafeLoadJobFromDisk(self, job_id, try_archived):
+ def SafeLoadJobFromDisk(self, job_id, try_archived, writable=None):
"""Load the given job file from disk.
Given a job file, read, load and restore it in a _QueuedJob format.
"""
try:
- return self._LoadJobFromDisk(job_id, try_archived)
+ return self._LoadJobFromDisk(job_id, try_archived, writable=writable)
except (errors.JobFileCorrupted, EnvironmentError):
logging.exception("Can't load/parse job %s", job_id)
return None
if self._queue_size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
raise errors.JobQueueFull()
- job = _QueuedJob(self, job_id, ops)
+ job = _QueuedJob(self, job_id, ops, True)
# Check priority
for idx, op in enumerate(job.ops):
raise errors.GenericError("Opcode %s has invalid priority %s, allowed"
" are %s" % (idx, op.priority, allowed))
+ dependencies = getattr(op.input, opcodes.DEPEND_ATTR, None)
+ if not opcodes.TNoRelativeJobDependencies(dependencies):
+ raise errors.GenericError("Opcode %s has invalid dependencies, must"
+ " match %s: %s" %
+ (idx, opcodes.TNoRelativeJobDependencies,
+ dependencies))
+
# Write to disk
self.UpdateJobUnlocked(job)
@see: L{_SubmitJobUnlocked}
"""
- job_id = self._NewSerialsUnlocked(1)[0]
- self._EnqueueJobs([self._SubmitJobUnlocked(job_id, ops)])
+ (job_id, ) = self._NewSerialsUnlocked(1)
+ self._EnqueueJobsUnlocked([self._SubmitJobUnlocked(job_id, ops)])
return job_id
@locking.ssynchronized(_LOCK)
@see: L{_SubmitJobUnlocked}
"""
- results = []
- added_jobs = []
all_job_ids = self._NewSerialsUnlocked(len(jobs))
- for job_id, ops in zip(all_job_ids, jobs):
- try:
- added_jobs.append(self._SubmitJobUnlocked(job_id, ops))
- status = True
- data = job_id
- except errors.GenericError, err:
- data = ("%s; opcodes %s" %
- (err, utils.CommaJoin(op.Summary() for op in ops)))
- status = False
- results.append((status, data))
- self._EnqueueJobs(added_jobs)
+ (results, added_jobs) = \
+ self._SubmitManyJobsUnlocked(jobs, all_job_ids, [])
+
+ self._EnqueueJobsUnlocked(added_jobs)
return results
+ @staticmethod
+ def _FormatSubmitError(msg, ops):
+ """Formats errors which occurred while submitting a job.
+
+ """
+ return ("%s; opcodes %s" %
+ (msg, utils.CommaJoin(op.Summary() for op in ops)))
+
+ @staticmethod
+ def _ResolveJobDependencies(resolve_fn, deps):
+ """Resolves relative job IDs in dependencies.
+
+ @type resolve_fn: callable
+ @param resolve_fn: Function to resolve a relative job ID
+ @type deps: list
+ @param deps: Dependencies
+ @rtype: list
+ @return: Resolved dependencies
+
+ """
+ result = []
+
+ for (dep_job_id, dep_status) in deps:
+ if ht.TRelativeJobId(dep_job_id):
+ assert ht.TInt(dep_job_id) and dep_job_id < 0
+ try:
+ job_id = resolve_fn(dep_job_id)
+ except IndexError:
+ # Abort
+ return (False, "Unable to resolve relative job ID %s" % dep_job_id)
+ else:
+ job_id = dep_job_id
+
+ result.append((job_id, dep_status))
+
+ return (True, result)
+
+ def _SubmitManyJobsUnlocked(self, jobs, job_ids, previous_job_ids):
+ """Create and store multiple jobs.
+
+ @see: L{_SubmitJobUnlocked}
+
+ """
+ results = []
+ added_jobs = []
+
+ def resolve_fn(job_idx, reljobid):
+ assert reljobid < 0
+ return (previous_job_ids + job_ids[:job_idx])[reljobid]
+
+ for (idx, (job_id, ops)) in enumerate(zip(job_ids, jobs)):
+ for op in ops:
+ if getattr(op, opcodes.DEPEND_ATTR, None):
+ (status, data) = \
+ self._ResolveJobDependencies(compat.partial(resolve_fn, idx),
+ op.depends)
+ if not status:
+ # Abort resolving dependencies
+ assert ht.TNonEmptyString(data), "No error message"
+ break
+ # Use resolved dependencies
+ op.depends = data
+ else:
+ try:
+ job = self._SubmitJobUnlocked(job_id, ops)
+ except errors.GenericError, err:
+ status = False
+ data = self._FormatSubmitError(str(err), ops)
+ else:
+ status = True
+ data = job_id
+ added_jobs.append(job)
+
+ results.append((status, data))
+
+ return (results, added_jobs)
+
+ @locking.ssynchronized(_LOCK)
def _EnqueueJobs(self, jobs):
"""Helper function to add jobs to worker pool's queue.
@param jobs: List of all jobs
"""
+ return self._EnqueueJobsUnlocked(jobs)
+
+ def _EnqueueJobsUnlocked(self, jobs):
+ """Helper function to add jobs to worker pool's queue.
+
+ @type jobs: list
+ @param jobs: List of all jobs
+
+ """
+ assert self._lock.is_owned(shared=0), "Must own lock in exclusive mode"
self._wpool.AddManyTasks([(job, ) for job in jobs],
priority=[job.CalcPriority() for job in jobs])
# Not using in-memory cache as doing so would require an exclusive lock
# Try to load from disk
- job = self.SafeLoadJobFromDisk(job_id, True)
+ job = self.SafeLoadJobFromDisk(job_id, True, writable=False)
+
+ assert not job.writable, "Got writable job" # pylint: disable=E1101
if job:
return job.CalcStatus()
if __debug__:
finalized = job.CalcStatus() in constants.JOBS_FINALIZED
assert (finalized ^ (job.end_timestamp is None))
+ assert job.writable, "Can't update read-only job"
filename = self._GetJobPath(job.id)
data = serializer.DumpJson(job.Serialize(), indent=False)
as such by the clients
"""
- load_fn = compat.partial(self.SafeLoadJobFromDisk, job_id, False)
+ load_fn = compat.partial(self.SafeLoadJobFromDisk, job_id, False,
+ writable=False)
helper = _WaitForJobChangesHelper()
logging.debug("Job %s not found", job_id)
return (False, "Job %s not found" % job_id)
+ assert job.writable, "Can't cancel read-only job"
+
(success, msg) = job.Cancel()
if success:
archive_jobs = []
rename_files = []
for job in jobs:
+ assert job.writable, "Can't archive read-only job"
+
if job.CalcStatus() not in constants.JOBS_FINALIZED:
logging.debug("Job %s is not yet done", job.id)
continue