+class _JobDependencyManager:
+ """Keeps track of job dependencies.
+
+ """
+ (WAIT,
+ ERROR,
+ CANCEL,
+ CONTINUE,
+ WRONGSTATUS) = range(1, 6)
+
+ 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 GetLockInfo(self, requested): # pylint: disable=W0613
+ """Retrieves information about waiting jobs.
+
+ @type requested: set
+ @param requested: Requested information, see C{query.LQ_*}
+
+ """
+ # No need to sort here, that's being done by the lock manager and query
+ # library. There are no priorities for notifying jobs, hence all show up as
+ # one item under "pending".
+ return [("job/%s" % job_id, None, None,
+ [("job", [job.id for job in waiters])])
+ for job_id, waiters in self._waiters.items()
+ if waiters]
+
+ @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: int
+ @param dep_job_id: ID of dependency job
+ @type dep_status: list
+ @param dep_status: Required status
+
+ """
+ assert ht.TJobId(job.id)
+ assert ht.TJobId(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)))
+
+ def _RemoveEmptyWaitersUnlocked(self):
+ """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]
+
+ def NotifyWaiters(self, job_id):
+ """Notifies all jobs waiting for a certain job ID.
+
+ @attention: Do not call until L{CheckAndRegister} returned a status other
+ than C{WAITDEP} for C{job_id}, or behaviour is undefined
+ @type job_id: int
+ @param job_id: Job ID
+
+ """
+ assert ht.TJobId(job_id)
+
+ self._lock.acquire()
+ try:
+ self._RemoveEmptyWaitersUnlocked()
+
+ jobs = self._waiters.pop(job_id, None)
+ finally:
+ self._lock.release()
+
+ 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)
+
+
projects / ganeti-local / blobdiff