path: root/src/waitress/channel.py

##############################################################################
#
# 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

        if flush:
            try:
                flush()
            except OSError:
                if self.adj.log_socket_errors:
                    self.logger.exception("Socket error")
                self.will_close = True
            except Exception:  # pragma: nocover
                self.logger.exception("Unexpected exception when flushing")
                self.will_close = True

        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 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):
        # 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()

                if self.total_outbufs_len < self.adj.outbuf_high_watermark:
                    self.outbuf_lock.notify()
            finally:
                self.outbuf_lock.release()

    def _flush_some(self):
        # 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)

                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:
                    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:
                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 = []