Merge pull request #297 from scrapy/downloader-gc

Add garbage collector to downloader

scrapy/core/downloader/__init__.py

@@ -3,7 +3,7 @@ import warnings
 from time import time
 from collections import deque
 
-from twisted.internet import reactor, defer
+from twisted.internet import reactor, defer, task
 
 from scrapy.utils.defer import mustbe_deferred
 from scrapy.utils.httpobj import urlparse_cached
@@ -35,6 +35,10 @@ class Slot(object):
             return random.uniform(0.5 * self.delay, 1.5 * self.delay)
         return self.delay
 
+    def close(self):
+        if self.latercall and self.latercall.active():
+            self.latercall.cancel()
+
 
 def _get_concurrency_delay(concurrency, spider, settings):
     delay = settings.getfloat('DOWNLOAD_DELAY')
@@ -71,6 +75,8 @@ class Downloader(object):
         self.domain_concurrency = self.settings.getint('CONCURRENT_REQUESTS_PER_DOMAIN')
         self.ip_concurrency = self.settings.getint('CONCURRENT_REQUESTS_PER_IP')
         self.middleware = DownloaderMiddlewareManager.from_crawler(crawler)
+        self._slot_gc_loop = task.LoopingCall(self._slot_gc)
+        self._slot_gc_loop.start(60)
 
     def fetch(self, request, spider):
         def _deactivate(response):
@@ -172,3 +178,14 @@ class Downloader(object):
 
     def is_idle(self):
         return not self.slots
+
+    def close(self):
+        self._slot_gc_loop.stop()
+        for slot in self.slots.itervalues():
+            slot.close()
+
+    def _slot_gc(self, age=60):
+        mintime = time() - age
+        for key, slot in self.slots.items():
+            if not slot.active and slot.lastseen + slot.delay < mintime:
+                self.slots.pop(key).close()

scrapy/core/engine.py

@@ -254,6 +254,9 @@ class ExecutionEngine(object):
 
         dfd = slot.close()
 
+        dfd.addBoth(lambda _: self.downloader.close())
+        dfd.addErrback(log.err, spider=spider)
+
         dfd.addBoth(lambda _: self.scraper.close_spider(spider))
         dfd.addErrback(log.err, spider=spider)
 

scrapy/tests/test_crawl.py

@@ -1,6 +1,6 @@
 import sys, time
 from twisted.internet import defer
-from twisted.trial.unittest import TestCase, SkipTest
+from twisted.trial.unittest import TestCase
 from subprocess import Popen, PIPE
 from scrapy.spider import BaseSpider
 from scrapy.http import Request
@@ -10,12 +10,13 @@ from scrapy.utils.test import get_crawler
 class FollowAllSpider(BaseSpider):
 
     name = 'follow'
-    start_urls = ["http://localhost:8998/follow?total=10&show=5&order=rand"]
     link_extractor = SgmlLinkExtractor()
 
-    def __init__(self):
+    def __init__(self, total=10, show=20, order="rand"):
         self.urls_visited = []
         self.times = []
+        url = "http://localhost:8998/follow?total=%s&show=%s&order=%s" % (total, show, order)
+        self.start_urls = [url]
 
     def parse(self, response):
         self.urls_visited.append(response.url)
@@ -48,13 +49,9 @@ class CrawlTestCase(TestCase):
 
     @defer.inlineCallbacks
     def test_delay(self):
-        # FIXME: this test fails because Scrapy leaves the reactor dirty with
-        # callLater calls when download delays are used. This test should be
-        # enabled after this bug is fixed.
-        raise SkipTest("disabled due to a reactor leak in the scrapy downloader")
-
         spider = FollowAllSpider()
-        yield docrawl(spider)
+        yield docrawl(spider, {"DOWNLOAD_DELAY": 0.3})
         t = spider.times[0]
-        for y in spider.times[1:]:
-            self.assertTrue(y-t > 0.5, "download delay too small: %s" % (y-t))
+        for t2 in spider.times[1:]:
+            self.assertTrue(t2-t > 0.15, "download delay too small: %s" % (t2-t))
+            t = t2