import logging
import errno
-import re
import time
import weakref
import threading
import itertools
try:
- # pylint: disable-msg=E0611
+ # pylint: disable=E0611
from pyinotify import pyinotify
except ImportError:
import pyinotify
@ivar writable: Whether the job is allowed to be modified
"""
- # pylint: disable-msg=W0212
+ # pylint: disable=W0212
__slots__ = ["queue", "id", "ops", "log_serial", "ops_iter", "cur_opctx",
"received_timestamp", "start_timestamp", "end_timestamp",
"__weakref__", "processor_lock", "writable"]
logging.exception("%s: Canceling job", opctx.log_prefix)
assert op.status == constants.OP_STATUS_CANCELING
return (constants.OP_STATUS_CANCELING, None)
- except Exception, err: # pylint: disable-msg=W0703
+ except Exception, err: # pylint: disable=W0703
logging.exception("%s: Caught exception in %s",
opctx.log_prefix, opctx.summary)
return (constants.OP_STATUS_ERROR, _EncodeOpError(err))
queue.release()
+def _EvaluateJobProcessorResult(depmgr, job, result):
+ """Looks at a result from L{_JobProcessor} for a job.
+
+ To be used in a L{_JobQueueWorker}.
+
+ """
+ if result == _JobProcessor.FINISHED:
+ # Notify waiting jobs
+ depmgr.NotifyWaiters(job.id)
+
+ elif result == _JobProcessor.DEFER:
+ # Schedule again
+ raise workerpool.DeferTask(priority=job.CalcPriority())
+
+ elif result == _JobProcessor.WAITDEP:
+ # No-op, dependency manager will re-schedule
+ pass
+
+ else:
+ raise errors.ProgrammerError("Job processor returned unknown status %s" %
+ (result, ))
+
+
class _JobQueueWorker(workerpool.BaseWorker):
"""The actual job workers.
"""
- def RunTask(self, job): # pylint: disable-msg=W0221
+ def RunTask(self, job): # pylint: disable=W0221
"""Job executor.
@type job: L{_QueuedJob}
wrap_execop_fn = compat.partial(self._WrapExecOpCode, setname_fn,
proc.ExecOpCode)
- result = _JobProcessor(queue, wrap_execop_fn, job)()
-
- if result == _JobProcessor.FINISHED:
- # Notify waiting jobs
- queue.depmgr.NotifyWaiters(job.id)
-
- elif result == _JobProcessor.DEFER:
- # Schedule again
- raise workerpool.DeferTask(priority=job.CalcPriority())
-
- elif result == _JobProcessor.WAITDEP:
- # No-op, dependency manager will re-schedule
- pass
-
- else:
- raise errors.ProgrammerError("Job processor returned unknown status %s" %
- (result, ))
+ _EvaluateJobProcessorResult(queue.depmgr, job,
+ _JobProcessor(queue, wrap_execop_fn, job)())
@staticmethod
def _WrapExecOpCode(setname_fn, execop_fn, op, *args, **kwargs):
self._lock = locking.SharedLock("JobDepMgr")
@locking.ssynchronized(_LOCK, shared=1)
- def GetLockInfo(self, requested): # pylint: disable-msg=W0613
+ def GetLockInfo(self, requested): # pylint: disable=W0613
"""Retrieves information about waiting jobs.
@type requested: set
" not one of '%s' as required" %
(dep_job_id, status, utils.CommaJoin(dep_status)))
- @locking.ssynchronized(_LOCK)
+ 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: string
@param job_id: Job ID
"""
assert ht.TString(job_id)
- jobs = self._waiters.pop(job_id, None)
+ 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)
- # 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 _RequireOpenQueue(fn):
"""Decorator for "public" functions.
"""
def wrapper(self, *args, **kwargs):
- # pylint: disable-msg=W0212
+ # pylint: disable=W0212
assert self._queue_filelock is not None, "Queue should be open"
return fn(self, *args, **kwargs)
return wrapper
-class JobQueue(object):
- """Queue used to manage the jobs.
+def _RequireNonDrainedQueue(fn):
+ """Decorator checking for a non-drained queue.
- @cvar _RE_JOB_FILE: regex matching the valid job file names
+ To be used with functions submitting new jobs.
"""
- _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
+ def wrapper(self, *args, **kwargs):
+ """Wrapper function.
+ @raise errors.JobQueueDrainError: if the job queue is marked for draining
+
+ """
+ # Ok when sharing the big job queue lock, as the drain file is created when
+ # the lock is exclusive.
+ # Needs access to protected member, pylint: disable=W0212
+ if self._drained:
+ raise errors.JobQueueDrainError("Job queue is drained, refusing job")
+
+ if not self._accepting_jobs:
+ raise errors.JobQueueError("Job queue is shutting down, refusing job")
+
+ return fn(self, *args, **kwargs)
+ return wrapper
+
+
+class JobQueue(object):
+ """Queue used to manage the jobs.
+
+ """
def __init__(self, context):
"""Constructor for JobQueue.
self.acquire = self._lock.acquire
self.release = self._lock.release
+ # Accept jobs by default
+ self._accepting_jobs = True
+
# Initialize the queue, and acquire the filelock.
# This ensures no other process is working on the job queue.
self._queue_filelock = jstore.InitAndVerifyQueue(must_lock=True)
# TODO: Check consistency across nodes
- self._queue_size = 0
+ self._queue_size = None
self._UpdateQueueSizeUnlocked()
+ assert ht.TInt(self._queue_size)
self._drained = jstore.CheckDrainFlag()
# Job dependencies
logging.info("Job queue inspection finished")
+ def _GetRpc(self, address_list):
+ """Gets RPC runner with context.
+
+ """
+ return rpc.JobQueueRunner(self.context, address_list)
+
@locking.ssynchronized(_LOCK)
@_RequireOpenQueue
def AddNode(self, node):
assert node_name != self._my_hostname
# Clean queue directory on added node
- result = rpc.RpcRunner.call_jobqueue_purge(node_name)
+ result = self._GetRpc(None).call_jobqueue_purge(node_name)
msg = result.fail_msg
if msg:
logging.warning("Cannot cleanup queue directory on node %s: %s",
# Upload current serial file
files.append(constants.JOB_QUEUE_SERIAL_FILE)
+ # Static address list
+ addrs = [node.primary_ip]
+
for file_name in files:
# Read file content
content = utils.ReadFile(file_name)
- result = rpc.RpcRunner.call_jobqueue_update([node_name],
- [node.primary_ip],
- file_name, content)
+ result = self._GetRpc(addrs).call_jobqueue_update([node_name], file_name,
+ content)
msg = result[node_name].fail_msg
if msg:
logging.error("Failed to upload file %s to node %s: %s",
if replicate:
names, addrs = self._GetNodeIp()
- result = rpc.RpcRunner.call_jobqueue_update(names, addrs, file_name, data)
+ result = self._GetRpc(addrs).call_jobqueue_update(names, file_name, data)
self._CheckRpcResult(result, self._nodes, "Updating %s" % file_name)
def _RenameFilesUnlocked(self, rename):
# ... and on all nodes
names, addrs = self._GetNodeIp()
- result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
+ result = self._GetRpc(addrs).call_jobqueue_rename(names, rename)
self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
@staticmethod
@return: a string representing the job identifier.
"""
- assert count > 0
+ assert ht.TPositiveInt(count)
+
# New number
serial = self._last_serial + count
return utils.PathJoin(constants.JOB_QUEUE_ARCHIVE_DIR,
cls._GetArchiveDirectory(job_id), "job-%s" % job_id)
- def _GetJobIDsUnlocked(self, sort=True):
+ @staticmethod
+ def _GetJobIDsUnlocked(sort=True):
"""Return all known job IDs.
The method only looks at disk because it's a requirement that all
"""
jlist = []
for filename in utils.ListVisibleFiles(constants.QUEUE_DIR):
- m = self._RE_JOB_FILE.match(filename)
+ m = constants.JOB_FILE_RE.match(filename)
if m:
jlist.append(m.group(1))
if sort:
try:
data = serializer.LoadJson(raw_data)
job = _QueuedJob.Restore(self, data, writable)
- except Exception, err: # pylint: disable-msg=W0703
+ except Exception, err: # pylint: disable=W0703
raise errors.JobFileCorrupted(err)
return job
@param ops: The list of OpCodes that will become the new job.
@rtype: L{_QueuedJob}
@return: the job object to be queued
- @raise errors.JobQueueDrainError: if the job queue is marked for draining
@raise errors.JobQueueFull: if the job queue has too many jobs in it
@raise errors.GenericError: If an opcode is not valid
"""
- # Ok when sharing the big job queue lock, as the drain file is created when
- # the lock is exclusive.
- if self._drained:
- raise errors.JobQueueDrainError("Job queue is drained, refusing job")
-
if self._queue_size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
raise errors.JobQueueFull()
@locking.ssynchronized(_LOCK)
@_RequireOpenQueue
+ @_RequireNonDrainedQueue
def SubmitJob(self, ops):
"""Create and store a new job.
@locking.ssynchronized(_LOCK)
@_RequireOpenQueue
+ @_RequireNonDrainedQueue
def SubmitManyJobs(self, jobs):
"""Create and store multiple jobs.
# Try to load from disk
job = self.SafeLoadJobFromDisk(job_id, True, writable=False)
- assert not job.writable, "Got writable job"
+ assert not job.writable, "Got writable job" # pylint: disable=E1101
if job:
return job.CalcStatus()
assert job.writable, "Can't update read-only job"
filename = self._GetJobPath(job.id)
- data = serializer.DumpJson(job.Serialize(), indent=False)
+ data = serializer.DumpJson(job.Serialize())
logging.debug("Writing job %s to %s", job.id, filename)
self._UpdateJobQueueFile(filename, data, replicate)
return jobs
@locking.ssynchronized(_LOCK)
+ def PrepareShutdown(self):
+ """Prepare to stop the job queue.
+
+ Disables execution of jobs in the workerpool and returns whether there are
+ any jobs currently running. If the latter is the case, the job queue is not
+ yet ready for shutdown. Once this function returns C{True} L{Shutdown} can
+ be called without interfering with any job. Queued and unfinished jobs will
+ be resumed next time.
+
+ Once this function has been called no new job submissions will be accepted
+ (see L{_RequireNonDrainedQueue}).
+
+ @rtype: bool
+ @return: Whether there are any running jobs
+
+ """
+ if self._accepting_jobs:
+ self._accepting_jobs = False
+
+ # Tell worker pool to stop processing pending tasks
+ self._wpool.SetActive(False)
+
+ return self._wpool.HasRunningTasks()
+
+ @locking.ssynchronized(_LOCK)
@_RequireOpenQueue
def Shutdown(self):
"""Stops the job queue.
projects / ganeti-local / blobdiff