Tornado源碼分析之HTTP服務(wù)請(qǐng)求解析

更新時(shí)間：2023年09月08日 10:08:27 作者：BruceChen7

這篇文章主要為大家介紹了Tornado源碼分析之HTTP服務(wù)請(qǐng)求解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

listen fd的讀事件回調(diào)

代碼版本 tornado1.2版本下httpserver.py

Tornado定義類HTTPServer來表示一個(gè)HTTP服務(wù)器，該類在構(gòu)造函數(shù)中會(huì)傳入事件循環(huán)ioloop，和Application對(duì)象。同時(shí)該HTTPServer提供了如下幾種方法：

listen() 表示該Server的監(jiān)聽方法，調(diào)用該方法時(shí)，通過調(diào)用bind已經(jīng)將套接字設(shè)置成non-blocking，并使用socket.SO_REUSEADDR。
bind() 方法表示Server綁定端口，并設(shè)置socket 為non-blocking.
start() 方法則是在ioloop中啟動(dòng)該服務(wù)器。在不給start()方法傳入任何參數(shù)的情況下，使用單進(jìn)程模型的IOLoop。

在start()方法啟動(dòng)服務(wù)器時(shí)，要向IOLoop中注冊(cè)對(duì)listen fd的可讀事件的回調(diào)，listen fd可讀，表示有新的客戶接入到HTTPServer中。我們來看看接入HTTPServer的事件回調(diào)_handle_events

def _handle_events(self, fd, events):
    while True:
        try:
            connection, address = self._socket.accept()
        except socket.error, e:
            if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
                return
            raise
        if self.ssl_options is not None:
            assert ssl, "Python 2.6+ and OpenSSL required for SSL"
            try:
                connection = ssl.wrap_socket(connection,
                                             server_side=True,
                                             do_handshake_on_connect=False,
                                             **self.ssl_options)
            except ssl.SSLError, err:
                if err.args[0] == ssl.SSL_ERROR_EOF:
                    return connection.close()
                else:
                    raise
            except socket.error, err:
                if err.args[0] == errno.ECONNABORTED:
                    return connection.close()
                else:
                    raise
        try:
            if self.ssl_options is not None:
                stream = iostream.SSLIOStream(connection, io_loop=self.io_loop)
            else:
                stream = iostream.IOStream(connection, io_loop=self.io_loop)
            HTTPConnection(stream, address, self.request_callback,
                           self.no_keep_alive, self.xheaders)
        except:
            logging.error("Error in connection callback", exc_info=True)

不看ssl處理部分。首先基于listen fd調(diào)用accept函數(shù)，獲取connection和address，注意到tornado處理了spurious wakeup的情況：

if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
    return

也就是說，當(dāng)前l(fā)isten fd，沒有數(shù)據(jù)可以讀了，表示沒有新的客戶連接過來，那么就應(yīng)該返回。回調(diào)函數(shù)_handle_events，創(chuàng)建了IOStream對(duì)象stream，

  stream = iostream.SSLIOStream(connection, io_loop=self.io_loop)

注意到創(chuàng)建IOStream對(duì)象會(huì)給單進(jìn)程的IOLoop添加新的回調(diào)函數(shù)，該函數(shù)是用來處理accept fd的讀事件，此時(shí)，在有客戶接入過來，那么IOLoop中的handler有如下幾個(gè)：

其中，

描述符４是listen fd 讀事件的處理回調(diào)
描述符7 是用來喚醒IOLoop _read_waker
描述符9 是新accept文件描述符讀事件的回調(diào).

accept fd 回調(diào)函數(shù)

 def _handle_events(self, fd, events):
        # 確保該連接還存在
        if not self.socket:
            logging.warning("Got events for closed stream %d", fd)
            return
        try:
            # 如果該連接的讀事件產(chǎn)生了，調(diào)用讀回調(diào)
            if events & self.io_loop.READ:
                self._handle_read()
            if not self.socket:
                return
            # 如果是寫事件
            if events & self.io_loop.WRITE:
                # 如果該socket是客戶端創(chuàng)建的socket, 其已經(jīng)被服務(wù)器處理
                if self._connecting:
                    # 連接服務(wù)器端
                    self._handle_connect()
                # 普通的寫事件回調(diào)
                self._handle_write()
            if not self.socket:
                return
            # 錯(cuò)誤
            if events & self.io_loop.ERROR:
                # IOLoop中刪除該文件描述符對(duì)應(yīng)的handler
                # 并關(guān)閉連接.
                self.close()
                return
            state = self.io_loop.ERROR
            # 如果正在讀
            if self.reading():
                # 添加狀態(tài)繼續(xù)讀
                state |= self.io_loop.READ
            if self.writing():
               # 如果正在寫，添加狀態(tài)寫
                state |= self.io_loop.WRITE
            if state != self._state:
                self._state = state
               # 更新對(duì)應(yīng)fd對(duì)應(yīng)的關(guān)注狀態(tài)
                self.io_loop.update_handler(self.socket.fileno(), self._state)
        except:
            logging.error("Uncaught exception, closing connection.",
                          exc_info=True)
            self.close()
            raise

HTTPConnection對(duì)象的創(chuàng)建

接著我們將思路放回到listen fd 的回調(diào)函數(shù)_handle_events中，在IOStream對(duì)象stream創(chuàng)建后，_handle_events將創(chuàng)建HTTPConnection

HTTPConnection(stream, address, self.request_callback,
                               self.no_keep_alive, self.xheaders)

在HTTPConnnection構(gòu)造函數(shù)中，可以看到起調(diào)用了

self.stream.read_until("\r\n\r\n", self._header_callback)

此時(shí)，HTTPConnection中的成員stream，表示的是accept fd中的stream，read_until函數(shù)將會(huì)從accept fd中一直讀到\r\n\n，然后調(diào)用_header_callback來解析HTTP的頭部字段。

read_until函數(shù)

read_until從套接字中就是讀到指定的分隔符為止

 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)

readl_until有兩種退出方式：一種是從buffer中讀到數(shù)據(jù)后，直接返回，另一種是將數(shù)據(jù)讀到buffer，_\read_to_buffer就是將數(shù)據(jù)讀到buffer，如果沒有數(shù)據(jù)，則將該socket的讀事件添加進(jìn)來。

accept fd獲取后，顯然該fd對(duì)應(yīng)的IOStream中的緩沖區(qū)為0，所以_read_from_buffer返回False，流程將執(zhí)行_read_to_buffer。

從緩沖區(qū)讀_read_from_buffer

  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
                self._run_callback(callback,
                                   self._consume(loc + delimiter_len))
                return True
        return False

read_from_buffer有兩種形式的讀：

一種是讀指定字節(jié)的數(shù)據(jù)
一種是讀到指定的分隔符

應(yīng)該注意的是_read_from_buffer中指定了回調(diào)函數(shù)，意思是從socket中讀的數(shù)據(jù)后，使用回調(diào)函數(shù)來消費(fèi)。應(yīng)該注意的是_read_from_buffer并跟socket打交道，其假設(shè)所有的數(shù)據(jù)已經(jīng)在buffer中了。

將數(shù)據(jù)讀到緩沖區(qū)

從socket中讀數(shù)據(jù)到緩沖區(qū)，使用的是_read_to_buffer。read_to_buffer實(shí)際調(diào)用的是_read_from_socket，其從non-blocking中讀一次，最多讀4096個(gè)字節(jié)。注意錯(cuò)誤的處理，當(dāng)我們發(fā)現(xiàn)從socket中讀，發(fā)生了其他的錯(cuò)誤（除了EAGAIN）的時(shí)候，就應(yīng)該關(guān)閉連接。

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
        self._read_buffer.append(chunk)
        if self._read_buffer_size() &gt;= self.max_buffer_size:
            logging.error("Reached maximum read buffer size")
            self.close()
            raise IOError("Reached maximum read buffer size")
        return len(chunk)

該函數(shù)總體來說就是講將據(jù)讀到bufferIOStream中的_read_buffer中，并返回實(shí)際讀的大小。注意錯(cuò)誤的處理：如果所讀的數(shù)據(jù)過大，那么也應(yīng)該關(guān)閉連接。

從socket讀一次數(shù)據(jù)_read_from_socket

 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):
　　　　  # 沒有數(shù)據(jù)　　　　　　
                return None
            else:
                raise
        if not chunk:
　　　 # 客戶端數(shù)據(jù)沒有了，那么表示客戶端關(guān)閉了連接
            self.close()
            return None
        return chunk

LT模式下從socket中讀數(shù)據(jù)_handle_read()

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.
                result = self._read_to_buffer()
            except Exception:
                self.close()
                return
            if result == 0:
                break
            else:
                if self._read_from_buffer():
                    return

在LT模式下從socket中讀分成了兩步：