Address Lucy's comments

add tests to check correctness of slot setermination
unmark requests after downloading
shorter
better exception message

scrapy/pqueues.py

@@ -36,6 +36,12 @@ def _scheduler_slot_write(request, slot):
 
 
 def _set_scheduler_slot(request):
+    """
+        >>> _set_scheduler_slot({'url':'http://foo.com'}) == _set_scheduler_slot({'url':'http://bar.com'})
+        False
+        >>> _set_scheduler_slot({'url':'http://foo.com'}) == _set_scheduler_slot({'url':'http://foo.com'})
+        True
+    """
     meta = _get_request_meta(request)
     slot = meta.get(SCHEDULER_SLOT_META_KEY, None)
 
@@ -141,9 +147,8 @@ class DownloaderAwarePriorityQueue(object):
         ip_concurrency = crawler.settings.getint(ip_concurrency_key, 0)
 
         if ip_concurrency > 0:
-            raise ValueError('"%s" does not support %s=%d' % (self.__class__,
-                                                              ip_concurrency_key,
-                                                              ip_concurrency))
+            raise ValueError('"%s" does not support setting %s' % (self.__class__,
+                                                                   ip_concurrency_key))
 
         def pqfactory(startprios=()):
             return PriorityAsTupleQueue(qfactory, startprios)
@@ -171,6 +176,9 @@ class DownloaderAwarePriorityQueue(object):
     def check_mark(self, request):
         return request.meta.get(self._DOWNLOADER_AWARE_PQ_ID, None) == id(self)
 
+    def unmark(self, request):
+        del request.meta[self._DOWNLOADER_AWARE_PQ_ID]
+
     def pop(self):
         slots = [(active_downloads, slot)
                  for slot, active_downloads in self._active_downloads.items()
@@ -194,6 +202,7 @@ class DownloaderAwarePriorityQueue(object):
     def on_response_download(self, response, request, spider):
         if not self.check_mark(request):
             return
+        self.unmark(request)
 
         slot = _scheduler_slot_read(request)
         if slot not in self._active_downloads or self._active_downloads[slot] <= 0:

tests/test_scheduler.py

@@ -248,8 +248,8 @@ class TestSchedulerWithDownloaderAwareOnDisk(BaseSchedulerOnDiskTester,
         self.close_scheduler()
         self.create_scheduler()
 
-        slots = list()
-        requests = list()
+        slots = []
+        requests = []
         while self.scheduler.has_pending_requests():
             request = self.scheduler.next_request()
             slots.append(_scheduler_slot_read(request))