diff --git a/docs/topics/broad-crawls.rst b/docs/topics/broad-crawls.rst index eb02086dc..b887b98af 100644 --- a/docs/topics/broad-crawls.rst +++ b/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:: diff --git a/docs/topics/settings.rst b/docs/topics/settings.rst index 062c4b2ca..4a5439bfc 100644 --- a/docs/topics/settings.rst +++ b/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 diff --git a/scrapy/core/downloader/__init__.py b/scrapy/core/downloader/__init__.py index 59c3ad074..4695d75f4 100644 --- a/scrapy/core/downloader/__init__.py +++ b/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) diff --git a/scrapy/core/scheduler.py b/scrapy/core/scheduler.py index eb790a67e..975aede0c 100644 --- a/scrapy/core/scheduler.py +++ b/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) diff --git a/scrapy/pqueues.py b/scrapy/pqueues.py new file mode 100644 index 000000000..6ecd1b51a --- /dev/null +++ b/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) diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py index a800d39ab..9986827d8 100644 --- a/scrapy/settings/default_settings.py +++ b/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 diff --git a/scrapy/squeues.py b/scrapy/squeues.py index d2074a457..30cc926e5 100644 --- a/scrapy/squeues.py +++ b/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 diff --git a/tests/test_crawler.py b/tests/test_crawler.py index 37cea3ad3..e811c5757 100644 --- a/tests/test_crawler.py +++ b/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): diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py new file mode 100644 index 000000000..e0e3600e5 --- /dev/null +++ b/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()