optimise scheluler dynamic select limit and improve task queue (#796)

* optimise scheduler select-limit and task queue

* fix test case in python2.6

* fix: time priority queue only compare exetime

* update:add test case for time priority queue

* optimise: add globally auto increasing value for task to keep priority queue in order

Author: plusplus1

pyspider/scheduler/scheduler.py

@@ -478,24 +478,51 @@ def _check_select(self):
         cnt = 0
         cnt_dict = dict()
         limit = self.LOOP_LIMIT
-        for project in itervalues(self.projects):
+
+        # dynamic assign select limit for each project, use qsize as weight
+        project_weights, total_weight = dict(), 0
+        for project in itervalues(self.projects):  # type:Project
             if not project.active:
                 continue
             # only check project pause when select new tasks, cronjob and new request still working
             if project.paused:
                 continue
             if project.waiting_get_info:
                 continue
+
+            # task queue
+            task_queue = project.task_queue  # type:TaskQueue
+            pro_weight = task_queue.size()
+            total_weight += pro_weight
+            project_weights[project.name] = pro_weight
+            pass
+
+        min_project_limit = int(limit / 10.)  # ensure minimum select limit for each project
+        max_project_limit = int(limit / 3.0)  # ensure maximum select limit for each project
+
+        for pro_name, pro_weight in iteritems(project_weights):
             if cnt >= limit:
                 break
 
+            project = self.projects[pro_name]  # type:Project
+
             # task queue
             task_queue = project.task_queue
             task_queue.check_update()
             project_cnt = 0
 
+            # calculate select limit for project
+            if total_weight < 1 or pro_weight < 1:
+                project_limit = min_project_limit
+            else:
+                project_limit = int((1.0 * pro_weight / total_weight) * limit)
+                if project_limit < min_project_limit:
+                    project_limit = min_project_limit
+                elif project_limit > max_project_limit:
+                    project_limit = max_project_limit
+
             # check send_buffer here. when not empty, out_queue may blocked. Not sending tasks
-            while cnt < limit and project_cnt < limit / 10:
+            while cnt < limit and project_cnt < project_limit:
                 taskid = task_queue.get()
                 if not taskid:
                     break

pyspider/scheduler/task_queue.py

@@ -5,10 +5,11 @@
 #         http://binux.me
 # Created on 2014-02-07 13:12:10
 
-import time
 import heapq
 import logging
 import threading
+import time
+
 try:
     from UserDict import DictMixin
 except ImportError:
@@ -24,8 +25,21 @@
     cmp = lambda x, y: (x > y) - (x < y)
 
 
+class AtomInt(object):
+    __value__ = 0
+    __mutex__ = threading.RLock()
+
+    @classmethod
+    def get_value(cls):
+        cls.__mutex__.acquire()
+        cls.__value__ = cls.__value__ + 1
+        value = cls.__value__
+        cls.__mutex__.release()
+        return value
+
+
 class InQueueTask(DictMixin):
-    __slots__ = ('taskid', 'priority', 'exetime')
+    __slots__ = ('taskid', 'priority', 'exetime', 'sequence')
     __getitem__ = lambda *x: getattr(*x)
     __setitem__ = lambda *x: setattr(*x)
     __iter__ = lambda self: iter(self.__slots__)
@@ -36,19 +50,23 @@ def __init__(self, taskid, priority=0, exetime=0):
         self.taskid = taskid
         self.priority = priority
         self.exetime = exetime
+        self.sequence = AtomInt.get_value()
 
     def __cmp__(self, other):
         if self.exetime == 0 and other.exetime == 0:
-            return -cmp(self.priority, other.priority)
+            diff = -cmp(self.priority, other.priority)
         else:
-            return cmp(self.exetime, other.exetime)
+            diff = cmp(self.exetime, other.exetime)
+
+        # compare in-queue sequence number finally if two element has the same
+        # priority or exetime
+        return diff if diff != 0 else cmp(self.sequence, other.sequence)
 
     def __lt__(self, other):
         return self.__cmp__(other) < 0
 
 
 class PriorityTaskQueue(Queue.Queue):
-
     '''
     TaskQueue
 
@@ -66,12 +84,10 @@ def _put(self, item, heappush=heapq.heappush):
         if item.taskid in self.queue_dict:
             task = self.queue_dict[item.taskid]
             changed = False
-            if item.priority > task.priority:
-                task.priority = item.priority
-                changed = True
-            if item.exetime < task.exetime:
-                task.exetime = item.exetime
+            if item < task:
                 changed = True
+            task.priority = max(item.priority, task.priority)
+            task.exetime = min(item.exetime, task.exetime)
             if changed:
                 self._resort()
         else:
@@ -113,7 +129,6 @@ def __delitem__(self, taskid):
 
 
 class TaskQueue(object):
-
     '''
     task queue for scheduler, have a priority queue and a time queue for delayed tasks
     '''
@@ -155,7 +170,7 @@ def _check_time_queue(self):
         now = time.time()
         self.mutex.acquire()
         while self.time_queue.qsize() and self.time_queue.top and self.time_queue.top.exetime < now:
-            task = self.time_queue.get_nowait()
+            task = self.time_queue.get_nowait()  # type: InQueueTask
             task.exetime = 0
             self.priority_queue.put(task)
         self.mutex.release()
@@ -173,9 +188,24 @@ def _check_processing(self):
         self.mutex.release()
 
     def put(self, taskid, priority=0, exetime=0):
-        '''Put a task into task queue'''
+        """
+        Put a task into task queue
+        
+        when use heap sort, if we put tasks(with the same priority and exetime=0) into queue,
+        the queue is not a strict FIFO queue, but more like a FILO stack.
+        It is very possible that when there are continuous big flow, the speed of select is 
+        slower than request, resulting in priority-queue accumulation in short time.
+        In this scenario, the tasks more earlier entering the priority-queue will not get 
+        processed until the request flow becomes small. 
+        
+        Thus, we store a global atom self increasing value into task.sequence which represent 
+        the task enqueue sequence. When the comparison of exetime and priority have no 
+        difference, we compare task.sequence to ensure that the entire queue is ordered.
+        """
         now = time.time()
+
         task = InQueueTask(taskid, priority, exetime)
+
         self.mutex.acquire()
         if taskid in self.priority_queue:
             self.priority_queue.put(task)
@@ -189,7 +219,9 @@ def put(self, taskid, priority=0, exetime=0):
             if exetime and exetime > now:
                 self.time_queue.put(task)
             else:
+                task.exetime = 0
                 self.priority_queue.put(task)
+
         self.mutex.release()
 
     def get(self):

tests/test_task_queue.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import time
+import unittest
+
+import six
+from six.moves import queue as Queue
+
+from pyspider.scheduler.task_queue import InQueueTask, TaskQueue
+
+
+class TestTaskQueue(unittest.TestCase):
+    """
+        TestTaskQueue
+    """
+
+    def test_task_queue_in_time_order(self):
+        tq = TaskQueue(rate=300, burst=1000)
+
+        queues = dict()
+        tasks = dict()
+
+        for i in range(0, 100):
+            it = InQueueTask(str(i), priority=int(i // 10), exetime=0)
+            tq.put(it.taskid, it.priority, it.exetime)
+
+            if it.priority not in queues:
+                queues[it.priority] = Queue.Queue()
+
+            q = queues[it.priority]  # type:Queue.Queue
+            q.put(it)
+            tasks[it.taskid] = it
+            six.print_('put, taskid=', it.taskid, 'priority=', it.priority, 'exetime=', it.exetime)
+        for i in range(0, 100):
+            task_id = tq.get()
+            task = tasks[task_id]
+            q = queues[task.priority]  # type: Queue.Queue
+            expect_task = q.get()
+            self.assertEqual(task_id, expect_task.taskid)
+            self.assertEqual(task.priority, int(9 - i // 10))
+            six.print_('get, taskid=', task.taskid, 'priority=', task.priority, 'exetime=', task.exetime)
+
+        self.assertEqual(tq.size(), 100)
+        self.assertEqual(tq.priority_queue.qsize(), 0)
+        self.assertEqual(tq.processing.qsize(), 100)
+        for q in six.itervalues(queues):  # type:Queue.Queue
+            self.assertEqual(q.qsize(), 0)
+        pass
+
+    pass
+
+
+class TestTimeQueue(unittest.TestCase):
+    def test_time_queue(self):
+
+        six.print_('Test time queue order by time only')
+
+        tq = TaskQueue(rate=300, burst=1000)
+
+        fifo_queue = Queue.Queue()
+
+        interval = 5.0 / 1000
+
+        for i in range(0, 20):
+            it = InQueueTask(str(i), priority=int(i // 10), exetime=time.time() + (i + 1) * interval)
+            tq.put(it.taskid, it.priority, it.exetime)
+            fifo_queue.put(it)
+            six.print_('put, taskid=', it.taskid, 'priority=', it.priority, 'exetime=', it.exetime)
+
+        self.assertEqual(tq.priority_queue.qsize(), 0)
+        self.assertEqual(tq.processing.qsize(), 0)
+        self.assertEqual(tq.time_queue.qsize(), 20)
+
+        for i in range(0, 20):
+            t1 = fifo_queue.get()
+            t2 = tq.time_queue.get()
+            self.assertEqual(t1.taskid, t2.taskid)
+            six.print_('get, taskid=', t2.taskid, 'priority=', t2.priority, 'exetime=', t2.exetime)
+        self.assertEqual(tq.priority_queue.qsize(), 0)
+        self.assertEqual(tq.processing.qsize(), 0)
+        self.assertEqual(tq.time_queue.qsize(), 0)
+
+        queues = dict()
+        tasks = dict()
+        for i in range(0, 20):
+            priority = int(i // 10)
+            it = InQueueTask(str(i), priority=priority, exetime=time.time() + (i + 1) * interval)
+            tq.put(it.taskid, it.priority, it.exetime)
+            tasks[it.taskid] = it
+
+            if priority not in queues:
+                queues[priority] = Queue.Queue()
+            q = queues[priority]
+            q.put(it)
+            pass
+
+        self.assertEqual(tq.priority_queue.qsize(), 0)
+        self.assertEqual(tq.processing.qsize(), 0)
+        self.assertEqual(tq.time_queue.qsize(), 20)
+
+        time.sleep(20 * interval)
+        tq.check_update()
+        self.assertEqual(tq.priority_queue.qsize(), 20)
+        self.assertEqual(tq.processing.qsize(), 0)
+        self.assertEqual(tq.time_queue.qsize(), 0)
+        for i in range(0, 20):
+            taskid = tq.get()
+            t1 = tasks[taskid]
+            t2 = queues[t1.priority].get()
+            self.assertEqual(t1.taskid, t2.taskid)
+
+        self.assertEqual(tq.priority_queue.qsize(), 0)
+        self.assertEqual(tq.processing.qsize(), 20)
+        self.assertEqual(tq.time_queue.qsize(), 0)
+
+        pass
+
+    pass
+
+
+if __name__ == '__main__':
+    unittest.main()