src/nepi/execution/scheduler.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program is distributed in the hope that it will be useful,
  11 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19
  20 import itertools
  21 import heapq
  22
  23 class TaskStatus:
  24     """ Execution state of the Task
  25     """
  26     NEW = 0
  27     DONE = 1
  28     ERROR = 2
  29
  30 class Task(object):
  31     """ A Task represents an operation to be executed by the
  32     ExperimentController scheduler
  33     """
  34
  35     def __init__(self, timestamp, callback):
  36         """
  37         :param timestamp: Future execution date of the operation
  38         :type timestamp: str
  39
  40         :param callback: A function to invoke in order to execute the operation
  41         :type callback: function
  42
  43         """
  44         self.id = None
  45         self.timestamp = timestamp
  46         self.callback = callback
  47         self.result = None
  48         self.status = TaskStatus.NEW
  49
  50 class HeapScheduler(object):
  51     """ Create a Heap Scheduler
  52
  53     .. note::
  54
  55         This class is thread safe.
  56         All calls to C Extensions are made atomic by the GIL in the CPython implementation.
  57         heapq.heappush, heapq.heappop, and list access are therefore thread-safe.
  58
  59     """
  60
  61     def __init__(self):
  62         super(HeapScheduler, self).__init__()
  63         self._queue = list()
  64         self._valid = set()
  65         self._idgen = itertools.count(1)
  66
  67     @property
  68     def pending(self):
  69         """ Returns the list of pending task ids """
  70         return self._valid
  71
  72     def schedule(self, task):
  73         """ Add a task to the queue ordered by task.timestamp and arrival order
  74
  75         :param task: task to schedule
  76         :type task: task
  77         """
  78         if task.id == None:
  79             task.id = self._idgen.next()
  80
  81         entry = (task.timestamp, task.id, task)
  82         self._valid.add(task.id)
  83         heapq.heappush(self._queue, entry)
  84         return task
  85
  86     def remove(self, tid):
  87         """ Remove a task form the queue
  88
  89         :param tid: Id of the task to be removed
  90         :type tid: int
  91
  92         """
  93         try:
  94             self._valid.remove(tid)
  95         except:
  96             pass
  97
  98     def next(self):
  99         """ Get the next task in the queue by timestamp and arrival order
 100         """
 101         while self._queue:
 102             try:
 103                 timestamp, tid, task = heapq.heappop(self._queue)
 104                 if tid in self._valid:
 105                     self.remove(tid)
 106                     return task
 107             except IndexError:
 108                 # heap empty
 109                 pass
 110         return None
 111