How do I stop Tornado web server?
In case you do no want to bother with threads, you could catch a keyboard interrupt signal :
try:
tornado.ioloop.IOLoop.instance().start()
# signal : CTRL + BREAK on windows or CTRL + C on linux
except KeyboardInterrupt:
tornado.ioloop.IOLoop.instance().stop()
I just ran into this and found this issue myself, and using info from this thread came up with the following. I simply took my working stand alone Tornado code (copied from all the examples) and moved the actual starting code into a function. I then called the function as a threading thread. My case different as the threading call was done from my existing code where I just imported the startTornado and stopTornado routines.
The suggestion above seemed to work great, so I figured I would supply the missing example code. I tested this code under Linux on a FC16 system (and fixed my initial type-o).
import tornado.ioloop, tornado.web
class Handler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world")
application = tornado.web.Application([ (r"/", Handler) ])
def startTornado():
application.listen(8888)
tornado.ioloop.IOLoop.instance().start()
def stopTornado():
tornado.ioloop.IOLoop.instance().stop()
if __name__ == "__main__":
import time, threading
threading.Thread(target=startTornado).start()
print "Your web server will self destruct in 2 minutes"
time.sleep(120)
stopTornado()
Hope this helps the next person.
Here is the solution how to stop Torando from another thread. Schildmeijer provided a good hint, but it took me a while to actually figure the final example that works.
Please see below:
import threading
import tornado.ioloop
import tornado.web
import time
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world!\n")
def start_tornado(*args, **kwargs):
application = tornado.web.Application([
(r"/", MainHandler),
])
application.listen(8888)
print "Starting Torando"
tornado.ioloop.IOLoop.instance().start()
print "Tornado finished"
def stop_tornado():
ioloop = tornado.ioloop.IOLoop.instance()
ioloop.add_callback(ioloop.stop)
print "Asked Tornado to exit"
def main():
t = threading.Thread(target=start_tornado)
t.start()
time.sleep(5)
stop_tornado()
t.join()
if __name__ == "__main__":
main()
There is a problem with Zaar Hai's solution, namely that it leaves the socket open. The reason I was looking for a solution to stop Tornado is I'm running unit tests against my app server and I needed a way to start/stop the server between tests to have a clear state (empty session, etc.). By leaving the socket open, the second test always ran into an Address already in use
error. So I came up with the following:
import logging as log
from time import sleep
from threading import Thread
import tornado
from tornado.httpserver import HTTPServer
server = None
thread = None
def start_app():
def start():
global server
server = HTTPServer(create_app())
server.listen(TEST_PORT, TEST_HOST)
tornado.ioloop.IOLoop.instance().start()
global thread
thread = Thread(target=start)
thread.start()
# wait for the server to fully initialize
sleep(0.5)
def stop_app():
server.stop()
# silence StreamClosedError Tornado is throwing after it is stopped
log.getLogger().setLevel(log.FATAL)
ioloop = tornado.ioloop.IOLoop.instance()
ioloop.add_callback(ioloop.stop)
thread.join()
So the main idea here is to keep a reference to the HTTPServer
instance and call its stop()
method. And create_app()
just returns an Application
instance configured with handlers. Now you can use these methods in your unit tests like this:
class FoobarTest(unittest.TestCase):
def setUp(self):
start_app()
def tearDown(self):
stop_app()
def test_foobar(self):
# here the server is up and running, so you can make requests to it
pass