#!/usr/bin/python # # Copyright (C) 2010, 2011, 2012 Google Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU # General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. """Script for testing ganeti.jqueue""" import os import sys import unittest import tempfile import shutil import errno import itertools import random import operator try: # pylint: disable=E0611 from pyinotify import pyinotify except ImportError: import pyinotify from ganeti import constants from ganeti import utils from ganeti import errors from ganeti import jqueue from ganeti import opcodes from ganeti import compat from ganeti import mcpu from ganeti import query from ganeti import workerpool import testutils class _FakeJob: def __init__(self, job_id, status): self.id = job_id self.writable = False self._status = status self._log = [] def SetStatus(self, status): self._status = status def AddLogEntry(self, msg): self._log.append((len(self._log), msg)) def CalcStatus(self): return self._status def GetInfo(self, fields): result = [] for name in fields: if name == "status": result.append(self._status) else: raise Exception("Unknown field") return result def GetLogEntries(self, newer_than): assert newer_than is None or newer_than >= 0 if newer_than is None: return self._log return self._log[newer_than:] class TestJobChangesChecker(unittest.TestCase): def testStatus(self): job = _FakeJob(9094, constants.JOB_STATUS_QUEUED) checker = jqueue._JobChangesChecker(["status"], None, None) self.assertEqual(checker(job), ([constants.JOB_STATUS_QUEUED], [])) job.SetStatus(constants.JOB_STATUS_RUNNING) self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], [])) job.SetStatus(constants.JOB_STATUS_SUCCESS) self.assertEqual(checker(job), ([constants.JOB_STATUS_SUCCESS], [])) # job.id is used by checker self.assertEqual(job.id, 9094) def testStatusWithPrev(self): job = _FakeJob(12807, constants.JOB_STATUS_QUEUED) checker = jqueue._JobChangesChecker(["status"], [constants.JOB_STATUS_QUEUED], None) self.assert_(checker(job) is None) job.SetStatus(constants.JOB_STATUS_RUNNING) self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], [])) def testFinalStatus(self): for status in constants.JOBS_FINALIZED: job = _FakeJob(2178711, status) checker = jqueue._JobChangesChecker(["status"], [status], None) # There won't be any changes in this status, hence it should signal # a change immediately self.assertEqual(checker(job), ([status], [])) def testLog(self): job = _FakeJob(9094, constants.JOB_STATUS_RUNNING) checker = jqueue._JobChangesChecker(["status"], None, None) self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], [])) job.AddLogEntry("Hello World") (job_info, log_entries) = checker(job) self.assertEqual(job_info, [constants.JOB_STATUS_RUNNING]) self.assertEqual(log_entries, [[0, "Hello World"]]) checker2 = jqueue._JobChangesChecker(["status"], job_info, len(log_entries)) self.assert_(checker2(job) is None) job.AddLogEntry("Foo Bar") job.SetStatus(constants.JOB_STATUS_ERROR) (job_info, log_entries) = checker2(job) self.assertEqual(job_info, [constants.JOB_STATUS_ERROR]) self.assertEqual(log_entries, [[1, "Foo Bar"]]) checker3 = jqueue._JobChangesChecker(["status"], None, None) (job_info, log_entries) = checker3(job) self.assertEqual(job_info, [constants.JOB_STATUS_ERROR]) self.assertEqual(log_entries, [[0, "Hello World"], [1, "Foo Bar"]]) class TestJobChangesWaiter(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp() self.filename = utils.PathJoin(self.tmpdir, "job-1") utils.WriteFile(self.filename, data="") def tearDown(self): shutil.rmtree(self.tmpdir) def _EnsureNotifierClosed(self, notifier): try: os.fstat(notifier._fd) except EnvironmentError, err: self.assertEqual(err.errno, errno.EBADF) else: self.fail("File descriptor wasn't closed") def testClose(self): for wait in [False, True]: waiter = jqueue._JobFileChangesWaiter(self.filename) try: if wait: waiter.Wait(0.001) finally: waiter.Close() # Ensure file descriptor was closed self._EnsureNotifierClosed(waiter._notifier) def testChangingFile(self): waiter = jqueue._JobFileChangesWaiter(self.filename) try: self.assertFalse(waiter.Wait(0.1)) utils.WriteFile(self.filename, data="changed") self.assert_(waiter.Wait(60)) finally: waiter.Close() self._EnsureNotifierClosed(waiter._notifier) def testChangingFile2(self): waiter = jqueue._JobChangesWaiter(self.filename) try: self.assertFalse(waiter._filewaiter) self.assert_(waiter.Wait(0.1)) self.assert_(waiter._filewaiter) # File waiter is now used, but there have been no changes self.assertFalse(waiter.Wait(0.1)) utils.WriteFile(self.filename, data="changed") self.assert_(waiter.Wait(60)) finally: waiter.Close() self._EnsureNotifierClosed(waiter._filewaiter._notifier) class _FailingWatchManager(pyinotify.WatchManager): """Subclass of L{pyinotify.WatchManager} which always fails to register. """ def add_watch(self, filename, mask): assert mask == (pyinotify.EventsCodes.ALL_FLAGS["IN_MODIFY"] | pyinotify.EventsCodes.ALL_FLAGS["IN_IGNORED"]) return { filename: -1, } class TestWaitForJobChangesHelper(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp() self.filename = utils.PathJoin(self.tmpdir, "job-2614226563") utils.WriteFile(self.filename, data="") def tearDown(self): shutil.rmtree(self.tmpdir) def _LoadWaitingJob(self): return _FakeJob(2614226563, constants.JOB_STATUS_WAITING) def _LoadLostJob(self): return None def testNoChanges(self): wfjc = jqueue._WaitForJobChangesHelper() # No change self.assertEqual(wfjc(self.filename, self._LoadWaitingJob, ["status"], [constants.JOB_STATUS_WAITING], None, 0.1), constants.JOB_NOTCHANGED) # No previous information self.assertEqual(wfjc(self.filename, self._LoadWaitingJob, ["status"], None, None, 1.0), ([constants.JOB_STATUS_WAITING], [])) def testLostJob(self): wfjc = jqueue._WaitForJobChangesHelper() self.assert_(wfjc(self.filename, self._LoadLostJob, ["status"], None, None, 1.0) is None) def testNonExistentFile(self): wfjc = jqueue._WaitForJobChangesHelper() filename = utils.PathJoin(self.tmpdir, "does-not-exist") self.assertFalse(os.path.exists(filename)) result = wfjc(filename, self._LoadLostJob, ["status"], None, None, 1.0, _waiter_cls=compat.partial(jqueue._JobChangesWaiter, _waiter_cls=NotImplemented)) self.assertTrue(result is None) def testInotifyError(self): jobfile_waiter_cls = \ compat.partial(jqueue._JobFileChangesWaiter, _inotify_wm_cls=_FailingWatchManager) jobchange_waiter_cls = \ compat.partial(jqueue._JobChangesWaiter, _waiter_cls=jobfile_waiter_cls) wfjc = jqueue._WaitForJobChangesHelper() # Test if failing to watch a job file (e.g. due to # fs.inotify.max_user_watches being too low) raises errors.InotifyError self.assertRaises(errors.InotifyError, wfjc, self.filename, self._LoadWaitingJob, ["status"], [constants.JOB_STATUS_WAITING], None, 1.0, _waiter_cls=jobchange_waiter_cls) class TestEncodeOpError(unittest.TestCase): def test(self): encerr = jqueue._EncodeOpError(errors.LockError("Test 1")) self.assert_(isinstance(encerr, tuple)) self.assertRaises(errors.LockError, errors.MaybeRaise, encerr) encerr = jqueue._EncodeOpError(errors.GenericError("Test 2")) self.assert_(isinstance(encerr, tuple)) self.assertRaises(errors.GenericError, errors.MaybeRaise, encerr) encerr = jqueue._EncodeOpError(NotImplementedError("Foo")) self.assert_(isinstance(encerr, tuple)) self.assertRaises(errors.OpExecError, errors.MaybeRaise, encerr) encerr = jqueue._EncodeOpError("Hello World") self.assert_(isinstance(encerr, tuple)) self.assertRaises(errors.OpExecError, errors.MaybeRaise, encerr) class TestQueuedOpCode(unittest.TestCase): def testDefaults(self): def _Check(op): self.assertFalse(hasattr(op.input, "dry_run")) self.assertEqual(op.priority, constants.OP_PRIO_DEFAULT) self.assertFalse(op.log) self.assert_(op.start_timestamp is None) self.assert_(op.exec_timestamp is None) self.assert_(op.end_timestamp is None) self.assert_(op.result is None) self.assertEqual(op.status, constants.OP_STATUS_QUEUED) op1 = jqueue._QueuedOpCode(opcodes.OpTestDelay()) _Check(op1) op2 = jqueue._QueuedOpCode.Restore(op1.Serialize()) _Check(op2) self.assertEqual(op1.Serialize(), op2.Serialize()) def testPriority(self): def _Check(op): assert constants.OP_PRIO_DEFAULT != constants.OP_PRIO_HIGH, \ "Default priority equals high priority; test can't work" self.assertEqual(op.priority, constants.OP_PRIO_HIGH) self.assertEqual(op.status, constants.OP_STATUS_QUEUED) inpop = opcodes.OpTagsGet(priority=constants.OP_PRIO_HIGH) op1 = jqueue._QueuedOpCode(inpop) _Check(op1) op2 = jqueue._QueuedOpCode.Restore(op1.Serialize()) _Check(op2) self.assertEqual(op1.Serialize(), op2.Serialize()) class TestQueuedJob(unittest.TestCase): def testNoOpCodes(self): self.assertRaises(errors.GenericError, jqueue._QueuedJob, None, 1, [], False) def testDefaults(self): job_id = 4260 ops = [ opcodes.OpTagsGet(), opcodes.OpTestDelay(), ] def _Check(job): self.assertTrue(job.writable) self.assertEqual(job.id, job_id) self.assertEqual(job.log_serial, 0) self.assert_(job.received_timestamp) self.assert_(job.start_timestamp is None) self.assert_(job.end_timestamp is None) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assert_(repr(job).startswith("<")) self.assertEqual(len(job.ops), len(ops)) self.assert_(compat.all(inp.__getstate__() == op.input.__getstate__() for (inp, op) in zip(ops, job.ops))) self.assertRaises(errors.OpPrereqError, job.GetInfo, ["unknown-field"]) self.assertEqual(job.GetInfo(["summary"]), [[op.input.Summary() for op in job.ops]]) self.assertFalse(job.archived) job1 = jqueue._QueuedJob(None, job_id, ops, True) _Check(job1) job2 = jqueue._QueuedJob.Restore(None, job1.Serialize(), True, False) _Check(job2) self.assertEqual(job1.Serialize(), job2.Serialize()) def testWritable(self): job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], False) self.assertFalse(job.writable) job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], True) self.assertTrue(job.writable) def testArchived(self): job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], False) self.assertFalse(job.archived) newjob = jqueue._QueuedJob.Restore(None, job.Serialize(), True, True) self.assertTrue(newjob.archived) newjob2 = jqueue._QueuedJob.Restore(None, newjob.Serialize(), True, False) self.assertFalse(newjob2.archived) def testPriority(self): job_id = 4283 ops = [ opcodes.OpTagsGet(priority=constants.OP_PRIO_DEFAULT), opcodes.OpTestDelay(), ] def _Check(job): self.assertEqual(job.id, job_id) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assert_(repr(job).startswith("<")) job = jqueue._QueuedJob(None, job_id, ops, True) _Check(job) self.assert_(compat.all(op.priority == constants.OP_PRIO_DEFAULT for op in job.ops)) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) # Increase first job.ops[0].priority -= 1 _Check(job) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 1) # Mark opcode as finished job.ops[0].status = constants.OP_STATUS_SUCCESS _Check(job) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) # Increase second job.ops[1].priority -= 10 self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 10) # Test increasing first job.ops[0].status = constants.OP_STATUS_RUNNING job.ops[0].priority -= 19 self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 20) def _JobForPriority(self, job_id): ops = [ opcodes.OpTagsGet(), opcodes.OpTestDelay(), opcodes.OpTagsGet(), opcodes.OpTestDelay(), ] job = jqueue._QueuedJob(None, job_id, ops, True) self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT for op in job.ops)) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) return job def testChangePriorityAllQueued(self): job = self._JobForPriority(24984) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertTrue(compat.all(op.status == constants.OP_STATUS_QUEUED for op in job.ops)) result = job.ChangePriority(-10) self.assertEqual(job.CalcPriority(), -10) self.assertTrue(compat.all(op.priority == -10 for op in job.ops)) self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) self.assertEqual(result, (True, ("Priorities of pending opcodes for job 24984 have" " been changed to -10"))) def testChangePriorityAllFinished(self): job = self._JobForPriority(16405) for (idx, op) in enumerate(job.ops): if idx > 2: op.status = constants.OP_STATUS_ERROR else: op.status = constants.OP_STATUS_SUCCESS self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) result = job.ChangePriority(-10) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT for op in job.ops)) self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) self.assertEqual(map(operator.attrgetter("status"), job.ops), [ constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_ERROR, ]) self.assertEqual(result, (False, "Job 16405 is finished")) def testChangePriorityCancelling(self): job = self._JobForPriority(31572) for (idx, op) in enumerate(job.ops): if idx > 1: op.status = constants.OP_STATUS_CANCELING else: op.status = constants.OP_STATUS_SUCCESS self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELING) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) result = job.ChangePriority(5) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT for op in job.ops)) self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) self.assertEqual(map(operator.attrgetter("status"), job.ops), [ constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_CANCELING, constants.OP_STATUS_CANCELING, ]) self.assertEqual(result, (False, "Job 31572 is cancelling")) def testChangePriorityFirstRunning(self): job = self._JobForPriority(1716215889) for (idx, op) in enumerate(job.ops): if idx == 0: op.status = constants.OP_STATUS_RUNNING else: op.status = constants.OP_STATUS_QUEUED self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) result = job.ChangePriority(7) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assertEqual(map(operator.attrgetter("priority"), job.ops), [constants.OP_PRIO_DEFAULT, 7, 7, 7]) self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) self.assertEqual(map(operator.attrgetter("status"), job.ops), [ constants.OP_STATUS_RUNNING, constants.OP_STATUS_QUEUED, constants.OP_STATUS_QUEUED, constants.OP_STATUS_QUEUED, ]) self.assertEqual(result, (True, ("Priorities of pending opcodes for job" " 1716215889 have been changed to 7"))) def testChangePriorityLastRunning(self): job = self._JobForPriority(1308) for (idx, op) in enumerate(job.ops): if idx == (len(job.ops) - 1): op.status = constants.OP_STATUS_RUNNING else: op.status = constants.OP_STATUS_SUCCESS self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) result = job.ChangePriority(-3) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT for op in job.ops)) self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) self.assertEqual(map(operator.attrgetter("status"), job.ops), [ constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_RUNNING, ]) self.assertEqual(result, (False, "Job 1308 had no pending opcodes")) def testChangePrioritySecondOpcodeRunning(self): job = self._JobForPriority(27701) self.assertEqual(len(job.ops), 4) job.ops[0].status = constants.OP_STATUS_SUCCESS job.ops[1].status = constants.OP_STATUS_RUNNING job.ops[2].status = constants.OP_STATUS_QUEUED job.ops[3].status = constants.OP_STATUS_QUEUED self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) result = job.ChangePriority(-19) self.assertEqual(job.CalcPriority(), -19) self.assertEqual(map(operator.attrgetter("priority"), job.ops), [constants.OP_PRIO_DEFAULT, constants.OP_PRIO_DEFAULT, -19, -19]) self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) self.assertEqual(map(operator.attrgetter("status"), job.ops), [ constants.OP_STATUS_SUCCESS, constants.OP_STATUS_RUNNING, constants.OP_STATUS_QUEUED, constants.OP_STATUS_QUEUED, ]) self.assertEqual(result, (True, ("Priorities of pending opcodes for job" " 27701 have been changed to -19"))) def testChangePriorityWithInconsistentJob(self): job = self._JobForPriority(30097) self.assertEqual(len(job.ops), 4) # This job is invalid (as it has two opcodes marked as running) and make # the call fail because an unprocessed opcode precedes a running one (which # should never happen in reality) job.ops[0].status = constants.OP_STATUS_SUCCESS job.ops[1].status = constants.OP_STATUS_RUNNING job.ops[2].status = constants.OP_STATUS_QUEUED job.ops[3].status = constants.OP_STATUS_RUNNING self.assertRaises(AssertionError, job.ChangePriority, 19) def testCalcStatus(self): def _Queued(ops): # The default status is "queued" self.assert_(compat.all(op.status == constants.OP_STATUS_QUEUED for op in ops)) def _Waitlock1(ops): ops[0].status = constants.OP_STATUS_WAITING def _Waitlock2(ops): ops[0].status = constants.OP_STATUS_SUCCESS ops[1].status = constants.OP_STATUS_SUCCESS ops[2].status = constants.OP_STATUS_WAITING def _Running(ops): ops[0].status = constants.OP_STATUS_SUCCESS ops[1].status = constants.OP_STATUS_RUNNING for op in ops[2:]: op.status = constants.OP_STATUS_QUEUED def _Canceling1(ops): ops[0].status = constants.OP_STATUS_SUCCESS ops[1].status = constants.OP_STATUS_SUCCESS for op in ops[2:]: op.status = constants.OP_STATUS_CANCELING def _Canceling2(ops): for op in ops: op.status = constants.OP_STATUS_CANCELING def _Canceled(ops): for op in ops: op.status = constants.OP_STATUS_CANCELED def _Error1(ops): for idx, op in enumerate(ops): if idx > 3: op.status = constants.OP_STATUS_ERROR else: op.status = constants.OP_STATUS_SUCCESS def _Error2(ops): for op in ops: op.status = constants.OP_STATUS_ERROR def _Success(ops): for op in ops: op.status = constants.OP_STATUS_SUCCESS tests = { constants.JOB_STATUS_QUEUED: [_Queued], constants.JOB_STATUS_WAITING: [_Waitlock1, _Waitlock2], constants.JOB_STATUS_RUNNING: [_Running], constants.JOB_STATUS_CANCELING: [_Canceling1, _Canceling2], constants.JOB_STATUS_CANCELED: [_Canceled], constants.JOB_STATUS_ERROR: [_Error1, _Error2], constants.JOB_STATUS_SUCCESS: [_Success], } def _NewJob(): job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay() for _ in range(10)], True) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assert_(compat.all(op.status == constants.OP_STATUS_QUEUED for op in job.ops)) return job for status in constants.JOB_STATUS_ALL: sttests = tests[status] assert sttests for fn in sttests: job = _NewJob() fn(job.ops) self.assertEqual(job.CalcStatus(), status) class _FakeDependencyManager: def __init__(self): self._checks = [] self._notifications = [] self._waiting = set() def AddCheckResult(self, job, dep_job_id, dep_status, result): self._checks.append((job, dep_job_id, dep_status, result)) def CountPendingResults(self): return len(self._checks) def CountWaitingJobs(self): return len(self._waiting) def GetNextNotification(self): return self._notifications.pop(0) def JobWaiting(self, job): return job in self._waiting def CheckAndRegister(self, job, dep_job_id, dep_status): (exp_job, exp_dep_job_id, exp_dep_status, result) = self._checks.pop(0) assert exp_job == job assert exp_dep_job_id == dep_job_id assert exp_dep_status == dep_status (result_status, _) = result if result_status == jqueue._JobDependencyManager.WAIT: self._waiting.add(job) elif result_status == jqueue._JobDependencyManager.CONTINUE: self._waiting.remove(job) return result def NotifyWaiters(self, job_id): self._notifications.append(job_id) class _DisabledFakeDependencyManager: def JobWaiting(self, _): return False def CheckAndRegister(self, *args): assert False, "Should not be called" def NotifyWaiters(self, _): pass class _FakeQueueForProc: def __init__(self, depmgr=None): self._acquired = False self._updates = [] self._submitted = [] self._accepting_jobs = True self._submit_count = itertools.count(1000) if depmgr: self.depmgr = depmgr else: self.depmgr = _DisabledFakeDependencyManager() def IsAcquired(self): return self._acquired def GetNextUpdate(self): return self._updates.pop(0) def GetNextSubmittedJob(self): return self._submitted.pop(0) def acquire(self, shared=0): assert shared == 1 self._acquired = True def release(self): assert self._acquired self._acquired = False def UpdateJobUnlocked(self, job, replicate=True): assert self._acquired, "Lock not acquired while updating job" self._updates.append((job, bool(replicate))) def SubmitManyJobs(self, jobs): assert not self._acquired, "Lock acquired while submitting jobs" job_ids = [self._submit_count.next() for _ in jobs] self._submitted.extend(zip(job_ids, jobs)) return job_ids def StopAcceptingJobs(self): self._accepting_jobs = False def AcceptingJobsUnlocked(self): return self._accepting_jobs class _FakeExecOpCodeForProc: def __init__(self, queue, before_start, after_start): self._queue = queue self._before_start = before_start self._after_start = after_start def __call__(self, op, cbs, timeout=None): assert isinstance(op, opcodes.OpTestDummy) assert not self._queue.IsAcquired(), \ "Queue lock not released when executing opcode" if self._before_start: self._before_start(timeout, cbs.CurrentPriority()) cbs.NotifyStart() if self._after_start: self._after_start(op, cbs) # Check again after the callbacks assert not self._queue.IsAcquired() if op.fail: raise errors.OpExecError("Error requested (%s)" % op.result) if hasattr(op, "submit_jobs") and op.submit_jobs is not None: return cbs.SubmitManyJobs(op.submit_jobs) return op.result class _JobProcessorTestUtils: def _CreateJob(self, queue, job_id, ops): job = jqueue._QueuedJob(queue, job_id, ops, True) self.assertFalse(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertEqual(len(ops), len(job.ops)) self.assert_(compat.all(op.input == inp for (op, inp) in zip(job.ops, ops))) self.assertEqual(job.GetInfo(["ops"]), [[op.__getstate__() for op in ops]]) return job class TestJobProcessor(unittest.TestCase, _JobProcessorTestUtils): def _GenericCheckJob(self, job): assert compat.all(isinstance(op.input, opcodes.OpTestDummy) for op in job.ops) self.assertEqual(job.GetInfo(["opstart", "opexec", "opend"]), [[op.start_timestamp for op in job.ops], [op.exec_timestamp for op in job.ops], [op.end_timestamp for op in job.ops]]) self.assertEqual(job.GetInfo(["received_ts", "start_ts", "end_ts"]), [job.received_timestamp, job.start_timestamp, job.end_timestamp]) self.assert_(job.start_timestamp) self.assert_(job.end_timestamp) self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp) def testSuccess(self): queue = _FakeQueueForProc() for (job_id, opcount) in [(25351, 1), (6637, 3), (24644, 10), (32207, 100)]: ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(opcount)] # Create job job = self._CreateJob(queue, job_id, ops) def _BeforeStart(timeout, priority): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertFalse(job.cur_opctx) def _AfterStart(op, cbs): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertFalse(job.cur_opctx) # Job is running, cancelling shouldn't be possible (success, _) = job.Cancel() self.assertFalse(success) opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) for idx in range(len(ops)): self.assertRaises(IndexError, queue.GetNextUpdate) result = jqueue._JobProcessor(queue, opexec, job)() self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) if idx == len(ops) - 1: # Last opcode self.assertEqual(result, jqueue._JobProcessor.FINISHED) else: self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) self.assertEqual(job.GetInfo(["opresult"]), [[op.input.result for op in job.ops]]) self.assertEqual(job.GetInfo(["opstatus"]), [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) self.assert_(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops)) self._GenericCheckJob(job) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) def testOpcodeError(self): queue = _FakeQueueForProc() testdata = [ (17077, 1, 0, 0), (1782, 5, 2, 2), (18179, 10, 9, 9), (4744, 10, 3, 8), (23816, 100, 39, 45), ] for (job_id, opcount, failfrom, failto) in testdata: # Prepare opcodes ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=(failfrom <= i and i <= failto)) for i in range(opcount)] # Create job job = self._CreateJob(queue, str(job_id), ops) opexec = _FakeExecOpCodeForProc(queue, None, None) for idx in range(len(ops)): self.assertRaises(IndexError, queue.GetNextUpdate) result = jqueue._JobProcessor(queue, opexec, job)() # queued to waitlock self.assertEqual(queue.GetNextUpdate(), (job, True)) # waitlock to running self.assertEqual(queue.GetNextUpdate(), (job, True)) # Opcode result self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) if idx in (failfrom, len(ops) - 1): # Last opcode self.assertEqual(result, jqueue._JobProcessor.FINISHED) break self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertRaises(IndexError, queue.GetNextUpdate) # Check job status self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR) self.assertEqual(job.GetInfo(["id"]), [job_id]) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_ERROR]) # Check opcode status data = zip(job.ops, job.GetInfo(["opstatus"])[0], job.GetInfo(["opresult"])[0]) for idx, (op, opstatus, opresult) in enumerate(data): if idx < failfrom: assert not op.input.fail self.assertEqual(opstatus, constants.OP_STATUS_SUCCESS) self.assertEqual(opresult, op.input.result) elif idx <= failto: assert op.input.fail self.assertEqual(opstatus, constants.OP_STATUS_ERROR) self.assertRaises(errors.OpExecError, errors.MaybeRaise, opresult) else: assert not op.input.fail self.assertEqual(opstatus, constants.OP_STATUS_ERROR) self.assertRaises(errors.OpExecError, errors.MaybeRaise, opresult) self.assert_(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops[:failfrom])) self._GenericCheckJob(job) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) def testCancelWhileInQueue(self): queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(5)] # Create job job_id = 17045 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) # Mark as cancelled (success, _) = job.Cancel() self.assert_(success) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(job.start_timestamp) self.assertTrue(job.end_timestamp) self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELED for op in job.ops)) # Serialize to check for differences before_proc = job.Serialize() # Simulate processor called in workerpool opexec = _FakeExecOpCodeForProc(queue, None, None) self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) # Check result self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) self.assertFalse(job.start_timestamp) self.assertTrue(job.end_timestamp) self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp for op in job.ops)) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_CANCELED for _ in job.ops], ["Job canceled by request" for _ in job.ops]]) # Must not have changed or written self.assertEqual(before_proc, job.Serialize()) self.assertRaises(IndexError, queue.GetNextUpdate) def testCancelWhileWaitlockInQueue(self): queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(5)] # Create job job_id = 8645 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) job.ops[0].status = constants.OP_STATUS_WAITING assert len(job.ops) == 5 self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) # Mark as cancelling (success, _) = job.Cancel() self.assert_(success) self.assertRaises(IndexError, queue.GetNextUpdate) self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING for op in job.ops)) opexec = _FakeExecOpCodeForProc(queue, None, None) self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) # Check result self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) self.assertFalse(job.start_timestamp) self.assert_(job.end_timestamp) self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp for op in job.ops)) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_CANCELED for _ in job.ops], ["Job canceled by request" for _ in job.ops]]) def testCancelWhileWaitlock(self): queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(5)] # Create job job_id = 11009 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) def _BeforeStart(timeout, priority): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) # Mark as cancelled (success, _) = job.Cancel() self.assert_(success) self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING for op in job.ops)) self.assertRaises(IndexError, queue.GetNextUpdate) def _AfterStart(op, cbs): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) # Check result self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) self.assert_(job.start_timestamp) self.assert_(job.end_timestamp) self.assertFalse(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops)) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_CANCELED for _ in job.ops], ["Job canceled by request" for _ in job.ops]]) def _TestCancelWhileSomething(self, cb): queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(5)] # Create job job_id = 24314 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) def _BeforeStart(timeout, priority): self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) # Mark as cancelled (success, _) = job.Cancel() self.assert_(success) self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING for op in job.ops)) cb(queue) def _AfterStart(op, cbs): self.fail("Should not reach this") opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) # Check result self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) self.assert_(job.start_timestamp) self.assert_(job.end_timestamp) self.assertFalse(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops)) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_CANCELED for _ in job.ops], ["Job canceled by request" for _ in job.ops]]) return queue def testCancelWhileWaitlockWithTimeout(self): def fn(_): # Fake an acquire attempt timing out raise mcpu.LockAcquireTimeout() self._TestCancelWhileSomething(fn) def testCancelDuringQueueShutdown(self): queue = self._TestCancelWhileSomething(lambda q: q.StopAcceptingJobs()) self.assertFalse(queue.AcceptingJobsUnlocked()) def testCancelWhileRunning(self): # Tests canceling a job with finished opcodes and more, unprocessed ones queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(3)] # Create job job_id = 28492 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) opexec = _FakeExecOpCodeForProc(queue, None, None) # Run one opcode self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.DEFER) # Job goes back to queued self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_QUEUED, constants.OP_STATUS_QUEUED], ["Res0", None, None]]) # Mark as cancelled (success, _) = job.Cancel() self.assert_(success) # Try processing another opcode (this will actually cancel the job) self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) # Check result self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) self.assertEqual(job.GetInfo(["id"]), [job_id]) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_CANCELED, constants.OP_STATUS_CANCELED], ["Res0", "Job canceled by request", "Job canceled by request"]]) def _TestQueueShutdown(self, queue, opexec, job, runcount): self.assertTrue(queue.AcceptingJobsUnlocked()) # Simulate shutdown queue.StopAcceptingJobs() self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.DEFER) # Check result self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_QUEUED]) self.assertFalse(job.cur_opctx) self.assertTrue(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp) self.assertTrue(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops[:runcount])) self.assertFalse(job.ops[runcount].end_timestamp) self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp for op in job.ops[(runcount + 1):])) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [(([constants.OP_STATUS_SUCCESS] * runcount) + ([constants.OP_STATUS_QUEUED] * (len(job.ops) - runcount))), (["Res%s" % i for i in range(runcount)] + ([None] * (len(job.ops) - runcount)))]) # Must have been written and replicated self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) def testQueueShutdownWhileRunning(self): # Tests shutting down the queue while a job is running queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(3)] # Create job job_id = 2718211587 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) opexec = _FakeExecOpCodeForProc(queue, None, None) self.assertRaises(IndexError, queue.GetNextUpdate) # Run one opcode self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.DEFER) # Job goes back to queued self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_QUEUED, constants.OP_STATUS_QUEUED], ["Res0", None, None]]) self.assertFalse(job.cur_opctx) # Writes for waiting, running and result for _ in range(3): self.assertEqual(queue.GetNextUpdate(), (job, True)) # Run second opcode self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.DEFER) # Job goes back to queued self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_QUEUED], ["Res0", "Res1", None]]) self.assertFalse(job.cur_opctx) # Writes for waiting, running and result for _ in range(3): self.assertEqual(queue.GetNextUpdate(), (job, True)) self._TestQueueShutdown(queue, opexec, job, 2) def testQueueShutdownWithLockTimeout(self): # Tests shutting down while a lock acquire times out queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(3)] # Create job job_id = 1304231178 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) acquire_timeout = False def _BeforeStart(timeout, priority): self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) if acquire_timeout: raise mcpu.LockAcquireTimeout() opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, None) self.assertRaises(IndexError, queue.GetNextUpdate) # Run one opcode self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.DEFER) # Job goes back to queued self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_QUEUED, constants.OP_STATUS_QUEUED], ["Res0", None, None]]) self.assertFalse(job.cur_opctx) # Writes for waiting, running and result for _ in range(3): self.assertEqual(queue.GetNextUpdate(), (job, True)) # The next opcode should have expiring lock acquires acquire_timeout = True self._TestQueueShutdown(queue, opexec, job, 1) def testQueueShutdownWhileInQueue(self): # This should never happen in reality (no new jobs are started by the # workerpool once a shutdown has been initiated), but it's better to test # the job processor for this scenario queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(5)] # Create job job_id = 2031 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertTrue(compat.all(op.status == constants.OP_STATUS_QUEUED for op in job.ops)) opexec = _FakeExecOpCodeForProc(queue, None, None) self._TestQueueShutdown(queue, opexec, job, 0) def testQueueShutdownWhileWaitlockInQueue(self): queue = _FakeQueueForProc() ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(5)] # Create job job_id = 53125685 job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) job.ops[0].status = constants.OP_STATUS_WAITING assert len(job.ops) == 5 self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertRaises(IndexError, queue.GetNextUpdate) opexec = _FakeExecOpCodeForProc(queue, None, None) self._TestQueueShutdown(queue, opexec, job, 0) def testPartiallyRun(self): # Tests calling the processor on a job that's been partially run before the # program was restarted queue = _FakeQueueForProc() opexec = _FakeExecOpCodeForProc(queue, None, None) for job_id, successcount in [(30697, 1), (2552, 4), (12489, 9)]: ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(10)] # Create job job = self._CreateJob(queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) for _ in range(successcount): self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.GetInfo(["opstatus"]), [[constants.OP_STATUS_SUCCESS for _ in range(successcount)] + [constants.OP_STATUS_QUEUED for _ in range(len(ops) - successcount)]]) self.assert_(job.ops_iter) # Serialize and restore (simulates program restart) newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False) self.assertFalse(newjob.ops_iter) self._TestPartial(newjob, successcount) def _TestPartial(self, job, successcount): self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp) queue = _FakeQueueForProc() opexec = _FakeExecOpCodeForProc(queue, None, None) for remaining in reversed(range(len(job.ops) - successcount)): result = jqueue._JobProcessor(queue, opexec, job)() self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) if remaining == 0: # Last opcode self.assertEqual(result, jqueue._JobProcessor.FINISHED) break self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) self.assertEqual(job.GetInfo(["opresult"]), [[op.input.result for op in job.ops]]) self.assertEqual(job.GetInfo(["opstatus"]), [[constants.OP_STATUS_SUCCESS for _ in job.ops]]) self.assert_(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops)) self._GenericCheckJob(job) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) # ... also after being restored job2 = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job2)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) def testProcessorOnRunningJob(self): ops = [opcodes.OpTestDummy(result="result", fail=False)] queue = _FakeQueueForProc() opexec = _FakeExecOpCodeForProc(queue, None, None) # Create job job = self._CreateJob(queue, 9571, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) job.ops[0].status = constants.OP_STATUS_RUNNING assert len(job.ops) == 1 self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) # Calling on running job must fail self.assertRaises(errors.ProgrammerError, jqueue._JobProcessor(queue, opexec, job)) def testLogMessages(self): # Tests the "Feedback" callback function queue = _FakeQueueForProc() messages = { 1: [ (None, "Hello"), (None, "World"), (constants.ELOG_MESSAGE, "there"), ], 4: [ (constants.ELOG_JQUEUE_TEST, (1, 2, 3)), (constants.ELOG_JQUEUE_TEST, ("other", "type")), ], } ops = [opcodes.OpTestDummy(result="Logtest%s" % i, fail=False, messages=messages.get(i, [])) for i in range(5)] # Create job job = self._CreateJob(queue, 29386, ops) def _BeforeStart(timeout, priority): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) def _AfterStart(op, cbs): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertRaises(AssertionError, cbs.Feedback, "too", "many", "arguments") for (log_type, msg) in op.messages: self.assertRaises(IndexError, queue.GetNextUpdate) if log_type: cbs.Feedback(log_type, msg) else: cbs.Feedback(msg) # Check for job update without replication self.assertEqual(queue.GetNextUpdate(), (job, False)) self.assertRaises(IndexError, queue.GetNextUpdate) opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) for remaining in reversed(range(len(job.ops))): self.assertRaises(IndexError, queue.GetNextUpdate) result = jqueue._JobProcessor(queue, opexec, job)() self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) if remaining == 0: # Last opcode self.assertEqual(result, jqueue._JobProcessor.FINISHED) break self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) self.assertEqual(job.GetInfo(["opresult"]), [[op.input.result for op in job.ops]]) logmsgcount = sum(len(m) for m in messages.values()) self._CheckLogMessages(job, logmsgcount) # Serialize and restore (simulates program restart) newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False) self._CheckLogMessages(newjob, logmsgcount) # Check each message prevserial = -1 for idx, oplog in enumerate(job.GetInfo(["oplog"])[0]): for (serial, timestamp, log_type, msg) in oplog: (exptype, expmsg) = messages.get(idx).pop(0) if exptype: self.assertEqual(log_type, exptype) else: self.assertEqual(log_type, constants.ELOG_MESSAGE) self.assertEqual(expmsg, msg) self.assert_(serial > prevserial) prevserial = serial def _CheckLogMessages(self, job, count): # Check serial self.assertEqual(job.log_serial, count) # No filter self.assertEqual(job.GetLogEntries(None), [entry for entries in job.GetInfo(["oplog"])[0] if entries for entry in entries]) # Filter with serial assert count > 3 self.assert_(job.GetLogEntries(3)) self.assertEqual(job.GetLogEntries(3), [entry for entries in job.GetInfo(["oplog"])[0] if entries for entry in entries][3:]) # No log message after highest serial self.assertFalse(job.GetLogEntries(count)) self.assertFalse(job.GetLogEntries(count + 3)) def testSubmitManyJobs(self): queue = _FakeQueueForProc() job_id = 15656 ops = [ opcodes.OpTestDummy(result="Res0", fail=False, submit_jobs=[]), opcodes.OpTestDummy(result="Res1", fail=False, submit_jobs=[ [opcodes.OpTestDummy(result="r1j0", fail=False)], ]), opcodes.OpTestDummy(result="Res2", fail=False, submit_jobs=[ [opcodes.OpTestDummy(result="r2j0o0", fail=False), opcodes.OpTestDummy(result="r2j0o1", fail=False), opcodes.OpTestDummy(result="r2j0o2", fail=False), opcodes.OpTestDummy(result="r2j0o3", fail=False)], [opcodes.OpTestDummy(result="r2j1", fail=False)], [opcodes.OpTestDummy(result="r2j3o0", fail=False), opcodes.OpTestDummy(result="r2j3o1", fail=False)], ]), ] # Create job job = self._CreateJob(queue, job_id, ops) def _BeforeStart(timeout, priority): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertFalse(job.cur_opctx) def _AfterStart(op, cbs): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertFalse(job.cur_opctx) # Job is running, cancelling shouldn't be possible (success, _) = job.Cancel() self.assertFalse(success) opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) for idx in range(len(ops)): self.assertRaises(IndexError, queue.GetNextUpdate) result = jqueue._JobProcessor(queue, opexec, job)() self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) if idx == len(ops) - 1: # Last opcode self.assertEqual(result, jqueue._JobProcessor.FINISHED) else: self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertRaises(IndexError, queue.GetNextUpdate) for idx, submitted_ops in enumerate(job_ops for op in ops for job_ops in op.submit_jobs): self.assertEqual(queue.GetNextSubmittedJob(), (1000 + idx, submitted_ops)) self.assertRaises(IndexError, queue.GetNextSubmittedJob) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) self.assertEqual(job.GetInfo(["opresult"]), [[[], [1000], [1001, 1002, 1003]]]) self.assertEqual(job.GetInfo(["opstatus"]), [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) self._GenericCheckJob(job) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) def testJobDependency(self): depmgr = _FakeDependencyManager() queue = _FakeQueueForProc(depmgr=depmgr) self.assertEqual(queue.depmgr, depmgr) prev_job_id = 22113 prev_job_id2 = 28102 job_id = 29929 ops = [ opcodes.OpTestDummy(result="Res0", fail=False, depends=[ [prev_job_id2, None], [prev_job_id, None], ]), opcodes.OpTestDummy(result="Res1", fail=False), ] # Create job job = self._CreateJob(queue, job_id, ops) def _BeforeStart(timeout, priority): if attempt == 0 or attempt > 5: # Job should only be updated when it wasn't waiting for another job self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertFalse(job.cur_opctx) def _AfterStart(op, cbs): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertFalse(job.cur_opctx) # Job is running, cancelling shouldn't be possible (success, _) = job.Cancel() self.assertFalse(success) opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) counter = itertools.count() while True: attempt = counter.next() self.assertRaises(IndexError, queue.GetNextUpdate) self.assertRaises(IndexError, depmgr.GetNextNotification) if attempt < 2: depmgr.AddCheckResult(job, prev_job_id2, None, (jqueue._JobDependencyManager.WAIT, "wait2")) elif attempt == 2: depmgr.AddCheckResult(job, prev_job_id2, None, (jqueue._JobDependencyManager.CONTINUE, "cont")) # The processor will ask for the next dependency immediately depmgr.AddCheckResult(job, prev_job_id, None, (jqueue._JobDependencyManager.WAIT, "wait")) elif attempt < 5: depmgr.AddCheckResult(job, prev_job_id, None, (jqueue._JobDependencyManager.WAIT, "wait")) elif attempt == 5: depmgr.AddCheckResult(job, prev_job_id, None, (jqueue._JobDependencyManager.CONTINUE, "cont")) if attempt == 2: self.assertEqual(depmgr.CountPendingResults(), 2) elif attempt > 5: self.assertEqual(depmgr.CountPendingResults(), 0) else: self.assertEqual(depmgr.CountPendingResults(), 1) result = jqueue._JobProcessor(queue, opexec, job)() if attempt == 0 or attempt >= 5: # Job should only be updated if there was an actual change self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(depmgr.CountPendingResults()) if attempt < 5: # Simulate waiting for other job self.assertEqual(result, jqueue._JobProcessor.WAITDEP) self.assertTrue(job.cur_opctx) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertRaises(IndexError, depmgr.GetNextNotification) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) continue if result == jqueue._JobProcessor.FINISHED: # Last opcode self.assertFalse(job.cur_opctx) break self.assertRaises(IndexError, depmgr.GetNextNotification) self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) self.assertEqual(job.GetInfo(["opresult"]), [[op.input.result for op in job.ops]]) self.assertEqual(job.GetInfo(["opstatus"]), [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) self.assertTrue(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops)) self._GenericCheckJob(job) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertRaises(IndexError, depmgr.GetNextNotification) self.assertFalse(depmgr.CountPendingResults()) self.assertFalse(depmgr.CountWaitingJobs()) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) def testJobDependencyCancel(self): depmgr = _FakeDependencyManager() queue = _FakeQueueForProc(depmgr=depmgr) self.assertEqual(queue.depmgr, depmgr) prev_job_id = 13623 job_id = 30876 ops = [ opcodes.OpTestDummy(result="Res0", fail=False), opcodes.OpTestDummy(result="Res1", fail=False, depends=[ [prev_job_id, None], ]), opcodes.OpTestDummy(result="Res2", fail=False), ] # Create job job = self._CreateJob(queue, job_id, ops) def _BeforeStart(timeout, priority): if attempt == 0 or attempt > 5: # Job should only be updated when it wasn't waiting for another job self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertFalse(job.cur_opctx) def _AfterStart(op, cbs): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertFalse(job.cur_opctx) # Job is running, cancelling shouldn't be possible (success, _) = job.Cancel() self.assertFalse(success) opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) counter = itertools.count() while True: attempt = counter.next() self.assertRaises(IndexError, queue.GetNextUpdate) self.assertRaises(IndexError, depmgr.GetNextNotification) if attempt == 0: # This will handle the first opcode pass elif attempt < 4: depmgr.AddCheckResult(job, prev_job_id, None, (jqueue._JobDependencyManager.WAIT, "wait")) elif attempt == 4: # Other job was cancelled depmgr.AddCheckResult(job, prev_job_id, None, (jqueue._JobDependencyManager.CANCEL, "cancel")) if attempt == 0: self.assertEqual(depmgr.CountPendingResults(), 0) else: self.assertEqual(depmgr.CountPendingResults(), 1) result = jqueue._JobProcessor(queue, opexec, job)() if attempt <= 1 or attempt >= 4: # Job should only be updated if there was an actual change self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(depmgr.CountPendingResults()) if attempt > 0 and attempt < 4: # Simulate waiting for other job self.assertEqual(result, jqueue._JobProcessor.WAITDEP) self.assertTrue(job.cur_opctx) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertRaises(IndexError, depmgr.GetNextNotification) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) continue if result == jqueue._JobProcessor.FINISHED: # Last opcode self.assertFalse(job.cur_opctx) break self.assertRaises(IndexError, depmgr.GetNextNotification) self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_CANCELED, constants.OP_STATUS_CANCELED], ["Res0", "Job canceled by request", "Job canceled by request"]]) self._GenericCheckJob(job) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertRaises(IndexError, depmgr.GetNextNotification) self.assertFalse(depmgr.CountPendingResults()) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) def testJobDependencyWrongstatus(self): depmgr = _FakeDependencyManager() queue = _FakeQueueForProc(depmgr=depmgr) self.assertEqual(queue.depmgr, depmgr) prev_job_id = 9741 job_id = 11763 ops = [ opcodes.OpTestDummy(result="Res0", fail=False), opcodes.OpTestDummy(result="Res1", fail=False, depends=[ [prev_job_id, None], ]), opcodes.OpTestDummy(result="Res2", fail=False), ] # Create job job = self._CreateJob(queue, job_id, ops) def _BeforeStart(timeout, priority): if attempt == 0 or attempt > 5: # Job should only be updated when it wasn't waiting for another job self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertFalse(job.cur_opctx) def _AfterStart(op, cbs): self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) self.assertFalse(job.cur_opctx) # Job is running, cancelling shouldn't be possible (success, _) = job.Cancel() self.assertFalse(success) opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) counter = itertools.count() while True: attempt = counter.next() self.assertRaises(IndexError, queue.GetNextUpdate) self.assertRaises(IndexError, depmgr.GetNextNotification) if attempt == 0: # This will handle the first opcode pass elif attempt < 4: depmgr.AddCheckResult(job, prev_job_id, None, (jqueue._JobDependencyManager.WAIT, "wait")) elif attempt == 4: # Other job failed depmgr.AddCheckResult(job, prev_job_id, None, (jqueue._JobDependencyManager.WRONGSTATUS, "w")) if attempt == 0: self.assertEqual(depmgr.CountPendingResults(), 0) else: self.assertEqual(depmgr.CountPendingResults(), 1) result = jqueue._JobProcessor(queue, opexec, job)() if attempt <= 1 or attempt >= 4: # Job should only be updated if there was an actual change self.assertEqual(queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertFalse(depmgr.CountPendingResults()) if attempt > 0 and attempt < 4: # Simulate waiting for other job self.assertEqual(result, jqueue._JobProcessor.WAITDEP) self.assertTrue(job.cur_opctx) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) self.assertRaises(IndexError, depmgr.GetNextNotification) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) continue if result == jqueue._JobProcessor.FINISHED: # Last opcode self.assertFalse(job.cur_opctx) break self.assertRaises(IndexError, depmgr.GetNextNotification) self.assertEqual(result, jqueue._JobProcessor.DEFER) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_ERROR]) self.assertEqual(job.GetInfo(["opstatus"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_ERROR, constants.OP_STATUS_ERROR]]), (opresult, ) = job.GetInfo(["opresult"]) self.assertEqual(len(opresult), len(ops)) self.assertEqual(opresult[0], "Res0") self.assertTrue(errors.GetEncodedError(opresult[1])) self.assertTrue(errors.GetEncodedError(opresult[2])) self._GenericCheckJob(job) self.assertRaises(IndexError, queue.GetNextUpdate) self.assertRaises(IndexError, depmgr.GetNextNotification) self.assertFalse(depmgr.CountPendingResults()) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, queue.GetNextUpdate) class TestEvaluateJobProcessorResult(unittest.TestCase): def testFinished(self): depmgr = _FakeDependencyManager() job = _IdOnlyFakeJob(30953) jqueue._EvaluateJobProcessorResult(depmgr, job, jqueue._JobProcessor.FINISHED) self.assertEqual(depmgr.GetNextNotification(), job.id) self.assertRaises(IndexError, depmgr.GetNextNotification) def testDefer(self): depmgr = _FakeDependencyManager() job = _IdOnlyFakeJob(11326, priority=5463) try: jqueue._EvaluateJobProcessorResult(depmgr, job, jqueue._JobProcessor.DEFER) except workerpool.DeferTask, err: self.assertEqual(err.priority, 5463) else: self.fail("Didn't raise exception") self.assertRaises(IndexError, depmgr.GetNextNotification) def testWaitdep(self): depmgr = _FakeDependencyManager() job = _IdOnlyFakeJob(21317) jqueue._EvaluateJobProcessorResult(depmgr, job, jqueue._JobProcessor.WAITDEP) self.assertRaises(IndexError, depmgr.GetNextNotification) def testOther(self): depmgr = _FakeDependencyManager() job = _IdOnlyFakeJob(5813) self.assertRaises(errors.ProgrammerError, jqueue._EvaluateJobProcessorResult, depmgr, job, "Other result") self.assertRaises(IndexError, depmgr.GetNextNotification) class _FakeTimeoutStrategy: def __init__(self, timeouts): self.timeouts = timeouts self.attempts = 0 self.last_timeout = None def NextAttempt(self): self.attempts += 1 if self.timeouts: timeout = self.timeouts.pop(0) else: timeout = None self.last_timeout = timeout return timeout class TestJobProcessorTimeouts(unittest.TestCase, _JobProcessorTestUtils): def setUp(self): self.queue = _FakeQueueForProc() self.job = None self.curop = None self.opcounter = None self.timeout_strategy = None self.retries = 0 self.prev_tsop = None self.prev_prio = None self.prev_status = None self.lock_acq_prio = None self.gave_lock = None self.done_lock_before_blocking = False def _BeforeStart(self, timeout, priority): job = self.job # If status has changed, job must've been written if self.prev_status != self.job.ops[self.curop].status: self.assertEqual(self.queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, self.queue.GetNextUpdate) self.assertFalse(self.queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) ts = self.timeout_strategy self.assert_(timeout is None or isinstance(timeout, (int, float))) self.assertEqual(timeout, ts.last_timeout) self.assertEqual(priority, job.ops[self.curop].priority) self.gave_lock = True self.lock_acq_prio = priority if (self.curop == 3 and job.ops[self.curop].priority == constants.OP_PRIO_HIGHEST + 3): # Give locks before running into blocking acquire assert self.retries == 7 self.retries = 0 self.done_lock_before_blocking = True return if self.retries > 0: self.assert_(timeout is not None) self.retries -= 1 self.gave_lock = False raise mcpu.LockAcquireTimeout() if job.ops[self.curop].priority == constants.OP_PRIO_HIGHEST: assert self.retries == 0, "Didn't exhaust all retries at highest priority" assert not ts.timeouts self.assert_(timeout is None) def _AfterStart(self, op, cbs): job = self.job # Setting to "running" requires an update self.assertEqual(self.queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, self.queue.GetNextUpdate) self.assertFalse(self.queue.IsAcquired()) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) # Job is running, cancelling shouldn't be possible (success, _) = job.Cancel() self.assertFalse(success) def _NextOpcode(self): self.curop = self.opcounter.next() self.prev_prio = self.job.ops[self.curop].priority self.prev_status = self.job.ops[self.curop].status def _NewTimeoutStrategy(self): job = self.job self.assertEqual(self.retries, 0) if self.prev_tsop == self.curop: # Still on the same opcode, priority must've been increased self.assertEqual(self.prev_prio, job.ops[self.curop].priority + 1) if self.curop == 1: # Normal retry timeouts = range(10, 31, 10) self.retries = len(timeouts) - 1 elif self.curop == 2: # Let this run into a blocking acquire timeouts = range(11, 61, 12) self.retries = len(timeouts) elif self.curop == 3: # Wait for priority to increase, but give lock before blocking acquire timeouts = range(12, 100, 14) self.retries = len(timeouts) self.assertFalse(self.done_lock_before_blocking) elif self.curop == 4: self.assert_(self.done_lock_before_blocking) # Timeouts, but no need to retry timeouts = range(10, 31, 10) self.retries = 0 elif self.curop == 5: # Normal retry timeouts = range(19, 100, 11) self.retries = len(timeouts) else: timeouts = [] self.retries = 0 assert len(job.ops) == 10 assert self.retries <= len(timeouts) ts = _FakeTimeoutStrategy(timeouts) self.timeout_strategy = ts self.prev_tsop = self.curop self.prev_prio = job.ops[self.curop].priority return ts def testTimeout(self): ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(10)] # Create job job_id = 15801 job = self._CreateJob(self.queue, job_id, ops) self.job = job self.opcounter = itertools.count(0) opexec = _FakeExecOpCodeForProc(self.queue, self._BeforeStart, self._AfterStart) tsf = self._NewTimeoutStrategy self.assertFalse(self.done_lock_before_blocking) while True: proc = jqueue._JobProcessor(self.queue, opexec, job, _timeout_strategy_factory=tsf) self.assertRaises(IndexError, self.queue.GetNextUpdate) if self.curop is not None: self.prev_status = self.job.ops[self.curop].status self.lock_acq_prio = None result = proc(_nextop_fn=self._NextOpcode) assert self.curop is not None # Input priority should never be set or modified self.assertFalse(compat.any(hasattr(op.input, "priority") for op in job.ops)) if result == jqueue._JobProcessor.FINISHED or self.gave_lock: # Got lock and/or job is done, result must've been written self.assertFalse(job.cur_opctx) self.assertEqual(self.queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, self.queue.GetNextUpdate) self.assertEqual(self.lock_acq_prio, job.ops[self.curop].priority) self.assert_(job.ops[self.curop].exec_timestamp) if result == jqueue._JobProcessor.FINISHED: self.assertFalse(job.cur_opctx) break self.assertEqual(result, jqueue._JobProcessor.DEFER) if self.curop == 0: self.assertEqual(job.ops[self.curop].start_timestamp, job.start_timestamp) if self.gave_lock: # Opcode finished, but job not yet done self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) else: # Did not get locks self.assert_(job.cur_opctx) self.assertEqual(job.cur_opctx._timeout_strategy._fn, self.timeout_strategy.NextAttempt) self.assertFalse(job.ops[self.curop].exec_timestamp) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) # If priority has changed since acquiring locks, the job must've been # updated if self.lock_acq_prio != job.ops[self.curop].priority: self.assertEqual(self.queue.GetNextUpdate(), (job, True)) self.assertRaises(IndexError, self.queue.GetNextUpdate) self.assert_(job.start_timestamp) self.assertFalse(job.end_timestamp) self.assertEqual(self.curop, len(job.ops) - 1) self.assertEqual(self.job, job) self.assertEqual(self.opcounter.next(), len(job.ops)) self.assert_(self.done_lock_before_blocking) self.assertRaises(IndexError, self.queue.GetNextUpdate) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) self.assertEqual(job.GetInfo(["opresult"]), [[op.input.result for op in job.ops]]) self.assertEqual(job.GetInfo(["opstatus"]), [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) self.assert_(compat.all(op.start_timestamp and op.end_timestamp for op in job.ops)) # Calling the processor on a finished job should be a no-op self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertRaises(IndexError, self.queue.GetNextUpdate) class TestJobProcessorChangePriority(unittest.TestCase, _JobProcessorTestUtils): def setUp(self): self.queue = _FakeQueueForProc() self.opexecprio = [] def _BeforeStart(self, timeout, priority): self.assertFalse(self.queue.IsAcquired()) self.opexecprio.append(priority) def testChangePriorityWhileRunning(self): # Tests changing the priority on a job while it has finished opcodes # (successful) and more, unprocessed ones ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) for i in range(3)] # Create job job_id = 3499 job = self._CreateJob(self.queue, job_id, ops) self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) opexec = _FakeExecOpCodeForProc(self.queue, self._BeforeStart, None) # Run first opcode self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), jqueue._JobProcessor.DEFER) # Job goes back to queued self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_QUEUED, constants.OP_STATUS_QUEUED], ["Res0", None, None]]) self.assertEqual(self.opexecprio.pop(0), constants.OP_PRIO_DEFAULT) self.assertRaises(IndexError, self.opexecprio.pop, 0) # Change priority self.assertEqual(job.ChangePriority(-10), (True, ("Priorities of pending opcodes for job 3499 have" " been changed to -10"))) self.assertEqual(job.CalcPriority(), -10) # Process second opcode self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), jqueue._JobProcessor.DEFER) self.assertEqual(self.opexecprio.pop(0), -10) self.assertRaises(IndexError, self.opexecprio.pop, 0) # Check status self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) self.assertEqual(job.CalcPriority(), -10) self.assertEqual(job.GetInfo(["id"]), [job_id]) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_QUEUED]) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_QUEUED], ["Res0", "Res1", None]]) # Change priority once more self.assertEqual(job.ChangePriority(5), (True, ("Priorities of pending opcodes for job 3499 have" " been changed to 5"))) self.assertEqual(job.CalcPriority(), 5) # Process third opcode self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), jqueue._JobProcessor.FINISHED) self.assertEqual(self.opexecprio.pop(0), 5) self.assertRaises(IndexError, self.opexecprio.pop, 0) # Check status self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) self.assertEqual(job.GetInfo(["id"]), [job_id]) self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) self.assertEqual(job.GetInfo(["opstatus", "opresult"]), [[constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS, constants.OP_STATUS_SUCCESS], ["Res0", "Res1", "Res2"]]) self.assertEqual(map(operator.attrgetter("priority"), job.ops), [constants.OP_PRIO_DEFAULT, -10, 5]) class _IdOnlyFakeJob: def __init__(self, job_id, priority=NotImplemented): self.id = str(job_id) self._priority = priority def CalcPriority(self): return self._priority class TestJobDependencyManager(unittest.TestCase): def setUp(self): self._status = [] self._queue = [] self.jdm = jqueue._JobDependencyManager(self._GetStatus, self._Enqueue) def _GetStatus(self, job_id): (exp_job_id, result) = self._status.pop(0) self.assertEqual(exp_job_id, job_id) return result def _Enqueue(self, jobs): self.assertFalse(self.jdm._lock.is_owned(), msg=("Must not own manager lock while re-adding jobs" " (potential deadlock)")) self._queue.append(jobs) def testNotFinalizedThenCancel(self): job = _IdOnlyFakeJob(17697) job_id = str(28625) self._status.append((job_id, constants.JOB_STATUS_RUNNING)) (result, _) = self.jdm.CheckAndRegister(job, job_id, []) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [ ("job/28625", None, None, [("job", [job.id])]) ]) self._status.append((job_id, constants.JOB_STATUS_CANCELED)) (result, _) = self.jdm.CheckAndRegister(job, job_id, []) self.assertEqual(result, self.jdm.CANCEL) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING])) def testNotFinalizedThenQueued(self): # This can happen on a queue shutdown job = _IdOnlyFakeJob(1320) job_id = str(22971) for i in range(5): if i > 2: self._status.append((job_id, constants.JOB_STATUS_QUEUED)) else: self._status.append((job_id, constants.JOB_STATUS_RUNNING)) (result, _) = self.jdm.CheckAndRegister(job, job_id, []) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [ ("job/22971", None, None, [("job", [job.id])]) ]) def testRequireCancel(self): job = _IdOnlyFakeJob(5278) job_id = str(9610) dep_status = [constants.JOB_STATUS_CANCELED] self._status.append((job_id, constants.JOB_STATUS_WAITING)) (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [ ("job/9610", None, None, [("job", [job.id])]) ]) self._status.append((job_id, constants.JOB_STATUS_CANCELED)) (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status) self.assertEqual(result, self.jdm.CONTINUE) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING])) def testRequireError(self): job = _IdOnlyFakeJob(21459) job_id = str(25519) dep_status = [constants.JOB_STATUS_ERROR] self._status.append((job_id, constants.JOB_STATUS_WAITING)) (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self._status.append((job_id, constants.JOB_STATUS_ERROR)) (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status) self.assertEqual(result, self.jdm.CONTINUE) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING])) def testRequireMultiple(self): dep_status = list(constants.JOBS_FINALIZED) for end_status in dep_status: job = _IdOnlyFakeJob(21343) job_id = str(14609) self._status.append((job_id, constants.JOB_STATUS_WAITING)) (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [ ("job/14609", None, None, [("job", [job.id])]) ]) self._status.append((job_id, end_status)) (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status) self.assertEqual(result, self.jdm.CONTINUE) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING])) def testNotify(self): job = _IdOnlyFakeJob(8227) job_id = str(4113) self._status.append((job_id, constants.JOB_STATUS_RUNNING)) (result, _) = self.jdm.CheckAndRegister(job, job_id, []) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self.jdm.NotifyWaiters(job_id) self.assertFalse(self._status) self.assertFalse(self.jdm._waiters) self.assertFalse(self.jdm.JobWaiting(job)) self.assertEqual(self._queue, [set([job])]) def testWrongStatus(self): job = _IdOnlyFakeJob(10102) job_id = str(1271) self._status.append((job_id, constants.JOB_STATUS_QUEUED)) (result, _) = self.jdm.CheckAndRegister(job, job_id, [constants.JOB_STATUS_SUCCESS]) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self._status.append((job_id, constants.JOB_STATUS_ERROR)) (result, _) = self.jdm.CheckAndRegister(job, job_id, [constants.JOB_STATUS_SUCCESS]) self.assertEqual(result, self.jdm.WRONGSTATUS) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) def testCorrectStatus(self): job = _IdOnlyFakeJob(24273) job_id = str(23885) self._status.append((job_id, constants.JOB_STATUS_QUEUED)) (result, _) = self.jdm.CheckAndRegister(job, job_id, [constants.JOB_STATUS_SUCCESS]) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set([job]), }) self._status.append((job_id, constants.JOB_STATUS_SUCCESS)) (result, _) = self.jdm.CheckAndRegister(job, job_id, [constants.JOB_STATUS_SUCCESS]) self.assertEqual(result, self.jdm.CONTINUE) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) def testFinalizedRightAway(self): job = _IdOnlyFakeJob(224) job_id = str(3081) self._status.append((job_id, constants.JOB_STATUS_SUCCESS)) (result, _) = self.jdm.CheckAndRegister(job, job_id, [constants.JOB_STATUS_SUCCESS]) self.assertEqual(result, self.jdm.CONTINUE) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, { job_id: set(), }) # Force cleanup self.jdm.NotifyWaiters("0") self.assertFalse(self.jdm._waiters) self.assertFalse(self._status) self.assertFalse(self._queue) def testMultipleWaiting(self): # Use a deterministic random generator rnd = random.Random(21402) job_ids = map(str, rnd.sample(range(1, 10000), 150)) waiters = dict((job_ids.pop(), set(map(_IdOnlyFakeJob, [job_ids.pop() for _ in range(rnd.randint(1, 20))]))) for _ in range(10)) # Ensure there are no duplicate job IDs assert not utils.FindDuplicates(waiters.keys() + [job.id for jobs in waiters.values() for job in jobs]) # Register all jobs as waiters for job_id, job in [(job_id, job) for (job_id, jobs) in waiters.items() for job in jobs]: self._status.append((job_id, constants.JOB_STATUS_QUEUED)) (result, _) = self.jdm.CheckAndRegister(job, job_id, [constants.JOB_STATUS_SUCCESS]) self.assertEqual(result, self.jdm.WAIT) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertTrue(self.jdm.JobWaiting(job)) self.assertEqual(self.jdm._waiters, waiters) def _MakeSet((name, mode, owner_names, pending)): return (name, mode, owner_names, [(pendmode, set(pend)) for (pendmode, pend) in pending]) def _CheckLockInfo(): info = self.jdm.GetLockInfo([query.LQ_PENDING]) self.assertEqual(sorted(map(_MakeSet, info)), sorted([ ("job/%s" % job_id, None, None, [("job", set([job.id for job in jobs]))]) for job_id, jobs in waiters.items() if jobs ])) _CheckLockInfo() # Notify in random order for job_id in rnd.sample(waiters, len(waiters)): # Remove from pending waiter list jobs = waiters.pop(job_id) for job in jobs: self._status.append((job_id, constants.JOB_STATUS_SUCCESS)) (result, _) = self.jdm.CheckAndRegister(job, job_id, [constants.JOB_STATUS_SUCCESS]) self.assertEqual(result, self.jdm.CONTINUE) self.assertFalse(self._status) self.assertFalse(self._queue) self.assertFalse(self.jdm.JobWaiting(job)) _CheckLockInfo() self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING])) assert not waiters def testSelfDependency(self): job = _IdOnlyFakeJob(18937) self._status.append((job.id, constants.JOB_STATUS_SUCCESS)) (result, _) = self.jdm.CheckAndRegister(job, job.id, []) self.assertEqual(result, self.jdm.ERROR) def testJobDisappears(self): job = _IdOnlyFakeJob(30540) job_id = str(23769) def _FakeStatus(_): raise errors.JobLost("#msg#") jdm = jqueue._JobDependencyManager(_FakeStatus, None) (result, _) = jdm.CheckAndRegister(job, job_id, []) self.assertEqual(result, self.jdm.ERROR) self.assertFalse(jdm.JobWaiting(job)) self.assertFalse(jdm.GetLockInfo([query.LQ_PENDING])) if __name__ == "__main__": testutils.GanetiTestProgram()