| ############################################################################## |
| # |
| # Copyright (c) 2001, 2002 Zope Foundation and Contributors. |
| # All Rights Reserved. |
| # |
| # This software is subject to the provisions of the Zope Public License, |
| # Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution. |
| # THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED |
| # WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED |
| # WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS |
| # FOR A PARTICULAR PURPOSE. |
| # |
| ############################################################################## |
| import socket |
| import threading |
| import time |
| import traceback |
| |
| from waitress.buffers import OverflowableBuffer, ReadOnlyFileBasedBuffer |
| from waitress.parser import HTTPRequestParser |
| from waitress.task import ErrorTask, WSGITask |
| from waitress.utilities import InternalServerError |
| |
| from . import wasyncore |
| |
| |
| class ClientDisconnected(Exception): |
| """Raised when attempting to write to a closed socket.""" |
| |
| |
| class HTTPChannel(wasyncore.dispatcher): |
| """ |
| Setting self.requests = [somerequest] prevents more requests from being |
| received until the out buffers have been flushed. |
| |
| Setting self.requests = [] allows more requests to be received. |
| """ |
| |
| task_class = WSGITask |
| error_task_class = ErrorTask |
| parser_class = HTTPRequestParser |
| |
| # A request that has not been received yet completely is stored here |
| request = None |
| last_activity = 0 # Time of last activity |
| will_close = False # set to True to close the socket. |
| close_when_flushed = False # set to True to close the socket when flushed |
| sent_continue = False # used as a latch after sending 100 continue |
| total_outbufs_len = 0 # total bytes ready to send |
| current_outbuf_count = 0 # total bytes written to current outbuf |
| |
| # |
| # ASYNCHRONOUS METHODS (including __init__) |
| # |
| |
| def __init__(self, server, sock, addr, adj, map=None): |
| self.server = server |
| self.adj = adj |
| self.outbufs = [OverflowableBuffer(adj.outbuf_overflow)] |
| self.creation_time = self.last_activity = time.time() |
| self.sendbuf_len = sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF) |
| |
| # requests_lock used to push/pop requests and modify the request that is |
| # currently being created |
| self.requests_lock = threading.Lock() |
| # outbuf_lock used to access any outbuf (expected to use an RLock) |
| self.outbuf_lock = threading.Condition() |
| |
| wasyncore.dispatcher.__init__(self, sock, map=map) |
| |
| # Don't let wasyncore.dispatcher throttle self.addr on us. |
| self.addr = addr |
| self.requests = [] |
| |
| def check_client_disconnected(self): |
| """ |
| This method is inserted into the environment of any created task so it |
| may occasionally check if the client has disconnected and interrupt |
| execution. |
| """ |
| |
| return not self.connected |
| |
| def writable(self): |
| # if there's data in the out buffer or we've been instructed to close |
| # the channel (possibly by our server maintenance logic), run |
| # handle_write |
| |
| return self.total_outbufs_len or self.will_close or self.close_when_flushed |
| |
| def handle_write(self): |
| # Precondition: there's data in the out buffer to be sent, or |
| # there's a pending will_close request |
| |
| if not self.connected: |
| # we dont want to close the channel twice |
| |
| return |
| |
| # try to flush any pending output |
| |
| if not self.requests: |
| # 1. There are no running tasks, so we don't need to try to lock |
| # the outbuf before sending |
| # 2. The data in the out buffer should be sent as soon as possible |
| # because it's either data left over from task output |
| # or a 100 Continue line sent within "received". |
| flush = self._flush_some |
| elif self.total_outbufs_len >= self.adj.send_bytes: |
| # 1. There's a running task, so we need to try to lock |
| # the outbuf before sending |
| # 2. Only try to send if the data in the out buffer is larger |
| # than self.adj_bytes to avoid TCP fragmentation |
| flush = self._flush_some_if_lockable |
| else: |
| # 1. There's not enough data in the out buffer to bother to send |
| # right now. |
| flush = None |
| |
| self._flush_exception(flush) |
| |
| if self.close_when_flushed and not self.total_outbufs_len: |
| self.close_when_flushed = False |
| self.will_close = True |
| |
| if self.will_close: |
| self.handle_close() |
| |
| def _flush_exception(self, flush, do_close=True): |
| if flush: |
| try: |
| return (flush(do_close=do_close), False) |
| except OSError: |
| if self.adj.log_socket_errors: |
| self.logger.exception("Socket error") |
| self.will_close = True |
| |
| return (False, True) |
| except Exception: # pragma: nocover |
| self.logger.exception("Unexpected exception when flushing") |
| self.will_close = True |
| |
| return (False, True) |
| |
| def readable(self): |
| # We might want to read more requests. We can only do this if: |
| # 1. We're not already about to close the connection. |
| # 2. We're not waiting to flush remaining data before closing the |
| # connection |
| # 3. There are not too many tasks already queued |
| # 4. There's no data in the output buffer that needs to be sent |
| # before we potentially create a new task. |
| |
| return not ( |
| self.will_close |
| or self.close_when_flushed |
| or len(self.requests) > self.adj.channel_request_lookahead |
| or self.total_outbufs_len |
| ) |
| |
| def handle_read(self): |
| try: |
| data = self.recv(self.adj.recv_bytes) |
| except OSError: |
| if self.adj.log_socket_errors: |
| self.logger.exception("Socket error") |
| self.handle_close() |
| |
| return |
| |
| if data: |
| self.last_activity = time.time() |
| self.received(data) |
| else: |
| # Client disconnected. |
| self.connected = False |
| |
| def send_continue(self): |
| """ |
| Send a 100-Continue header to the client. This is either called from |
| receive (if no requests are running and the client expects it) or at |
| the end of service (if no more requests are queued and a request has |
| been read partially that expects it). |
| """ |
| self.request.expect_continue = False |
| outbuf_payload = b"HTTP/1.1 100 Continue\r\n\r\n" |
| num_bytes = len(outbuf_payload) |
| with self.outbuf_lock: |
| self.outbufs[-1].append(outbuf_payload) |
| self.current_outbuf_count += num_bytes |
| self.total_outbufs_len += num_bytes |
| self.sent_continue = True |
| self._flush_some() |
| self.request.completed = False |
| |
| def received(self, data): |
| """ |
| Receives input asynchronously and assigns one or more requests to the |
| channel. |
| """ |
| |
| if not data: |
| return False |
| |
| with self.requests_lock: |
| while data: |
| if self.request is None: |
| self.request = self.parser_class(self.adj) |
| n = self.request.received(data) |
| |
| # if there are requests queued, we can not send the continue |
| # header yet since the responses need to be kept in order |
| |
| if ( |
| self.request.expect_continue |
| and self.request.headers_finished |
| and not self.requests |
| and not self.sent_continue |
| ): |
| self.send_continue() |
| |
| if self.request.completed: |
| # The request (with the body) is ready to use. |
| self.sent_continue = False |
| |
| if not self.request.empty: |
| self.requests.append(self.request) |
| |
| if len(self.requests) == 1: |
| # self.requests was empty before so the main thread |
| # is in charge of starting the task. Otherwise, |
| # service() will add a new task after each request |
| # has been processed |
| self.server.add_task(self) |
| self.request = None |
| |
| if n >= len(data): |
| break |
| data = data[n:] |
| |
| return True |
| |
| def _flush_some_if_lockable(self, do_close=True): |
| # Since our task may be appending to the outbuf, we try to acquire |
| # the lock, but we don't block if we can't. |
| |
| if self.outbuf_lock.acquire(False): |
| try: |
| self._flush_some(do_close=do_close) |
| |
| if self.total_outbufs_len < self.adj.outbuf_high_watermark: |
| self.outbuf_lock.notify() |
| finally: |
| self.outbuf_lock.release() |
| |
| def _flush_some(self, do_close=True): |
| # Send as much data as possible to our client |
| |
| sent = 0 |
| dobreak = False |
| |
| while True: |
| outbuf = self.outbufs[0] |
| # use outbuf.__len__ rather than len(outbuf) FBO of not getting |
| # OverflowError on 32-bit Python |
| outbuflen = outbuf.__len__() |
| |
| while outbuflen > 0: |
| chunk = outbuf.get(self.sendbuf_len) |
| num_sent = self.send(chunk, do_close=do_close) |
| |
| if num_sent: |
| outbuf.skip(num_sent, True) |
| outbuflen -= num_sent |
| sent += num_sent |
| self.total_outbufs_len -= num_sent |
| else: |
| # failed to write anything, break out entirely |
| dobreak = True |
| |
| break |
| else: |
| # self.outbufs[-1] must always be a writable outbuf |
| |
| if len(self.outbufs) > 1: |
| toclose = self.outbufs.pop(0) |
| try: |
| toclose.close() |
| except Exception: |
| self.logger.exception("Unexpected error when closing an outbuf") |
| else: |
| # caught up, done flushing for now |
| dobreak = True |
| |
| if dobreak: |
| break |
| |
| if sent: |
| self.last_activity = time.time() |
| |
| return True |
| |
| return False |
| |
| def handle_close(self): |
| with self.outbuf_lock: |
| for outbuf in self.outbufs: |
| try: |
| outbuf.close() |
| except Exception: |
| self.logger.exception( |
| "Unknown exception while trying to close outbuf" |
| ) |
| self.total_outbufs_len = 0 |
| self.connected = False |
| self.outbuf_lock.notify() |
| wasyncore.dispatcher.close(self) |
| |
| def add_channel(self, map=None): |
| """See wasyncore.dispatcher |
| |
| This hook keeps track of opened channels. |
| """ |
| wasyncore.dispatcher.add_channel(self, map) |
| self.server.active_channels[self._fileno] = self |
| |
| def del_channel(self, map=None): |
| """See wasyncore.dispatcher |
| |
| This hook keeps track of closed channels. |
| """ |
| fd = self._fileno # next line sets this to None |
| wasyncore.dispatcher.del_channel(self, map) |
| ac = self.server.active_channels |
| |
| if fd in ac: |
| del ac[fd] |
| |
| # |
| # SYNCHRONOUS METHODS |
| # |
| |
| def write_soon(self, data): |
| if not self.connected: |
| # if the socket is closed then interrupt the task so that it |
| # can cleanup possibly before the app_iter is exhausted |
| raise ClientDisconnected |
| |
| if data: |
| # the async mainloop might be popping data off outbuf; we can |
| # block here waiting for it because we're in a task thread |
| with self.outbuf_lock: |
| self._flush_outbufs_below_high_watermark() |
| |
| if not self.connected: |
| raise ClientDisconnected |
| num_bytes = len(data) |
| |
| if data.__class__ is ReadOnlyFileBasedBuffer: |
| # they used wsgi.file_wrapper |
| self.outbufs.append(data) |
| nextbuf = OverflowableBuffer(self.adj.outbuf_overflow) |
| self.outbufs.append(nextbuf) |
| self.current_outbuf_count = 0 |
| else: |
| if self.current_outbuf_count >= self.adj.outbuf_high_watermark: |
| # rotate to a new buffer if the current buffer has hit |
| # the watermark to avoid it growing unbounded |
| nextbuf = OverflowableBuffer(self.adj.outbuf_overflow) |
| self.outbufs.append(nextbuf) |
| self.current_outbuf_count = 0 |
| self.outbufs[-1].append(data) |
| self.current_outbuf_count += num_bytes |
| self.total_outbufs_len += num_bytes |
| |
| if self.total_outbufs_len >= self.adj.send_bytes: |
| (flushed, exception) = self._flush_exception( |
| self._flush_some, do_close=False |
| ) |
| |
| if ( |
| exception |
| or not flushed |
| or self.total_outbufs_len >= self.adj.send_bytes |
| ): |
| self.server.pull_trigger() |
| |
| return num_bytes |
| |
| return 0 |
| |
| def _flush_outbufs_below_high_watermark(self): |
| # check first to avoid locking if possible |
| |
| if self.total_outbufs_len > self.adj.outbuf_high_watermark: |
| with self.outbuf_lock: |
| (_, exception) = self._flush_exception(self._flush_some, do_close=False) |
| |
| if exception: |
| # An exception happened while flushing, wake up the main |
| # thread, then wait for it to decide what to do next |
| # (probably close the socket, and then just return) |
| self.server.pull_trigger() |
| self.outbuf_lock.wait() |
| |
| return |
| |
| while ( |
| self.connected |
| and self.total_outbufs_len > self.adj.outbuf_high_watermark |
| ): |
| self.server.pull_trigger() |
| self.outbuf_lock.wait() |
| |
| def service(self): |
| """Execute one request. If there are more, we add another task to the |
| server at the end.""" |
| |
| request = self.requests[0] |
| |
| if request.error: |
| task = self.error_task_class(self, request) |
| else: |
| task = self.task_class(self, request) |
| |
| try: |
| if self.connected: |
| task.service() |
| else: |
| task.close_on_finish = True |
| except ClientDisconnected: |
| self.logger.info("Client disconnected while serving %s" % task.request.path) |
| task.close_on_finish = True |
| except Exception: |
| self.logger.exception("Exception while serving %s" % task.request.path) |
| |
| if not task.wrote_header: |
| if self.adj.expose_tracebacks: |
| body = traceback.format_exc() |
| else: |
| body = "The server encountered an unexpected internal server error" |
| req_version = request.version |
| req_headers = request.headers |
| err_request = self.parser_class(self.adj) |
| err_request.error = InternalServerError(body) |
| # copy some original request attributes to fulfill |
| # HTTP 1.1 requirements |
| err_request.version = req_version |
| try: |
| err_request.headers["CONNECTION"] = req_headers["CONNECTION"] |
| except KeyError: |
| pass |
| task = self.error_task_class(self, err_request) |
| try: |
| task.service() # must not fail |
| except ClientDisconnected: |
| task.close_on_finish = True |
| else: |
| task.close_on_finish = True |
| |
| if task.close_on_finish: |
| with self.requests_lock: |
| self.close_when_flushed = True |
| |
| for request in self.requests: |
| request.close() |
| self.requests = [] |
| else: |
| # before processing a new request, ensure there is not too |
| # much data in the outbufs waiting to be flushed |
| # NB: currently readable() returns False while we are |
| # flushing data so we know no new requests will come in |
| # that we need to account for, otherwise it'd be better |
| # to do this check at the start of the request instead of |
| # at the end to account for consecutive service() calls |
| |
| if len(self.requests) > 1: |
| self._flush_outbufs_below_high_watermark() |
| |
| # this is a little hacky but basically it's forcing the |
| # next request to create a new outbuf to avoid sharing |
| # outbufs across requests which can cause outbufs to |
| # not be deallocated regularly when a connection is open |
| # for a long time |
| |
| if self.current_outbuf_count > 0: |
| self.current_outbuf_count = self.adj.outbuf_high_watermark |
| |
| request.close() |
| |
| # Add new task to process the next request |
| with self.requests_lock: |
| self.requests.pop(0) |
| |
| if self.connected and self.requests: |
| self.server.add_task(self) |
| elif ( |
| self.connected |
| and self.request is not None |
| and self.request.expect_continue |
| and self.request.headers_finished |
| and not self.sent_continue |
| ): |
| # A request waits for a signal to continue, but we could |
| # not send it until now because requests were being |
| # processed and the output needs to be kept in order |
| self.send_continue() |
| |
| if self.connected: |
| self.server.pull_trigger() |
| |
| self.last_activity = time.time() |
| |
| def cancel(self): |
| """Cancels all pending / active requests""" |
| self.will_close = True |
| self.connected = False |
| self.last_activity = time.time() |
| self.requests = [] |
