Source code for gevent._fileobjectcommon

"""
gevent internals.
"""
from __future__ import absolute_import, print_function, division

try:
    from errno import EBADF
except ImportError:
    EBADF = 9

import io
import functools
import sys
import os

from gevent.hub import _get_hub_noargs as get_hub
from gevent._compat import PY2
from gevent._compat import integer_types
from gevent._compat import reraise
from gevent._compat import fspath
from gevent.lock import Semaphore, DummySemaphore

class cancel_wait_ex(IOError):

    def __init__(self):
        IOError.__init__(
            self,
            EBADF, 'File descriptor was closed in another greenlet')

class FileObjectClosed(IOError):

    def __init__(self):
        IOError.__init__(
            self,
            EBADF, 'Bad file descriptor (FileObject was closed)')

class UniversalNewlineBytesWrapper(io.TextIOWrapper):
    """
    Uses TextWrapper to decode universal newlines, but returns the
    results as bytes.

    This is for Python 2 where the 'rU' mode did that.
    """
    mode = None
    def __init__(self, fobj, line_buffering):
        # latin-1 has the ability to round-trip arbitrary bytes.
        io.TextIOWrapper.__init__(self, fobj, encoding='latin-1',
                                  newline=None,
                                  line_buffering=line_buffering)

    def read(self, *args, **kwargs):
        result = io.TextIOWrapper.read(self, *args, **kwargs)
        return result.encode('latin-1')

    def readline(self, limit=-1):
        result = io.TextIOWrapper.readline(self, limit)
        return result.encode('latin-1')

    def __iter__(self):
        # readlines() is implemented in terms of __iter__
        # and TextIOWrapper.__iter__ checks that readline returns
        # a unicode object, which we don't, so we override
        return self

    def __next__(self):
        line = self.readline()
        if not line:
            raise StopIteration
        return line

    next = __next__


class FlushingBufferedWriter(io.BufferedWriter):

    def write(self, b):
        ret = io.BufferedWriter.write(self, b)
        self.flush()
        return ret


class WriteallMixin(object):

    def writeall(self, value):
        """
        Similar to :meth:`socket.socket.sendall`, ensures that all the contents of
        *value* have been written (though not necessarily flushed) before returning.

        Returns the length of *value*.

        .. versionadded:: 20.12.0
        """
        # Do we need to play the same get_memory games we do with sockets?
        # And what about chunking for large values? See _socketcommon.py
        write = super(WriteallMixin, self).write

        total = len(value)
        while value:
            l = len(value)
            w = write(value)
            if w == l:
                break
            value = value[w:]
        return total


class FileIO(io.FileIO):
    """A subclass that we can dynamically assign __class__ for."""
    __slots__ = ()


class WriteIsWriteallMixin(WriteallMixin):

    def write(self, value):
        return self.writeall(value)


class WriteallFileIO(WriteIsWriteallMixin, io.FileIO):
    pass


class OpenDescriptor(object): # pylint:disable=too-many-instance-attributes
    """
    Interprets the arguments to `open`. Internal use only.

    Originally based on code in the stdlib's _pyio.py (Python implementation of
    the :mod:`io` module), but modified for gevent:

    - Native strings are returned on Python 2 when neither
      'b' nor 't' are in the mode string and no encoding is specified.
    - Universal newlines work in that mode.
    - Allows externally unbuffered text IO.

    :keyword bool atomic_write: If true, then if the opened, wrapped, stream
        is unbuffered (meaning that ``write`` can produce short writes and the return
        value needs to be checked), then the implementation will be adjusted so that
        ``write`` behaves like Python 2 on a built-in file object and writes the
        entire value. Only set this on Python 2; the only intended user is
        :class:`gevent.subprocess.Popen`.
    """

    @staticmethod
    def _collapse_arg(pref_name, preferred_val, old_name, old_val, default):
        # We could play tricks with the callers ``locals()`` to avoid having to specify
        # the name (which we only use for error handling) but ``locals()`` may be slow and
        # inhibit JIT (on PyPy), so we just write it out long hand.
        if preferred_val is not None and old_val is not None:
            raise TypeError("Cannot specify both %s=%s and %s=%s" % (
                pref_name, preferred_val,
                old_name, old_val
            ))
        if preferred_val is None and old_val is None:
            return default
        return preferred_val if preferred_val is not None else old_val

    def __init__(self, fobj, mode='r', bufsize=None, close=None,
                 encoding=None, errors=None, newline=None,
                 buffering=None, closefd=None,
                 atomic_write=False):
        # Based on code in the stdlib's _pyio.py from 3.8.
        # pylint:disable=too-many-locals,too-many-branches,too-many-statements

        closefd = self._collapse_arg('closefd', closefd, 'close', close, True)
        del close
        buffering = self._collapse_arg('buffering', buffering, 'bufsize', bufsize, -1)
        del bufsize

        if not hasattr(fobj, 'fileno'):
            if not isinstance(fobj, integer_types):
                # Not a fd. Support PathLike on Python 2 and Python <= 3.5.
                fobj = fspath(fobj)
            if not isinstance(fobj, (str, bytes) + integer_types): # pragma: no cover
                raise TypeError("invalid file: %r" % fobj)
            if isinstance(fobj, (str, bytes)):
                closefd = True

        if not isinstance(mode, str):
            raise TypeError("invalid mode: %r" % mode)
        if not isinstance(buffering, integer_types):
            raise TypeError("invalid buffering: %r" % buffering)
        if encoding is not None and not isinstance(encoding, str):
            raise TypeError("invalid encoding: %r" % encoding)
        if errors is not None and not isinstance(errors, str):
            raise TypeError("invalid errors: %r" % errors)

        modes = set(mode)
        if modes - set("axrwb+tU") or len(mode) > len(modes):
            raise ValueError("invalid mode: %r" % mode)

        creating = "x" in modes
        reading = "r" in modes
        writing = "w" in modes
        appending = "a" in modes
        updating = "+" in modes
        text = "t" in modes
        binary = "b" in modes
        universal = 'U' in modes

        can_write = creating or writing or appending or updating

        if universal:
            if can_write:
                raise ValueError("mode U cannot be combined with 'x', 'w', 'a', or '+'")
            # Just because the stdlib deprecates this, no need for us to do so as well.
            # Especially not while we still support Python 2.
            # import warnings
            # warnings.warn("'U' mode is deprecated",
            #               DeprecationWarning, 4)
            reading = True
        if text and binary:
            raise ValueError("can't have text and binary mode at once")
        if creating + reading + writing + appending > 1:
            raise ValueError("can't have read/write/append mode at once")
        if not (creating or reading or writing or appending):
            raise ValueError("must have exactly one of read/write/append mode")
        if binary and encoding is not None:
            raise ValueError("binary mode doesn't take an encoding argument")
        if binary and errors is not None:
            raise ValueError("binary mode doesn't take an errors argument")
        if binary and newline is not None:
            raise ValueError("binary mode doesn't take a newline argument")
        if binary and buffering == 1:
            import warnings
            warnings.warn("line buffering (buffering=1) isn't supported in binary "
                          "mode, the default buffer size will be used",
                          RuntimeWarning, 4)

        self._fobj = fobj
        self.fileio_mode = (
            (creating and "x" or "")
            + (reading and "r" or "")
            + (writing and "w" or "")
            + (appending and "a" or "")
            + (updating and "+" or "")
        )
        self.mode = self.fileio_mode + ('t' if text else '') + ('b' if binary else '')

        self.creating = creating
        self.reading = reading
        self.writing = writing
        self.appending = appending
        self.updating = updating
        self.text = text
        self.binary = binary
        self.can_write = can_write
        self.can_read = reading or updating
        self.native = (
            not self.text and not self.binary # Neither t nor b given.
            and not encoding and not errors # And no encoding or error handling either.
        )
        self.universal = universal

        self.buffering = buffering
        self.encoding = encoding
        self.errors = errors
        self.newline = newline
        self.closefd = closefd
        self.atomic_write = atomic_write

    default_buffer_size = io.DEFAULT_BUFFER_SIZE

    _opened = None
    _opened_raw = None

    def is_fd(self):
        return isinstance(self._fobj, integer_types)

    def opened(self):
        """
        Return the :meth:`wrapped` file object.
        """
        if self._opened is None:
            raw = self.opened_raw()
            try:
                self._opened = self.__wrapped(raw)
            except:
                # XXX: This might be a bug? Could we wind up closing
                # something we shouldn't close?
                raw.close()
                raise
        return self._opened

    def _raw_object_is_new(self, raw):
        return self._fobj is not raw

    def opened_raw(self):
        if self._opened_raw is None:
            self._opened_raw = self._do_open_raw()
        return self._opened_raw

    def _do_open_raw(self):
        if hasattr(self._fobj, 'fileno'):
            return self._fobj
        # io.FileIO doesn't allow assigning to its __class__,
        # and we can't know for sure here whether we need the atomic write()
        # method or not (it depends on the layers on top of us),
        # so we use a subclass that *does* allow assigning.
        return FileIO(self._fobj, self.fileio_mode, self.closefd)

    @staticmethod
    def is_buffered(stream):
        return (
            # buffering happens internally in the text codecs
            isinstance(stream, (io.BufferedIOBase, io.TextIOBase))
            or (hasattr(stream, 'buffer') and stream.buffer is not None)
        )

    @classmethod
    def buffer_size_for_stream(cls, stream):
        result = cls.default_buffer_size
        try:
            bs = os.fstat(stream.fileno()).st_blksize
        except (OSError, AttributeError):
            pass
        else:
            if bs > 1:
                result = bs
        return result

    def __buffered(self, stream, buffering):
        if self.updating:
            Buffer = io.BufferedRandom
        elif self.creating or self.writing or self.appending:
            Buffer = io.BufferedWriter
        elif self.reading:
            Buffer = io.BufferedReader
        else: # prgama: no cover
            raise ValueError("unknown mode: %r" % self.mode)

        try:
            result = Buffer(stream, buffering)
        except AttributeError:
            # Python 2 file() objects don't have the readable/writable
            # attributes. But they handle their own buffering.
            result = stream

        return result

    def _make_atomic_write(self, result, raw):
        # The idea was to swizzle the class with one that defines
        # write() to call writeall(). This avoids setting any
        # attribute on the return object, avoids an additional layer
        # of proxying, and avoids any reference cycles (if setting a
        # method on the object).
        #
        # However, this is not possible with the built-in io classes
        # (static types defined in C cannot have __class__ assigned).
        # Fortunately, we need this only for the specific case of
        # opening a file descriptor (subprocess.py) on Python 2, in
        # which we fully control the types involved.
        #
        # So rather than attempt that, we only implement exactly what we need.
        if result is not raw or self._raw_object_is_new(raw):
            if result.__class__ is FileIO:
                result.__class__ = WriteallFileIO
            else: # pragma: no cover
                raise NotImplementedError(
                    "Don't know how to make %s have atomic write. "
                    "Please open a gevent issue with your use-case." % (
                        result
                    )
                )
        return result

    def __wrapped(self, raw):
        """
        Wraps the raw IO object (`RawIOBase` or `io.TextIOBase`) in
        buffers, text decoding, and newline handling.
        """
        if self.binary and isinstance(raw, io.TextIOBase):
            # Can't do it. The TextIO object will have its own buffer, and
            # trying to read from the raw stream or the buffer without going through
            # the TextIO object is likely to lead to problems with the codec.
            raise ValueError("Unable to perform binary IO on top of text IO stream")

        result = raw
        buffering = self.buffering

        line_buffering = False
        if buffering == 1 or buffering < 0 and raw.isatty():
            buffering = -1
            line_buffering = True
        if buffering < 0:
            buffering = self.buffer_size_for_stream(result)

        if buffering < 0: # pragma: no cover
            raise ValueError("invalid buffering size")

        if buffering != 0 and not self.is_buffered(result):
            # Need to wrap our own buffering around it. If it
            # is already buffered, don't do so.
            result = self.__buffered(result, buffering)

        if not self.binary:
            # Either native or text at this point.
            if PY2 and self.native:
                # Neither text mode nor binary mode specified.
                if self.universal:
                    # universal was requested, e.g., 'rU'
                    result = UniversalNewlineBytesWrapper(result, line_buffering)
            else:
                # Python 2 and text mode, or Python 3 and either text or native (both are the same)
                if not isinstance(raw, io.TextIOBase):
                    # Avoid double-wrapping a TextIOBase in another TextIOWrapper.
                    # That tends not to work. See https://github.com/gevent/gevent/issues/1542
                    result = io.TextIOWrapper(result, self.encoding, self.errors, self.newline,
                                              line_buffering)

        if result is not raw or self._raw_object_is_new(raw):
            # Set the mode, if possible, but only if we created a new
            # object.
            try:
                result.mode = self.mode
            except (AttributeError, TypeError):
                # AttributeError: No such attribute
                # TypeError: Readonly attribute (py2)
                pass

        if (
                self.atomic_write
                and not self.is_buffered(result)
                and not isinstance(result, WriteIsWriteallMixin)
        ):
            # Let subclasses have a say in how they make this atomic, and
            # whether or not they do so even if we're actually returning the raw object.
            result = self._make_atomic_write(result, raw)

        return result


class _ClosedIO(object):
    # Used for FileObjectBase._io when FOB.close()
    # is called. Lets us drop references to ``_io``
    # for GC/resource cleanup reasons, but keeps some useful
    # information around.
    __slots__ = ('name',)

    def __init__(self, io_obj):
        try:
            self.name = io_obj.name
        except AttributeError:
            pass

    def __getattr__(self, name):
        if name == 'name':
            # We didn't set it in __init__ because there wasn't one
            raise AttributeError
        raise FileObjectClosed

    def __bool__(self):
        return False
    __nonzero__ = __bool__


class FileObjectBase(object):
    """
    Internal base class to ensure a level of consistency
    between :class:`~.FileObjectPosix`, :class:`~.FileObjectThread`
    and :class:`~.FileObjectBlock`.
    """

    # List of methods we delegate to the wrapping IO object, if they
    # implement them and we do not.
    _delegate_methods = (
        # General methods
        'flush',
        'fileno',
        'writable',
        'readable',
        'seek',
        'seekable',
        'tell',

        # Read
        'read',
        'readline',
        'readlines',
        'read1',

        # Write.
        # Note that we do not extend WriteallMixin,
        # so writeall will be copied, if it exists, and
        # wrapped.
        'write',
        'writeall',
        'writelines',
        'truncate',
    )


    _io = None

    def __init__(self, descriptor):
        # type: (OpenDescriptor) -> None
        self._io = descriptor.opened()
        # We don't actually use this property ourself, but we save it (and
        # pass it along) for compatibility.
        self._close = descriptor.closefd
        self._do_delegate_methods()


    io = property(lambda s: s._io,
                  # Historically we either hand-wrote all the delegation methods
                  # to use self.io, or we simply used __getattr__ to look them up at
                  # runtime. This meant people could change the io attribute on the fly
                  # and it would mostly work (subprocess.py used to do that). We don't recommend
                  # that, but we still support it.
                  lambda s, nv: setattr(s, '_io', nv) or s._do_delegate_methods())

    def _do_delegate_methods(self):
        for meth_name in self._delegate_methods:
            meth = getattr(self._io, meth_name, None)
            implemented_by_class = hasattr(type(self), meth_name)
            if meth and not implemented_by_class:
                setattr(self, meth_name, self._wrap_method(meth))
            elif hasattr(self, meth_name) and not implemented_by_class:
                delattr(self, meth_name)

    def _wrap_method(self, method):
        """
        Wrap a method we're copying into our dictionary from the underlying
        io object to do something special or different, if necessary.
        """
        return method

    @property
    def closed(self):
        """True if the file is closed"""
        return isinstance(self._io, _ClosedIO)

    def close(self):
        if isinstance(self._io, _ClosedIO):
            return

        fobj = self._io
        self._io = _ClosedIO(self._io)
        try:
            self._do_close(fobj, self._close)
        finally:
            fobj = None
            # Remove delegate methods to drop remaining references to
            # _io.
            d = self.__dict__
            for meth_name in self._delegate_methods:
                d.pop(meth_name, None)

    def _do_close(self, fobj, closefd):
        raise NotImplementedError()

    def __getattr__(self, name):
        return getattr(self._io, name)

    def __repr__(self):
        return '<%s at 0x%x %s_fobj=%r%s>' % (
            self.__class__.__name__,
            id(self),
            'closed' if self.closed else '',
            self.io,
            self._extra_repr()
        )

    def _extra_repr(self):
        return ''

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

    def __iter__(self):
        return self

    def __next__(self):
        line = self.readline()
        if not line:
            raise StopIteration
        return line

    next = __next__

    def __bool__(self):
        return True

    __nonzero__ = __bool__


[docs]class FileObjectBlock(FileObjectBase): """ FileObjectBlock() A simple synchronous wrapper around a file object. Adds no concurrency or gevent compatibility. """ def __init__(self, fobj, *args, **kwargs): descriptor = OpenDescriptor(fobj, *args, **kwargs) FileObjectBase.__init__(self, descriptor) def _do_close(self, fobj, closefd): fobj.close()
[docs]class FileObjectThread(FileObjectBase): """ FileObjectThread() A file-like object wrapping another file-like object, performing all blocking operations on that object in a background thread. .. caution:: Attempting to change the threadpool or lock of an existing FileObjectThread has undefined consequences. .. versionchanged:: 1.1b1 The file object is closed using the threadpool. Note that whether or not this action is synchronous or asynchronous is not documented. """ def __init__(self, *args, **kwargs): """ :keyword bool lock: If True (the default) then all operations will be performed one-by-one. Note that this does not guarantee that, if using this file object from multiple threads/greenlets, operations will be performed in any particular order, only that no two operations will be attempted at the same time. You can also pass your own :class:`gevent.lock.Semaphore` to synchronize file operations with an external resource. :keyword bool closefd: If True (the default) then when this object is closed, the underlying object is closed as well. If *fobj* is a path, then *closefd* must be True. """ lock = kwargs.pop('lock', True) threadpool = kwargs.pop('threadpool', None) descriptor = OpenDescriptor(*args, **kwargs) self.threadpool = threadpool or get_hub().threadpool self.lock = lock if self.lock is True: self.lock = Semaphore() elif not self.lock: self.lock = DummySemaphore() if not hasattr(self.lock, '__enter__'): raise TypeError('Expected a Semaphore or boolean, got %r' % type(self.lock)) self.__io_holder = [descriptor.opened()] # signal for _wrap_method FileObjectBase.__init__(self, descriptor) def _do_close(self, fobj, closefd): self.__io_holder[0] = None # for _wrap_method try: with self.lock: self.threadpool.apply(fobj.flush) finally: if closefd: # Note that we're not taking the lock; older code # did fobj.close() without going through the threadpool at all, # so acquiring the lock could potentially introduce deadlocks # that weren't present before. Avoiding the lock doesn't make # the existing race condition any worse. # We wrap the close in an exception handler and re-raise directly # to avoid the (common, expected) IOError from being logged by the pool def close(_fobj=fobj): try: _fobj.close() except: # pylint:disable=bare-except return sys.exc_info() finally: _fobj = None del fobj exc_info = self.threadpool.apply(close) del close if exc_info: reraise(*exc_info) def _do_delegate_methods(self): FileObjectBase._do_delegate_methods(self) self.__io_holder[0] = self._io def _extra_repr(self): return ' threadpool=%r' % (self.threadpool,) def _wrap_method(self, method): # NOTE: We are careful to avoid introducing a refcycle # within self. Our wrapper cannot refer to self. io_holder = self.__io_holder lock = self.lock threadpool = self.threadpool @functools.wraps(method) def thread_method(*args, **kwargs): if io_holder[0] is None: # This is different than FileObjectPosix, etc, # because we want to save the expensive trip through # the threadpool. raise FileObjectClosed with lock: return threadpool.apply(method, args, kwargs) return thread_method