Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tornado/iostream.py: 14%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#
2# Copyright 2009 Facebook
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16"""Utility classes to write to and read from non-blocking files and sockets.
18Contents:
20* `BaseIOStream`: Generic interface for reading and writing.
21* `IOStream`: Implementation of BaseIOStream using non-blocking sockets.
22* `SSLIOStream`: SSL-aware version of IOStream.
23* `PipeIOStream`: Pipe-based IOStream implementation.
24"""
26import asyncio
27import collections
28import errno
29import io
30import numbers
31import os
32import re
33import socket
34import ssl
35import sys
36import typing
37from collections.abc import Awaitable, Callable
38from re import Pattern
39from types import TracebackType
40from typing import (
41 Any,
42 TypeVar,
43)
45from tornado import ioloop
46from tornado.concurrent import Future, future_set_result_unless_cancelled
47from tornado.log import gen_log
48from tornado.netutil import _client_ssl_defaults, _server_ssl_defaults, ssl_wrap_socket
49from tornado.util import errno_from_exception
51_IOStreamType = TypeVar("_IOStreamType", bound="IOStream")
53# These errnos indicate that a connection has been abruptly terminated.
54# They should be caught and handled less noisily than other errors.
55_ERRNO_CONNRESET = (errno.ECONNRESET, errno.ECONNABORTED, errno.EPIPE, errno.ETIMEDOUT)
57if hasattr(errno, "WSAECONNRESET"):
58 _ERRNO_CONNRESET += ( # type: ignore
59 errno.WSAECONNRESET, # type: ignore
60 errno.WSAECONNABORTED, # type: ignore
61 errno.WSAETIMEDOUT, # type: ignore
62 )
64if sys.platform == "darwin":
65 # OSX appears to have a race condition that causes send(2) to return
66 # EPROTOTYPE if called while a socket is being torn down:
67 # http://erickt.github.io/blog/2014/11/19/adventures-in-debugging-a-potential-osx-kernel-bug/
68 # Since the socket is being closed anyway, treat this as an ECONNRESET
69 # instead of an unexpected error.
70 _ERRNO_CONNRESET += (errno.EPROTOTYPE,) # type: ignore
72_WINDOWS = sys.platform.startswith("win")
75class StreamClosedError(IOError):
76 """Exception raised by `IOStream` methods when the stream is closed.
78 Note that the close callback is scheduled to run *after* other
79 callbacks on the stream (to allow for buffered data to be processed),
80 so you may see this error before you see the close callback.
82 The ``real_error`` attribute contains the underlying error that caused
83 the stream to close (if any).
85 .. versionchanged:: 4.3
86 Added the ``real_error`` attribute.
87 """
89 def __init__(self, real_error: BaseException | None = None) -> None:
90 super().__init__("Stream is closed")
91 self.real_error = real_error
94class UnsatisfiableReadError(Exception):
95 """Exception raised when a read cannot be satisfied.
97 Raised by ``read_until`` and ``read_until_regex`` with a ``max_bytes``
98 argument.
99 """
101 pass
104class StreamBufferFullError(Exception):
105 """Exception raised by `IOStream` methods when the buffer is full."""
108class _StreamBuffer:
109 """
110 A specialized buffer that tries to avoid copies when large pieces
111 of data are encountered.
112 """
114 def __init__(self) -> None:
115 # A sequence of (False, bytearray) and (True, memoryview) objects
116 self._buffers: collections.deque[tuple[bool, bytearray | memoryview]] = (
117 collections.deque()
118 )
119 # Position in the first buffer
120 self._first_pos = 0
121 self._size = 0
123 def __len__(self) -> int:
124 return self._size
126 # Data above this size will be appended separately instead
127 # of extending an existing bytearray
128 _large_buf_threshold = 2048
130 def append(self, data: bytes | bytearray | memoryview) -> None:
131 """
132 Append the given piece of data (should be a buffer-compatible object).
133 """
134 size = len(data)
135 if size > self._large_buf_threshold:
136 # Always make a new view (even of a memoryview) so we can
137 # release it when we're done with it.
138 self._buffers.append((True, memoryview(data)))
139 elif size > 0:
140 if self._buffers:
141 is_memview, b = self._buffers[-1]
142 new_buf = is_memview or len(b) >= self._large_buf_threshold
143 else:
144 new_buf = True
145 if new_buf:
146 self._buffers.append((False, bytearray(data)))
147 else:
148 b += data # type: ignore
150 self._size += size
152 def peek(self, size: int) -> memoryview:
153 """
154 Get a view over at most ``size`` bytes (possibly fewer) at the
155 current buffer position.
157 The caller must release the returned view (e.g. with a ``with``
158 statement) before the buffer is modified again.
159 """
160 assert size > 0
161 try:
162 is_memview, b = self._buffers[0]
163 except IndexError:
164 return memoryview(b"")
166 pos = self._first_pos
167 if is_memview:
168 return typing.cast(memoryview, b[pos : pos + size])
169 else:
170 # Release the full view immediately so that only the returned
171 # slice holds a reference to the bytearray's buffer. (The
172 # bytearray cannot be resized while any view of it is
173 # alive, and on PyPy views are not freed until garbage collection)
174 with memoryview(b) as m:
175 return m[pos : pos + size]
177 def advance(self, size: int) -> None:
178 """
179 Advance the current buffer position by ``size`` bytes.
180 """
181 assert 0 < size <= self._size
182 self._size -= size
183 pos = self._first_pos
185 buffers = self._buffers
186 while buffers and size > 0:
187 is_large, b = buffers[0]
188 b_remain = len(b) - size - pos
189 if b_remain <= 0:
190 buffers.popleft()
191 size -= len(b) - pos
192 if is_large:
193 # Release our view of the caller's data so it can be
194 # resized (if it is a bytearray) without waiting for GC.
195 typing.cast(memoryview, b).release()
196 pos = 0
197 elif is_large:
198 pos += size
199 size = 0
200 else:
201 pos += size
202 del typing.cast(bytearray, b)[:pos]
203 pos = 0
204 size = 0
206 assert size == 0
207 self._first_pos = pos
209 def clear(self) -> None:
210 """
211 Discard all buffered data, releasing our views of large buffers.
212 """
213 for is_large, b in self._buffers:
214 if is_large:
215 typing.cast(memoryview, b).release()
216 self._buffers.clear()
217 self._first_pos = 0
218 self._size = 0
221class BaseIOStream:
222 """A utility class to write to and read from a non-blocking file or socket.
224 We support a non-blocking ``write()`` and a family of ``read_*()``
225 methods. When the operation completes, the ``Awaitable`` will resolve
226 with the data read (or ``None`` for ``write()``). All outstanding
227 ``Awaitables`` will resolve with a `StreamClosedError` when the
228 stream is closed; `.BaseIOStream.set_close_callback` can also be used
229 to be notified of a closed stream.
231 When a stream is closed due to an error, the IOStream's ``error``
232 attribute contains the exception object.
234 Subclasses must implement `fileno`, `close_fd`, `write_to_fd`,
235 `read_from_fd`, and optionally `get_fd_error`.
237 """
239 def __init__(
240 self,
241 max_buffer_size: int | None = None,
242 read_chunk_size: int | None = None,
243 max_write_buffer_size: int | None = None,
244 ) -> None:
245 """`BaseIOStream` constructor.
247 :arg max_buffer_size: Maximum amount of incoming data to buffer;
248 defaults to 100MB.
249 :arg read_chunk_size: Amount of data to read at one time from the
250 underlying transport; defaults to 64KB.
251 :arg max_write_buffer_size: Amount of outgoing data to buffer;
252 defaults to unlimited.
254 .. versionchanged:: 4.0
255 Add the ``max_write_buffer_size`` parameter. Changed default
256 ``read_chunk_size`` to 64KB.
257 .. versionchanged:: 5.0
258 The ``io_loop`` argument (deprecated since version 4.1) has been
259 removed.
260 """
261 self.io_loop = ioloop.IOLoop.current()
262 self.max_buffer_size = max_buffer_size or 104857600
263 # A chunk size that is too close to max_buffer_size can cause
264 # spurious failures.
265 self.read_chunk_size = min(read_chunk_size or 65536, self.max_buffer_size // 2)
266 self.max_write_buffer_size = max_write_buffer_size
267 self.error: BaseException | None = None
268 self._read_buffer = bytearray()
269 self._read_buffer_size = 0
270 self._user_read_buffer = False
271 self._after_user_read_buffer: bytearray | None = None
272 self._write_buffer = _StreamBuffer()
273 self._total_write_index = 0
274 self._total_write_done_index = 0
275 self._read_delimiter: bytes | None = None
276 self._read_regex: Pattern | None = None
277 self._read_max_bytes: int | None = None
278 self._read_bytes: int | None = None
279 self._read_partial = False
280 self._read_until_close = False
281 self._read_future: Future | None = None
282 self._write_futures: collections.deque[tuple[int, Future[None]]] = (
283 collections.deque()
284 )
285 self._close_callback: Callable[[], None] | None = None
286 self._connect_future: Future[IOStream] | None = None
287 # _ssl_connect_future should be defined in SSLIOStream
288 # but it's here so we can clean it up in _signal_closed
289 # TODO: refactor that so subclasses can add additional futures
290 # to be cancelled.
291 self._ssl_connect_future: Future[SSLIOStream] | None = None
292 self._connecting = False
293 self._state: int | None = None
294 self._closed = False
296 def fileno(self) -> int | ioloop._Selectable:
297 """Returns the file descriptor for this stream."""
298 raise NotImplementedError()
300 def close_fd(self) -> None:
301 """Closes the file underlying this stream.
303 ``close_fd`` is called by `BaseIOStream` and should not be called
304 elsewhere; other users should call `close` instead.
305 """
306 raise NotImplementedError()
308 def write_to_fd(self, data: memoryview) -> int:
309 """Attempts to write ``data`` to the underlying file.
311 Returns the number of bytes written.
312 """
313 raise NotImplementedError()
315 def read_from_fd(self, buf: bytearray | memoryview) -> int | None:
316 """Attempts to read from the underlying file.
318 Reads up to ``len(buf)`` bytes, storing them in the buffer.
319 Returns the number of bytes read. Returns None if there was
320 nothing to read (the socket returned `~errno.EWOULDBLOCK` or
321 equivalent), and zero on EOF.
323 .. versionchanged:: 5.0
325 Interface redesigned to take a buffer and return a number
326 of bytes instead of a freshly-allocated object.
327 """
328 raise NotImplementedError()
330 def get_fd_error(self) -> Exception | None:
331 """Returns information about any error on the underlying file.
333 This method is called after the `.IOLoop` has signaled an error on the
334 file descriptor, and should return an Exception (such as `socket.error`
335 with additional information, or None if no such information is
336 available.
337 """
338 return None
340 def read_until_regex(
341 self, regex: bytes, max_bytes: int | None = None
342 ) -> Awaitable[bytes]:
343 """Asynchronously read until we have matched the given regex.
345 The result includes the data that matches the regex and anything
346 that came before it.
348 If ``max_bytes`` is not None, the connection will be closed
349 if more than ``max_bytes`` bytes have been read and the regex is
350 not satisfied.
352 .. versionchanged:: 4.0
353 Added the ``max_bytes`` argument. The ``callback`` argument is
354 now optional and a `.Future` will be returned if it is omitted.
356 .. versionchanged:: 6.0
358 The ``callback`` argument was removed. Use the returned
359 `.Future` instead.
361 """
362 future = self._start_read()
363 self._read_regex = re.compile(regex)
364 self._read_max_bytes = max_bytes
365 try:
366 self._try_inline_read()
367 except UnsatisfiableReadError as e:
368 # Handle this the same way as in _handle_events.
369 gen_log.info("Unsatisfiable read, closing connection: %s" % e)
370 self.close(exc_info=e)
371 return future
372 except:
373 # Ensure that the future doesn't log an error because its
374 # failure was never examined.
375 future.add_done_callback(lambda f: f.exception())
376 raise
377 return future
379 def read_until(
380 self, delimiter: bytes, max_bytes: int | None = None
381 ) -> Awaitable[bytes]:
382 """Asynchronously read until we have found the given delimiter.
384 The result includes all the data read including the delimiter.
386 If ``max_bytes`` is not None, the connection will be closed
387 if more than ``max_bytes`` bytes have been read and the delimiter
388 is not found.
390 .. versionchanged:: 4.0
391 Added the ``max_bytes`` argument. The ``callback`` argument is
392 now optional and a `.Future` will be returned if it is omitted.
394 .. versionchanged:: 6.0
396 The ``callback`` argument was removed. Use the returned
397 `.Future` instead.
398 """
399 future = self._start_read()
400 self._read_delimiter = delimiter
401 self._read_max_bytes = max_bytes
402 try:
403 self._try_inline_read()
404 except UnsatisfiableReadError as e:
405 # Handle this the same way as in _handle_events.
406 gen_log.info("Unsatisfiable read, closing connection: %s" % e)
407 self.close(exc_info=e)
408 return future
409 except:
410 future.add_done_callback(lambda f: f.exception())
411 raise
412 return future
414 def read_bytes(self, num_bytes: int, partial: bool = False) -> Awaitable[bytes]:
415 """Asynchronously read a number of bytes.
417 If ``partial`` is true, data is returned as soon as we have
418 any bytes to return (but never more than ``num_bytes``)
420 .. versionchanged:: 4.0
421 Added the ``partial`` argument. The callback argument is now
422 optional and a `.Future` will be returned if it is omitted.
424 .. versionchanged:: 6.0
426 The ``callback`` and ``streaming_callback`` arguments have
427 been removed. Use the returned `.Future` (and
428 ``partial=True`` for ``streaming_callback``) instead.
430 """
431 future = self._start_read()
432 assert isinstance(num_bytes, numbers.Integral)
433 self._read_bytes = num_bytes
434 self._read_partial = partial
435 try:
436 self._try_inline_read()
437 except:
438 future.add_done_callback(lambda f: f.exception())
439 raise
440 return future
442 def read_into(self, buf: bytearray, partial: bool = False) -> Awaitable[int]:
443 """Asynchronously read a number of bytes.
445 ``buf`` must be a writable buffer into which data will be read.
447 If ``partial`` is true, the callback is run as soon as any bytes
448 have been read. Otherwise, it is run when the ``buf`` has been
449 entirely filled with read data.
451 .. versionadded:: 5.0
453 .. versionchanged:: 6.0
455 The ``callback`` argument was removed. Use the returned
456 `.Future` instead.
458 """
459 future = self._start_read()
461 # First copy data already in read buffer
462 available_bytes = self._read_buffer_size
463 n = len(buf)
464 if available_bytes >= n:
465 with memoryview(self._read_buffer) as m, m[:n] as src:
466 buf[:] = src
467 del self._read_buffer[:n]
468 self._after_user_read_buffer = self._read_buffer
469 elif available_bytes > 0:
470 with memoryview(self._read_buffer) as m:
471 buf[:available_bytes] = m
473 # Set up the supplied buffer as our temporary read buffer.
474 # The original (if it had any data remaining) has been
475 # saved for later.
476 self._user_read_buffer = True
477 self._read_buffer = buf
478 self._read_buffer_size = available_bytes
479 self._read_bytes = n
480 self._read_partial = partial
482 try:
483 self._try_inline_read()
484 except:
485 future.add_done_callback(lambda f: f.exception())
486 raise
487 return future
489 def read_until_close(self) -> Awaitable[bytes]:
490 """Asynchronously reads all data from the socket until it is closed.
492 This will buffer all available data until ``max_buffer_size``
493 is reached. If flow control or cancellation are desired, use a
494 loop with `read_bytes(partial=True) <.read_bytes>` instead.
496 .. versionchanged:: 4.0
497 The callback argument is now optional and a `.Future` will
498 be returned if it is omitted.
500 .. versionchanged:: 6.0
502 The ``callback`` and ``streaming_callback`` arguments have
503 been removed. Use the returned `.Future` (and `read_bytes`
504 with ``partial=True`` for ``streaming_callback``) instead.
506 """
507 future = self._start_read()
508 if self.closed():
509 self._finish_read(self._read_buffer_size)
510 return future
511 self._read_until_close = True
512 try:
513 self._try_inline_read()
514 except:
515 future.add_done_callback(lambda f: f.exception())
516 raise
517 return future
519 def write(self, data: bytes | memoryview) -> "Future[None]":
520 """Asynchronously write the given data to this stream.
522 This method returns a `.Future` that resolves (with a result
523 of ``None``) when the write has been completed.
525 The ``data`` argument may be of type `bytes` or `memoryview`.
527 .. versionchanged:: 4.0
528 Now returns a `.Future` if no callback is given.
530 .. versionchanged:: 4.5
531 Added support for `memoryview` arguments.
533 .. versionchanged:: 6.0
535 The ``callback`` argument was removed. Use the returned
536 `.Future` instead.
538 """
539 self._check_closed()
540 if data:
541 if isinstance(data, memoryview):
542 # Make sure that ``len(data) == data.nbytes``
543 data = data.cast("B")
544 if (
545 self.max_write_buffer_size is not None
546 and len(self._write_buffer) + len(data) > self.max_write_buffer_size
547 ):
548 raise StreamBufferFullError("Reached maximum write buffer size")
549 self._write_buffer.append(data)
550 self._total_write_index += len(data)
551 future: Future[None] = Future()
552 future.add_done_callback(lambda f: f.exception())
553 self._write_futures.append((self._total_write_index, future))
554 if not self._connecting:
555 self._handle_write()
556 if self._write_buffer:
557 self._add_io_state(self.io_loop.WRITE)
558 self._maybe_add_error_listener()
559 return future
561 def set_close_callback(self, callback: Callable[[], None] | None) -> None:
562 """Call the given callback when the stream is closed.
564 This mostly is not necessary for applications that use the
565 `.Future` interface; all outstanding ``Futures`` will resolve
566 with a `StreamClosedError` when the stream is closed. However,
567 it is still useful as a way to signal that the stream has been
568 closed while no other read or write is in progress.
570 Unlike other callback-based interfaces, ``set_close_callback``
571 was not removed in Tornado 6.0.
572 """
573 self._close_callback = callback
574 self._maybe_add_error_listener()
576 def close(
577 self,
578 exc_info: (
579 None
580 | bool
581 | BaseException
582 | tuple[
583 type[BaseException] | None,
584 BaseException | None,
585 TracebackType | None,
586 ]
587 ) = False,
588 ) -> None:
589 """Close this stream.
591 If ``exc_info`` is true, set the ``error`` attribute to the current
592 exception from `sys.exc_info` (or if ``exc_info`` is a tuple,
593 use that instead of `sys.exc_info`).
594 """
595 if not self.closed():
596 if exc_info:
597 if isinstance(exc_info, tuple):
598 self.error = exc_info[1]
599 elif isinstance(exc_info, BaseException):
600 self.error = exc_info
601 else:
602 exc_info = sys.exc_info()
603 if any(exc_info):
604 self.error = exc_info[1]
605 if self._read_until_close:
606 self._read_until_close = False
607 if self.error is None or self._is_connreset(self.error):
608 # A connection reset is treated as a normal close
609 # throughout this class (on some platforms, notably
610 # windows, a peer that closes cleanly may still be
611 # reported as a reset), so deliver the buffered data as
612 # the result of the read.
613 self._finish_read(self._read_buffer_size)
614 # Otherwise the stream is closing because of a real error, so
615 # leave the read future pending for _signal_closed() to fail
616 # with StreamClosedError(real_error=self.error). Resolving it
617 # with the buffered data would report a truncated result as if
618 # it were complete.
619 elif self._read_future is not None:
620 # resolve reads that are pending and ready to complete
621 try:
622 pos = self._find_read_pos()
623 except UnsatisfiableReadError:
624 pass
625 else:
626 if pos is not None:
627 self._read_from_buffer(pos)
628 if self._state is not None:
629 self.io_loop.remove_handler(self.fileno())
630 self._state = None
631 self.close_fd()
632 self._closed = True
633 self._signal_closed()
635 def _signal_closed(self) -> None:
636 futures: list[Future] = []
637 if self._read_future is not None:
638 futures.append(self._read_future)
639 self._read_future = None
640 futures += [future for _, future in self._write_futures]
641 self._write_futures.clear()
642 if self._connect_future is not None:
643 futures.append(self._connect_future)
644 self._connect_future = None
645 for future in futures:
646 if not future.done():
647 future.set_exception(StreamClosedError(real_error=self.error))
648 # Reference the exception to silence warnings. Annoyingly,
649 # this raises if the future was cancelled, but just
650 # returns any other error.
651 try:
652 future.exception()
653 except asyncio.CancelledError:
654 pass
655 if self._ssl_connect_future is not None:
656 # _ssl_connect_future expects to see the real exception (typically
657 # an ssl.SSLError), not just StreamClosedError.
658 if not self._ssl_connect_future.done():
659 if self.error is not None:
660 self._ssl_connect_future.set_exception(self.error)
661 else:
662 self._ssl_connect_future.set_exception(StreamClosedError())
663 try:
664 self._ssl_connect_future.exception()
665 except asyncio.CancelledError:
666 pass
667 self._ssl_connect_future = None
668 if self._close_callback is not None:
669 cb = self._close_callback
670 self._close_callback = None
671 self.io_loop.add_callback(cb)
672 # Clear the buffers so they can be cleared immediately even
673 # if the IOStream object is kept alive by a reference cycle.
674 # TODO: Clear the read buffer too; it currently breaks some tests.
675 if self._write_buffer is not None:
676 # Release our views of the caller's data from large writes
677 # that did not complete, so the caller can resize it (on PyPy,
678 # views are not freed until garbage collection).
679 self._write_buffer.clear()
680 self._write_buffer = None # type: ignore
682 def reading(self) -> bool:
683 """Returns ``True`` if we are currently reading from the stream."""
684 return self._read_future is not None
686 def writing(self) -> bool:
687 """Returns ``True`` if we are currently writing to the stream."""
688 return bool(self._write_buffer)
690 def closed(self) -> bool:
691 """Returns ``True`` if the stream has been closed."""
692 return self._closed
694 def set_nodelay(self, value: bool) -> None:
695 """Sets the no-delay flag for this stream.
697 By default, data written to TCP streams may be held for a time
698 to make the most efficient use of bandwidth (according to
699 Nagle's algorithm). The no-delay flag requests that data be
700 written as soon as possible, even if doing so would consume
701 additional bandwidth.
703 This flag is currently defined only for TCP-based ``IOStreams``.
705 .. versionadded:: 3.1
706 """
707 pass
709 def _handle_connect(self) -> None:
710 raise NotImplementedError()
712 def _handle_events(self, fd: int | ioloop._Selectable, events: int) -> None:
713 if self.closed():
714 gen_log.warning("Got events for closed stream %s", fd)
715 return
716 try:
717 if self._connecting:
718 # Most IOLoops will report a write failed connect
719 # with the WRITE event, but SelectIOLoop reports a
720 # READ as well so we must check for connecting before
721 # either.
722 self._handle_connect()
723 if self.closed():
724 return
725 if events & self.io_loop.READ:
726 self._handle_read()
727 if self.closed():
728 return
729 if events & self.io_loop.WRITE:
730 self._handle_write()
731 if self.closed():
732 return
733 if events & self.io_loop.ERROR:
734 self.error = self.get_fd_error()
735 # We may have queued up a user callback in _handle_read or
736 # _handle_write, so don't close the IOStream until those
737 # callbacks have had a chance to run.
738 self.io_loop.add_callback(self.close)
739 return
740 state = self.io_loop.ERROR
741 if self.reading():
742 state |= self.io_loop.READ
743 if self.writing():
744 state |= self.io_loop.WRITE
745 if state == self.io_loop.ERROR and self._read_buffer_size == 0:
746 # If the connection is idle, listen for reads too so
747 # we can tell if the connection is closed. If there is
748 # data in the read buffer we won't run the close callback
749 # yet anyway, so we don't need to listen in this case.
750 state |= self.io_loop.READ
751 if state != self._state:
752 assert (
753 self._state is not None
754 ), "shouldn't happen: _handle_events without self._state"
755 self._state = state
756 self.io_loop.update_handler(self.fileno(), self._state)
757 except UnsatisfiableReadError as e:
758 gen_log.info("Unsatisfiable read, closing connection: %s" % e)
759 self.close(exc_info=e)
760 except Exception as e:
761 gen_log.error("Uncaught exception, closing connection.", exc_info=True)
762 self.close(exc_info=e)
763 raise
765 def _read_to_buffer_loop(self) -> int | None:
766 # This method is called from _handle_read and _try_inline_read.
767 if self._read_bytes is not None:
768 target_bytes: int | None = self._read_bytes
769 elif self._read_max_bytes is not None:
770 target_bytes = self._read_max_bytes
771 elif self.reading():
772 # For read_until without max_bytes, or
773 # read_until_close, read as much as we can before
774 # scanning for the delimiter.
775 target_bytes = None
776 else:
777 target_bytes = 0
778 next_find_pos = 0
779 while not self.closed():
780 # Read from the socket until we get EWOULDBLOCK or equivalent.
781 # SSL sockets do some internal buffering, and if the data is
782 # sitting in the SSL object's buffer select() and friends
783 # can't see it; the only way to find out if it's there is to
784 # try to read it.
785 if self._read_to_buffer() == 0:
786 break
788 # If we've read all the bytes we can use, break out of
789 # this loop.
791 # If we've reached target_bytes, we know we're done.
792 if target_bytes is not None and self._read_buffer_size >= target_bytes:
793 break
795 # Otherwise, we need to call the more expensive find_read_pos.
796 # It's inefficient to do this on every read, so instead
797 # do it on the first read and whenever the read buffer
798 # size has doubled.
799 if self._read_buffer_size >= next_find_pos:
800 pos = self._find_read_pos()
801 if pos is not None:
802 return pos
803 next_find_pos = self._read_buffer_size * 2
804 return self._find_read_pos()
806 def _handle_read(self) -> None:
807 try:
808 pos = self._read_to_buffer_loop()
809 except UnsatisfiableReadError:
810 raise
811 except asyncio.CancelledError:
812 raise
813 except Exception as e:
814 gen_log.warning("error on read: %s" % e)
815 self.close(exc_info=e)
816 return
817 if pos is not None:
818 self._read_from_buffer(pos)
820 def _start_read(self) -> Future:
821 if self._read_future is not None:
822 # It is an error to start a read while a prior read is unresolved.
823 # However, if the prior read is unresolved because the stream was
824 # closed without satisfying it, it's better to raise
825 # StreamClosedError instead of AssertionError. In particular, this
826 # situation occurs in harmless situations in http1connection.py and
827 # an AssertionError would be logged noisily.
828 #
829 # On the other hand, it is legal to start a new read while the
830 # stream is closed, in case the read can be satisfied from the
831 # read buffer. So we only want to check the closed status of the
832 # stream if we need to decide what kind of error to raise for
833 # "already reading".
834 #
835 # These conditions have proven difficult to test; we have no
836 # unittests that reliably verify this behavior so be careful
837 # when making changes here. See #2651 and #2719.
838 self._check_closed()
839 assert self._read_future is None, "Already reading"
840 self._read_future = Future()
841 return self._read_future
843 def _finish_read(self, size: int) -> None:
844 if self._user_read_buffer:
845 self._read_buffer = self._after_user_read_buffer or bytearray()
846 self._after_user_read_buffer = None
847 self._read_buffer_size = len(self._read_buffer)
848 self._user_read_buffer = False
849 result: int | bytes = size
850 else:
851 result = self._consume(size)
852 if self._read_future is not None:
853 future = self._read_future
854 self._read_future = None
855 future_set_result_unless_cancelled(future, result)
856 self._maybe_add_error_listener()
858 def _try_inline_read(self) -> None:
859 """Attempt to complete the current read operation from buffered data.
861 If the read can be completed without blocking, schedules the
862 read callback on the next IOLoop iteration; otherwise starts
863 listening for reads on the socket.
864 """
865 # See if we've already got the data from a previous read
866 pos = self._find_read_pos()
867 if pos is not None:
868 self._read_from_buffer(pos)
869 return
870 self._check_closed()
871 pos = self._read_to_buffer_loop()
872 if pos is not None:
873 self._read_from_buffer(pos)
874 return
875 # We couldn't satisfy the read inline, so make sure we're
876 # listening for new data unless the stream is closed.
877 if not self.closed():
878 self._add_io_state(ioloop.IOLoop.READ)
880 def _read_to_buffer(self) -> int | None:
881 """Reads from the socket and appends the result to the read buffer.
883 Returns the number of bytes read. Returns 0 if there is nothing
884 to read (i.e. the read returns EWOULDBLOCK or equivalent). On
885 error closes the socket and raises an exception.
886 """
887 buf: bytearray | None = None
888 try:
889 while True:
890 try:
891 if self._user_read_buffer:
892 # Release the views explicitly so the caller can
893 # resize their buffer after the read completes
894 # (PyPy does not free them until garbage collection).
895 with (
896 memoryview(self._read_buffer) as m,
897 m[self._read_buffer_size :] as view,
898 ):
899 bytes_read = self.read_from_fd(view)
900 else:
901 buf = bytearray(self.read_chunk_size)
902 bytes_read = self.read_from_fd(buf)
903 except OSError as e:
904 # ssl.SSLError is a subclass of socket.error
905 if self._is_connreset(e):
906 # Treat ECONNRESET as a connection close rather than
907 # an error to minimize log spam (the exception will
908 # be available on self.error for apps that care).
909 self.close(exc_info=e)
910 return None
911 self.close(exc_info=e)
912 raise
913 break
914 if bytes_read is None:
915 return 0
916 elif bytes_read == 0:
917 self.close()
918 return 0
919 if not self._user_read_buffer:
920 assert buf is not None
921 with memoryview(buf) as m, m[:bytes_read] as view:
922 self._read_buffer += view
923 self._read_buffer_size += bytes_read
924 finally:
925 # Break the reference to buf so we don't waste a chunk's worth of
926 # memory in case an exception hangs on to our stack frame.
927 del buf
928 if self._read_buffer_size > self.max_buffer_size:
929 gen_log.error("Reached maximum read buffer size")
930 buffer_full_error = StreamBufferFullError(
931 "Reached maximum read buffer size"
932 )
933 self.close(exc_info=buffer_full_error)
934 raise buffer_full_error
935 return bytes_read
937 def _read_from_buffer(self, pos: int) -> None:
938 """Attempts to complete the currently-pending read from the buffer.
940 The argument is either a position in the read buffer or None,
941 as returned by _find_read_pos.
942 """
943 self._read_bytes = self._read_delimiter = self._read_regex = None
944 self._read_partial = False
945 self._finish_read(pos)
947 def _find_read_pos(self) -> int | None:
948 """Attempts to find a position in the read buffer that satisfies
949 the currently-pending read.
951 Returns a position in the buffer if the current read can be satisfied,
952 or None if it cannot.
953 """
954 if self._read_bytes is not None and (
955 self._read_buffer_size >= self._read_bytes
956 or (self._read_partial and self._read_buffer_size > 0)
957 ):
958 num_bytes = min(self._read_bytes, self._read_buffer_size)
959 return num_bytes
960 elif self._read_delimiter is not None:
961 # Multi-byte delimiters (e.g. '\r\n') may straddle two
962 # chunks in the read buffer, so we can't easily find them
963 # without collapsing the buffer. However, since protocols
964 # using delimited reads (as opposed to reads of a known
965 # length) tend to be "line" oriented, the delimiter is likely
966 # to be in the first few chunks. Merge the buffer gradually
967 # since large merges are relatively expensive and get undone in
968 # _consume().
969 if self._read_buffer:
970 loc = self._read_buffer.find(self._read_delimiter)
971 if loc != -1:
972 delimiter_len = len(self._read_delimiter)
973 self._check_max_bytes(self._read_delimiter, loc + delimiter_len)
974 return loc + delimiter_len
975 self._check_max_bytes(self._read_delimiter, self._read_buffer_size)
976 elif self._read_regex is not None:
977 if self._read_buffer:
978 m = self._read_regex.search(self._read_buffer)
979 if m is not None:
980 loc = m.end()
981 self._check_max_bytes(self._read_regex, loc)
982 return loc
983 self._check_max_bytes(self._read_regex, self._read_buffer_size)
984 return None
986 def _check_max_bytes(self, delimiter: bytes | Pattern, size: int) -> None:
987 if self._read_max_bytes is not None and size > self._read_max_bytes:
988 raise UnsatisfiableReadError(
989 "delimiter %r not found within %d bytes"
990 % (delimiter, self._read_max_bytes)
991 )
993 def _handle_write(self) -> None:
994 while True:
995 size = len(self._write_buffer)
996 if not size:
997 break
998 assert size > 0
999 try:
1000 if _WINDOWS:
1001 # On windows, socket.send blows up if given a
1002 # write buffer that's too large, instead of just
1003 # returning the number of bytes it was able to
1004 # process. Therefore we must not call socket.send
1005 # with more than 128KB at a time.
1006 size = 128 * 1024
1008 with self._write_buffer.peek(size) as data:
1009 num_bytes = self.write_to_fd(data)
1010 if num_bytes == 0:
1011 break
1012 self._write_buffer.advance(num_bytes)
1013 self._total_write_done_index += num_bytes
1014 except BlockingIOError:
1015 break
1016 except OSError as e:
1017 if not self._is_connreset(e):
1018 # Broken pipe errors are usually caused by connection
1019 # reset, and its better to not log EPIPE errors to
1020 # minimize log spam
1021 gen_log.warning("Write error on %s: %s", self.fileno(), e)
1022 self.close(exc_info=e)
1023 return
1025 while self._write_futures:
1026 index, future = self._write_futures[0]
1027 if index > self._total_write_done_index:
1028 break
1029 self._write_futures.popleft()
1030 future_set_result_unless_cancelled(future, None)
1032 def _consume(self, loc: int) -> bytes:
1033 # Consume loc bytes from the read buffer and return them
1034 if loc == 0:
1035 return b""
1036 assert loc <= self._read_buffer_size
1037 buf = self._read_buffer
1038 if loc == len(buf):
1039 # Taking the whole buffer is the most common case, and bytes()
1040 # copies it without a memoryview.
1041 b = bytes(buf)
1042 elif loc <= 2048:
1043 # Copying a small slice twice is cheaper than creating and
1044 # releasing memoryviews.
1045 b = bytes(buf[:loc])
1046 else:
1047 # Slice the bytearray buffer into bytes, without intermediate
1048 # copying. The views must be released before the bytearray is
1049 # resized (explicitly, since PyPy does not free them until
1050 # garbage collection).
1051 with memoryview(buf) as m, m[:loc] as view:
1052 b = view.tobytes()
1053 self._read_buffer_size -= loc
1054 del buf[:loc]
1055 return b
1057 def _check_closed(self) -> None:
1058 if self.closed():
1059 raise StreamClosedError(real_error=self.error)
1061 def _maybe_add_error_listener(self) -> None:
1062 # This method is part of an optimization: to detect a connection that
1063 # is closed when we're not actively reading or writing, we must listen
1064 # for read events. However, it is inefficient to do this when the
1065 # connection is first established because we are going to read or write
1066 # immediately anyway. Instead, we insert checks at various times to
1067 # see if the connection is idle and add the read listener then.
1068 if self._state is None or self._state == ioloop.IOLoop.ERROR:
1069 if (
1070 not self.closed()
1071 and self._read_buffer_size == 0
1072 and self._close_callback is not None
1073 ):
1074 self._add_io_state(ioloop.IOLoop.READ)
1076 def _add_io_state(self, state: int) -> None:
1077 """Adds `state` (IOLoop.{READ,WRITE} flags) to our event handler.
1079 Implementation notes: Reads and writes have a fast path and a
1080 slow path. The fast path reads synchronously from socket
1081 buffers, while the slow path uses `_add_io_state` to schedule
1082 an IOLoop callback.
1084 To detect closed connections, we must have called
1085 `_add_io_state` at some point, but we want to delay this as
1086 much as possible so we don't have to set an `IOLoop.ERROR`
1087 listener that will be overwritten by the next slow-path
1088 operation. If a sequence of fast-path ops do not end in a
1089 slow-path op, (e.g. for an @asynchronous long-poll request),
1090 we must add the error handler.
1092 TODO: reevaluate this now that callbacks are gone.
1094 """
1095 if self.closed():
1096 # connection has been closed, so there can be no future events
1097 return
1098 if self._state is None:
1099 self._state = ioloop.IOLoop.ERROR | state
1100 self.io_loop.add_handler(self.fileno(), self._handle_events, self._state)
1101 elif not self._state & state:
1102 self._state = self._state | state
1103 self.io_loop.update_handler(self.fileno(), self._state)
1105 def _is_connreset(self, exc: BaseException) -> bool:
1106 """Return ``True`` if exc is ECONNRESET or equivalent.
1108 May be overridden in subclasses.
1109 """
1110 return (
1111 isinstance(exc, (socket.error, IOError))
1112 and errno_from_exception(exc) in _ERRNO_CONNRESET
1113 )
1116class IOStream(BaseIOStream):
1117 r"""Socket-based `IOStream` implementation.
1119 This class supports the read and write methods from `BaseIOStream`
1120 plus a `connect` method.
1122 The ``socket`` parameter may either be connected or unconnected.
1123 For server operations the socket is the result of calling
1124 `socket.accept <socket.socket.accept>`. For client operations the
1125 socket is created with `socket.socket`, and may either be
1126 connected before passing it to the `IOStream` or connected with
1127 `IOStream.connect`.
1129 A very simple (and broken) HTTP client using this class:
1131 .. testcode::
1133 import socket
1134 import tornado
1136 async def main():
1137 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
1138 stream = tornado.iostream.IOStream(s)
1139 await stream.connect(("friendfeed.com", 80))
1140 await stream.write(b"GET / HTTP/1.0\r\nHost: friendfeed.com\r\n\r\n")
1141 header_data = await stream.read_until(b"\r\n\r\n")
1142 headers = {}
1143 for line in header_data.split(b"\r\n"):
1144 parts = line.split(b":")
1145 if len(parts) == 2:
1146 headers[parts[0].strip()] = parts[1].strip()
1147 body_data = await stream.read_bytes(int(headers[b"Content-Length"]))
1148 print(body_data)
1149 stream.close()
1151 if __name__ == '__main__':
1152 asyncio.run(main())
1154 """
1156 def __init__(self, socket: socket.socket, *args: Any, **kwargs: Any) -> None:
1157 self.socket = socket
1158 self.socket.setblocking(False)
1159 super().__init__(*args, **kwargs)
1161 def fileno(self) -> int | ioloop._Selectable:
1162 return self.socket
1164 def close_fd(self) -> None:
1165 self.socket.close()
1166 self.socket = None # type: ignore
1168 def get_fd_error(self) -> Exception | None:
1169 errno = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
1170 return socket.error(errno, os.strerror(errno))
1172 def read_from_fd(self, buf: bytearray | memoryview) -> int | None:
1173 try:
1174 return self.socket.recv_into(buf, len(buf))
1175 except BlockingIOError:
1176 return None
1177 finally:
1178 del buf
1180 def write_to_fd(self, data: memoryview) -> int:
1181 try:
1182 return self.socket.send(data) # type: ignore
1183 finally:
1184 # Avoid keeping to data, which can be a memoryview.
1185 # See https://github.com/tornadoweb/tornado/pull/2008
1186 del data
1188 def connect(
1189 self: _IOStreamType, address: Any, server_hostname: str | None = None
1190 ) -> "Future[_IOStreamType]":
1191 """Connects the socket to a remote address without blocking.
1193 May only be called if the socket passed to the constructor was
1194 not previously connected. The address parameter is in the
1195 same format as for `socket.connect <socket.socket.connect>` for
1196 the type of socket passed to the IOStream constructor,
1197 e.g. an ``(ip, port)`` tuple. Hostnames are accepted here,
1198 but will be resolved synchronously and block the IOLoop.
1199 If you have a hostname instead of an IP address, the `.TCPClient`
1200 class is recommended instead of calling this method directly.
1201 `.TCPClient` will do asynchronous DNS resolution and handle
1202 both IPv4 and IPv6.
1204 If ``callback`` is specified, it will be called with no
1205 arguments when the connection is completed; if not this method
1206 returns a `.Future` (whose result after a successful
1207 connection will be the stream itself).
1209 In SSL mode, the ``server_hostname`` parameter will be used
1210 for certificate validation (unless disabled in the
1211 ``ssl_options``) and SNI.
1213 Note that it is safe to call `IOStream.write
1214 <BaseIOStream.write>` while the connection is pending, in
1215 which case the data will be written as soon as the connection
1216 is ready. Calling `IOStream` read methods before the socket is
1217 connected works on some platforms but is non-portable.
1219 .. versionchanged:: 4.0
1220 If no callback is given, returns a `.Future`.
1222 .. versionchanged:: 4.2
1223 SSL certificates are validated by default; pass
1224 ``ssl_options=dict(cert_reqs=ssl.CERT_NONE)`` or a
1225 suitably-configured `ssl.SSLContext` to the
1226 `SSLIOStream` constructor to disable.
1228 .. versionchanged:: 6.0
1230 The ``callback`` argument was removed. Use the returned
1231 `.Future` instead.
1233 """
1234 self._connecting = True
1235 future: Future[_IOStreamType] = Future()
1236 self._connect_future = typing.cast("Future[IOStream]", future)
1237 try:
1238 self.socket.connect(address)
1239 except BlockingIOError:
1240 # In non-blocking mode we expect connect() to raise an
1241 # exception with EINPROGRESS or EWOULDBLOCK.
1242 pass
1243 except OSError as e:
1244 # On freebsd, other errors such as ECONNREFUSED may be
1245 # returned immediately when attempting to connect to
1246 # localhost, so handle them the same way as an error
1247 # reported later in _handle_connect.
1248 if future is None:
1249 gen_log.warning("Connect error on fd %s: %s", self.socket.fileno(), e)
1250 self.close(exc_info=e)
1251 return future
1252 self._add_io_state(self.io_loop.WRITE)
1253 return future
1255 def start_tls(
1256 self,
1257 server_side: bool,
1258 ssl_options: dict[str, Any] | ssl.SSLContext | None = None,
1259 server_hostname: str | None = None,
1260 ) -> "Future[SSLIOStream]":
1261 """Convert this `IOStream` to an `SSLIOStream`.
1263 This enables protocols that begin in clear-text mode and
1264 switch to SSL after some initial negotiation (such as the
1265 ``STARTTLS`` extension to SMTP and IMAP).
1267 This method cannot be used if there are outstanding reads
1268 or writes on the stream, or if there is any data in the
1269 IOStream's buffer (data in the operating system's socket
1270 buffer is allowed). This means it must generally be used
1271 immediately after reading or writing the last clear-text
1272 data. It can also be used immediately after connecting,
1273 before any reads or writes.
1275 The ``ssl_options`` argument may be either an `ssl.SSLContext`
1276 object or a dictionary of keyword arguments for the
1277 `ssl.SSLContext.wrap_socket` function. The ``server_hostname`` argument
1278 will be used for certificate validation unless disabled
1279 in the ``ssl_options``.
1281 This method returns a `.Future` whose result is the new
1282 `SSLIOStream`. After this method has been called,
1283 any other operation on the original stream is undefined.
1285 If a close callback is defined on this stream, it will be
1286 transferred to the new stream.
1288 .. versionadded:: 4.0
1290 .. versionchanged:: 4.2
1291 SSL certificates are validated by default; pass
1292 ``ssl_options=dict(cert_reqs=ssl.CERT_NONE)`` or a
1293 suitably-configured `ssl.SSLContext` to disable.
1295 .. versionchanged:: 6.6
1296 Cancelling the returned `.Future` before the handshake completes
1297 now closes the new stream and its socket.
1298 """
1299 if (
1300 self._read_future
1301 or self._write_futures
1302 or self._connect_future
1303 or self._closed
1304 or self._read_buffer
1305 or self._write_buffer
1306 ):
1307 raise ValueError("IOStream is not idle; cannot convert to SSL")
1308 if ssl_options is None:
1309 if server_side:
1310 ssl_options = _server_ssl_defaults()
1311 else:
1312 ssl_options = _client_ssl_defaults()
1314 socket = self.socket
1315 self.io_loop.remove_handler(socket)
1316 self.socket = None # type: ignore
1317 socket = ssl_wrap_socket(
1318 socket,
1319 ssl_options,
1320 server_hostname=server_hostname,
1321 server_side=server_side,
1322 do_handshake_on_connect=False,
1323 )
1324 orig_close_callback = self._close_callback
1325 self._close_callback = None
1327 future: Future[SSLIOStream] = Future()
1328 ssl_stream = SSLIOStream(socket, ssl_options=ssl_options)
1329 ssl_stream.set_close_callback(orig_close_callback)
1330 ssl_stream._ssl_connect_future = future
1331 ssl_stream.max_buffer_size = self.max_buffer_size
1332 ssl_stream.read_chunk_size = self.read_chunk_size
1333 # The socket now belongs to ssl_stream, which the caller can't see
1334 # until the handshake completes. If the caller gives up (e.g. on a
1335 # timeout) and cancels the future, close the stream so the socket
1336 # is not leaked.
1337 future.add_done_callback(
1338 lambda f: ssl_stream.close() if f.cancelled() else None
1339 )
1340 return future
1342 def _handle_connect(self) -> None:
1343 try:
1344 err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
1345 except OSError as e:
1346 # Hurd doesn't allow SO_ERROR for loopback sockets because all
1347 # errors for such sockets are reported synchronously.
1348 if errno_from_exception(e) == errno.ENOPROTOOPT:
1349 err = 0
1350 if err != 0:
1351 self.error = socket.error(err, os.strerror(err))
1352 # IOLoop implementations may vary: some of them return
1353 # an error state before the socket becomes writable, so
1354 # in that case a connection failure would be handled by the
1355 # error path in _handle_events instead of here.
1356 if self._connect_future is None:
1357 gen_log.warning(
1358 "Connect error on fd %s: %s",
1359 self.socket.fileno(),
1360 errno.errorcode[err],
1361 )
1362 self.close()
1363 return
1364 if self._connect_future is not None:
1365 future = self._connect_future
1366 self._connect_future = None
1367 future_set_result_unless_cancelled(future, self)
1368 self._connecting = False
1370 def set_nodelay(self, value: bool) -> None:
1371 if self.socket is not None and self.socket.family in (
1372 socket.AF_INET,
1373 socket.AF_INET6,
1374 ):
1375 try:
1376 self.socket.setsockopt(
1377 socket.IPPROTO_TCP, socket.TCP_NODELAY, 1 if value else 0
1378 )
1379 except OSError as e:
1380 # Sometimes setsockopt will fail if the socket is closed
1381 # at the wrong time. This can happen with HTTPServer
1382 # resetting the value to ``False`` between requests.
1383 if e.errno != errno.EINVAL and not self._is_connreset(e):
1384 raise
1387class SSLIOStream(IOStream):
1388 """A utility class to write to and read from a non-blocking SSL socket.
1390 If the socket passed to the constructor is already connected,
1391 it should be wrapped with::
1393 ssl.SSLContext(...).wrap_socket(sock, do_handshake_on_connect=False, **kwargs)
1395 before constructing the `SSLIOStream`. Unconnected sockets will be
1396 wrapped when `IOStream.connect` is finished.
1397 """
1399 socket: ssl.SSLSocket
1401 def __init__(self, *args: Any, **kwargs: Any) -> None:
1402 """The ``ssl_options`` keyword argument may either be an
1403 `ssl.SSLContext` object or a dictionary of keywords arguments
1404 for `ssl.SSLContext.wrap_socket`
1405 """
1406 self._ssl_options = kwargs.pop("ssl_options", None)
1407 if self._ssl_options is None:
1408 self._ssl_options = _client_ssl_defaults()
1409 super().__init__(*args, **kwargs)
1410 self._ssl_accepting = True
1411 self._handshake_reading = False
1412 self._handshake_writing = False
1413 self._server_hostname: str | None = None
1415 # If the socket is already connected, attempt to start the handshake.
1416 try:
1417 self.socket.getpeername()
1418 except OSError:
1419 pass
1420 else:
1421 # Indirectly start the handshake, which will run on the next
1422 # IOLoop iteration and then the real IO state will be set in
1423 # _handle_events.
1424 self._add_io_state(self.io_loop.WRITE)
1426 def reading(self) -> bool:
1427 return self._handshake_reading or super().reading()
1429 def writing(self) -> bool:
1430 return self._handshake_writing or super().writing()
1432 def _do_ssl_handshake(self) -> None:
1433 # Based on code from test_ssl.py in the python stdlib
1434 try:
1435 self._handshake_reading = False
1436 self._handshake_writing = False
1437 self.socket.do_handshake()
1438 except ssl.SSLError as err:
1439 if err.args[0] == ssl.SSL_ERROR_WANT_READ:
1440 self._handshake_reading = True
1441 return
1442 elif err.args[0] == ssl.SSL_ERROR_WANT_WRITE:
1443 self._handshake_writing = True
1444 return
1445 elif err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN):
1446 return self.close(exc_info=err)
1447 elif err.args[0] in (ssl.SSL_ERROR_SSL, ssl.SSL_ERROR_SYSCALL):
1448 try:
1449 peer = self.socket.getpeername()
1450 except Exception:
1451 peer = "(not connected)"
1452 gen_log.warning(
1453 "SSL Error on %s %s: %s", self.socket.fileno(), peer, err
1454 )
1455 return self.close(exc_info=err)
1456 raise
1457 except OSError as err:
1458 # Some port scans (e.g. nmap in -sT mode) have been known
1459 # to cause do_handshake to raise EBADF and ENOTCONN, so make
1460 # those errors quiet as well.
1461 # https://groups.google.com/forum/?fromgroups#!topic/python-tornado/ApucKJat1_0
1462 # Errno 0 is also possible in some cases (nc -z).
1463 # https://github.com/tornadoweb/tornado/issues/2504
1464 if self._is_connreset(err) or err.args[0] in (
1465 0,
1466 errno.EBADF,
1467 errno.ENOTCONN,
1468 ):
1469 return self.close(exc_info=err)
1470 raise
1471 except AttributeError as err:
1472 # On Linux, if the connection was reset before the call to
1473 # wrap_socket, do_handshake will fail with an
1474 # AttributeError.
1475 return self.close(exc_info=err)
1476 else:
1477 self._ssl_accepting = False
1478 # Prior to the introduction of SNI, this is where we would check
1479 # the server's claimed hostname.
1480 assert ssl.HAS_SNI
1481 self._finish_ssl_connect()
1483 def _finish_ssl_connect(self) -> None:
1484 if self._ssl_connect_future is not None:
1485 future = self._ssl_connect_future
1486 self._ssl_connect_future = None
1487 future_set_result_unless_cancelled(future, self)
1489 def _handle_read(self) -> None:
1490 if self._ssl_accepting:
1491 self._do_ssl_handshake()
1492 return
1493 super()._handle_read()
1495 def _handle_write(self) -> None:
1496 if self._ssl_accepting:
1497 self._do_ssl_handshake()
1498 return
1499 super()._handle_write()
1501 def connect(
1502 self, address: tuple, server_hostname: str | None = None
1503 ) -> "Future[SSLIOStream]":
1504 self._server_hostname = server_hostname
1505 # Ignore the result of connect(). If it fails,
1506 # wait_for_handshake will raise an error too. This is
1507 # necessary for the old semantics of the connect callback
1508 # (which takes no arguments). In 6.0 this can be refactored to
1509 # be a regular coroutine.
1510 # TODO: This is trickier than it looks, since if write()
1511 # is called with a connect() pending, we want the connect
1512 # to resolve before the write. Or do we care about this?
1513 # (There's a test for it, but I think in practice users
1514 # either wait for the connect before performing a write or
1515 # they don't care about the connect Future at all)
1516 fut = super().connect(address)
1517 fut.add_done_callback(lambda f: f.exception())
1518 return self.wait_for_handshake()
1520 def _handle_connect(self) -> None:
1521 # Call the superclass method to check for errors.
1522 super()._handle_connect()
1523 if self.closed():
1524 return
1525 # When the connection is complete, wrap the socket for SSL
1526 # traffic. Note that we do this by overriding _handle_connect
1527 # instead of by passing a callback to super().connect because
1528 # user callbacks are enqueued asynchronously on the IOLoop,
1529 # but since _handle_events calls _handle_connect immediately
1530 # followed by _handle_write we need this to be synchronous.
1531 #
1532 # The IOLoop will get confused if we swap out self.socket while the
1533 # fd is registered, so remove it now and re-register after
1534 # wrap_socket().
1535 self.io_loop.remove_handler(self.socket)
1536 old_state = self._state
1537 assert old_state is not None
1538 self._state = None
1539 self.socket = ssl_wrap_socket(
1540 self.socket,
1541 self._ssl_options,
1542 server_hostname=self._server_hostname,
1543 do_handshake_on_connect=False,
1544 server_side=False,
1545 )
1546 self._add_io_state(old_state)
1548 def wait_for_handshake(self) -> "Future[SSLIOStream]":
1549 """Wait for the initial SSL handshake to complete.
1551 If a ``callback`` is given, it will be called with no
1552 arguments once the handshake is complete; otherwise this
1553 method returns a `.Future` which will resolve to the
1554 stream itself after the handshake is complete.
1556 Once the handshake is complete, information such as
1557 the peer's certificate and NPN/ALPN selections may be
1558 accessed on ``self.socket``.
1560 This method is intended for use on server-side streams
1561 or after using `IOStream.start_tls`; it should not be used
1562 with `IOStream.connect` (which already waits for the
1563 handshake to complete). It may only be called once per stream.
1565 .. versionadded:: 4.2
1567 .. versionchanged:: 6.0
1569 The ``callback`` argument was removed. Use the returned
1570 `.Future` instead.
1572 """
1573 if self._ssl_connect_future is not None:
1574 raise RuntimeError("Already waiting")
1575 future = self._ssl_connect_future = Future()
1576 if not self._ssl_accepting:
1577 self._finish_ssl_connect()
1578 return future
1580 def write_to_fd(self, data: memoryview) -> int:
1581 # clip buffer size at 1GB since SSL sockets only support upto 2GB
1582 # this change in behaviour is transparent, since the function is
1583 # already expected to (possibly) write less than the provided buffer
1584 if len(data) >> 30:
1585 data = memoryview(data)[: 1 << 30]
1586 try:
1587 return self.socket.send(data) # type: ignore
1588 except ssl.SSLError as e:
1589 if e.args[0] == ssl.SSL_ERROR_WANT_WRITE:
1590 # In Python 3.5+, SSLSocket.send raises a WANT_WRITE error if
1591 # the socket is not writeable; we need to transform this into
1592 # an EWOULDBLOCK socket.error or a zero return value,
1593 # either of which will be recognized by the caller of this
1594 # method. Prior to Python 3.5, an unwriteable socket would
1595 # simply return 0 bytes written.
1596 return 0
1597 raise
1598 finally:
1599 # Avoid keeping to data, which can be a memoryview.
1600 # See https://github.com/tornadoweb/tornado/pull/2008
1601 del data
1603 def read_from_fd(self, buf: bytearray | memoryview) -> int | None:
1604 try:
1605 if self._ssl_accepting:
1606 # If the handshake hasn't finished yet, there can't be anything
1607 # to read (attempting to read may or may not raise an exception
1608 # depending on the SSL version)
1609 return None
1610 # clip buffer size at 1GB since SSL sockets only support upto 2GB
1611 # this change in behaviour is transparent, since the function is
1612 # already expected to (possibly) read less than the provided buffer
1613 if len(buf) >> 30:
1614 buf = memoryview(buf)[: 1 << 30]
1615 try:
1616 return self.socket.recv_into(buf, len(buf))
1617 except ssl.SSLError as e:
1618 # SSLError is a subclass of socket.error, so this except
1619 # block must come first.
1620 if e.args[0] == ssl.SSL_ERROR_WANT_READ:
1621 return None
1622 else:
1623 raise
1624 except BlockingIOError:
1625 return None
1626 finally:
1627 del buf
1629 def _is_connreset(self, e: BaseException) -> bool:
1630 if isinstance(e, ssl.SSLError) and e.args[0] == ssl.SSL_ERROR_EOF:
1631 return True
1632 return super()._is_connreset(e)
1635class PipeIOStream(BaseIOStream):
1636 """Pipe-based `IOStream` implementation.
1638 The constructor takes an integer file descriptor (such as one returned
1639 by `os.pipe`) rather than an open file object. Pipes are generally
1640 one-way, so a `PipeIOStream` can be used for reading or writing but not
1641 both.
1643 ``PipeIOStream`` is only available on Unix-based platforms.
1644 """
1646 def __init__(self, fd: int, *args: Any, **kwargs: Any) -> None:
1647 self.fd = fd
1648 self._fio = io.FileIO(self.fd, "r+")
1649 if sys.platform == "win32":
1650 # The form and placement of this assertion is important to mypy.
1651 # A plain assert statement isn't recognized here. If the assertion
1652 # were earlier it would worry that the attributes of self aren't
1653 # set on windows. If it were missing it would complain about
1654 # the absence of the set_blocking function.
1655 raise AssertionError("PipeIOStream is not supported on Windows")
1656 os.set_blocking(fd, False)
1657 super().__init__(*args, **kwargs)
1659 def fileno(self) -> int:
1660 return self.fd
1662 def close_fd(self) -> None:
1663 self._fio.close()
1665 def write_to_fd(self, data: memoryview) -> int:
1666 try:
1667 return os.write(self.fd, data) # type: ignore
1668 finally:
1669 # Avoid keeping to data, which can be a memoryview.
1670 # See https://github.com/tornadoweb/tornado/pull/2008
1671 del data
1673 def read_from_fd(self, buf: bytearray | memoryview) -> int | None:
1674 try:
1675 return self._fio.readinto(buf) # type: ignore
1676 except OSError as e:
1677 if errno_from_exception(e) == errno.EBADF:
1678 # If the writing half of a pipe is closed, select will
1679 # report it as readable but reads will fail with EBADF.
1680 self.close(exc_info=e)
1681 return None
1682 else:
1683 raise
1684 finally:
1685 del buf
1688def doctests() -> Any:
1689 import doctest
1691 return doctest.DocTestSuite()