From 2434000cda588e21d274e1c82965f1154328642a Mon Sep 17 00:00:00 2001 From: Pablo Hoffman Date: Tue, 13 Jan 2009 01:14:40 +0000 Subject: [PATCH] * ported PriorityQueue and PriorityStack to use heapq instead of queue.Queue + bisect which was up to 5x slower! * added test case for PriorityStack (only PriorityQueue had before) * changed Priority{Stack,Queue} API to just push(), pop(), and made them iterable --HG-- extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%40712 --- .../trunk/scrapy/core/scheduler/schedulers.py | 27 +++++----- .../scrapy/tests/test_utils_datatypes.py | 36 ++++++++----- scrapy/trunk/scrapy/utils/datatypes.py | 52 +++++++------------ 3 files changed, 55 insertions(+), 60 deletions(-) diff --git a/scrapy/trunk/scrapy/core/scheduler/schedulers.py b/scrapy/trunk/scrapy/core/scheduler/schedulers.py index 81d2c9f82..5abf936b7 100644 --- a/scrapy/trunk/scrapy/core/scheduler/schedulers.py +++ b/scrapy/trunk/scrapy/core/scheduler/schedulers.py @@ -14,17 +14,16 @@ from scrapy.conf import settings class Scheduler(object) : """ - The scheduler decides what to scrape, how fast, and in what order. - The scheduler schedules websites and pages to be scraped. Individual - web pages that are to be scraped are batched up into a "run" for a website. - As the domain is being scraped, pages that are discovered are added to the - scheduler. The scheduler must not allow the same page to be requested - multiple times within the same batch. + The scheduler decides what to scrape next. In other words, it defines the + crawling order. The scheduler schedules websites and requests to be + scraped. Individual web pages that are to be scraped are batched up into a + "run" for a website. As the domain is being scraped, pages that are + discovered are added to the scheduler. Typical usage: - * next_availble_domain() called to find out when there is something to do - * begin_domain() called to commence scraping a website + * next_available_domain() called to find out when there is something to do + * open_domain() called to commence scraping a website * enqueue_request() called multiple times when new links found * next_request() called multiple times when there is capacity to process urls * close_domain() called when there are no more pages or upon error @@ -66,14 +65,14 @@ class Scheduler(object) : def domain_has_pending(self, domain): if domain in self.pending_requests: - return not self.pending_requests[domain].empty() + return bool(self.pending_requests[domain]) def next_domain(self) : """ Return next domain available to scrape and remove it from available domains queue """ if self.pending_domains_count: - priority, domain = self.domains_queue.get_nowait() + domain, priority = self.domains_queue.pop() if self.pending_domains_count[domain] == 1: del self.pending_domains_count[domain] else: @@ -88,7 +87,7 @@ class Scheduler(object) : domain can be scheduled twice, either with the same or with different priority. """ - self.domains_queue.put(domain, priority=priority) + self.domains_queue.push(domain, priority) if domain not in self.pending_domains_count: self.pending_domains_count[domain] = 1 else: @@ -114,7 +113,7 @@ class Scheduler(object) : if request.dont_filter or added: deferred = defer.Deferred() - self.pending_requests[domain].put((request, deferred), priority) + self.pending_requests[domain].push((request, deferred), priority) return deferred else: return defer_fail(IgnoreRequest('Skipped (already visited): %s' % request)) @@ -133,8 +132,8 @@ class Scheduler(object) : None should be returned if there are no more request pending for the domain passed. """ pending_list = self.pending_requests.get(domain) - if pending_list and not pending_list.empty(): - return pending_list.get_nowait()[1] + if pending_list: + return pending_list.pop()[0] else: return (None, None) diff --git a/scrapy/trunk/scrapy/tests/test_utils_datatypes.py b/scrapy/trunk/scrapy/tests/test_utils_datatypes.py index 629eaf2ff..ef3239308 100644 --- a/scrapy/trunk/scrapy/tests/test_utils_datatypes.py +++ b/scrapy/trunk/scrapy/tests/test_utils_datatypes.py @@ -1,26 +1,34 @@ import unittest -from scrapy.utils.datatypes import PriorityQueue + +from scrapy.utils.datatypes import PriorityQueue, PriorityStack class DatatypesTestCase(unittest.TestCase): def test_priority_queue(self): + + input = [('five', 5), ('three-1', 3), ('three-2', 3), ('six', 6), ('one', 1)] + output = [('one', 1), ('three-1', 3), ('three-2', 3), ('five', 5), ('six', 6)] + pq = PriorityQueue() + for item, prio in input: + pq.push(item, prio) + out = [] + while pq: + out.append(pq.pop()) + self.assertEqual(out, output) - pq.put('b', priority=1) - pq.put('a', priority=1) - pq.put('c', priority=1) - pq.put('z', priority=0) - pq.put('d', priority=2) + def test_priority_stack(self): - v = [] - p = [] - while not pq.empty(): - priority, value = pq.get() - v.append(value) - p.append(priority) + input = [('five', 5), ('three-1', 3), ('three-2', 3), ('six', 6), ('one', 1)] + output = [('one', 1), ('three-2', 3), ('three-1', 3), ('five', 5), ('six', 6)] - self.assertEqual(v, ['z', 'b', 'a', 'c', 'd']) - self.assertEqual(p, [0, 1, 1, 1, 2]) + pq = PriorityStack() + for item, prio in input: + pq.push(item, prio) + out = [] + while pq: + out.append(pq.pop()) + self.assertEqual(out, output) if __name__ == "__main__": unittest.main() diff --git a/scrapy/trunk/scrapy/utils/datatypes.py b/scrapy/trunk/scrapy/utils/datatypes.py index ab32a82ce..7e75d5aee 100644 --- a/scrapy/trunk/scrapy/utils/datatypes.py +++ b/scrapy/trunk/scrapy/utils/datatypes.py @@ -5,11 +5,11 @@ Python Standard Library. This module must not depend on any module outside the Standard Library. """ +import time import copy import gzip -import Queue -import bisect from cStringIO import StringIO +from heapq import heappush, heappop class MergeDict(object): """ @@ -407,43 +407,31 @@ class CaselessDict(dict): def pop(self, key, def_val=None): return dict.pop(self, self.normkey(key), def_val) +class PriorityQueue(object): + """A simple priority queue""" -class PriorityQueue(Queue.Queue): + def __init__(self): + self.items = [] - def __len__(self): - return self.qsize() + def push(self, item, priority=0): + heappush(self.items, (priority, time.time(), item)) - def _init(self, maxsize): - self.maxsize = maxsize - # Python 2.5 uses collections.deque, but we can't because - # we need insert(pos, item) for our priority stuff - self.queue = [] + def pop(self): + priority, _, item = heappop(self.items) + return item, priority - def put(self, item, priority=0, block=True, timeout=None): - """Puts an item onto the queue with a numeric priority (default is zero). - - Note that we are "shadowing" the original Queue.Queue put() method here. - """ - Queue.Queue.put(self, (priority, item), block, timeout) - - def _put(self, item): - """Override of the Queue._put to support prioritisation.""" - # Priorities must be integers! - priority = int(item[0]) - - # Using a tuple (priority+1,) finds us the correct insertion - # position to maintain the existing ordering. - self.queue.insert(bisect.bisect_left(self.queue, (priority+1,)), item) - - def _get(self): - """Override of Queue._get(). Strips the priority.""" - return self.queue.pop(0) + def __iter__(self): + return ((priority, item) for priority, _, item in self.items) + def __nonzero__(self): + return bool(self.items) class PriorityStack(PriorityQueue): - def _put(self, item): - priority = int(item[0]) - self.queue.insert(bisect.bisect_left(self.queue, (priority,)), item) + """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): + heappush(self.items, (priority, -time.time(), item)) class gzStringIO: """a file like object, similar to StringIO, but gzip-compressed."""