@ivar id: the job ID
@type ops: list
@ivar ops: the list of _QueuedOpCode that constitute the job
- @type run_op_index: int
- @ivar run_op_index: the currently executing opcode, or -1 if
- we didn't yet start executing
@type log_serial: int
@ivar log_serial: holds the index for the next log entry
@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 lock_status: In-memory locking information for debugging
@ivar change: a Condition variable we use for waiting for job changes
"""
- __slots__ = ["queue", "id", "ops", "run_op_index", "log_serial",
+ __slots__ = ["queue", "id", "ops", "log_serial",
"received_timestamp", "start_timestamp", "end_timestamp",
- "change",
+ "lock_status", "change",
"__weakref__"]
def __init__(self, queue, job_id, ops):
self.queue = queue
self.id = job_id
self.ops = [_QueuedOpCode(op) for op in ops]
- self.run_op_index = -1
self.log_serial = 0
self.received_timestamp = TimeStampNow()
self.start_timestamp = None
self.end_timestamp = None
+ # In-memory attributes
+ self.lock_status = None
+
# Condition to wait for changes
self.change = threading.Condition(self.queue._lock)
obj = _QueuedJob.__new__(cls)
obj.queue = queue
obj.id = state["id"]
- obj.run_op_index = state["run_op_index"]
obj.received_timestamp = state.get("received_timestamp", None)
obj.start_timestamp = state.get("start_timestamp", None)
obj.end_timestamp = state.get("end_timestamp", None)
+ # In-memory attributes
+ obj.lock_status = None
+
obj.ops = []
obj.log_serial = 0
for op_state in state["ops"]:
return {
"id": self.id,
"ops": [op.Serialize() for op in self.ops],
- "run_op_index": self.run_op_index,
"start_timestamp": self.start_timestamp,
"end_timestamp": self.end_timestamp,
"received_timestamp": self.received_timestamp,
not_marked = False
-class _OpCodeExecCallbacks(mcpu.OpExecCbBase):
+class _OpExecCallbacks(mcpu.OpExecCbBase):
def __init__(self, queue, job, op):
"""Initializes this class.
assert self._op.status in (constants.OP_STATUS_WAITLOCK,
constants.OP_STATUS_CANCELING)
+ # All locks are acquired by now
+ self._job.lock_status = None
+
# Cancel here if we were asked to
if self._op.status == constants.OP_STATUS_CANCELING:
raise CancelJob()
finally:
self._queue.release()
+ def ReportLocks(self, msg):
+ """Write locking information to the job.
+
+ Called whenever the LU processor is waiting for a lock or has acquired one.
+
+ """
+ # Not getting the queue lock because this is a single assignment
+ self._job.lock_status = msg
+
class _JobQueueWorker(workerpool.BaseWorker):
"""The actual job workers.
if op.status == constants.OP_STATUS_CANCELED:
raise CancelJob()
assert op.status == constants.OP_STATUS_QUEUED
- job.run_op_index = idx
op.status = constants.OP_STATUS_WAITLOCK
op.result = None
op.start_timestamp = TimeStampNow()
# Make sure not to hold queue lock while calling ExecOpCode
result = proc.ExecOpCode(input_opcode,
- _OpCodeExecCallbacks(queue, job, op))
+ _OpExecCallbacks(queue, job, op))
queue.acquire()
try:
queue.acquire()
try:
try:
- job.run_op_index = -1
+ job.lock_status = None
job.end_timestamp = TimeStampNow()
queue.UpdateJobUnlocked(job)
finally:
status = job.CalcStatus()
finally:
queue.release()
+
logging.info("Worker %s finished job %s, status = %s",
self.worker_id, job_id, status)
# Clean queue directory on added node
result = rpc.RpcRunner.call_jobqueue_purge(node_name)
- msg = result.RemoteFailMsg()
+ msg = result.fail_msg
if msg:
logging.warning("Cannot cleanup queue directory on node %s: %s",
node_name, msg)
result = rpc.RpcRunner.call_jobqueue_update([node_name],
[node.primary_ip],
file_name, content)
- msg = result[node_name].RemoteFailMsg()
+ msg = result[node_name].fail_msg
if msg:
logging.error("Failed to upload file %s to node %s: %s",
file_name, node_name, msg)
success = []
for node in nodes:
- msg = result[node].RemoteFailMsg()
+ msg = result[node].fail_msg
if msg:
failed.append(node)
logging.error("RPC call %s failed on node %s: %s",
return results
-
@_RequireOpenQueue
def UpdateJobUnlocked(self, job):
"""Update a job's on disk storage.
row.append(job.start_timestamp)
elif fname == "end_ts":
row.append(job.end_timestamp)
+ elif fname == "lock_status":
+ row.append(job.lock_status)
elif fname == "summary":
row.append([op.input.Summary() for op in job.ops])
else: