引言：在分析了部分IOLoop，了解了其工作原理后，就可以看看建立在IOLoop上層的IOStream。IOStream主要提供的功能就是異步的讀寫操作。
IOStream提供的接口有幾個：
1.read_bytes(bytes,callback)
這個接口就是在有固定的字節的數據到來的時候調用回調函數
2.read_until(delimiter,callback)
這個接口的作用是在讀取到固定的字符序列結尾后調用回調函數 callback
3.write(data)
異步寫，就是將數據拷貝到應用層的緩沖區，由IOLoop下層統一調度
看看源碼的Demo
from tornado import ioloop
from tornado import iostream
import socket
#當connect()函數完成后，調用本函數
#本函數在非阻塞異步寫 write 后
#異步讀取以\r\n\r\n 結尾的數據，並調用回調 on_headers
def send_request():
stream.write("GET / HTTP/1.0\r\nHost: friendfeed.com\r\n\r\n")
stream.read_until("\r\n\r\n", on_headers)
#當讀取到以\r\n\r\n結尾的數據的時候，觸發 on_headers回調函數，
#並傳入參數 data（次data數據以\r\n\r\n結尾） 到 on_headers中
def on_headers(data):
headers = {}
for line in data.split("\r\n"):
parts = line.split(":")
if len(parts) == 2:
headers[parts[0].strip()] = parts[1].strip()
#讀取完header后，根據讀取到的length 觸發 on_body 函數，整個過程都是異步的
stream.read_bytes(int(headers["Content-Length"]), on_body)
def on_body(data):
print data
stream.close()
ioloop.IOLoop.instance().stop()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
#創建一個流的實例，綁定描述符 s
stream = iostream.IOStream(s)
#連接並放置相應的回調 send_request，這里后面的操作都是純異步實現的
#connect完成后觸發 send_request的調用
stream.connect(("friendfeed.com", 80), send_request)
ioloop.IOLoop.instance().start()
如代碼所示，上述過程全部是異步的，這也是為什么tornado比其他同類python的開源框架快的原因
IOStream類的源碼注釋：
class IOStream(object):
"""A utility class to write to and read from a non-blocking socket.
We support three methods: write(), read_until(), and read_bytes().
All of the methods take callbacks (since writing and reading are
non-blocking and asynchronous). read_until() reads the socket until
a given delimiter, and read_bytes() reads until a specified number
of bytes have been read from the socket.
The socket parameter may either be connected or unconnected. For
server operations the socket is the result of calling socket.accept().
For client operations the socket is created with socket.socket(),
and may either be connected before passing it to the IOStream or
connected with IOStream.connect.
A very simple (and broken) HTTP client using this class:
from tornado import ioloop
from tornado import iostream
import socket
#當connect()函數完成后，調用本函數
#本函數在非阻塞異步寫 write 后
#異步讀取以\r\n\r\n 結尾的數據，並調用回調 on_headers
def send_request():
stream.write("GET / HTTP/1.0\r\nHost: friendfeed.com\r\n\r\n")
stream.read_until("\r\n\r\n", on_headers)
#當讀取到以\r\n\r\n結尾的數據的時候，觸發 on_headers回調函數，
#並傳入參數 data（次data數據以\r\n\r\n結尾） 到 on_headers中
def on_headers(data):
headers = {}
for line in data.split("\r\n"):
parts = line.split(":")
if len(parts) == 2:
headers[parts[0].strip()] = parts[1].strip()
#讀取完header后，根據讀取到的length 觸發 on_body 函數，整個過程都是異步的
stream.read_bytes(int(headers["Content-Length"]), on_body)
def on_body(data):
print data
stream.close()
ioloop.IOLoop.instance().stop()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
#創建一個流的實例，綁定描述符 s
stream = iostream.IOStream(s)
#連接並放置相應的回調 send_request，這里后面的操作都是純異步實現的
#connect完成后觸發 send_request的調用
stream.connect(("friendfeed.com", 80), send_request)
ioloop.IOLoop.instance().start()
"""
def __init__(self, socket, io_loop=None, max_buffer_size=104857600,
read_chunk_size=4096):
self.socket = socket
self.socket.setblocking(False)
#
self.io_loop = io_loop or ioloop.IOLoop.instance()
self.max_buffer_size = max_buffer_size
self.read_chunk_size = read_chunk_size
#collections 是一個內部類，提供一些抽象類操作
self._read_buffer = collections.deque()
self._write_buffer = collections.deque()
self._write_buffer_frozen = False
self._read_delimiter = None
self._read_bytes = None
#讀寫調用的回調函數
self._read_callback = None
self._write_callback = None
self._close_callback = None
self._connect_callback = None
self._connecting = False
self._state = self.io_loop.ERROR
#注冊次流buffer的 callback 到IOLoop 中
#self._handle_events 就是此 fd 的回調
#相應的事件類型就是 self._state
#一般情況下這個狀態是讀和寫都會回調統一的
# self._handle_events()
with stack_context.NullContext():
self.io_loop.add_handler(
self.socket.fileno(), self._handle_events, self._state)
def connect(self, address, callback=None):
"""Connects the socket to a remote address without blocking.
May only be called if the socket passed to the constructor was
not previously connected. The address parameter is in the
same format as for socket.connect, i.e. a (host, port) tuple.
If callback is specified, it will be called when the
connection is completed.
Note that it is safe to call IOStream.write while the
connection is pending, in which case the data will be written
as soon as the connection is ready. Calling IOStream read
methods before the socket is connected works on some platforms
but is non-portable.
"""
self._connecting = True
try:
self.socket.connect(address)
except socket.error, e:
# In non-blocking mode connect() always raises an exception
if e.args[0] not in (errno.EINPROGRESS, errno.EWOULDBLOCK):
raise
self._connect_callback = stack_context.wrap(callback)
self._add_io_state(self.io_loop.WRITE)
def read_until(self, delimiter, callback):
"""Call callback when we read the given delimiter."""
assert not self._read_callback, "Already reading"
self._read_delimiter = delimiter
self._read_callback = stack_context.wrap(callback)
while True:
# See if we've already got the data from a previous read
if self._read_from_buffer():
return
self._check_closed()
if self._read_to_buffer() == 0:
break
self._add_io_state(self.io_loop.READ)
def read_bytes(self, num_bytes, callback):
"""Call callback when we read the given number of bytes."""
assert not self._read_callback, "Already reading"
if num_bytes == 0:
callback("")
return
self._read_bytes = num_bytes
self._read_callback = stack_context.wrap(callback)
while True:
if self._read_from_buffer():
return
self._check_closed()
if self._read_to_buffer() == 0:
break
self._add_io_state(self.io_loop.READ)
#異步write 很簡單，在IOStream層只需要將數據append 到本地buffer
#注意不是append 到socket的buffer
#設置回調函數
#在異步的寫的過程中，遵循的原則就是能寫多少就寫多少
def write(self, data, callback=None):
"""Write the given data to this stream.
If callback is given, we call it when all of the buffered write
data has been successfully written to the stream. If there was
previously buffered write data and an old write callback, that
callback is simply overwritten with this new callback.
"""
self._check_closed()
#直接將數據append 到_wriet_buffer 中
self._write_buffer.append(data)
#設置狀態為監聽可寫的事件的時候將數據發出去
#下面的工作是由IOLoop層來完成的
self._add_io_state(self.io_loop.WRITE)
#注冊回調函數上去
self._write_callback = stack_context.wrap(callback)
def set_close_callback(self, callback):
"""Call the given callback when the stream is closed."""
self._close_callback = stack_context.wrap(callback)
def close(self):
"""Close this stream."""
if self.socket is not None:
self.io_loop.remove_handler(self.socket.fileno())
self.socket.close()
self.socket = None
if self._close_callback:
self._run_callback(self._close_callback)
def reading(self):
"""Returns true if we are currently reading from the stream."""
return self._read_callback is not None
def writing(self):
"""Returns true if we are currently writing to the stream."""
return bool(self._write_buffer)
def closed(self):
return self.socket is None
#_handle_events 作為上層IOLoop的add_handle()的參數，傳遞到IOLoop中檢測事件發生
#后回調此函數，READ , WRIET , ERROR 事件都采用這個統一的回調
def _handle_events(self, fd, events):
if not self.socket:
logging.warning("Got events for closed stream %d", fd)
return
#根據相應的狀態（事件類型）來統一的調度不同的回調函數
#一般情況下就是調用 read , write
try:
#處理注冊在此描述符上的讀事件
if events & self.io_loop.READ:
self._handle_read()
if not self.socket:
return
#epoll第一階段調用回調的過程中，會將事件類型和fd統一的傳入函數
#見 IOLoop.py 中 start()函數
#注意這里的 connecting 標志位
#在服務端的話是不需要此標志位的
#可讀事件就是回調 _handle_wriet()
if events & self.io_loop.WRITE:
if self._connecting:
self._handle_connect()
self._handle_write()
if not self.socket:
return
#出錯處理
if events & self.io_loop.ERROR:
self.close()
return
state = self.io_loop.ERROR
if self.reading():
state |= self.io_loop.READ
if self.writing():
state |= self.io_loop.WRITE
if state != self._state:
self._state = state
self.io_loop.update_handler(self.socket.fileno(), self._state)
except:
logging.error("Uncaught exception, closing connection.",
exc_info=True)
self.close()
raise
def _run_callback(self, callback, *args, **kwargs):
try:
# Use a NullContext to ensure that all StackContexts are run
# inside our blanket exception handler rather than outside.
with stack_context.NullContext():
callback(*args, **kwargs)
except:
logging.error("Uncaught exception, closing connection.",
exc_info=True)
# Close the socket on an uncaught exception from a user callback
# (It would eventually get closed when the socket object is
# gc'd, but we don't want to rely on gc happening before we
# run out of file descriptors)
self.close()
# Re-raise the exception so that IOLoop.handle_callback_exception
# can see it and log the error
raise
#這個函數是在_handle_events中被調用的
#當描述符可讀后，統一采取統一的方式讀取
#此時的 fd 作為類的全局變量，所以在調用 _read_to_buffer()的時候不需要傳遞這個參數
#_read_to_buffer() 是類方法，將從 socket 中讀取的數據讀入 buffer 中
#其中_read_to_buffer() 會調用 _read_from_socket 然后將數據轉儲到 buffer中統一管理
def _handle_read(self):
while True:
try:
# Read from the socket until we get EWOULDBLOCK or equivalent.
# SSL sockets do some internal buffering, and if the data is
# sitting in the SSL object's buffer select() and friends
# can't see it; the only way to find out if it's there is to
# try to read it.
#從socket讀取數據到本地buffer
#此同步調用其實是因為在這里的上層函數 _handle_read是異步被回調的
#所以這里能夠確定此時的 socket 是可讀的
result = self._read_to_buffer()
except Exception:
self.close()
return
if result == 0:
break
else:
#在處理完數據並將數據通過一系列處理后，調用_read_from_buffer()將數據從buffer
#中取出，然后回調 read_bytes()或者是 read_untils() 中的回調函數
if self._read_from_buffer():
return
#從socket 讀取數據到 chunk
def _read_from_socket(self):
"""Attempts to read from the socket.
Returns the data read or None if there is nothing to read.
May be overridden in subclasses.
"""
try:
chunk = self.socket.recv(self.read_chunk_size)
except socket.error, e:
if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
return None
else:
raise
if not chunk:
self.close()
return None
return chunk
#_read_to_buffer 將調用 _read_from_socket中得到的數據chunk 加入到buffer中，此buffer
#是接收buffer，即 _read_buffer
def _read_to_buffer(self):
"""Reads from the socket and appends the result to the read buffer.
Returns the number of bytes read. Returns 0 if there is nothing
to read (i.e. the read returns EWOULDBLOCK or equivalent). On
error closes the socket and raises an exception.
"""
try:
chunk = self._read_from_socket()
except socket.error, e:
# ssl.SSLError is a subclass of socket.error
logging.warning("Read error on %d: %s",
self.socket.fileno(), e)
self.close()
raise
if chunk is None:
return 0
#將 chunk 數據append 到 _read_buffer
self._read_buffer.append(chunk)
if self._read_buffer_size() >= self.max_buffer_size:
logging.error("Reached maximum read buffer size")
self.close()
raise IOError("Reached maximum read buffer size")
return len(chunk)
#_read_from_buffer 是根據兩個判斷標准來確定需要讀取多少數據
# 一 是根據結尾符號比如說 "\r\n\r\n" 作為結尾
# 二 是根據需要讀取的字符數為依據
def _read_from_buffer(self):
"""Attempts to complete the currently-pending read from the buffer.
Returns True if the read was completed.
"""
#以需要讀取的字符數為標准讀取數據
if self._read_bytes:
if self._read_buffer_size() >= self._read_bytes:
num_bytes = self._read_bytes
callback = self._read_callback
self._read_callback = None
self._read_bytes = None
#調用回調函數
self._run_callback(callback, self._consume(num_bytes))
return True
#判斷字符結尾符號，以此作為讀取的結束符
elif self._read_delimiter:
_merge_prefix(self._read_buffer, sys.maxint)
loc = self._read_buffer[0].find(self._read_delimiter)
if loc != -1:
callback = self._read_callback
delimiter_len = len(self._read_delimiter)
self._read_callback = None
self._read_delimiter = None
#同樣的是調用注冊在此buffer 上的描述符
self._run_callback(callback,
self._consume(loc + delimiter_len))
return True
return False
def _handle_connect(self):
if self._connect_callback is not None:
callback = self._connect_callback
self._connect_callback = None
self._run_callback(callback)
self._connecting = False
#_handle_write 也是作為 _handle_events里面處理可寫事件的函數
def _handle_write(self):
while self._write_buffer:
try:
if not self._write_buffer_frozen:
# On windows, socket.send blows up if given a
# write buffer that's too large, instead of just
# returning the number of bytes it was able to
# process. Therefore we must not call socket.send
# with more than 128KB at a time.
_merge_prefix(self._write_buffer, 128 * 1024)
#將buffer鏈的數據一並發出去
num_bytes = self.socket.send(self._write_buffer[0])
self._write_buffer_frozen = False
_merge_prefix(self._write_buffer, num_bytes)
self._write_buffer.popleft()
except socket.error, e:
if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
# With OpenSSL, after send returns EWOULDBLOCK,
# the very same string object must be used on the
# next call to send. Therefore we suppress
# merging the write buffer after an EWOULDBLOCK.
# A cleaner solution would be to set
# SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER, but this is
# not yet accessible from python
# (http://bugs.python.org/issue8240)
#這里的套接口寫過程當中如果遇到了不可寫的情況，會先停止
#但是此時未發送的buffer 的數據任然在，所以不會導致數據丟失
#因為在發送的時候是有多少發送多少，因為在負責不大的情況下，套接口
#是一直可寫的，所以 epoll返回的都是可寫狀態
self._write_buffer_frozen = True
break
else:
logging.warning("Write error on %d: %s",
self.socket.fileno(), e)
self.close()
return
#如果有 wriet_callback 則調用相應的回調函數
if not self._write_buffer and self._write_callback:
callback = self._write_callback
self._write_callback = None
self._run_callback(callback)
def _consume(self, loc):
_merge_prefix(self._read_buffer, loc)
return self._read_buffer.popleft()
def _check_closed(self):
if not self.socket:
raise IOError("Stream is closed")
def _add_io_state(self, state):
if self.socket is None:
# connection has been closed, so there can be no future events
return
if not self._state & state:
self._state = self._state | state
self.io_loop.update_handler(self.socket.fileno(), self._state)
def _read_buffer_size(self):
return sum(len(chunk) for chunk in self._read_buffer)
調用過程如下所示
結論：由於時間原因，沒有將整個過程敘述的很明白，請諒解。