Synnefo » snf-ganeti

# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc.

import threading

from ganeti import ht

               "__weakref__", "processor_lock"]

    obj.processor_lock = threading.Lock()

    # Create local copy to modify

    if getattr(op.input, opcodes.DEPEND_ATTR, None):

      self.jobdeps = op.input.depends[:]

    else:

      self.jobdeps = None

  @staticmethod

  def _CheckDependencies(queue, job, opctx):

    """Checks if an opcode has dependencies and if so, processes them.

    @type queue: L{JobQueue}

    @param queue: Queue object

    @type job: L{_QueuedJob}

    @param job: Job object

    @type opctx: L{_OpExecContext}

    @param opctx: Opcode execution context

    @rtype: bool

    @return: Whether opcode will be re-scheduled by dependency tracker

    """

    op = opctx.op

    result = False

    while opctx.jobdeps:

      (dep_job_id, dep_status) = opctx.jobdeps[0]

      (depresult, depmsg) = queue.depmgr.CheckAndRegister(job, dep_job_id,

                                                          dep_status)

      assert ht.TNonEmptyString(depmsg), "No dependency message"

      logging.info("%s: %s", opctx.log_prefix, depmsg)

      if depresult == _JobDependencyManager.CONTINUE:

        # Remove dependency and continue

        opctx.jobdeps.pop(0)

      elif depresult == _JobDependencyManager.WAIT:

        # Need to wait for notification, dependency tracker will re-add job

        # to workerpool

        result = True

        break

      elif depresult == _JobDependencyManager.CANCEL:

        # Job was cancelled, cancel this job as well

        job.Cancel()

        assert op.status == constants.OP_STATUS_CANCELING

        break

      elif depresult in (_JobDependencyManager.WRONGSTATUS,

                         _JobDependencyManager.ERROR):

        # Job failed or there was an error, this job must fail

        op.status = constants.OP_STATUS_ERROR

        op.result = _EncodeOpError(errors.OpExecError(depmsg))

        break

      else:

        raise errors.ProgrammerError("Unknown dependency result '%s'" %

                                     depresult)

    return result

      waitjob = None

        assert waitjob is None

        # Check if waiting for a job is necessary

        waitjob = self._CheckDependencies(queue, job, opctx)

        assert op.status in (constants.OP_STATUS_WAITLOCK,

                             constants.OP_STATUS_CANCELING,

                             constants.OP_STATUS_ERROR)

        if not (waitjob or op.status in (constants.OP_STATUS_CANCELING,

                                         constants.OP_STATUS_ERROR)):

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

                       opctx.log_prefix, opctx.summary)

          assert not opctx.jobdeps, "Not all dependencies were removed"

          queue.release()

          try:

            (op_status, op_result) = self._ExecOpCodeUnlocked(opctx)

          finally:

            queue.acquire(shared=1)

          op.status = op_status

          op.result = op_result

          assert not waitjob

      if op.status == constants.OP_STATUS_WAITLOCK or waitjob:

        if not waitjob and opctx.CheckPriorityIncrease():

        assert not waitjob

          # TODO: Check locking

          queue.depmgr.NotifyWaiters(job.id)

      assert not waitjob or queue.depmgr.JobWaiting(job)

      return bool(waitjob)

    # Ensure only one worker is active on a single job. If a job registers for

    # a dependency job, and the other job notifies before the first worker is

    # done, the job can end up in the tasklist more than once.

    job.processor_lock.acquire()

    try:

      return self._RunTaskInner(job)

    finally:

      job.processor_lock.release()

  def _RunTaskInner(self, job):

    """Executes a job.

    Must be called with per-job lock acquired.

    """

class _JobDependencyManager:

  """Keeps track of job dependencies.

  """

  (WAIT,

   ERROR,

   CANCEL,

   CONTINUE,

   WRONGSTATUS) = range(1, 6)

  # TODO: Export waiter information to lock monitor

  def __init__(self, getstatus_fn, enqueue_fn):

    """Initializes this class.

    """

    self._getstatus_fn = getstatus_fn

    self._enqueue_fn = enqueue_fn

    self._waiters = {}

    self._lock = locking.SharedLock("JobDepMgr")

  @locking.ssynchronized(_LOCK, shared=1)

  def JobWaiting(self, job):

    """Checks if a job is waiting.

    """

    return compat.any(job in jobs

                      for jobs in self._waiters.values())

  @locking.ssynchronized(_LOCK)

  def CheckAndRegister(self, job, dep_job_id, dep_status):

    """Checks if a dependency job has the requested status.

    If the other job is not yet in a finalized status, the calling job will be

    notified (re-added to the workerpool) at a later point.

    @type job: L{_QueuedJob}

    @param job: Job object

    @type dep_job_id: string

    @param dep_job_id: ID of dependency job

    @type dep_status: list

    @param dep_status: Required status

    """

    assert ht.TString(job.id)

    assert ht.TString(dep_job_id)

    assert ht.TListOf(ht.TElemOf(constants.JOBS_FINALIZED))(dep_status)

    if job.id == dep_job_id:

      return (self.ERROR, "Job can't depend on itself")

    # Get status of dependency job

    try:

      status = self._getstatus_fn(dep_job_id)

    except errors.JobLost, err:

      return (self.ERROR, "Dependency error: %s" % err)

    assert status in constants.JOB_STATUS_ALL

    job_id_waiters = self._waiters.setdefault(dep_job_id, set())

    if status not in constants.JOBS_FINALIZED:

      # Register for notification and wait for job to finish

      job_id_waiters.add(job)

      return (self.WAIT,

              "Need to wait for job %s, wanted status '%s'" %

              (dep_job_id, dep_status))

    # Remove from waiters list

    if job in job_id_waiters:

      job_id_waiters.remove(job)

    if (status == constants.JOB_STATUS_CANCELED and

        constants.JOB_STATUS_CANCELED not in dep_status):

      return (self.CANCEL, "Dependency job %s was cancelled" % dep_job_id)

    elif not dep_status or status in dep_status:

      return (self.CONTINUE,

              "Dependency job %s finished with status '%s'" %

              (dep_job_id, status))

    else:

      return (self.WRONGSTATUS,

              "Dependency job %s finished with status '%s',"

              " not one of '%s' as required" %

              (dep_job_id, status, utils.CommaJoin(dep_status)))

  @locking.ssynchronized(_LOCK)

  def NotifyWaiters(self, job_id):

    """Notifies all jobs waiting for a certain job ID.

    @type job_id: string

    @param job_id: Job ID

    """

    assert ht.TString(job_id)

    jobs = self._waiters.pop(job_id, None)

    if jobs:

      # Re-add jobs to workerpool

      logging.debug("Re-adding %s jobs which were waiting for job %s",

                    len(jobs), job_id)

      self._enqueue_fn(jobs)

    # Remove all jobs without actual waiters

    for job_id in [job_id for (job_id, waiters) in self._waiters.items()

                   if not waiters]:

      del self._waiters[job_id]

    # Job dependencies

    self.depmgr = _JobDependencyManager(self._GetJobStatusForDependencies,

                                        self._EnqueueJobs)

  def _GetJobStatusForDependencies(self, job_id):

    """Gets the status of a job for dependencies.

    @type job_id: string

    @param job_id: Job ID

    @raise errors.JobLost: If job can't be found

    """

    if not isinstance(job_id, basestring):

      job_id = self._FormatJobID(job_id)

    # Not using in-memory cache as doing so would require an exclusive lock

    # Try to load from disk

    job = self.SafeLoadJobFromDisk(job_id, True)

    if job:

      return job.CalcStatus()

    raise errors.JobLost("Job %s not found" % job_id)