Tornado blocking analysis

requires.txt

gevent
redis
requests
tornado
tornadoredis

tornado_server.py

#!/usr/env/python
# -*- coding: utf-8 -*-

ENABLE_GEVENT = False

if ENABLE_GEVENT:
  import gevent
  import gevent.monkey; gevent.monkey.patch_all(dns=False, aggressive=False)

import redis
import threading
import functools

import tornado.httpserver
import tornado.web
import tornado.ioloop
import tornado.gen
import tornado.options

import logging
logger = logging.getLogger(__name__)


redis = redis.StrictRedis()

#import tornadoredis
#redis = tornadoredis.Client()
#redis.connect()


# Asynchronous gevent decorator
def gasynchronous(func):
  @functools.wraps(func)
  def f(self, *args, **kwargs):
      #self._auto_finish = False
      return gevent.spawn(func, self, *args, **kwargs)
  return f

class MainHandler(tornado.web.RequestHandler):

  def get(self):
    num = self.get_argument('num')
    self.write("Request: %s" % num)

class AsyncHandler(tornado.web.RequestHandler):

  waiters = []

#  @gasynchronous
  @tornado.web.asynchronous
  def get(self):
    self.num = self.get_argument('num')
    logger.debug('Request received %s' % self.num)
    self.wait_for_new_message(self.on_message)

  def wait_for_new_message(self, callback):
    AsyncHandler.waiters.append(callback)

  def on_message(self, message):
    self.write("%s\n" % message)
    self.write("Finished: %s\n" % self.num)
    self.finish()


class PublishHandler(tornado.web.RequestHandler):

  def get(self):
    message = self.get_argument('message', 'Mesage is empty')
    logger.info('New message: %s' % message)
    self.new_message(message)

  def new_message(self, message):
    cls = AsyncHandler
    for callback in cls.waiters:
      callback(message)
    AsyncHandler.waiters = []

class TornadoServer(threading.Thread):

  port = 8889

  def __init__(self):
    threading.Thread.__init__(self)
    self.daemon = True

  def run(self):
    logger.info('Starting tornado on port %d' % self.port)
    application.listen(self.port)
    tornado.ioloop.IOLoop.instance().start()

  def stop(self):
    tornado.ioloop.IOLoop.instance().stop()
    logger.warn('Tornado stoped!')


  def add_periodic_callback(self, callback, interval=5000):
    tornado.ioloop.PeriodicCallback(callback, interval).start()

#@tornado.gen.engine
def blocking_task():
  logger.debug('Blocking...')
  data = redis.get('large_data')
  #data = yield tornado.gen.Task(redis.get, 'large_data')
  logger.debug('Got: %d'% len(data))
  print 'Got: %d'% len(data)

application = tornado.web.Application([
    (r"/", MainHandler),
    (r"/wait", AsyncHandler),
    (r"/pub", PublishHandler),
])

@tornado.gen.engine
def add_large_data():
  import cStringIO
  output = cStringIO.StringIO()
  for i in range(0, 80*1024):
    output.write('#'*1024)

  redis.set('large_data', output.getvalue())
  #yield tornado.gen.Task(redis.set, 'large_data', output.getvalue())
  output.close()
  logger.info("Data loaded!")

add_large_data()

def main():

  tornado.options.parse_command_line()
  server = TornadoServer()
  #server.add_periodic_callback(blocking_task, 10)
  server.run()

if __name__ == '__main__':
  main()

tornado_workers.py

#!/usr/env/python
# -*- coding: utf-8 -*-


ENABLE_GEVENT = False

if ENABLE_GEVENT:
  import gevent
  import gevent.monkey; gevent.monkey.patch_all(dns=False, aggressive=False)

import threading
import requests

import logging
logger = logging.getLogger(__name__)

PORT = 8889

def worker(num, path='/'):
  url = 'http://localhost:%d%s?num=%d' % (PORT, path, num)
  logger.info('Requesting %s ' % url)
  r = requests.get(url)
  print r.text

def thread_requests():

  logger.info('Start requests...')

  threads = []
  for i in range(100):
    t = threading.Thread(target=worker, args=(i,'/wait'))
    t.daemon = True
    threads.append(t)
    t.start()

  requests.get('http://localhost:%d/pub' % PORT)

  for t in threads:
    t.join()

def gevent_requests():
  logger.info('Start requests (gevent)...')
  jobs = [gevent.spawn(worker, i, path='/wait') for i in range(100)]

  gevent.sleep(1)
  requests.get('http://localhost:%d/pub' % PORT)


  gevent.joinall(jobs)


if __name__ == '__main__':
  logging.basicConfig(level=logging.INFO)

  if ENABLE_GEVENT:
    gevent_requests()
  else:
    thread_requests()