@ivar change: a Condition variable we use for waiting for job changes
"""
+ # pylint: disable-msg=W0212
__slots__ = ["queue", "id", "ops", "log_serial",
"received_timestamp", "start_timestamp", "end_timestamp",
"lock_status", "change",
"""The actual job workers.
"""
- def RunTask(self, job):
+ def RunTask(self, job): # pylint: disable-msg=W0221
"""Job executor.
This functions processes a job. It is closely tied to the _QueuedJob and
"""
logging.info("Worker %s processing job %s",
self.worker_id, job.id)
- proc = mcpu.Processor(self.pool.queue.context)
+ proc = mcpu.Processor(self.pool.queue.context, job.id)
queue = job.queue
try:
try:
self.queue = queue
-class JobQueue(object):
- """Queue used to manage the jobs.
+def _RequireOpenQueue(fn):
+ """Decorator for "public" functions.
- @cvar _RE_JOB_FILE: regex matching the valid job file names
+ This function should be used for all 'public' functions. That is,
+ functions usually called from other classes. Note that this should
+ be applied only to methods (not plain functions), since it expects
+ that the decorated function is called with a first argument that has
+ a '_queue_lock' argument.
- """
- _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
+ @warning: Use this decorator only after utils.LockedMethod!
- def _RequireOpenQueue(fn):
- """Decorator for "public" functions.
+ Example::
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def Example(self):
+ pass
- This function should be used for all 'public' functions. That is,
- functions usually called from other classes.
+ """
+ def wrapper(self, *args, **kwargs):
+ # pylint: disable-msg=W0212
+ assert self._queue_lock is not None, "Queue should be open"
+ return fn(self, *args, **kwargs)
+ return wrapper
- @warning: Use this decorator only after utils.LockedMethod!
- Example::
- @utils.LockedMethod
- @_RequireOpenQueue
- def Example(self):
- pass
+class JobQueue(object):
+ """Queue used to manage the jobs.
- """
- def wrapper(self, *args, **kwargs):
- assert self._queue_lock is not None, "Queue should be open"
- return fn(self, *args, **kwargs)
- return wrapper
+ @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.
except KeyError:
pass
- def _CheckRpcResult(self, result, nodes, failmsg):
+ @staticmethod
+ def _CheckRpcResult(result, nodes, failmsg):
"""Verifies the status of an RPC call.
Since we aim to keep consistency should this node (the current
msg = result[node].fail_msg
if msg:
failed.append(node)
- logging.error("RPC call %s failed on node %s: %s",
- result[node].call, node, msg)
+ logging.error("RPC call %s (%s) failed on node %s: %s",
+ result[node].call, failmsg, node, msg)
else:
success.append(node)
result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
- def _FormatJobID(self, job_id):
+ @staticmethod
+ def _FormatJobID(job_id):
"""Convert a job ID to string format.
Currently this just does C{str(job_id)} after performing some
@return: the list of job IDs
"""
+ # pylint: disable-msg=W0613
jlist = [self._ExtractJobID(name) for name in self._ListJobFiles()]
jlist = utils.NiceSort(jlist)
return jlist
try:
job = _QueuedJob.Restore(self, data)
- except Exception, err:
+ except Exception, err: # pylint: disable-msg=W0703
new_path = self._GetArchivedJobPath(job_id)
if filepath == new_path:
# job already archived (future case)
self._RenameFilesUnlocked(rename_files)
logging.debug("Successfully archived job(s) %s",
- ", ".join(job.id for job in archive_jobs))
+ utils.CommaJoin(job.id for job in archive_jobs))
return len(archive_jobs)
return (archived_count, len(all_job_ids) - last_touched - 1)
- def _GetJobInfoUnlocked(self, job, fields):
+ @staticmethod
+ def _GetJobInfoUnlocked(job, fields):
"""Returns information about a job.
@type job: L{_QueuedJob}