tornado_blocking_iostream_read.py

from tornado.ioloop import IOLoop
from tornado.iostream import IOStream
from tornado.netutil import TCPServer
from tornado.gen import engine, Task
from tornado.util import b

import socket
import time
import sys

MB   = 0x100000
dlen = 2 * MB
data = b("A") * dlen + "\r\n"
port = 16661

class Connection(object):
    finished_reads = 0

    def __init__(self, stream):
        self.stream = stream
        self.stream.write("welcome\r\n")
        stream.read_until("\r\n", self.read_finished)

    def read_finished(self, data):
        print len(data)
        self.__class__.finished_reads += 1
        self.stream.write("read: %d\r\n" % self.__class__.finished_reads)

class Server(TCPServer):
    def handle_stream(self, stream, address):
        stream.consecutive_read_iterations = 8
        Connection(stream)

class Client(object):
    finished_reqs = 0

    def __init__(self, host, callback):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        self.stream = IOStream(s)
        self.stream.connect((host, port))
        self.stream.read_until("\r\n", self.got_greeting)
        self.callback = callback

    def got_greeting(self, greeting):
        sys.stderr.write(greeting)
        self.stream.write(data)
        self.stream.read_until("\r\n", self.finished)
        self.callback()

    def finished(self, res):
        sys.stderr.write(res)
        self.stream.close()

        self.__class__.finished_reqs += 1
        if self.__class__.finished_reqs == 2:
            IOLoop.instance().stop()


@engine
def spawn_clients():
    for x in range(2):
        yield Task(Client, sys.argv[1])

if len(sys.argv) >= 2:
    spawn_clients()
else:
    Server().listen(port)

IOLoop.instance().start()