@param job: the job to be processed
"""
- logging.debug("Worker %s processing job %s",
+ logging.info("Worker %s processing job %s",
self.worker_id, job.id)
proc = mcpu.Processor(self.pool.queue.context)
self.queue = queue = job.queue
try:
count = len(job.ops)
for idx, op in enumerate(job.ops):
+ op_summary = op.input.Summary()
try:
- logging.debug("Op %s/%s: Starting %s", idx + 1, count, op)
+ logging.info("Op %s/%s: Starting opcode %s", idx + 1, count,
+ op_summary)
queue.acquire()
try:
+ 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
finally:
queue.release()
- logging.debug("Op %s/%s: Successfully finished %s",
- idx + 1, count, op)
+ logging.info("Op %s/%s: Successfully finished opcode %s",
+ idx + 1, count, op_summary)
except CancelJob:
# Will be handled further up
raise
op.status = constants.OP_STATUS_ERROR
op.result = str(err)
op.end_timestamp = TimeStampNow()
- logging.debug("Op %s/%s: Error in %s", idx + 1, count, op)
+ logging.info("Op %s/%s: Error in opcode %s", idx + 1, count,
+ op_summary)
finally:
queue.UpdateJobUnlocked(job)
finally:
status = job.CalcStatus()
finally:
queue.release()
- logging.debug("Worker %s finished job %s, status = %s",
- self.worker_id, job_id, status)
+ logging.info("Worker %s finished job %s, status = %s",
+ self.worker_id, job_id, status)
class _JobQueueWorkerPool(workerpool.WorkerPool):
"""
try:
for op in job.ops:
- op.status = constants.OP_STATUS_ERROR
+ op.status = constants.OP_STATUS_CANCELED
op.result = "Job canceled by request"
finally:
self.UpdateJobUnlocked(job)
logging.debug("Job %s not found", job_id)
return False
- return self._ArchiveJobUnlocked([job]) == 1
+ return self._ArchiveJobsUnlocked([job]) == 1
@utils.LockedMethod
@_RequireOpenQueue