Merge pull request #3520 from whalebot-helmsman/round-robin-scheduler-tested

[MRG+1] Downloader-aware Priority Queue for Scrapy

docs/topics/broad-crawls.rst

@@ -39,6 +39,17 @@ you need to keep in mind when using Scrapy for doing broad crawls, along with
 concrete suggestions of Scrapy settings to tune in order to achieve an
 efficient broad crawl.
 
+Use the right :setting:`SCHEDULER_PRIORITY_QUEUE`
+=================================================
+
+Scrapy’s default scheduler priority queue is ``'scrapy.pqueues.ScrapyPriorityQueue'``.
+It works best during single-domain crawl. It does not work well with crawling
+many different domains in parallel
+
+To apply the recommended priority queue use::
+
+    SCHEDULER_PRIORITY_QUEUE = 'scrapy.pqueues.DownloaderAwarePriorityQueue'
+
 Increase concurrency
 ====================
 
@@ -85,7 +96,7 @@ When doing broad crawls you are often only interested in the crawl rates you
 get and any errors found. These stats are reported by Scrapy when using the
 ``INFO`` log level. In order to save CPU (and log storage requirements) you
 should not use ``DEBUG`` log level when preforming large broad crawls in
-production. Using ``DEBUG`` level when developing your (broad) crawler may be 
+production. Using ``DEBUG`` level when developing your (broad) crawler may be
 fine though.
 
 To set the log level use::

docs/topics/settings.rst

@@ -1155,9 +1155,14 @@ Type of in-memory queue used by scheduler. Other available type is:
 
 SCHEDULER_PRIORITY_QUEUE
 ------------------------
-Default: ``'queuelib.PriorityQueue'``
+Default: ``'scrapy.pqueues.ScrapyPriorityQueue'``
 
-Type of priority queue used by scheduler.
+Type of priority queue used by the scheduler. Another available type is
+``scrapy.pqueues.DownloaderAwarePriorityQueue``.
+``scrapy.pqueues.DownloaderAwarePriorityQueue`` works better than
+``scrapy.pqueues.ScrapyPriorityQueue`` when you crawl many different
+domains in parallel. But currently ``scrapy.pqueues.DownloaderAwarePriorityQueue``
+does not work together with :setting:`CONCURRENT_REQUESTS_PER_IP`.
 
 .. setting:: SPIDER_CONTRACTS
 

scrapy/core/downloader/__init__.py

@@ -75,6 +75,8 @@ def _get_concurrency_delay(concurrency, spider, settings):
 
 class Downloader(object):
 
+    DOWNLOAD_SLOT = 'download_slot'
+
     def __init__(self, crawler):
         self.settings = crawler.settings
         self.signals = crawler.signals
@@ -111,8 +113,8 @@ class Downloader(object):
         return key, self.slots[key]
 
     def _get_slot_key(self, request, spider):
-        if 'download_slot' in request.meta:
-            return request.meta['download_slot']
+        if self.DOWNLOAD_SLOT in request.meta:
+            return request.meta[self.DOWNLOAD_SLOT]
 
         key = urlparse_cached(request).hostname or ''
         if self.ip_concurrency:
@@ -122,7 +124,7 @@ class Downloader(object):
 
     def _enqueue_request(self, request, spider):
         key, slot = self._get_slot(request, spider)
-        request.meta['download_slot'] = key
+        request.meta[self.DOWNLOAD_SLOT] = key
 
         def _deactivate(response):
             slot.active.remove(request)

scrapy/core/scheduler.py

@@ -1,19 +1,46 @@
 import os
 import json
 import logging
+import warnings
 from os.path import join, exists
 
-from scrapy.utils.reqser import request_to_dict, request_from_dict
+from queuelib import PriorityQueue
+
 from scrapy.utils.misc import load_object, create_instance
 from scrapy.utils.job import job_dir
+from scrapy.utils.deprecate import ScrapyDeprecationWarning
+
 
 logger = logging.getLogger(__name__)
 
 
 class Scheduler(object):
+    """
+    Scrapy Scheduler. It allows to enqueue requests and then get
+    a next request to download. Scheduler is also handling duplication
+    filtering, via dupefilter.
 
+    Prioritization and queueing is not performed by the Scheduler.
+    User sets ``priority`` field for each Request, and a PriorityQueue
+    (defined by :setting:`SCHEDULER_PRIORITY_QUEUE`) uses these priorities
+    to dequeue requests in a desired order.
+
+    Scheduler uses two PriorityQueue instances, configured to work in-memory
+    and on-disk (optional). When on-disk queue is present, it is used by
+    default, and an in-memory queue is used as a fallback for cases where
+    a disk queue can't handle a request (can't serialize it).
+
+    :setting:`SCHEDULER_MEMORY_QUEUE` and
+    :setting:`SCHEDULER_DISK_QUEUE` allow to specify lower-level queue classes
+    which PriorityQueue instances would be instantiated with, to keep requests
+    on disk and in memory respectively.
+
+    Overall, Scheduler is an object which holds several PriorityQueue instances
+    (in-memory and on-disk) and implements fallback logic for them.
+    Also, it handles dupefilters.
+    """
     def __init__(self, dupefilter, jobdir=None, dqclass=None, mqclass=None,
-                 logunser=False, stats=None, pqclass=None):
+                 logunser=False, stats=None, pqclass=None, crawler=None):
         self.df = dupefilter
         self.dqdir = self._dqdir(jobdir)
         self.pqclass = pqclass
@@ -21,6 +48,7 @@ class Scheduler(object):
         self.mqclass = mqclass
         self.logunser = logunser
         self.stats = stats
+        self.crawler = crawler
 
     @classmethod
     def from_crawler(cls, crawler):
@@ -28,26 +56,35 @@ class Scheduler(object):
         dupefilter_cls = load_object(settings['DUPEFILTER_CLASS'])
         dupefilter = create_instance(dupefilter_cls, settings, crawler)
         pqclass = load_object(settings['SCHEDULER_PRIORITY_QUEUE'])
+        if pqclass is PriorityQueue:
+            warnings.warn("SCHEDULER_PRIORITY_QUEUE='queuelib.PriorityQueue'"
+                          " is no longer supported because of API changes; "
+                          "please use 'scrapy.pqueues.ScrapyPriorityQueue'",
+                          ScrapyDeprecationWarning)
+            from scrapy.pqueues import ScrapyPriorityQueue
+            pqclass = ScrapyPriorityQueue
+
         dqclass = load_object(settings['SCHEDULER_DISK_QUEUE'])
         mqclass = load_object(settings['SCHEDULER_MEMORY_QUEUE'])
-        logunser = settings.getbool('LOG_UNSERIALIZABLE_REQUESTS', settings.getbool('SCHEDULER_DEBUG'))
+        logunser = settings.getbool('LOG_UNSERIALIZABLE_REQUESTS',
+                                    settings.getbool('SCHEDULER_DEBUG'))
         return cls(dupefilter, jobdir=job_dir(settings), logunser=logunser,
-                   stats=crawler.stats, pqclass=pqclass, dqclass=dqclass, mqclass=mqclass)
+                   stats=crawler.stats, pqclass=pqclass, dqclass=dqclass,
+                   mqclass=mqclass, crawler=crawler)
 
     def has_pending_requests(self):
         return len(self) > 0
 
     def open(self, spider):
         self.spider = spider
-        self.mqs = self.pqclass(self._newmq)
+        self.mqs = self._mq()
         self.dqs = self._dq() if self.dqdir else None
         return self.df.open()
 
     def close(self, reason):
         if self.dqs:
-            prios = self.dqs.close()
-            with open(join(self.dqdir, 'active.json'), 'w') as f:
-                json.dump(prios, f)
+            state = self.dqs.close()
+            self._write_dqs_state(self.dqdir, state)
         return self.df.close(reason)
 
     def enqueue_request(self, request):
@@ -82,8 +119,7 @@ class Scheduler(object):
         if self.dqs is None:
             return
         try:
-            reqd = request_to_dict(request, self.spider)
-            self.dqs.push(reqd, -request.priority)
+            self.dqs.push(request, -request.priority)
         except ValueError as e:  # non serializable request
             if self.logunser:
                 msg = ("Unable to serialize request: %(request)s - reason:"
@@ -103,32 +139,51 @@ class Scheduler(object):
 
     def _dqpop(self):
         if self.dqs:
-            d = self.dqs.pop()
-            if d:
-                return request_from_dict(d, self.spider)
+            return self.dqs.pop()
 
     def _newmq(self, priority):
+        """ Factory for creating memory queues. """
         return self.mqclass()
 
     def _newdq(self, priority):
-        return self.dqclass(join(self.dqdir, 'p%s' % priority))
+        """ Factory for creating disk queues. """
+        path = join(self.dqdir, 'p%s' % (priority, ))
+        return self.dqclass(path)
+
+    def _mq(self):
+        """ Create a new priority queue instance, with in-memory storage """
+        return create_instance(self.pqclass, None, self.crawler, self._newmq,
+                               serialize=False)
 
     def _dq(self):
-        activef = join(self.dqdir, 'active.json')
-        if exists(activef):
-            with open(activef) as f:
-                prios = json.load(f)
-        else:
-            prios = ()
-        q = self.pqclass(self._newdq, startprios=prios)
+        """ Create a new priority queue instance, with disk storage """
+        state = self._read_dqs_state(self.dqdir)
+        q = create_instance(self.pqclass,
+                            None,
+                            self.crawler,
+                            self._newdq,
+                            state,
+                            serialize=True)
         if q:
             logger.info("Resuming crawl (%(queuesize)d requests scheduled)",
                         {'queuesize': len(q)}, extra={'spider': self.spider})
         return q
 
     def _dqdir(self, jobdir):
+        """ Return a folder name to keep disk queue state at """
         if jobdir:
             dqdir = join(jobdir, 'requests.queue')
             if not exists(dqdir):
                 os.makedirs(dqdir)
             return dqdir
+
+    def _read_dqs_state(self, dqdir):
+        path = join(dqdir, 'active.json')
+        if not exists(path):
+            return ()
+        with open(path) as f:
+            return json.load(f)
+
+    def _write_dqs_state(self, dqdir, state):
+        with open(join(dqdir, 'active.json'), 'w') as f:
+            json.dump(state, f)

scrapy/pqueues.py

@@ -0,0 +1,193 @@
+import hashlib
+import logging
+from collections import namedtuple
+
+from queuelib import PriorityQueue
+
+from scrapy.utils.reqser import request_to_dict, request_from_dict
+
+
+logger = logging.getLogger(__name__)
+
+
+def _path_safe(text):
+    """
+    Return a filesystem-safe version of a string ``text``
+
+    >>> _path_safe('simple.org').startswith('simple.org')
+    True
+    >>> _path_safe('dash-underscore_.org').startswith('dash-underscore_.org')
+    True
+    >>> _path_safe('some@symbol?').startswith('some_symbol_')
+    True
+    """
+    pathable_slot = "".join([c if c.isalnum() or c in '-._' else '_'
+                             for c in text])
+    # as we replace some letters we can get collision for different slots
+    # add we add unique part
+    unique_slot = hashlib.md5(text.encode('utf8')).hexdigest()
+    return '-'.join([pathable_slot, unique_slot])
+
+
+class _Priority(namedtuple("_Priority", ["priority", "slot"])):
+    """ Slot-specific priority. It is a hack - ``(priority, slot)`` tuple
+    which can be used instead of int priorities in queues:
+
+    * they are ordered in the same way - order is still by priority value,
+      min(prios) works;
+    * str(p) representation is guaranteed to be different when slots
+      are different - this is important because str(p) is used to create
+      queue files on disk;
+    * they have readable str(p) representation which is safe
+      to use as a file name.
+    """
+    __slots__ = ()
+
+    def __str__(self):
+        return '%s_%s' % (self.priority, _path_safe(str(self.slot)))
+
+
+class _SlotPriorityQueues(object):
+    """ Container for multiple priority queues. """
+    def __init__(self, pqfactory, slot_startprios=None):
+        """
+        ``pqfactory`` is a factory for creating new PriorityQueues.
+        It must be a function which accepts a single optional ``startprios``
+        argument, with a list of priorities to create queues for.
+
+        ``slot_startprios`` is a ``{slot: startprios}`` dict.
+        """
+        self.pqfactory = pqfactory
+        self.pqueues = {}  # slot -> priority queue
+        for slot, startprios in (slot_startprios or {}).items():
+            self.pqueues[slot] = self.pqfactory(startprios)
+
+    def pop_slot(self, slot):
+        """ Pop an object from a priority queue for this slot """
+        queue = self.pqueues[slot]
+        request = queue.pop()
+        if len(queue) == 0:
+            del self.pqueues[slot]
+        return request
+
+    def push_slot(self, slot, obj, priority):
+        """ Push an object to a priority queue for this slot """
+        if slot not in self.pqueues:
+            self.pqueues[slot] = self.pqfactory()
+        queue = self.pqueues[slot]
+        queue.push(obj, priority)
+
+    def close(self):
+        active = {slot: queue.close()
+                  for slot, queue in self.pqueues.items()}
+        self.pqueues.clear()
+        return active
+
+    def __len__(self):
+        return sum(len(x) for x in self.pqueues.values()) if self.pqueues else 0
+
+    def __contains__(self, slot):
+        return slot in self.pqueues
+
+
+class ScrapyPriorityQueue(PriorityQueue):
+    """
+    PriorityQueue which works with scrapy.Request instances and
+    can optionally convert them to/from dicts before/after putting to a queue.
+    """
+    def __init__(self, crawler, qfactory, startprios=(), serialize=False):
+        super(ScrapyPriorityQueue, self).__init__(qfactory, startprios)
+        self.serialize = serialize
+        self.spider = crawler.spider
+
+    @classmethod
+    def from_crawler(cls, crawler, qfactory, startprios=(), serialize=False):
+        return cls(crawler, qfactory, startprios, serialize)
+
+    def push(self, request, priority=0):
+        if self.serialize:
+            request = request_to_dict(request, self.spider)
+        super(ScrapyPriorityQueue, self).push(request, priority)
+
+    def pop(self):
+        request = super(ScrapyPriorityQueue, self).pop()
+        if request and self.serialize:
+            request = request_from_dict(request, self.spider)
+        return request
+
+
+class DownloaderInterface(object):
+
+    def __init__(self, crawler):
+        self.downloader = crawler.engine.downloader
+
+    def stats(self, possible_slots):
+        return [(self._active_downloads(slot), slot)
+                for slot in possible_slots]
+
+    def get_slot_key(self, request):
+        return self.downloader._get_slot_key(request, None)
+
+    def _active_downloads(self, slot):
+        """ Return a number of requests in a Downloader for a given slot """
+        if slot not in self.downloader.slots:
+            return 0
+        return len(self.downloader.slots[slot].active)
+
+
+class DownloaderAwarePriorityQueue(object):
+    """ PriorityQueue which takes Downlaoder activity in account:
+    domains (slots) with the least amount of active downloads are dequeued
+    first.
+    """
+
+    @classmethod
+    def from_crawler(cls, crawler, qfactory, slot_startprios=None, serialize=False):
+        return cls(crawler, qfactory, slot_startprios, serialize)
+
+    def __init__(self, crawler, qfactory, slot_startprios=None, serialize=False):
+        if crawler.settings.getint('CONCURRENT_REQUESTS_PER_IP') != 0:
+            raise ValueError('"%s" does not support CONCURRENT_REQUESTS_PER_IP'
+                             % (self.__class__,))
+
+        if slot_startprios and not isinstance(slot_startprios, dict):
+            raise ValueError("DownloaderAwarePriorityQueue accepts "
+                             "``slot_startprios`` as a dict; %r instance "
+                             "is passed. Most likely, it means the state is"
+                             "created by an incompatible priority queue. "
+                             "Only a crawl started with the same priority "
+                             "queue class can be resumed." %
+                             slot_startprios.__class__)
+
+        slot_startprios = {
+            slot: [_Priority(p, slot) for p in startprios]
+            for slot, startprios in (slot_startprios or {}).items()}
+
+        def pqfactory(startprios=()):
+            return ScrapyPriorityQueue(crawler, qfactory, startprios, serialize)
+        self._slot_pqueues = _SlotPriorityQueues(pqfactory, slot_startprios)
+        self.serialize = serialize
+        self._downloader_interface = DownloaderInterface(crawler)
+
+    def pop(self):
+        stats = self._downloader_interface.stats(self._slot_pqueues.pqueues)
+
+        if not stats:
+            return
+
+        slot = min(stats)[1]
+        request = self._slot_pqueues.pop_slot(slot)
+        return request
+
+    def push(self, request, priority):
+        slot = self._downloader_interface.get_slot_key(request)
+        priority_slot = _Priority(priority=priority, slot=slot)
+        self._slot_pqueues.push_slot(slot, request, priority_slot)
+
+    def close(self):
+        active = self._slot_pqueues.close()
+        return {slot: [p.priority for p in startprios]
+                for slot, startprios in active.items()}
+
+    def __len__(self):
+        return len(self._slot_pqueues)

scrapy/settings/default_settings.py

@@ -246,7 +246,7 @@ ROBOTSTXT_OBEY = False
 SCHEDULER = 'scrapy.core.scheduler.Scheduler'
 SCHEDULER_DISK_QUEUE = 'scrapy.squeues.PickleLifoDiskQueue'
 SCHEDULER_MEMORY_QUEUE = 'scrapy.squeues.LifoMemoryQueue'
-SCHEDULER_PRIORITY_QUEUE = 'queuelib.PriorityQueue'
+SCHEDULER_PRIORITY_QUEUE = 'scrapy.pqueues.ScrapyPriorityQueue'
 
 SPIDER_LOADER_CLASS = 'scrapy.spiderloader.SpiderLoader'
 SPIDER_LOADER_WARN_ONLY = False

scrapy/squeues.py

@@ -7,6 +7,7 @@ from six.moves import cPickle as pickle
 
 from queuelib import queue
 
+
 def _serializable_queue(queue_class, serialize, deserialize):
 
     class SerializableQueue(queue_class):
@@ -22,6 +23,7 @@ def _serializable_queue(queue_class, serialize, deserialize):
 
     return SerializableQueue
 
+
 def _pickle_serialize(obj):
     try:
         return pickle.dumps(obj, protocol=2)
@@ -31,13 +33,14 @@ def _pickle_serialize(obj):
     except (pickle.PicklingError, AttributeError, TypeError) as e:
         raise ValueError(str(e))
 
-PickleFifoDiskQueue = _serializable_queue(queue.FifoDiskQueue, \
+
+PickleFifoDiskQueue = _serializable_queue(queue.FifoDiskQueue,
     _pickle_serialize, pickle.loads)
-PickleLifoDiskQueue = _serializable_queue(queue.LifoDiskQueue, \
+PickleLifoDiskQueue = _serializable_queue(queue.LifoDiskQueue,
     _pickle_serialize, pickle.loads)
-MarshalFifoDiskQueue = _serializable_queue(queue.FifoDiskQueue, \
+MarshalFifoDiskQueue = _serializable_queue(queue.FifoDiskQueue,
     marshal.dumps, marshal.loads)
-MarshalLifoDiskQueue = _serializable_queue(queue.LifoDiskQueue, \
+MarshalLifoDiskQueue = _serializable_queue(queue.LifoDiskQueue,
     marshal.dumps, marshal.loads)
 FifoMemoryQueue = queue.FifoMemoryQueue
 LifoMemoryQueue = queue.LifoMemoryQueue

tests/test_crawler.py

@@ -1,5 +1,4 @@
 import logging
-import tempfile
 import warnings
 
 from twisted.internet import defer
@@ -38,7 +37,11 @@ class CrawlerTestCase(BaseCrawlerTest):
             self.assertIsInstance(spiders, sl_cls)
 
             self.crawler.spiders
-            self.assertEqual(len(w), 1, "Warn deprecated access only once")
+            is_one_warning = len(w) == 1
+            if not is_one_warning:
+                for warning in w:
+                    print(warning)
+            self.assertTrue(is_one_warning, "Warn deprecated access only once")
 
     def test_populate_spidercls_settings(self):
         spider_settings = {'TEST1': 'spider', 'TEST2': 'spider'}
@@ -179,8 +182,12 @@ class CrawlerRunnerTestCase(BaseCrawlerTest):
                 'SPIDER_MANAGER_CLASS': 'tests.test_crawler.CustomSpiderLoader'
             })
             self.assertIsInstance(runner.spider_loader, CustomSpiderLoader)
-            self.assertEqual(len(w), 1)
+            is_one_warning = len(w) == 1
+            if not is_one_warning:
+                for warning in w:
+                    print(warning)
             self.assertIn('Please use SPIDER_LOADER_CLASS', str(w[0].message))
+            self.assertTrue(is_one_warning)
 
     def test_crawl_rejects_spider_objects(self):
         with raises(ValueError):

tests/test_scheduler.py

@@ -0,0 +1,342 @@
+import shutil
+import tempfile
+import unittest
+import collections
+
+from twisted.internet import defer
+from twisted.trial.unittest import TestCase
+
+from scrapy.crawler import Crawler
+from scrapy.core.downloader import Downloader
+from scrapy.core.scheduler import Scheduler
+from scrapy.http import Request
+from scrapy.spiders import Spider
+from scrapy.utils.httpobj import urlparse_cached
+from scrapy.utils.test import get_crawler
+from tests.mockserver import MockServer
+
+
+MockEngine = collections.namedtuple('MockEngine', ['downloader'])
+MockSlot = collections.namedtuple('MockSlot', ['active'])
+
+
+class MockDownloader(object):
+    def __init__(self):
+        self.slots = dict()
+
+    def _get_slot_key(self, request, spider):
+        if Downloader.DOWNLOAD_SLOT in request.meta:
+            return request.meta[Downloader.DOWNLOAD_SLOT]
+
+        return urlparse_cached(request).hostname or ''
+
+    def increment(self, slot_key):
+        slot = self.slots.setdefault(slot_key, MockSlot(active=list()))
+        slot.active.append(1)
+
+    def decrement(self, slot_key):
+        slot = self.slots.get(slot_key)
+        slot.active.pop()
+
+    def close(self):
+        pass
+
+
+class MockCrawler(Crawler):
+    def __init__(self, priority_queue_cls, jobdir):
+
+        settings = dict(
+                LOG_UNSERIALIZABLE_REQUESTS=False,
+                SCHEDULER_DISK_QUEUE='scrapy.squeues.PickleLifoDiskQueue',
+                SCHEDULER_MEMORY_QUEUE='scrapy.squeues.LifoMemoryQueue',
+                SCHEDULER_PRIORITY_QUEUE=priority_queue_cls,
+                JOBDIR=jobdir,
+                DUPEFILTER_CLASS='scrapy.dupefilters.BaseDupeFilter'
+                )
+        super(MockCrawler, self).__init__(Spider, settings)
+        self.engine = MockEngine(downloader=MockDownloader())
+
+
+class SchedulerHandler(object):
+    priority_queue_cls = None
+    jobdir = None
+
+    def create_scheduler(self):
+        self.mock_crawler = MockCrawler(self.priority_queue_cls, self.jobdir)
+        self.scheduler = Scheduler.from_crawler(self.mock_crawler)
+        self.spider = Spider(name='spider')
+        self.scheduler.open(self.spider)
+
+    def close_scheduler(self):
+        self.scheduler.close('finished')
+        self.mock_crawler.stop()
+        self.mock_crawler.engine.downloader.close()
+
+    def setUp(self):
+        self.create_scheduler()
+
+    def tearDown(self):
+        self.close_scheduler()
+
+
+_PRIORITIES = [("http://foo.com/a", -2),
+               ("http://foo.com/d", 1),
+               ("http://foo.com/b", -1),
+               ("http://foo.com/c", 0),
+               ("http://foo.com/e", 2)]
+
+
+_URLS = {"http://foo.com/a", "http://foo.com/b", "http://foo.com/c"}
+
+
+class BaseSchedulerInMemoryTester(SchedulerHandler):
+    def test_length(self):
+        self.assertFalse(self.scheduler.has_pending_requests())
+        self.assertEqual(len(self.scheduler), 0)
+
+        for url in _URLS:
+            self.scheduler.enqueue_request(Request(url))
+
+        self.assertTrue(self.scheduler.has_pending_requests())
+        self.assertEqual(len(self.scheduler), len(_URLS))
+
+    def test_dequeue(self):
+        for url in _URLS:
+            self.scheduler.enqueue_request(Request(url))
+
+        urls = set()
+        while self.scheduler.has_pending_requests():
+            urls.add(self.scheduler.next_request().url)
+
+        self.assertEqual(urls, _URLS)
+
+    def test_dequeue_priorities(self):
+        for url, priority in _PRIORITIES:
+            self.scheduler.enqueue_request(Request(url, priority=priority))
+
+        priorities = list()
+        while self.scheduler.has_pending_requests():
+            priorities.append(self.scheduler.next_request().priority)
+
+        self.assertEqual(priorities,
+                         sorted([x[1] for x in _PRIORITIES], key=lambda x: -x))
+
+
+class BaseSchedulerOnDiskTester(SchedulerHandler):
+
+    def setUp(self):
+        self.jobdir = tempfile.mkdtemp()
+        self.create_scheduler()
+
+    def tearDown(self):
+        self.close_scheduler()
+
+        shutil.rmtree(self.jobdir)
+        self.jobdir = None
+
+    def test_length(self):
+        self.assertFalse(self.scheduler.has_pending_requests())
+        self.assertEqual(len(self.scheduler), 0)
+
+        for url in _URLS:
+            self.scheduler.enqueue_request(Request(url))
+
+        self.close_scheduler()
+        self.create_scheduler()
+
+        self.assertTrue(self.scheduler.has_pending_requests())
+        self.assertEqual(len(self.scheduler), len(_URLS))
+
+    def test_dequeue(self):
+        for url in _URLS:
+            self.scheduler.enqueue_request(Request(url))
+
+        self.close_scheduler()
+        self.create_scheduler()
+
+        urls = set()
+        while self.scheduler.has_pending_requests():
+            urls.add(self.scheduler.next_request().url)
+
+        self.assertEqual(urls, _URLS)
+
+    def test_dequeue_priorities(self):
+        for url, priority in _PRIORITIES:
+            self.scheduler.enqueue_request(Request(url, priority=priority))
+
+        self.close_scheduler()
+        self.create_scheduler()
+
+        priorities = list()
+        while self.scheduler.has_pending_requests():
+            priorities.append(self.scheduler.next_request().priority)
+
+        self.assertEqual(priorities,
+                         sorted([x[1] for x in _PRIORITIES], key=lambda x: -x))
+
+
+class TestSchedulerInMemory(BaseSchedulerInMemoryTester, unittest.TestCase):
+    priority_queue_cls = 'scrapy.pqueues.ScrapyPriorityQueue'
+
+
+class TestSchedulerOnDisk(BaseSchedulerOnDiskTester, unittest.TestCase):
+    priority_queue_cls = 'scrapy.pqueues.ScrapyPriorityQueue'
+
+
+_URLS_WITH_SLOTS = [("http://foo.com/a", 'a'),
+                    ("http://foo.com/b", 'a'),
+                    ("http://foo.com/c", 'b'),
+                    ("http://foo.com/d", 'b'),
+                    ("http://foo.com/e", 'c'),
+                    ("http://foo.com/f", 'c')]
+
+
+class TestMigration(unittest.TestCase):
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir)
+
+    def _migration(self, tmp_dir):
+        prev_scheduler_handler = SchedulerHandler()
+        prev_scheduler_handler.priority_queue_cls = 'scrapy.pqueues.ScrapyPriorityQueue'
+        prev_scheduler_handler.jobdir = tmp_dir
+
+        prev_scheduler_handler.create_scheduler()
+        for url in _URLS:
+            prev_scheduler_handler.scheduler.enqueue_request(Request(url))
+        prev_scheduler_handler.close_scheduler()
+
+        next_scheduler_handler = SchedulerHandler()
+        next_scheduler_handler.priority_queue_cls = 'scrapy.pqueues.DownloaderAwarePriorityQueue'
+        next_scheduler_handler.jobdir = tmp_dir
+
+        next_scheduler_handler.create_scheduler()
+
+    def test_migration(self):
+        with self.assertRaises(ValueError):
+            self._migration(self.tmpdir)
+
+
+def _is_scheduling_fair(enqueued_slots, dequeued_slots):
+    """
+    We enqueued same number of requests for every slot.
+    Assert correct order, e.g.
+
+    >>> enqueued = ['a', 'b', 'c'] * 2
+    >>> correct = ['a', 'c', 'b', 'b', 'a', 'c']
+    >>> incorrect = ['a', 'a', 'b', 'c', 'c', 'b']
+    >>> _is_scheduling_fair(enqueued, correct)
+    True
+    >>> _is_scheduling_fair(enqueued, incorrect)
+    False
+    """
+    if len(dequeued_slots) != len(enqueued_slots):
+        return False
+
+    slots_number = len(set(enqueued_slots))
+    for i in range(0, len(dequeued_slots), slots_number):
+        part = dequeued_slots[i:i + slots_number]
+        if len(part) != len(set(part)):
+            return False
+
+    return True
+
+
+class DownloaderAwareSchedulerTestMixin(object):
+    priority_queue_cls = 'scrapy.pqueues.DownloaderAwarePriorityQueue'
+    reopen = False
+
+    def test_logic(self):
+        for url, slot in _URLS_WITH_SLOTS:
+            request = Request(url)
+            request.meta[Downloader.DOWNLOAD_SLOT] = slot
+            self.scheduler.enqueue_request(request)
+
+        if self.reopen:
+            self.close_scheduler()
+            self.create_scheduler()
+
+        dequeued_slots = list()
+        requests = []
+        downloader = self.mock_crawler.engine.downloader
+        while self.scheduler.has_pending_requests():
+            request = self.scheduler.next_request()
+            # pylint: disable=protected-access
+            slot = downloader._get_slot_key(request, None)
+            dequeued_slots.append(slot)
+            downloader.increment(slot)
+            requests.append(request)
+
+        for request in requests:
+            # pylint: disable=protected-access
+            slot = downloader._get_slot_key(request, None)
+            downloader.decrement(slot)
+
+        self.assertTrue(_is_scheduling_fair(list(s for u, s in _URLS_WITH_SLOTS),
+                                            dequeued_slots))
+        self.assertEqual(sum(len(s.active) for s in downloader.slots.values()), 0)
+
+
+class TestSchedulerWithDownloaderAwareInMemory(DownloaderAwareSchedulerTestMixin,
+                                               BaseSchedulerInMemoryTester,
+                                               unittest.TestCase):
+    pass
+
+
+class TestSchedulerWithDownloaderAwareOnDisk(DownloaderAwareSchedulerTestMixin,
+                                             BaseSchedulerOnDiskTester,
+                                             unittest.TestCase):
+    reopen = True
+
+
+class StartUrlsSpider(Spider):
+
+    def __init__(self, start_urls):
+        self.start_urls = start_urls
+        super(StartUrlsSpider, self).__init__(start_urls)
+
+    def parse(self, response):
+        pass
+
+
+class TestIntegrationWithDownloaderAwareInMemory(TestCase):
+    def setUp(self):
+        self.crawler = get_crawler(
+                    StartUrlsSpider,
+                    {'SCHEDULER_PRIORITY_QUEUE': 'scrapy.pqueues.DownloaderAwarePriorityQueue',
+                     'DUPEFILTER_CLASS': 'scrapy.dupefilters.BaseDupeFilter'}
+                    )
+
+    @defer.inlineCallbacks
+    def tearDown(self):
+        yield self.crawler.stop()
+
+    @defer.inlineCallbacks
+    def test_integration_downloader_aware_priority_queue(self):
+        with MockServer() as mockserver:
+
+            url = mockserver.url("/status?n=200", is_secure=False)
+            start_urls = [url] * 6
+            yield self.crawler.crawl(start_urls)
+            self.assertEqual(self.crawler.stats.get_value('downloader/response_count'),
+                             len(start_urls))
+
+
+class TestIncompatibility(unittest.TestCase):
+
+    def _incompatible(self):
+        settings = dict(
+                SCHEDULER_PRIORITY_QUEUE='scrapy.pqueues.DownloaderAwarePriorityQueue',
+                CONCURRENT_REQUESTS_PER_IP=1
+                )
+        crawler = Crawler(Spider, settings)
+        scheduler = Scheduler.from_crawler(crawler)
+        spider = Spider(name='spider')
+        scheduler.open(spider)
+
+    def test_incompatibility(self):
+        with self.assertRaises(ValueError):
+            self._incompatible()