diff --git a/docs/topics/scrapyd.rst b/docs/topics/scrapyd.rst index e78b5c76e..4b9e12fb8 100644 --- a/docs/topics/scrapyd.rst +++ b/docs/topics/scrapyd.rst @@ -27,16 +27,17 @@ How Scrapyd works ================= Scrapyd is an application (typically run as a daemon) that continually polls -for projects that need to run (ie. those projects that have spiders enqueued). +for spiders that need to run (ie. those projects that have spiders enqueued). -When a project needs to run, a Scrapy process is started for that project using -something similar to the typical ``scrapy crawl``` command, and it continues to -run until it finishes processing all spiders form the spider queue. +When a spider needs to run, a process is started to crawl the spider:: + + scrapy crawl my spider Scrapyd also runs multiple processes in parallel, allocating them in a fixed number of "slots", which defaults to the number of cpu processors available in -the system, but this can be changed with the ``max_proc`` option. It also -starts as many processes as possible to handle the load. +the system, but this can be changed with the ``max_proc`` and +``max_proc_per_cpu`` options. It also starts as many processes as possible to +handle the load. In addition to dispatching and managing processes, Scrapyd provides a :ref:`JSON web service ` to upload new project versions @@ -144,7 +145,14 @@ max_proc -------- The maximum number of concurrent Scrapy process that will be started. If unset -or ``0`` it will use the number of cpus available in the system. +or ``0`` it will use the number of cpus available in the system mulitplied by +the value in ``max_proc_per_cpu`` option. Defaults to ``0``. + +max_proc_per_cpu +---------------- + +The maximum number of concurrent Scrapy process that will be started per cpu. +Defaults to ``4``. debug ----- @@ -363,7 +371,7 @@ Example request:: Example reponse:: - {"status": "ok", "spiders": ["spider1", "spider2", "spider3"]} + {"status": "ok", "spiders": 3} schedule.json ------------- diff --git a/scrapyd/default_scrapyd.conf b/scrapyd/default_scrapyd.conf index b0f046a77..69ad18aa5 100644 --- a/scrapyd/default_scrapyd.conf +++ b/scrapyd/default_scrapyd.conf @@ -3,6 +3,7 @@ eggs_dir = eggs logs_dir = logs dbs_dir = dbs max_proc = 0 +max_proc_per_cpu = 4 http_port = 6800 debug = off egg_runner = scrapyd.eggrunner diff --git a/scrapyd/launcher.py b/scrapyd/launcher.py index 06762944e..845a35f8b 100644 --- a/scrapyd/launcher.py +++ b/scrapyd/launcher.py @@ -7,12 +7,16 @@ from twisted.application.service import Service from twisted.python import log from scrapy.utils.py26 import cpu_count +from scrapy.utils.python import unicode_to_str +from scrapyd.utils import get_crawl_args from .interfaces import IPoller, IEggStorage, IEnvironment class Launcher(Service): def __init__(self, config, app): - self.max_proc = config.getint('max_proc', 0) or cpu_count() + self.max_proc = config.getint('max_proc', 0) + if not self.max_proc: + self.max_proc = cpu_count() * config.getint('max_proc_per_cpu') self.egg_runner = config.get('egg_runner', 'scrapyd.eggrunner') self.app = app @@ -39,12 +43,14 @@ class Launcher(Service): return eggpath def _spawn_process(self, message, slot): - project = message['project'] + project = unicode_to_str(message['project']) + spider = unicode_to_str(message['spider']) eggpath = self._get_eggpath(project) args = [sys.executable, '-m', self.egg_runner, 'crawl'] + args += get_crawl_args(message) e = self.app.getComponent(IEnvironment) env = e.get_environment(message, slot, eggpath) - pp = ScrapyProcessProtocol(eggpath, slot) + pp = ScrapyProcessProtocol(eggpath, slot, project, spider) pp.deferred.addBoth(self._process_finished, eggpath, slot) reactor.spawnProcess(pp, sys.executable, args=args, env=env) @@ -56,10 +62,12 @@ class Launcher(Service): class ScrapyProcessProtocol(protocol.ProcessProtocol): - def __init__(self, eggfile, slot): + def __init__(self, eggfile, slot, project, spider): self.eggfile = eggfile self.slot = slot self.pid = None + self.project = project + self.spider = spider self.deferred = defer.Deferred() def outReceived(self, data): @@ -80,5 +88,6 @@ class ScrapyProcessProtocol(protocol.ProcessProtocol): self.deferred.callback(self) def log(self, msg): - msg += "slot=%r pid=%r egg=%r" % (self.slot, self.pid, self.eggfile) + msg += "project=%r spider=%r slot=%r pid=%r egg=%r" % (self.project, \ + self.spider, self.slot, self.pid, self.eggfile) log.msg(msg, system="Launcher") diff --git a/scrapyd/poller.py b/scrapyd/poller.py index 7a6ec4668..05c9feaa3 100644 --- a/scrapyd/poller.py +++ b/scrapyd/poller.py @@ -18,7 +18,8 @@ class QueuePoller(object): return for p, q in self.queues.iteritems(): if q.count(): - return self.dq.put(self._message(p)) + msg = q.pop() + return self.dq.put(self._message(msg, p)) def next(self): return self.dq.get() @@ -26,5 +27,8 @@ class QueuePoller(object): def update_projects(self): self.queues = get_spider_queues(self.config) - def _message(self, project): - return {'project': str(project)} + def _message(self, queue_msg, project): + d = queue_msg.copy() + d['project'] = project + d['spider'] = d.pop('name') + return d diff --git a/scrapyd/tests/test_poller.py b/scrapyd/tests/test_poller.py index 36bbf04a5..0cae9c13c 100644 --- a/scrapyd/tests/test_poller.py +++ b/scrapyd/tests/test_poller.py @@ -37,5 +37,5 @@ class QueuePollerTest(unittest.TestCase): self.poller.poll() self.queues['mybot1'].pop() self.poller.poll() - self.failUnlessEqual(d1.result, {'project': 'mybot1'}) - self.failUnlessEqual(d2.result, {'project': 'mybot2'}) + self.failUnlessEqual(d1.result, {'project': 'mybot1', 'spider': 'spider1'}) + self.failUnlessEqual(d2.result, {'project': 'mybot2', 'spider': 'spider2'}) diff --git a/scrapyd/tests/test_utils.py b/scrapyd/tests/test_utils.py new file mode 100644 index 000000000..68aed0053 --- /dev/null +++ b/scrapyd/tests/test_utils.py @@ -0,0 +1,13 @@ +import unittest + +from scrapyd.utils import get_crawl_args + +class UtilsTest(unittest.TestCase): + + def test_get_crawl_args(self): + msg = {'project': 'lolo', 'spider': 'lala'} + self.assertEqual(get_crawl_args(msg), ['lala']) + msg = {'project': 'lolo', 'spider': 'lala', 'arg1': 'val1'} + cargs = get_crawl_args(msg) + self.assertEqual(cargs, ['lala', '-a', 'arg1=val1']) + assert all(isinstance(x, str) for x in cargs), cargs diff --git a/scrapyd/utils.py b/scrapyd/utils.py index 1ab368aa5..6adf1faba 100644 --- a/scrapyd/utils.py +++ b/scrapyd/utils.py @@ -2,6 +2,7 @@ import os from ConfigParser import NoSectionError from scrapy.spiderqueue import SqliteSpiderQueue +from scrapy.utils.python import stringify_dict, unicode_to_str def get_spider_queues(config): """Return a dict of Spider Quees keyed by project name""" @@ -28,3 +29,15 @@ def get_project_list(config): except NoSectionError: pass return projects + +def get_crawl_args(message): + """Return the command-line arguments to use for the scrapy crawl process + that will be started for this message + """ + msg = message.copy() + args = [unicode_to_str(msg['spider'])] + del msg['project'], msg['spider'] + for k, v in stringify_dict(msg).items(): + args += ['-a'] + args += ['%s=%s' % (k, v)] + return args diff --git a/scrapyd/webservice.py b/scrapyd/webservice.py index cf1464876..2e71c5f03 100644 --- a/scrapyd/webservice.py +++ b/scrapyd/webservice.py @@ -46,7 +46,7 @@ class AddVersion(WsResource): eggstorage = self.root.app.getComponent(IEggStorage) eggstorage.put(eggf, project, version) self.root.update_projects() - return {"status": "ok", "spiders": spiders} + return {"status": "ok", "spiders": len(spiders)} class ListProjects(WsResource):