fix xmlrpc server not stopped bug, check xmlrpc server stop

pyspider/fetcher/tornado_fetcher.py

@@ -542,7 +542,8 @@ def quit(self):
         self._running = False
         self._quit = True
         self.ioloop.add_callback(self.ioloop.stop)
-        if hasattr(self, 'xmlrpc_ioloop'):
+        if hasattr(self, 'xmlrpc_server'):
+            self.xmlrpc_ioloop.add_callback(self.xmlrpc_server.stop)
             self.xmlrpc_ioloop.add_callback(self.xmlrpc_ioloop.stop)
 
     def size(self):
@@ -578,8 +579,8 @@ def dump_counter(_time, _type):
 
         container = tornado.wsgi.WSGIContainer(application)
         self.xmlrpc_ioloop = tornado.ioloop.IOLoop()
-        http_server = tornado.httpserver.HTTPServer(container, io_loop=self.xmlrpc_ioloop)
-        http_server.listen(port=port, address=bind)
+        self.xmlrpc_server = tornado.httpserver.HTTPServer(container, io_loop=self.xmlrpc_ioloop)
+        self.xmlrpc_server.listen(port=port, address=bind)
         self.xmlrpc_ioloop.start()
 
     def on_fetch(self, type, task):

pyspider/libs/utils.py

@@ -8,6 +8,7 @@
 import logging
 import hashlib
 import datetime
+import socket
 import base64
 
 import six
@@ -409,3 +410,12 @@ def python_console(namespace=None):
         namespace.update(caller.f_locals)
 
     return get_python_console(namespace=namespace).interact()
+
+
+def check_port_open(port, addr='127.0.0.1'):
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    result = sock.connect_ex((addr, port))
+    if result == 0:
+        return True
+    else:
+        return False

pyspider/scheduler/scheduler.py

@@ -450,8 +450,9 @@ def quit(self):
         '''Set quit signal'''
         self._quit = True
         # stop xmlrpc server
-        if hasattr(self, 'ioloop'):
-            self.ioloop.add_callback(self.ioloop.stop)
+        if hasattr(self, 'xmlrpc_server'):
+            self.xmlrpc_ioloop.add_callback(self.xmlrpc_server.stop)
+            self.xmlrpc_ioloop.add_callback(self.xmlrpc_ioloop.stop)
 
     def run_once(self):
         '''comsume queues and feed tasks to fetcher, once'''
@@ -578,10 +579,10 @@ def get_active_tasks(project=None, limit=100):
         import tornado.httpserver
 
         container = tornado.wsgi.WSGIContainer(application)
-        self.ioloop = tornado.ioloop.IOLoop()
-        http_server = tornado.httpserver.HTTPServer(container, io_loop=self.ioloop)
-        http_server.listen(port=port, address=bind)
-        self.ioloop.start()
+        self.xmlrpc_ioloop = tornado.ioloop.IOLoop()
+        self.xmlrpc_server = tornado.httpserver.HTTPServer(container, io_loop=self.xmlrpc_ioloop)
+        self.xmlrpc_server.listen(port=port, address=bind)
+        self.xmlrpc_ioloop.start()
 
     def on_request(self, task):
         if self.INQUEUE_LIMIT and len(self.task_queue[task['project']]) >= self.INQUEUE_LIMIT:

tests/test_fetcher.py

@@ -90,6 +90,13 @@ def tearDownClass(self):
             self.phantomjs.wait()
         self.rpc._quit()
         self.thread.join()
+
+        assert not utils.check_port_open(5000)
+        assert not utils.check_port_open(23333)
+        assert not utils.check_port_open(24444)
+        assert not utils.check_port_open(25555)
+        assert not utils.check_port_open(14887)
+
         time.sleep(1)
 
     def test_10_http_get(self):

tests/test_run.py

@@ -39,6 +39,12 @@ def tearDownClass(self):
         self.httpbin_thread.terminate()
         self.httpbin_thread.join()
 
+        assert not utils.check_port_open(5000)
+        assert not utils.check_port_open(23333)
+        assert not utils.check_port_open(24444)
+        assert not utils.check_port_open(25555)
+        assert not utils.check_port_open(14887)
+
         shutil.rmtree('./data/tests', ignore_errors=True)
 
     def test_10_cli(self):
@@ -323,6 +329,11 @@ def tearDownClass(self):
         self.scheduler_thread.join()
         time.sleep(1)
 
+        assert not utils.check_port_open(5000)
+        assert not utils.check_port_open(23333)
+        assert not utils.check_port_open(24444)
+        assert not utils.check_port_open(25555)
+
         shutil.rmtree('./data/tests', ignore_errors=True)
 
     def test_10_send_message(self):

tests/test_scheduler.py

@@ -14,6 +14,7 @@
 logging.config.fileConfig("pyspider/logging.conf")
 
 from pyspider.scheduler.task_queue import TaskQueue
+from pyspider.libs import utils
 
 
 class TestTaskQueue(unittest.TestCase):
@@ -157,6 +158,11 @@ def tearDownClass(self):
         shutil.rmtree('./data/tests', ignore_errors=True)
         time.sleep(1)
 
+        assert not utils.check_port_open(5000)
+        assert not utils.check_port_open(self.scheduler_xmlrpc_port)
+        assert not utils.check_port_open(24444)
+        assert not utils.check_port_open(25555)
+
     def test_10_new_task_ignore(self):
         self.newtask_queue.put({
             'taskid': 'taskid',

tests/test_webui.py

@@ -39,7 +39,7 @@ def setUpClass(self):
         self.threads = []
 
         ctx = run.scheduler.make_context('scheduler', [], self.ctx)
-        scheduler = run.scheduler.invoke(ctx)
+        self.scheduler = scheduler = run.scheduler.invoke(ctx)
         self.threads.append(run_in_thread(scheduler.xmlrpc_run))
         self.threads.append(run_in_thread(scheduler.run))
 
@@ -81,6 +81,12 @@ def tearDownClass(self):
         self.httpbin_thread.terminate()
         self.httpbin_thread.join()
 
+        assert not utils.check_port_open(5000)
+        assert not utils.check_port_open(23333)
+        assert not utils.check_port_open(24444)
+        assert not utils.check_port_open(25555)
+        assert not utils.check_port_open(14887)
+
         shutil.rmtree('./data/tests', ignore_errors=True)
 
     def test_10_index_page(self):