Synnefo » snf-ganeti

Job priorities

~~~~~~~~~~~~~~

Current state and shortcomings

++++++++++++++++++++++++++++++

.. TODO: Describe current situation

Proposed changes

++++++++++++++++

.. TODO: Describe changes to job queue and potentially client programs

Worker pool

^^^^^^^^^^^

To support job priorities in the job queue, the worker pool underlying

the job queue must be enhanced to support task priorities. Currently

tasks are processed in the order they are added to the queue (but, due

to their nature, they don't necessarily finish in that order). All tasks

are equal. To support tasks with higher or lower priority, a few changes

have to be made to the queue inside a worker pool.

Each task is assigned a priority when added to the queue. This priority

can not be changed until the task is executed (this is fine as in all

current use-cases, tasks are added to a pool and then forgotten about

until they're done).

A task's priority can be compared to Unix' process priorities. The lower

the priority number, the closer to the queue's front it is. A task with

priority 0 is going to be run before one with priority 10. Tasks with

the same priority are executed in the order in which they were added.

While a task is running it can query its own priority. If it's not ready

yet for finishing, it can raise an exception to defer itself, optionally

changing its own priority. This is useful for the following cases:

- A task is trying to acquire locks, but those locks are still held by

  other tasks. By deferring itself, the task gives others a chance to

  run. This is especially useful when all workers are busy.

- If a task decides it hasn't gotten its locks in a long time, it can

  start to increase its own priority.

- Tasks waiting for long-running operations running asynchronously could

  defer themselves while waiting for a long-running operation.

With these changes, the job queue will be able to implement per-job

priorities.

import heapq

from ganeti import errors

_DEFAULT_PRIORITY = 0

class DeferTask(Exception):

  """Special exception class to defer a task.

  This class can be raised by L{BaseWorker.RunTask} to defer the execution of a

  task. Optionally, the priority of the task can be changed.

  """

  def __init__(self, priority=None):

    """Initializes this class.

    @type priority: number

    @param priority: New task priority (None means no change)

    """

    Exception.__init__(self)

    self.priority = priority

  def GetCurrentPriority(self):

    """Returns the priority of the current task.

    Should only be called from within L{RunTask}.

    """

    self.pool._lock.acquire()

    try:

      assert self._HasRunningTaskUnlocked()

      (priority, _, _) = self._current_task

      return priority

    finally:

      self.pool._lock.release()

      defer = None

        (priority, _, args) = self._current_task

          # Run the actual task

          assert defer is None

          logging.debug("Starting task %r, priority %s", args, priority)

          self.RunTask(*args) # pylint: disable-msg=W0142

          logging.debug("Done with task %r, priority %s", args, priority)

        except DeferTask, err:

          defer = err

          if defer.priority is None:

            # Use same priority

            defer.priority = priority

          logging.debug("Deferring task %r, new priority %s", defer.priority)

          assert self._HasRunningTaskUnlocked()

          if defer:

            assert self._current_task

            # Schedule again for later run

            (_, _, args) = self._current_task

            pool._AddTaskUnlocked(args, defer.priority)

    self._counter = 0

    self._tasks = []

  def _AddTaskUnlocked(self, args, priority):

    """Adds a task to the internal queue.

    @type args: sequence

    @param args: Arguments passed to L{BaseWorker.RunTask}

    @type priority: number

    @param priority: Task priority

    """

    assert isinstance(priority, (int, long)), "Priority must be numeric"

    # This counter is used to ensure elements are processed in their

    # incoming order. For processing they're sorted by priority and then

    # counter.

    self._counter += 1

    heapq.heappush(self._tasks, (priority, self._counter, args))

  def AddTask(self, args, priority=_DEFAULT_PRIORITY):

    @type priority: number

    @param priority: Task priority

      self._AddTaskUnlocked(args, priority)

  def AddManyTasks(self, tasks, priority=_DEFAULT_PRIORITY):

    @type priority: number or list of numbers

    @param priority: Priority for all added tasks or a list with the priority

                     for each task

    assert (isinstance(priority, (int, long)) or

            compat.all(isinstance(prio, (int, long)) for prio in priority)), \

           "Priority must be numeric or be a list of numeric values"

    if isinstance(priority, (int, long)):

      priority = [priority] * len(tasks)

    elif len(priority) != len(tasks):

      raise errors.ProgrammerError("Number of priorities (%s) doesn't match"

                                   " number of tasks (%s)" %

                                   (len(priority), len(tasks)))

      assert compat.all(isinstance(prio, (int, long)) for prio in priority)

      assert len(tasks) == len(priority)

      for args, priority in zip(tasks, priority):

        self._AddTaskUnlocked(args, priority)

      return heapq.heappop(self._tasks)

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

import random

from ganeti import errors

class CountingContext(object):

class ListBuilderContext:

  def __init__(self):

    self.lock = threading.Lock()

    self.result = []

    self.prioresult = {}

class ListBuilderWorker(workerpool.BaseWorker):

  def RunTask(self, ctx, data):

    ctx.lock.acquire()

    try:

      ctx.result.append((self.GetCurrentPriority(), data))

      ctx.prioresult.setdefault(self.GetCurrentPriority(), []).append(data)

    finally:

      ctx.lock.release()

class DeferringTaskContext:

  def __init__(self):

    self.lock = threading.Lock()

    self.prioresult = {}

    self.samepriodefer = {}

class DeferringWorker(workerpool.BaseWorker):

  def RunTask(self, ctx, num, targetprio):

    ctx.lock.acquire()

    try:

      if num in ctx.samepriodefer:

        del ctx.samepriodefer[num]

        raise workerpool.DeferTask()

      if self.GetCurrentPriority() > targetprio:

        raise workerpool.DeferTask(priority=self.GetCurrentPriority() - 1)

      ctx.prioresult.setdefault(self.GetCurrentPriority(), set()).add(num)

    finally:

      ctx.lock.release()

  def testPriorityChecksum(self):

    # Tests whether all tasks are run and, since we're only using a single

    # thread, whether everything is started in order and respects the priority

    wp = workerpool.WorkerPool("Test", 1, ChecksumBaseWorker)

    try:

      self._CheckWorkerCount(wp, 1)

      ctx = ChecksumContext()

      data = {}

      tasks = []

      priorities = []

      for i in range(1, 333):

        prio = i % 7

        tasks.append((ctx, i))

        priorities.append(prio)

        data.setdefault(prio, []).append(i)

      wp.AddManyTasks(tasks, priority=priorities)

      wp.Quiesce()

      self._CheckNoTasks(wp)

      # Check sum

      ctx.lock.acquire()

      try:

        checksum = ChecksumContext.CHECKSUM_START

        for priority in sorted(data.keys()):

          for i in data[priority]:

            checksum = ChecksumContext.UpdateChecksum(checksum, i)

        self.assertEqual(checksum, ctx.checksum)

      finally:

        ctx.lock.release()

      self._CheckWorkerCount(wp, 1)

    finally:

      wp.TerminateWorkers()

      self._CheckWorkerCount(wp, 0)

  def testPriorityListManyTasks(self):

    # Tests whether all tasks are run and, since we're only using a single

    # thread, whether everything is started in order and respects the priority

    wp = workerpool.WorkerPool("Test", 1, ListBuilderWorker)

    try:

      self._CheckWorkerCount(wp, 1)

      ctx = ListBuilderContext()

      # Use static seed for this test

      rnd = random.Random(0)

      data = {}

      tasks = []

      priorities = []

      for i in range(1, 333):

        prio = int(rnd.random() * 10)

        tasks.append((ctx, i))

        priorities.append(prio)

        data.setdefault(prio, []).append((prio, i))

      wp.AddManyTasks(tasks, priority=priorities)

      self.assertRaises(errors.ProgrammerError, wp.AddManyTasks,

                        [("x", ), ("y", )], priority=[1] * 5)

      wp.Quiesce()

      self._CheckNoTasks(wp)

      # Check result

      ctx.lock.acquire()

      try:

        expresult = []

        for priority in sorted(data.keys()):

          expresult.extend(data[priority])

        self.assertEqual(expresult, ctx.result)

      finally:

        ctx.lock.release()

      self._CheckWorkerCount(wp, 1)

    finally:

      wp.TerminateWorkers()

      self._CheckWorkerCount(wp, 0)

  def testPriorityListSingleTasks(self):

    # Tests whether all tasks are run and, since we're only using a single

    # thread, whether everything is started in order and respects the priority

    wp = workerpool.WorkerPool("Test", 1, ListBuilderWorker)

    try:

      self._CheckWorkerCount(wp, 1)

      ctx = ListBuilderContext()

      # Use static seed for this test

      rnd = random.Random(26279)

      data = {}

      for i in range(1, 333):

        prio = int(rnd.random() * 30)

        wp.AddTask((ctx, i), priority=prio)

        data.setdefault(prio, []).append(i)

        # Cause some distortion

        if i % 11 == 0:

          time.sleep(.001)

        if i % 41 == 0:

          wp.Quiesce()

      wp.Quiesce()

      self._CheckNoTasks(wp)

      # Check result

      ctx.lock.acquire()

      try:

        self.assertEqual(data, ctx.prioresult)

      finally:

        ctx.lock.release()

      self._CheckWorkerCount(wp, 1)

    finally:

      wp.TerminateWorkers()

      self._CheckWorkerCount(wp, 0)

  def testPriorityListSingleTasks(self):

    # Tests whether all tasks are run and, since we're only using a single

    # thread, whether everything is started in order and respects the priority

    wp = workerpool.WorkerPool("Test", 1, ListBuilderWorker)

    try:

      self._CheckWorkerCount(wp, 1)

      ctx = ListBuilderContext()

      # Use static seed for this test

      rnd = random.Random(26279)

      data = {}

      for i in range(1, 333):

        prio = int(rnd.random() * 30)

        wp.AddTask((ctx, i), priority=prio)

        data.setdefault(prio, []).append(i)

        # Cause some distortion

        if i % 11 == 0:

          time.sleep(.001)

        if i % 41 == 0:

          wp.Quiesce()

      wp.Quiesce()

      self._CheckNoTasks(wp)

      # Check result

      ctx.lock.acquire()

      try:

        self.assertEqual(data, ctx.prioresult)

      finally:

        ctx.lock.release()

      self._CheckWorkerCount(wp, 1)

    finally:

      wp.TerminateWorkers()

      self._CheckWorkerCount(wp, 0)

  def testDeferTask(self):

    # Tests whether all tasks are run and, since we're only using a single

    # thread, whether everything is started in order and respects the priority

    wp = workerpool.WorkerPool("Test", 1, DeferringWorker)

    try:

      self._CheckWorkerCount(wp, 1)

      ctx = DeferringTaskContext()

      # Use static seed for this test

      rnd = random.Random(14921)

      data = {}

      for i in range(1, 333):

        ctx.lock.acquire()

        try:

          if i % 5 == 0:

            ctx.samepriodefer[i] = True

        finally:

          ctx.lock.release()

        prio = int(rnd.random() * 30)

        wp.AddTask((ctx, i, prio), priority=50)

        data.setdefault(prio, set()).add(i)

        # Cause some distortion

        if i % 24 == 0:

          time.sleep(.001)

        if i % 31 == 0:

          wp.Quiesce()

      wp.Quiesce()

      self._CheckNoTasks(wp)

      # Check result

      ctx.lock.acquire()

      try:

        self.assertEqual(data, ctx.prioresult)

      finally:

        ctx.lock.release()

      self._CheckWorkerCount(wp, 1)

    finally:

      wp.TerminateWorkers()

      self._CheckWorkerCount(wp, 0)