pym/_emerge/SequentialTaskQueue.py

   1 # Copyright 1999-2012 Gentoo Foundation
   2 # Distributed under the terms of the GNU General Public License v2
   3
   4 import sys
   5 from _emerge.SlotObject import SlotObject
   6 from collections import deque
   7 class SequentialTaskQueue(SlotObject):
   8
   9         __slots__ = ("max_jobs", "running_tasks") + \
  10                 ("_dirty", "_scheduling", "_task_queue")
  11
  12         def __init__(self, **kwargs):
  13                 SlotObject.__init__(self, **kwargs)
  14                 self._task_queue = deque()
  15                 self.running_tasks = set()
  16                 if self.max_jobs is None:
  17                         self.max_jobs = 1
  18                 self._dirty = True
  19
  20         def add(self, task):
  21                 self._task_queue.append(task)
  22                 self._dirty = True
  23                 self.schedule()
  24
  25         def addFront(self, task):
  26                 self._task_queue.appendleft(task)
  27                 self._dirty = True
  28                 self.schedule()
  29
  30         def schedule(self):
  31
  32                 if not self._dirty:
  33                         return False
  34
  35                 if not self:
  36                         return False
  37
  38                 if self._scheduling:
  39                         # Ignore any recursive schedule() calls triggered via
  40                         # self._task_exit().
  41                         return False
  42
  43                 self._scheduling = True
  44
  45                 task_queue = self._task_queue
  46                 running_tasks = self.running_tasks
  47                 max_jobs = self.max_jobs
  48                 state_changed = False
  49
  50                 while task_queue and \
  51                         (max_jobs is True or len(running_tasks) < max_jobs):
  52                         task = task_queue.popleft()
  53                         cancelled = getattr(task, "cancelled", None)
  54                         if not cancelled:
  55                                 running_tasks.add(task)
  56                                 task.addExitListener(self._task_exit)
  57                                 task.start()
  58                         state_changed = True
  59
  60                 self._dirty = False
  61                 self._scheduling = False
  62
  63                 return state_changed
  64
  65         def _task_exit(self, task):
  66                 """
  67                 Since we can always rely on exit listeners being called, the set of
  68                 running tasks is always pruned automatically and there is never any need
  69                 to actively prune it.
  70                 """
  71                 self.running_tasks.remove(task)
  72                 if self._task_queue:
  73                         self._dirty = True
  74                         self.schedule()
  75
  76         def clear(self):
  77                 self._task_queue.clear()
  78                 running_tasks = self.running_tasks
  79                 while running_tasks:
  80                         task = running_tasks.pop()
  81                         task.removeExitListener(self._task_exit)
  82                         task.cancel()
  83                 self._dirty = False
  84
  85         def __bool__(self):
  86                 return bool(self._task_queue or self.running_tasks)
  87
  88         if sys.hexversion < 0x3000000:
  89                 __nonzero__ = __bool__
  90
  91         def __len__(self):
  92                 return len(self._task_queue) + len(self.running_tasks)