SIO-1930 Implement PrioritizingScheduler

Change-Id: I78b2d6ade32b450f486c59e9be231bb86e797219

setup.py

@@ -19,6 +19,7 @@
         'Twisted>=15.2.1',
         'enum34',  # backport from py3
         'supervisor>=3.3.1',
+        'sortedcontainers',
     ],
 
     setup_requires = [

sio/sioworkersd/scheduler/__init__.py

@@ -4,6 +4,23 @@ class Scheduler(object):
     def __init__(self, manager):
         self.manager = manager
 
+    def __unicode__(self):
+        """Admin-friendly text representation of the queue.
+        Used for debugging and displaying in admin panel."""
+        raise NotImplementedError()
+
+    def updateContest(self, contest_uid, priority, weight):
+        """Update contest prioriy and weight in scheduler memory."""
+        pass
+
+    def addWorker(self, worker_id):
+        """Will be called when a new worker appears."""
+        pass
+
+    def delWorker(self, worker_id):
+        """Will be called when a worker disappears."""
+        pass
+
     def addTask(self, env):
         """Add a new task to queue."""
         raise NotImplementedError()
@@ -12,16 +29,11 @@ def delTask(self, task_id):
         """Will be called when a task is completed or cancelled."""
         raise NotImplementedError()
 
-    def __unicode__(self):
-        """Admin-friendly text representation of the queue.
-        Used for debugging and displaying in admin panel."""
-        raise NotImplementedError()
-
     def schedule(self):
         """Return a list of tasks to be executed now, as a list of pairs
         (task_id, worker_id)."""
         raise NotImplementedError()
 
 
-def get_default_scheduler_class_name():
-    return 'sio.sioworkersd.scheduler.fifo.FIFOScheduler'
+def getDefaultSchedulerClassName():
+    return 'sio.sioworkersd.scheduler.prioritizing.PrioritizingScheduler'

sio/sioworkersd/scheduler/fifo.py

@@ -18,7 +18,7 @@ def __init__(self, wid, wdata):
         self.can_run_cpu_exec = wdata.can_run_cpu_exec
         self.is_running_cpu_exec = wdata.is_running_cpu_exec
 
-    def can_run(self, task):
+    def canRun(self, task):
         if self.is_running_cpu_exec or (self.tasks_count >= self.concurrency):
             return False
         else:
@@ -28,7 +28,7 @@ def can_run(self, task):
                 return True
 
     def assign(self, task):
-        assert self.can_run(task)
+        assert self.canRun(task)
         if task.type == 'cpu':
             self.is_running_cpu_exec = True
         self.tasks_count += 1
@@ -71,7 +71,7 @@ def delTask(self, tid):
     def __unicode__(self):
         return unicode(self.queue)
 
-    def _schedule_queue_with(self, queue, workers):
+    def _scheduleQueueWith(self, queue, workers):
         """Schedule tasks from a queue using given workers.
         """
         result = []
@@ -85,7 +85,7 @@ def _schedule_queue_with(self, queue, workers):
             else:
                 workers_queue = workers['vcpu']
             # Some workers may have changed, skip as many as needed.
-            while workers_queue and not workers_queue[-1].can_run(queue[-1]):
+            while workers_queue and not workers_queue[-1].canRun(queue[-1]):
                 workers_queue.pop()
             if not workers_queue:
                 break
@@ -120,8 +120,8 @@ def schedule(self):
         workers['cpu+vcpu'] = deque(sorted(workers['cpu+vcpu'],
             key=lambda w: w.concurrency, reverse=True))
 
-        result = self._schedule_queue_with(self.queues['cpu+vcpu'], workers)
+        result = self._scheduleQueueWith(self.queues['cpu+vcpu'], workers)
         while workers['vcpu'] and workers['vcpu'][0].can_run_cpu_exec:
             workers['vcpu'].popleft()
-        result += self._schedule_queue_with(self.queues['vcpu'], workers)
+        result += self._scheduleQueueWith(self.queues['vcpu'], workers)
         return result