Synnefo » snf-ganeti

  __slots__ = ["queue", "id", "ops", "log_serial", "current_op",

    self.current_op = None

    obj.current_op = None

class _JobProcessor(object):

  def __init__(self, queue, opexec_fn, job):

    """Initializes this class.

    """

    self.queue = queue

    self.opexec_fn = opexec_fn

    self.job = job

  @staticmethod

  def _FindNextOpcode(job):

    """Locates the next opcode to run.

    @type job: L{_QueuedJob}

    @param job: Job object

    """

    # 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.current_op is None:

      job.current_op = enumerate(job.ops)

    # Find next opcode to run

    while True:

      try:

        (idx, op) = job.current_op.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")

      log_prefix = "Op %s/%s" % (idx + 1, len(job.ops))

      summary = op.input.Summary()

      if op.status == constants.OP_STATUS_CANCELED:

        # Cancelled jobs are handled by the caller

        assert not compat.any(i.status != constants.OP_STATUS_CANCELED

                              for i in job.ops[idx:])

      elif op.status in constants.OPS_FINALIZED:

        # 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",

                     log_prefix, summary)

        continue

      return (idx, op, log_prefix, 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 job: L{_QueuedOpCode}

    @param job: Opcode object

    """

    assert op in job.ops

    op.status = constants.OP_STATUS_WAITLOCK

    op.result = None

    op.start_timestamp = TimeStampNow()

    if job.start_timestamp is None:

      job.start_timestamp = op.start_timestamp

  def _ExecOpCodeUnlocked(self, log_prefix, op, summary):

    """Processes one opcode and returns the result.

    """

    assert op.status == constants.OP_STATUS_WAITLOCK

    try:

      # Make sure not to hold queue lock while calling ExecOpCode

      result = self.opexec_fn(op.input,

                              _OpExecCallbacks(self.queue, self.job, op))

    except CancelJob:

      logging.exception("%s: Canceling job", log_prefix)

      assert op.status == constants.OP_STATUS_CANCELING

      return (constants.OP_STATUS_CANCELING, None)

    except Exception, err: # pylint: disable-msg=W0703

      logging.exception("%s: Caught exception in %s", log_prefix, summary)

      return (constants.OP_STATUS_ERROR, _EncodeOpError(err))

    else:

      logging.debug("%s: %s successful", log_prefix, summary)

      return (constants.OP_STATUS_SUCCESS, result)

  def __call__(self):

    """Continues execution of a job.

    @rtype: bool

    @return: True if job is finished, False if processor needs to be called

             again

    """

    queue = self.queue

    job = self.job

    logging.debug("Processing job %s", job.id)

    queue.acquire(shared=1)

    try:

      opcount = len(job.ops)

      (opidx, op, log_prefix, op_summary) = self._FindNextOpcode(job)

      # Consistency check

      assert compat.all(i.status in (constants.OP_STATUS_QUEUED,

                                     constants.OP_STATUS_CANCELED)

                        for i in job.ops[opidx:])

      assert op.status in (constants.OP_STATUS_QUEUED,

                           constants.OP_STATUS_WAITLOCK,

                           constants.OP_STATUS_CANCELED)

      if op.status != constants.OP_STATUS_CANCELED:

        # Prepare to start opcode

        self._MarkWaitlock(job, op)

        assert op.status == constants.OP_STATUS_WAITLOCK

        assert job.CalcStatus() == constants.JOB_STATUS_WAITLOCK

        # Write to disk

        queue.UpdateJobUnlocked(job)

        logging.info("%s: opcode %s waiting for locks", log_prefix, op_summary)

        queue.release()

        try:

          (op_status, op_result) = \

            self._ExecOpCodeUnlocked(log_prefix, op, op_summary)

        finally:

          queue.acquire(shared=1)

        # Finalize opcode

        op.end_timestamp = TimeStampNow()

        op.status = op_status

        op.result = op_result

        if op.status == constants.OP_STATUS_CANCELING:

          assert not compat.any(i.status != constants.OP_STATUS_CANCELING

                                for i in job.ops[opidx:])

        else:

          assert op.status in constants.OPS_FINALIZED

      # Ensure all opcodes so far have been successful

      assert (opidx == 0 or

              compat.all(i.status == constants.OP_STATUS_SUCCESS

                         for i in job.ops[:opidx]))

      if op.status == constants.OP_STATUS_SUCCESS:

        finalize = False

      elif op.status == constants.OP_STATUS_ERROR:

        # Ensure failed opcode has an exception as its result

        assert errors.GetEncodedError(job.ops[opidx].result)

        to_encode = errors.OpExecError("Preceding opcode failed")

        job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,

                              _EncodeOpError(to_encode))

        finalize = True

        # Consistency check

        assert compat.all(i.status == constants.OP_STATUS_ERROR and

                          errors.GetEncodedError(i.result)

                          for i in job.ops[opidx:])

      elif op.status == constants.OP_STATUS_CANCELING:

        job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,

                              "Job canceled by request")

        finalize = True

      elif op.status == constants.OP_STATUS_CANCELED:

        finalize = True

      else:

        raise errors.ProgrammerError("Unknown status '%s'" % op.status)

      # Finalizing or last opcode?

      if finalize or opidx == (opcount - 1):

        # All opcodes have been run, finalize job

        job.end_timestamp = TimeStampNow()

      # Write to disk. If the job status is final, this is the final write

      # allowed. Once the file has been written, it can be archived anytime.

      queue.UpdateJobUnlocked(job)

      if finalize or opidx == (opcount - 1):

        logging.info("Finished job %s, status = %s", job.id, job.CalcStatus())

        return True

      return False

    finally:

      queue.release()

    This functions processes a job. It is closely tied to the L{_QueuedJob} and

    L{_QueuedOpCode} classes.

    queue = job.queue

    assert queue == self.pool.queue

    proc = mcpu.Processor(queue.context, job.id)

    if not _JobProcessor(queue, proc.ExecOpCode, job)():

      # Schedule again

      raise workerpool.DeferTask()