From 549725215e79fbd3c7b3590c2a47fc9c0ad30a1b Mon Sep 17 00:00:00 2001 From: Pablo Hoffman Date: Tue, 2 Aug 2011 11:57:55 -0300 Subject: [PATCH] Initial support for a persistent scheduler, to support pausing and resuming crawls. * requests are serialized (using marshal by default) and stored on disk, using one queue per priority * request priorities must be integers now * breadh-first and depth-first crawling orders can now be configured through a new DEPTH_PRIORITY setting (see doc). backwards compatilibty with SCHEDULER_ORDER was kept. * requests that can't be serialized (for example, non serializable callbacks) are always kept in memory queues * adapted crawl spider to work with persitent scheduler --- docs/faq.rst | 7 +- docs/topics/settings.rst | 44 ++++----- docs/topics/spider-middleware.rst | 2 + scrapy/contrib/dupefilter.py | 37 ------- scrapy/contrib/spidermiddleware/depth.py | 21 +++- scrapy/contrib/spiders/crawl.py | 17 ++-- scrapy/core/scheduler.py | 95 ++++++++++++++---- scrapy/dupefilter.py | 46 +++++++++ scrapy/http/request/__init__.py | 3 +- scrapy/settings/default_settings.py | 6 +- scrapy/squeue.py | 15 +++ scrapy/tests/test_dupefilter.py | 32 ++---- scrapy/tests/test_utils_datatypes.py | 73 +------------- scrapy/tests/test_utils_pqueue.py | 74 ++++++++++++++ scrapy/tests/test_utils_queue.py | 121 +++++++++++++++++++++++ scrapy/tests/test_utils_reqser.py | 75 ++++++++++++++ scrapy/utils/datatypes.py | 78 --------------- scrapy/utils/engine.py | 3 +- scrapy/utils/job.py | 7 ++ scrapy/utils/pqueue.py | 58 +++++++++++ scrapy/utils/queue.py | 121 +++++++++++++++++++++++ scrapy/utils/reqser.py | 72 ++++++++++++++ 22 files changed, 738 insertions(+), 269 deletions(-) delete mode 100644 scrapy/contrib/dupefilter.py create mode 100644 scrapy/dupefilter.py create mode 100644 scrapy/squeue.py create mode 100644 scrapy/tests/test_utils_pqueue.py create mode 100644 scrapy/tests/test_utils_queue.py create mode 100644 scrapy/tests/test_utils_reqser.py create mode 100644 scrapy/utils/job.py create mode 100644 scrapy/utils/pqueue.py create mode 100644 scrapy/utils/queue.py create mode 100644 scrapy/utils/reqser.py diff --git a/docs/faq.rst b/docs/faq.rst index 7085a585b..67c5241be 100644 --- a/docs/faq.rst +++ b/docs/faq.rst @@ -84,10 +84,11 @@ How can I simulate a user login in my spider? See :ref:`topics-request-response-ref-request-userlogin`. -Can I crawl in breadth-first order instead of depth-first order? ----------------------------------------------------------------- +Does Scrapy crawl in breath-first or depth-first order? +------------------------------------------------------- -Yes, there's a setting for that: :setting:`SCHEDULER_ORDER`. +It crawls on breath-first order by default, but you can change it to +depth-first order by setting the :setting:`DEPTH_PRIORITY` setting to ``-1``. My Scrapy crawler has memory leaks. What can I do? -------------------------------------------------- diff --git a/docs/topics/settings.rst b/docs/topics/settings.rst index 51bf9286a..4743d9b40 100644 --- a/docs/topics/settings.rst +++ b/docs/topics/settings.rst @@ -331,6 +331,22 @@ Default: ``0`` The maximum depth that will be allowed to crawl for any site. If zero, no limit will be imposed. +.. setting:: DEPTH_PRIORITY + +DEPTH_PRIORITY +-------------- + +Default: ``1`` + +An integer that is used to set the request priority based on request depth. + +To crawl in `breath-first order`_, set :setting:`DEPTH_PRIORITY` to ``1``. + +To crawl in `depth-first order`_, set :setting:`DEPTH_PRIORITY` to ``-1``. + +To disable any priority adjustment based on depth, set +:setting:`DEPTH_PRIORITY` to ``0``. + .. setting:: DEPTH_STATS DEPTH_STATS @@ -468,12 +484,12 @@ The amount of time (in secs) that the downloader will wait before timing out. DUPEFILTER_CLASS ---------------- -Default: ``'scrapy.contrib.dupefilter.RequestFingerprintDupeFilter'`` +Default: ``'scrapy.dupefilter.RFPDupeFilter'`` The class used to detect and filter duplicate requests. -The default (``RequestFingerprintDupeFilter``) filters based on request fingerprint -(using ``scrapy.utils.request.request_fingerprint``) and grouping per domain. +The default (``RFPDupeFilter``) filters based on request fingerprint using +the ``scrapy.utils.request.request_fingerprint`` function. .. setting:: EDITOR @@ -827,26 +843,6 @@ Default: ``'scrapy.core.scheduler.Scheduler'`` The scheduler to use for crawling. -.. setting:: SCHEDULER_ORDER - -SCHEDULER_ORDER ---------------- - -Default: ``'DFO'`` - -Scope: ``scrapy.core.scheduler`` - -The order to use for the crawling scheduler. Available orders are: - -* ``'BFO'``: `Breadth-first order`_ - typically consumes more memory but - reaches most relevant pages earlier. - -* ``'DFO'``: `Depth-first order`_ - typically consumes less memory than - but takes longer to reach most relevant pages. - -.. _Breadth-first order: http://en.wikipedia.org/wiki/Breadth-first_search -.. _Depth-first order: http://en.wikipedia.org/wiki/Depth-first_search - .. setting:: SPIDER_MIDDLEWARES SPIDER_MIDDLEWARES @@ -997,3 +993,5 @@ Default: ``"%s/%s" % (BOT_NAME, BOT_VERSION)`` The default User-Agent to use when crawling, unless overridden. .. _Amazon web services: http://aws.amazon.com/ +.. _breadth-first order: http://en.wikipedia.org/wiki/Breadth-first_search +.. _depth-first order: http://en.wikipedia.org/wiki/Depth-first_search diff --git a/docs/topics/spider-middleware.rst b/docs/topics/spider-middleware.rst index c859ce7c9..66059e6af 100644 --- a/docs/topics/spider-middleware.rst +++ b/docs/topics/spider-middleware.rst @@ -164,6 +164,8 @@ DepthMiddleware * :setting:`DEPTH_LIMIT` - The maximum depth that will be allowed to crawl for any site. If zero, no limit will be imposed. * :setting:`DEPTH_STATS` - Whether to collect depth stats. + * :setting:`DEPTH_PRIORITY` - Whether to prioritize the requests based on + their depth, to crawl in breadh-first or depth-first order. HttpErrorMiddleware ------------------- diff --git a/scrapy/contrib/dupefilter.py b/scrapy/contrib/dupefilter.py deleted file mode 100644 index 7bb449665..000000000 --- a/scrapy/contrib/dupefilter.py +++ /dev/null @@ -1,37 +0,0 @@ -""" -Dupe Filter classes implement a mechanism for filtering duplicate requests. -They must implement the following method: - -* request_seen(request, dont_record=False) - return ``True`` if the request was seen before, or ``False`` otherwise. If - ``dont_record`` is ``True`` the request must not be recorded as seen. - -""" - -from scrapy.utils.request import request_fingerprint - - -class BaseDupeFilter(object): - - @classmethod - def from_settings(cls, settings): - return cls() - - def request_seen(self, request, dont_record=False): - return False - - -class RequestFingerprintDupeFilter(BaseDupeFilter): - """Duplicate filter using scrapy.utils.request.request_fingerprint""" - - def __init__(self): - super(RequestFingerprintDupeFilter, self).__init__() - self.fingerprints = set() - - def request_seen(self, request, dont_record=False): - fp = request_fingerprint(request) - if fp in self.fingerprints: - return True - if not dont_record: - self.fingerprints.add(fp) - return False diff --git a/scrapy/contrib/spidermiddleware/depth.py b/scrapy/contrib/spidermiddleware/depth.py index 8ff52cc44..883d828ee 100644 --- a/scrapy/contrib/spidermiddleware/depth.py +++ b/scrapy/contrib/spidermiddleware/depth.py @@ -4,32 +4,49 @@ Depth Spider Middleware See documentation in docs/topics/spider-middleware.rst """ +import warnings + from scrapy import log from scrapy.http import Request class DepthMiddleware(object): - def __init__(self, maxdepth, stats=None, verbose_stats=False): + def __init__(self, maxdepth, stats=None, verbose_stats=False, prio=1): self.maxdepth = maxdepth self.stats = stats self.verbose_stats = verbose_stats + self.prio = prio @classmethod def from_settings(cls, settings): maxdepth = settings.getint('DEPTH_LIMIT') usestats = settings.getbool('DEPTH_STATS') verbose = settings.getbool('DEPTH_STATS_VERBOSE') + sorder = settings['SCHEDULER_ORDER'] + if sorder: + # XXX: backwards compatibility with old SCHEDULER_ORDER setting + # will be removed on Scrapy 0.15 + warnings.warn("SCHEDULER_ORDER setting is deprecated, " \ + "use DEPTH_PRIORITY instead", DeprecationWarning) + if sorder == 'BFO': + prio = 1 + elif sorder == 'DFO': + prio = -1 + else: + prio = settings.getint('DEPTH_PRIORITY') if usestats: from scrapy.stats import stats else: stats = None - return cls(maxdepth, stats, verbose) + return cls(maxdepth, stats, verbose, prio) def process_spider_output(self, response, result, spider): def _filter(request): if isinstance(request, Request): depth = response.request.meta['depth'] + 1 request.meta['depth'] = depth + if self.prio: + request.priority += depth * self.prio if self.maxdepth and depth > self.maxdepth: log.msg("Ignoring link (depth > %d): %s " % (self.maxdepth, request.url), \ level=log.DEBUG, spider=spider) diff --git a/scrapy/contrib/spiders/crawl.py b/scrapy/contrib/spiders/crawl.py index 0da70919b..30c17e8c3 100644 --- a/scrapy/contrib/spiders/crawl.py +++ b/scrapy/contrib/spiders/crawl.py @@ -6,7 +6,6 @@ See documentation in docs/topics/spiders.rst """ import copy -from functools import partial from scrapy.http import Request from scrapy.utils.spider import iterate_spider_output @@ -38,7 +37,7 @@ class CrawlSpider(BaseSpider): self._compile_rules() def parse(self, response): - return self._response_downloaded(response, self.parse_start_url, cb_kwargs={}, follow=True) + return self._parse_response(response, self.parse_start_url, cb_kwargs={}, follow=True) def parse_start_url(self, response): return [] @@ -48,19 +47,21 @@ class CrawlSpider(BaseSpider): def _requests_to_follow(self, response): seen = set() - for rule in self._rules: + for n, rule in enumerate(self._rules): links = [l for l in rule.link_extractor.extract_links(response) if l not in seen] if links and rule.process_links: links = rule.process_links(links) seen = seen.union(links) for link in links: - callback = partial(self._response_downloaded, callback=rule.callback, \ - cb_kwargs=rule.cb_kwargs, follow=rule.follow) - r = Request(url=link.url, callback=callback) - r.meta['link_text'] = link.text + r = Request(url=link.url, callback='_response_downloaded') + r.meta.update(rule=n, link_text=link.text) yield rule.process_request(r) - def _response_downloaded(self, response, callback, cb_kwargs, follow): + def _response_downloaded(self, response): + rule = self._rules[response.meta['rule']] + return self._parse_response(response, rule.callback, rule.cb_kwargs, rule.follow) + + def _parse_response(self, response, callback, cb_kwargs, follow=True): if callback: cb_res = callback(response, **cb_kwargs) or () cb_res = self.process_results(response, cb_res) diff --git a/scrapy/core/scheduler.py b/scrapy/core/scheduler.py index 1e2741e8c..f660aa44e 100644 --- a/scrapy/core/scheduler.py +++ b/scrapy/core/scheduler.py @@ -1,33 +1,92 @@ -from scrapy.utils.datatypes import PriorityQueue, PriorityStack +from __future__ import with_statement + +from os.path import join, exists + +from scrapy.utils.queue import MemoryQueue +from scrapy.utils.pqueue import PriorityQueue +from scrapy.utils.reqser import request_to_dict, request_from_dict from scrapy.utils.misc import load_object +from scrapy.utils.job import job_dir +from scrapy.utils.py26 import json +from scrapy.stats import stats +from scrapy import log class Scheduler(object): - def __init__(self, dupefilter, dfo=False): - self.dupefilter = dupefilter - Queue = PriorityStack if dfo else PriorityQueue - self.pending_requests = Queue() + def __init__(self, dupefilter, jobdir=None, dqclass=None): + self.df = dupefilter + self.dqdir = join(jobdir, 'requests.queue') if jobdir else None + self.dqclass = dqclass @classmethod def from_settings(cls, settings): - dfo = settings['SCHEDULER_ORDER'].upper() == 'DFO' dupefilter_cls = load_object(settings['DUPEFILTER_CLASS']) dupefilter = dupefilter_cls.from_settings(settings) - return cls(dupefilter, dfo=dfo) + dqclass = load_object(settings['SCHEDULER_DISK_QUEUE']) + return cls(dupefilter, job_dir(settings), dqclass) def has_pending_requests(self): - return bool(self.pending_requests) - - def enqueue_request(self, request): - if request.dont_filter or not self.dupefilter.request_seen(request): - self.pending_requests.push(request, -request.priority) - - def next_request(self): - if self.pending_requests: - return self.pending_requests.pop()[0] + return self.dqs or self.mqs def open(self, spider): - pass + self.spider = spider + self.mqs = PriorityQueue(self._newmq) + self.dqs = self._dq() if self.dqdir else None + return self.df.open() def close(self, reason): - pass + if self.dqs: + prios = self.dqs.close() + with open(join(self.dqdir, 'active.json'), 'w') as f: + json.dump(prios, f) + return self.df.close() + + def enqueue_request(self, request): + if not request.dont_filter and self.df.request_seen(request): + return + if not self._dqpush(request): + self._mqpush(request) + + def next_request(self): + return self.mqs.pop() or self._dqpop() + + def _dqpush(self, request): + if self.dqs is None: + return + try: + reqd = request_to_dict(request, self.spider) + except ValueError: # non serializable request + return + else: + self.dqs.push(reqd, request.priority) + stats.inc_value('scheduler/disk_enqueued', spider=self.spider) + return True + + def _mqpush(self, request): + stats.inc_value('scheduler/memory_enqueued', spider=self.spider) + self.mqs.push(request, request.priority) + + def _dqpop(self): + if self.dqs: + d = self.dqs.pop() + if d: + return request_from_dict(d, self.spider) + + def _newmq(self, priority): + return MemoryQueue() + + def _newdq(self, priority): + return self.dqclass(join(self.dqdir, 'p%s' % priority)) + + def _dq(self): + activef = join(self.dqdir, 'active.json') + if exists(activef): + with open(activef) as f: + prios = json.load(f) + else: + prios = () + q = PriorityQueue(self._newdq, startprios=prios) + if q: + log.msg("Resuming crawl (%d requests scheduled)" % len(q), \ + spider=self.spider) + return q diff --git a/scrapy/dupefilter.py b/scrapy/dupefilter.py new file mode 100644 index 000000000..69d95b181 --- /dev/null +++ b/scrapy/dupefilter.py @@ -0,0 +1,46 @@ +import os +from scrapy.utils.request import request_fingerprint +from scrapy.utils.job import job_dir + + +class BaseDupeFilter(object): + + @classmethod + def from_settings(cls, settings): + return cls() + + def request_seen(self, request): + return False + + def open(self): # can return deferred + pass + + def close(self): # can return a deferred + pass + + +class RFPDupeFilter(BaseDupeFilter): + """Request Fingerprint duplicates filter""" + + def __init__(self, path=None): + self.file = None + self.fingerprints = set() + if path: + self.file = open(os.path.join(path, 'requests.seen'), 'a+') + self.fingerprints.update(x.rstrip() for x in self.file) + + @classmethod + def from_settings(cls, settings): + return cls(job_dir(settings)) + + def request_seen(self, request): + fp = request_fingerprint(request) + if fp in self.fingerprints: + return True + self.fingerprints.add(fp) + if self.file: + self.file.write(fp + os.linesep) + + def close(self): + if self.file: + self.file.close() diff --git a/scrapy/http/request/__init__.py b/scrapy/http/request/__init__.py index ea6b76858..3fa41ccb7 100644 --- a/scrapy/http/request/__init__.py +++ b/scrapy/http/request/__init__.py @@ -21,13 +21,14 @@ class Request(object_ref): '__weakref__'] def __init__(self, url, callback=None, method='GET', headers=None, body=None, - cookies=None, meta=None, encoding='utf-8', priority=0.0, + cookies=None, meta=None, encoding='utf-8', priority=0, dont_filter=False, errback=None): self._encoding = encoding # this one has to be set first self.method = str(method).upper() self._set_url(url) self._set_body(body) + assert isinstance(priority, int), "Request priority not an integer: %r" % priority self.priority = priority assert callback or not errback, "Cannot use errback without a callback" diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py index 0862e5989..bfeb98aad 100644 --- a/scrapy/settings/default_settings.py +++ b/scrapy/settings/default_settings.py @@ -47,6 +47,7 @@ DEFAULT_RESPONSE_ENCODING = 'ascii' DEPTH_LIMIT = 0 DEPTH_STATS = True +DEPTH_PRIORITY = 1 DOWNLOAD_DELAY = 0 @@ -85,7 +86,7 @@ DOWNLOADER_MIDDLEWARES_BASE = { DOWNLOADER_STATS = True -DUPEFILTER_CLASS = 'scrapy.contrib.dupefilter.RequestFingerprintDupeFilter' +DUPEFILTER_CLASS = 'scrapy.dupefilter.RFPDupeFilter' try: EDITOR = os.environ['EDITOR'] @@ -219,8 +220,7 @@ RETRY_PRIORITY_ADJUST = -1 ROBOTSTXT_OBEY = False SCHEDULER = 'scrapy.core.scheduler.Scheduler' - -SCHEDULER_ORDER = 'DFO' +SCHEDULER_DISK_QUEUE = 'scrapy.squeue.MarshalDiskQueue' SELECTORS_BACKEND = None # possible values: libxml2, lxml diff --git a/scrapy/squeue.py b/scrapy/squeue.py new file mode 100644 index 000000000..957a4af11 --- /dev/null +++ b/scrapy/squeue.py @@ -0,0 +1,15 @@ +""" +Scheduler disk-based queues +""" + +import marshal + +from scrapy.utils.queue import DiskQueue + +class MarshalDiskQueue(DiskQueue): + + def push(self, obj): + super(MarshalDiskQueue, self).push(marshal.dumps(obj)) + + def pop(self): + return marshal.loads(super(MarshalDiskQueue, self).pop()) diff --git a/scrapy/tests/test_dupefilter.py b/scrapy/tests/test_dupefilter.py index fa9eef332..571c93c82 100644 --- a/scrapy/tests/test_dupefilter.py +++ b/scrapy/tests/test_dupefilter.py @@ -1,37 +1,23 @@ import unittest from scrapy.http import Request -from scrapy.spider import BaseSpider -from scrapy.contrib.dupefilter import RequestFingerprintDupeFilter, NullDupeFilter +from scrapy.dupefilter import RFPDupeFilter -class RequestFingerprintDupeFilterTest(unittest.TestCase): +class RFPDupeFilterTest(unittest.TestCase): def test_filter(self): - spider = BaseSpider('foo') - filter = RequestFingerprintDupeFilter() - filter.open_spider(spider) + filter = RFPDupeFilter() + filter.open() r1 = Request('http://scrapytest.org/1') r2 = Request('http://scrapytest.org/2') r3 = Request('http://scrapytest.org/2') - assert not filter.request_seen(spider, r1) - assert filter.request_seen(spider, r1) + assert not filter.request_seen(r1) + assert filter.request_seen(r1) - assert not filter.request_seen(spider, r2) - assert filter.request_seen(spider, r3) + assert not filter.request_seen(r2) + assert filter.request_seen(r3) - filter.close_spider(spider) - - -class NullDupeFilterTest(unittest.TestCase): - - def test_filter(self): - spider = BaseSpider('foo') - filter = NullDupeFilter() - filter.open_spider(spider) - - r1 = Request('http://scrapytest.org/1') - assert not filter.request_seen(spider, r1) - filter.close_spider(spider) + filter.close() diff --git a/scrapy/tests/test_utils_datatypes.py b/scrapy/tests/test_utils_datatypes.py index 19473971c..c8233f8fa 100644 --- a/scrapy/tests/test_utils_datatypes.py +++ b/scrapy/tests/test_utils_datatypes.py @@ -1,81 +1,10 @@ import copy import unittest -from scrapy.utils.datatypes import PriorityQueue, PriorityStack, CaselessDict +from scrapy.utils.datatypes import CaselessDict __doctests__ = ['scrapy.utils.datatypes'] -# (ITEM, PRIORITY) -INPUT = [(1, -5), (30, -1), (80, -3), (4, 1), (6, 3), (20, 0), (50, -1)] - -class PriorityQueueTestCase(unittest.TestCase): - - output = [(1, -5), (80, -3), (30, -1), (50, -1), (20, 0), (4, 1), (6, 3)] - - def test_popping(self): - pq = PriorityQueue() - for item, pr in INPUT: - pq.push(item, pr) - l = [] - while pq: - l.append(pq.pop()) - self.assertEquals(l, self.output) - - def test_iter(self): - pq = PriorityQueue() - for item, pr in INPUT: - pq.push(item, pr) - result = [x for x in pq] - self.assertEquals(result, self.output) - - def test_nonzero(self): - pq = PriorityQueue() - pq.push(80, -1) - pq.push(20, 0) - pq.push(30, 1) - - pq.pop() - self.assertEquals(bool(pq), True) - pq.pop() - self.assertEquals(bool(pq), True) - pq.pop() - self.assertEquals(bool(pq), False) - - def test_len(self): - pq = PriorityQueue() - pq.push(80, -1) - pq.push(20, 0) - pq.push(30, 1) - - self.assertEquals(len(pq), 3) - pq.pop() - self.assertEquals(len(pq), 2) - pq.pop() - self.assertEquals(len(pq), 1) - pq.pop() - self.assertEquals(len(pq), 0) - -class PriorityStackTestCase(unittest.TestCase): - - output = [(1, -5), (80, -3), (50, -1), (30, -1), (20, 0), (4, 1), (6, 3)] - - def test_popping(self): - pq = PriorityStack() - for item, pr in INPUT: - pq.push(item, pr) - l = [] - while pq: - l.append(pq.pop()) - self.assertEquals(l, self.output) - - def test_iter(self): - pq = PriorityStack() - for item, pr in INPUT: - pq.push(item, pr) - result = [x for x in pq] - self.assertEquals(result, self.output) - - class CaselessDictTest(unittest.TestCase): def test_init(self): diff --git a/scrapy/tests/test_utils_pqueue.py b/scrapy/tests/test_utils_pqueue.py new file mode 100644 index 000000000..7343339ca --- /dev/null +++ b/scrapy/tests/test_utils_pqueue.py @@ -0,0 +1,74 @@ +import unittest + +from scrapy.utils.pqueue import PriorityQueue +from scrapy.utils.queue import MemoryQueue + + +class TestMemoryQueue(MemoryQueue): + + def __init__(self): + super(TestMemoryQueue, self).__init__() + self.closed = False + + def close(self): + self.closed = True + +class PriorityQueueTest(unittest.TestCase): + + def setUp(self): + qfactory = lambda x: TestMemoryQueue() + self.q = PriorityQueue(qfactory) + + def test_push_pop_noprio(self): + self.q.push('a') + self.q.push('b') + self.q.push('c') + self.assertEqual(self.q.pop(), 'a') + self.assertEqual(self.q.pop(), 'b') + self.assertEqual(self.q.pop(), 'c') + self.assertEqual(self.q.pop(), None) + + def test_push_pop_prio(self): + self.q.push('a', 3) + self.q.push('b', 1) + self.q.push('c', 2) + self.q.push('d', 1) + self.assertEqual(self.q.pop(), 'b') + self.assertEqual(self.q.pop(), 'd') + self.assertEqual(self.q.pop(), 'c') + self.assertEqual(self.q.pop(), 'a') + self.assertEqual(self.q.pop(), None) + + def test_len_nonzero(self): + assert not self.q + self.assertEqual(len(self.q), 0) + self.q.push('a', 3) + assert self.q + self.q.push('b', 1) + self.q.push('c', 2) + self.q.push('d', 1) + self.assertEqual(len(self.q), 4) + self.q.pop() + self.q.pop() + self.q.pop() + self.q.pop() + assert not self.q + self.assertEqual(len(self.q), 0) + + def test_close(self): + self.q.push('a', 3) + self.q.push('b', 1) + self.q.push('c', 2) + self.q.push('d', 1) + iqueues = self.q.queues.values() + self.assertEqual(sorted(self.q.close()), [1, 2, 3]) + assert all(q.closed for q in iqueues) + + def test_popped_internal_queues_closed(self): + self.q.push('a', 3) + self.q.push('b', 1) + self.q.push('c', 2) + p1queue = self.q.queues[1] + self.assertEqual(self.q.pop(), 'b') + self.q.close() + assert p1queue.closed diff --git a/scrapy/tests/test_utils_queue.py b/scrapy/tests/test_utils_queue.py new file mode 100644 index 000000000..1a18a9860 --- /dev/null +++ b/scrapy/tests/test_utils_queue.py @@ -0,0 +1,121 @@ +import os, glob +from twisted.trial import unittest + +from scrapy.utils.queue import MemoryQueue, DiskQueue + +class MemoryQueueTest(unittest.TestCase): + + def queue(self): + return MemoryQueue() + + def test_empty(self): + """Empty queue test""" + q = self.queue() + assert q.pop() is None + + def test_push_pop1(self): + """Basic push/pop test""" + q = self.queue() + q.push('a') + q.push('b') + q.push('c') + self.assertEqual(q.pop(), 'a') + self.assertEqual(q.pop(), 'b') + self.assertEqual(q.pop(), 'c') + self.assertEqual(q.pop(), None) + + def test_push_pop2(self): + """Test interleaved push and pops""" + q = self.queue() + q.push('a') + q.push('b') + q.push('c') + q.push('d') + self.assertEqual(q.pop(), 'a') + self.assertEqual(q.pop(), 'b') + q.push('e') + self.assertEqual(q.pop(), 'c') + self.assertEqual(q.pop(), 'd') + self.assertEqual(q.pop(), 'e') + + def test_len(self): + q = self.queue() + self.assertEqual(len(q), 0) + q.push('a') + self.assertEqual(len(q), 1) + q.push('b') + q.push('c') + self.assertEqual(len(q), 3) + q.pop() + q.pop() + q.pop() + self.assertEqual(len(q), 0) + + +class DiskQueueTest(MemoryQueueTest): + + chunksize = 100000 + + def setUp(self): + self.qdir = self.mktemp() + + def queue(self): + return DiskQueue(self.qdir, chunksize=self.chunksize) + + def test_close_open(self): + """Test closing and re-opening keeps state""" + q = self.queue() + q.push('a') + q.push('b') + q.push('c') + q.push('d') + self.assertEqual(q.pop(), 'a') + self.assertEqual(q.pop(), 'b') + q.close() + del q + q = self.queue() + self.assertEqual(len(q), 2) + q.push('e') + self.assertEqual(q.pop(), 'c') + self.assertEqual(q.pop(), 'd') + q.close() + del q + q = self.queue() + self.assertEqual(q.pop(), 'e') + self.assertEqual(len(q), 0) + + def test_chunks(self): + """Test chunks are created and removed""" + q = self.queue() + for x in range(5): + q.push(str(x)) + chunks = glob.glob(os.path.join(self.qdir, 'q*')) + self.assertEqual(len(chunks), 5/self.chunksize + 1) + for x in range(5): + q.pop() + chunks = glob.glob(os.path.join(self.qdir, 'q*')) + self.assertEqual(len(chunks), 1) + + def test_cleanup(self): + """Test queue dir is removed if queue is empty""" + q = self.queue() + assert os.path.exists(self.qdir) + for x in range(5): + q.push(str(x)) + for x in range(5): + q.pop() + q.close() + assert not os.path.exists(self.qdir) + + +class ChunkSize1DiskQueueTest(DiskQueueTest): + chunksize = 1 + +class ChunkSize2DiskQueueTest(DiskQueueTest): + chunksize = 2 + +class ChunkSize3DiskQueueTest(DiskQueueTest): + chunksize = 3 + +class ChunkSize4DiskQueueTest(DiskQueueTest): + chunksize = 4 diff --git a/scrapy/tests/test_utils_reqser.py b/scrapy/tests/test_utils_reqser.py new file mode 100644 index 000000000..31d512d08 --- /dev/null +++ b/scrapy/tests/test_utils_reqser.py @@ -0,0 +1,75 @@ +import unittest + +from scrapy.http import Request +from scrapy.spider import BaseSpider +from scrapy.utils.reqser import request_to_dict, request_from_dict + +class RequestSerializationTest(unittest.TestCase): + + def setUp(self): + self.spider = TestSpider() + + def test_basic(self): + r = Request("http://www.example.com") + self._assert_serializes_ok(r) + + def test_all_attributes(self): + r = Request("http://www.example.com", + callback='parse_item', + errback='handle_error', + method="POST", + body="some body", + headers={'content-encoding': 'text/html; charset=latin-1'}, + cookies={'currency': 'usd'}, + encoding='latin-1', + priority=20, + meta={'a': 'b'}) + self._assert_serializes_ok(r) + + def test_latin1_body(self): + r = Request("http://www.example.com", body="\xa3") + self._assert_serializes_ok(r) + + def test_utf8_body(self): + r = Request("http://www.example.com", body="\xc2\xa3") + self._assert_serializes_ok(r) + + def _assert_serializes_ok(self, request, spider=None): + d = request_to_dict(request, spider=spider) + request2 = request_from_dict(d, spider=spider) + self._assert_same_request(request, request2) + + def _assert_same_request(self, r1, r2): + self.assertEqual(r1.url, r2.url) + self.assertEqual(r1.callback, r2.callback) + self.assertEqual(r1.errback, r2.errback) + self.assertEqual(r1.method, r2.method) + self.assertEqual(r1.body, r2.body) + self.assertEqual(r1.headers, r2.headers) + self.assertEqual(r1.cookies, r2.cookies) + self.assertEqual(r1.meta, r2.meta) + self.assertEqual(r1._encoding, r2._encoding) + self.assertEqual(r1.priority, r2.priority) + self.assertEqual(r1.dont_filter, r2.dont_filter) + + def test_callback_serialization(self): + r = Request("http://www.example.com", callback=self.spider.parse_item, \ + errback=self.spider.handle_error) + self._assert_serializes_ok(r, spider=self.spider) + + def test_unserializable_callback1(self): + r = Request("http://www.example.com", callback=lambda x: x) + self.assertRaises(ValueError, request_to_dict, r) + self.assertRaises(ValueError, request_to_dict, r, spider=self.spider) + + def test_unserializable_callback2(self): + r = Request("http://www.example.com", callback=self.spider.parse_item) + self.assertRaises(ValueError, request_to_dict, r) + + +class TestSpider(BaseSpider): + name = 'test' + def parse_item(self, response): + pass + def handle_error(self, failure): + pass diff --git a/scrapy/utils/datatypes.py b/scrapy/utils/datatypes.py index 73945ea3e..1c1b54f41 100644 --- a/scrapy/utils/datatypes.py +++ b/scrapy/utils/datatypes.py @@ -268,84 +268,6 @@ class MergeDict(object): return self.__copy__() -class PriorityQueue(object): - """Priority queue using a deque for priority 0""" - - def __init__(self): - self.negitems = defaultdict(deque) - self.pzero = deque() - self.positems = defaultdict(deque) - - def push(self, item, priority=0): - if priority == 0: - self.pzero.appendleft(item) - elif priority < 0: - self.negitems[priority].appendleft(item) - else: - self.positems[priority].appendleft(item) - - def pop(self): - if self.negitems: - priorities = self.negitems.keys() - priorities.sort() - for priority in priorities: - deq = self.negitems[priority] - if deq: - t = (deq.pop(), priority) - if not deq: - del self.negitems[priority] - return t - elif self.pzero: - return (self.pzero.pop(), 0) - else: - priorities = self.positems.keys() - priorities.sort() - for priority in priorities: - deq = self.positems[priority] - if deq: - t = (deq.pop(), priority) - if not deq: - del self.positems[priority] - return t - raise IndexError("pop from an empty queue") - - def clear(self): - self.negitems.clear() - self.pzero.clear() - self.positems.clear() - - def __len__(self): - total = sum(len(v) for v in self.negitems.values()) + \ - len(self.pzero) + \ - sum(len(v) for v in self.positems.values()) - return total - - def __iter__(self): - gen_negs = ((i, priority) - for priority in sorted(self.negitems.keys()) - for i in reversed(self.negitems[priority])) - gen_zeros = ((item,0) for item in self.pzero) - gen_pos = ((i, priority) - for priority in sorted(self.positems.keys()) - for i in reversed(self.positems[priority])) - return chain(gen_negs, gen_zeros, gen_pos) - - def __nonzero__(self): - return bool(self.negitems or self.pzero or self.positems) - -class PriorityStack(PriorityQueue): - """A simple priority stack which is similar to PriorityQueue but pops its - items in reverse order (for the same priority)""" - - def push(self, item, priority=0): - if priority == 0: - self.pzero.append(item) - elif priority < 0: - self.negitems[priority].append(item) - else: - self.positems[priority].append(item) - - class LocalCache(OrderedDict): """Dictionary with a finite number of keys. diff --git a/scrapy/utils/engine.py b/scrapy/utils/engine.py index 56d7147b9..6692f06fa 100644 --- a/scrapy/utils/engine.py +++ b/scrapy/utils/engine.py @@ -24,7 +24,8 @@ def get_engine_status(engine=None): "engine.spider_is_idle(spider)", "engine.slots[spider].closing", "len(engine.slots[spider].inprogress)", - "len(engine.scheduler.pending_requests[spider])", + "len(engine.slots[spider].scheduler.dq)", + "len(engine.slots[spider].scheduler.mq)", "len(engine.scraper.slots[spider].queue)", "len(engine.scraper.slots[spider].active)", "engine.scraper.slots[spider].active_size", diff --git a/scrapy/utils/job.py b/scrapy/utils/job.py new file mode 100644 index 000000000..389fde73a --- /dev/null +++ b/scrapy/utils/job.py @@ -0,0 +1,7 @@ +import os + +def job_dir(settings): + path = settings['JOBDIR'] + if path and not os.path.exists(path): + os.makedirs(path) + return path diff --git a/scrapy/utils/pqueue.py b/scrapy/utils/pqueue.py new file mode 100644 index 000000000..bf613ea6c --- /dev/null +++ b/scrapy/utils/pqueue.py @@ -0,0 +1,58 @@ +class PriorityQueue(object): + """A priority queue implemented using multiple internal queues (typically, + FIFO queues). The internal queue must implement the following methods: + + * push(obj) + * pop() + * close() + * __len__() + + The constructor receives a qfactory argument, which is a callable used to + instantiate a new (internal) queue when a new priority is allocated. The + qfactory function is called with the priority number as first and only + argument. + + Only integer priorities should be used. Lower numbers are higher + priorities. + """ + + def __init__(self, qfactory, startprios=()): + self.queues = {} + self.qfactory = qfactory + for p in startprios: + q = self.qfactory(p) + if q: + self.queues[p] = q + self.curprio = min(startprios) if startprios else None + + def push(self, obj, priority=0): + try: + q = self.queues[priority] + except KeyError: + self.queues[priority] = q = self.qfactory(priority) + q.push(obj) + if priority < self.curprio or self.curprio is None: + self.curprio = priority + + def pop(self): + if self.curprio is None: + return + q = self.queues[self.curprio] + m = q.pop() + if not q: + q = self.queues.pop(self.curprio) + q.close() + prios = self.queues.keys() + self.curprio = min(prios) if prios else None + return m + + def close(self): + for q in self.queues.values(): + q.close() + return self.queues.keys() + + def __len__(self): + return sum(len(x) for x in self.queues.values()) if self.queues else 0 + + def __nonzero__(self): + return bool(self.queues) diff --git a/scrapy/utils/queue.py b/scrapy/utils/queue.py new file mode 100644 index 000000000..09fce3c6b --- /dev/null +++ b/scrapy/utils/queue.py @@ -0,0 +1,121 @@ +from __future__ import with_statement + +import os +import struct +import glob +from collections import deque + +from scrapy.utils.py26 import json + + +class MemoryQueue(object): + """Memory FIFO queue.""" + + def __init__(self): + self.q = deque() + + def push(self, obj): + self.q.appendleft(obj) + + def pop(self): + if self.q: + return self.q.pop() + + def close(self): + pass + + def __len__(self): + return len(self.q) + + +class DiskQueue(object): + """Persistent FIFO queue.""" + + szhdr_format = ">L" + szhdr_size = struct.calcsize(szhdr_format) + + def __init__(self, path, chunksize=100000): + self.path = path + if not os.path.exists(path): + os.makedirs(path) + self.info = self._loadinfo(chunksize) + self.chunksize = self.info['chunksize'] + self.headf = self._openchunk(self.info['head'][0], 'ab+') + self.tailf = self._openchunk(self.info['tail'][0]) + self.tailf.seek(self.info['tail'][2]) + + def push(self, string): + hnum, hpos = self.info['head'] + hpos += 1 + szhdr = struct.pack(self.szhdr_format, len(string)) + os.write(self.headf.fileno(), szhdr + string) + if hpos == self.chunksize: + hpos = 0 + hnum += 1 + self.headf.close() + self.headf = self._openchunk(hnum, 'ab+') + self.info['size'] += 1 + self.info['head'] = hnum, hpos + + def _openchunk(self, number, mode='r'): + return open(os.path.join(self.path, 'q%05d' % number), mode) + + def pop(self): + tnum, tcnt, toffset = self.info['tail'] + if [tnum, tcnt] >= self.info['head']: + return + tfd = self.tailf.fileno() + szhdr = os.read(tfd, self.szhdr_size) + if not szhdr: + return + size, = struct.unpack(self.szhdr_format, szhdr) + data = os.read(tfd, size) + tcnt += 1 + toffset += self.szhdr_size + size + if tcnt == self.chunksize and tnum <= self.info['head'][0]: + tcnt = toffset = 0 + tnum += 1 + self.tailf.close() + os.remove(self.tailf.name) + self.tailf = self._openchunk(tnum) + self.info['size'] -= 1 + self.info['tail'] = tnum, tcnt, toffset + return data + + def close(self): + self.headf.close() + self.tailf.close() + self._saveinfo(self.info) + if len(self) == 0: + self._cleanup() + + def __len__(self): + return self.info['size'] + + def _loadinfo(self, chunksize): + infopath = self._infopath() + if os.path.exists(infopath): + with open(infopath) as f: + info = json.load(f) + else: + info = { + 'chunksize': chunksize, + 'size': 0, + 'tail': [0, 0, 0], + 'head': [0, 0], + } + return info + + def _saveinfo(self, info): + with open(self._infopath(), 'w') as f: + json.dump(info, f) + + def _infopath(self): + return os.path.join(self.path, 'info.json') + + def _cleanup(self): + for x in glob.glob(os.path.join(self.path, 'q*')): + os.remove(x) + os.remove(os.path.join(self.path, 'info.json')) + if not os.listdir(self.path): + os.rmdir(self.path) diff --git a/scrapy/utils/reqser.py b/scrapy/utils/reqser.py new file mode 100644 index 000000000..d02585a32 --- /dev/null +++ b/scrapy/utils/reqser.py @@ -0,0 +1,72 @@ +""" +Helper functions for serializing (and deserializing) requests. +""" + +from scrapy.http import Request + +def request_to_dict(request, spider=None): + """Convert Request object to a dict. + + If a spider is given, it will try to find out the name of the spider method + used in the callback and store that as the callback. + """ + cb = request.callback + if callable(cb): + cb = _find_method(spider, cb) + eb = request.errback + if callable(eb): + eb = _find_method(spider, eb) + d = { + 'url': request.url.decode('ascii'), # urls should be safe (safe_string_url) + 'callback': cb, + 'errback': eb, + 'method': request.method, + 'headers': dict(request.headers), + 'body': request.body, + 'cookies': request.cookies, + 'meta': request.meta, + '_encoding': request._encoding, + 'priority': request.priority, + 'dont_filter': request.dont_filter, + } + return d + + +def request_from_dict(d, spider=None): + """Create Request object from a dict. + + If a spider is given, it will try to resolve the callbacks looking at the + spider for methods with the same name. + """ + cb = d['callback'] + if cb and spider: + cb = _get_method(spider, cb) + eb = d['errback'] + if eb and spider: + eb = _get_method(spider, eb) + return Request( + url=d['url'].encode('ascii'), + callback=cb, + errback=eb, + method=d['method'], + headers=d['headers'], + body=d['body'], + cookies=d['cookies'], + meta=d['meta'], + encoding=d['_encoding'], + priority=d['priority'], + dont_filter=d['dont_filter']) + + +def _find_method(obj, func): + if obj and hasattr(func, 'im_self') and func.im_self is obj: + return func.im_func.__name__ + else: + raise ValueError("Function %s is not a method of: %s" % (func, obj)) + +def _get_method(obj, name): + name = str(name) + try: + return getattr(obj, name) + except AttributeError: + raise ValueError("Method %r not found in: %s" % (name, obj))