Synnefo » snf-ganeti

  __slots__ = ["queue", "id", "ops", "log_serial", "ops_iter", "cur_opctx",

    obj.cur_opctx = None

class _TimeoutStrategyWrapper:

  def __init__(self, fn):

    """Initializes this class.

    """

    self._fn = fn

    self._next = None

  def _Advance(self):

    """Gets the next timeout if necessary.

    """

    if self._next is None:

      self._next = self._fn()

  def Peek(self):

    """Returns the next timeout.

    """

    self._Advance()

    return self._next

  def Next(self):

    """Returns the current timeout and advances the internal state.

    """

    self._Advance()

    result = self._next

    self._next = None

    return result

  def __init__(self, op, index, log_prefix, timeout_strategy_factory):

    self._timeout_strategy_factory = timeout_strategy_factory

    self._ResetTimeoutStrategy()

  def _ResetTimeoutStrategy(self):

    """Creates a new timeout strategy.

    """

    self._timeout_strategy = \

      _TimeoutStrategyWrapper(self._timeout_strategy_factory().NextAttempt)

  def CheckPriorityIncrease(self):

    """Checks whether priority can and should be increased.

    Called when locks couldn't be acquired.

    """

    op = self.op

    # Exhausted all retries and next round should not use blocking acquire

    # for locks?

    if (self._timeout_strategy.Peek() is None and

        op.priority > constants.OP_PRIO_HIGHEST):

      logging.debug("Increasing priority")

      op.priority -= 1

      self._ResetTimeoutStrategy()

      return True

    return False

  def GetNextLockTimeout(self):

    """Returns the next lock acquire timeout.

    """

    return self._timeout_strategy.Next()

  def __init__(self, queue, opexec_fn, job,

               _timeout_strategy_factory=mcpu.LockAttemptTimeoutStrategy):

    self._timeout_strategy_factory = _timeout_strategy_factory

  def _FindNextOpcode(job, timeout_strategy_factory):

    @param timeout_strategy_factory: Callable to create new timeout strategy

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

                             timeout_strategy_factory)

    timeout = opctx.GetNextLockTimeout()

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

                              timeout=timeout)

    except mcpu.LockAcquireTimeout:

      assert timeout is not None, "Received timeout for blocking acquire"

      logging.debug("Couldn't acquire locks in %0.6fs", timeout)

      assert op.status == constants.OP_STATUS_WAITLOCK

      return (constants.OP_STATUS_QUEUED, None)

  def __call__(self, _nextop_fn=None):

    @param _nextop_fn: Callback function for tests

      # Is a previous opcode still pending?

      if job.cur_opctx:

        opctx = job.cur_opctx

      else:

        if __debug__ and _nextop_fn:

          _nextop_fn()

        opctx = self._FindNextOpcode(job, self._timeout_strategy_factory)

      assert (op.priority <= constants.OP_PRIO_LOWEST and

              op.priority >= constants.OP_PRIO_HIGHEST)

        if op.status == constants.OP_STATUS_QUEUED:

          # Couldn't get locks in time

          assert not op.end_timestamp

          # Finalize opcode

          op.end_timestamp = TimeStampNow()

          if op.status == constants.OP_STATUS_CANCELING:

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

                                  for i in job.ops[opctx.index:])

          else:

            assert op.status in constants.OPS_FINALIZED

      if op.status == constants.OP_STATUS_QUEUED:

        opctx.CheckPriorityIncrease()

        # Keep around for another round

        job.cur_opctx = opctx

        assert (op.priority <= constants.OP_PRIO_LOWEST and

                op.priority >= constants.OP_PRIO_HIGHEST)

        # In no case must the status be finalized here

        assert job.CalcStatus() == constants.JOB_STATUS_QUEUED

        queue.UpdateJobUnlocked(job)

        # Ensure all opcodes so far have been successful

        assert (opctx.index == 0 or

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

                           for i in job.ops[:opctx.index]))

        # Reset context

        job.cur_opctx = None

        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[opctx.index].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[opctx.index:])

        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 opctx.index == (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 opctx.index == (opcount - 1):

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

          return True

      raise workerpool.DeferTask(priority=job.CalcPriority())