Synnefo » snf-ganeti

  @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 change: a Condition variable we use for waiting for job changes

  """

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

               "received_timestamp", "start_timestamp", "end_timestamp",

               "__weakref__"]

  def __init__(self, queue, job_id, ops):

    self.start_timestamp = None

    self.end_timestamp = None

    # Condition to wait for changes

    self.change = threading.Condition(self.queue._lock)

    obj.start_timestamp = state.get("start_timestamp", None)

    obj.end_timestamp = state.get("end_timestamp", None)

    obj.ops = []

    obj.log_serial = 0

    for op_state in state["ops"]:

      not_marked = False

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

    """Initializes this class.

      assert self._op.status in (constants.OP_STATUS_WAITLOCK,

                                 constants.OP_STATUS_CANCELING)

      # Cancel here if we were asked to

      if self._op.status == constants.OP_STATUS_CANCELING:

        raise CancelJob()

    finally:

      self._queue.release()

class _JobQueueWorker(workerpool.BaseWorker):

  """The actual job workers.

            # Make sure not to hold queue lock while calling ExecOpCode

            result = proc.ExecOpCode(input_opcode,

            queue.acquire()

            try:

      queue.acquire()

      try:

        try:

          job.run_op_index = -1

          job.end_timestamp = TimeStampNow()

          queue.UpdateJobUnlocked(job)

          status = job.CalcStatus()

      finally:

        queue.release()

      logging.info("Worker %s finished job %s, status = %s",

                   self.worker_id, job_id, status)

    return results

  @_RequireOpenQueue

  def UpdateJobUnlocked(self, job):

    """Update a job's on disk storage.