python-eventlet/eventlet/greenio/py3.py

   1 import _pyio as _original_pyio
   2 import errno
   3 import os as _original_os
   4 import socket as _original_socket
   5 from io import (
   6     BufferedRandom as _OriginalBufferedRandom,
   7     BufferedReader as _OriginalBufferedReader,
   8     BufferedWriter as _OriginalBufferedWriter,
   9     DEFAULT_BUFFER_SIZE,
  10     TextIOWrapper as _OriginalTextIOWrapper,
  11     IOBase as _OriginalIOBase,
  12 )
  13 from types import FunctionType
  14
  15 from eventlet.greenio.base import (
  16     _operation_on_closed_file,
  17     greenpipe_doc,
  18     set_nonblocking,
  19     SOCKET_BLOCKING,
  20 )
  21 from eventlet.hubs import notify_close, notify_opened, IOClosed, trampoline
  22 from eventlet.support import get_errno, six
  23
  24 __all__ = ['_fileobject', 'GreenPipe']
  25
  26 # TODO get rid of this, it only seems like the original _fileobject
  27 _fileobject = _original_socket.SocketIO
  28
  29 # Large part of the following code is copied from the original
  30 # eventlet.greenio module
  31
  32
  33 class GreenFileIO(_OriginalIOBase):
  34     def __init__(self, name, mode='r', closefd=True, opener=None):
  35         if isinstance(name, int):
  36             fileno = name
  37             self._name = "<fd:%d>" % fileno
  38         else:
  39             assert isinstance(name, six.string_types)
  40             with open(name, mode) as fd:
  41                 self._name = fd.name
  42                 fileno = _original_os.dup(fd.fileno())
  43
  44         notify_opened(fileno)
  45         self._fileno = fileno
  46         self._mode = mode
  47         self._closed = False
  48         set_nonblocking(self)
  49         self._seekable = None
  50
  51     @property
  52     def closed(self):
  53         return self._closed
  54
  55     def seekable(self):
  56         if self._seekable is None:
  57             try:
  58                 _original_os.lseek(self._fileno, 0, _original_os.SEEK_CUR)
  59             except IOError as e:
  60                 if get_errno(e) == errno.ESPIPE:
  61                     self._seekable = False
  62                 else:
  63                     raise
  64             else:
  65                 self._seekable = True
  66
  67         return self._seekable
  68
  69     def readable(self):
  70         return 'r' in self._mode or '+' in self._mode
  71
  72     def writable(self):
  73         return 'w' in self._mode or '+' in self._mode
  74
  75     def fileno(self):
  76         return self._fileno
  77
  78     def read(self, size=-1):
  79         if size == -1:
  80             return self.readall()
  81
  82         while True:
  83             try:
  84                 return _original_os.read(self._fileno, size)
  85             except OSError as e:
  86                 if get_errno(e) not in SOCKET_BLOCKING:
  87                     raise IOError(*e.args)
  88             self._trampoline(self, read=True)
  89
  90     def readall(self):
  91         buf = []
  92         while True:
  93             try:
  94                 chunk = _original_os.read(self._fileno, DEFAULT_BUFFER_SIZE)
  95                 if chunk == b'':
  96                     return b''.join(buf)
  97                 buf.append(chunk)
  98             except OSError as e:
  99                 if get_errno(e) not in SOCKET_BLOCKING:
 100                     raise IOError(*e.args)
 101             self._trampoline(self, read=True)
 102
 103     def readinto(self, b):
 104         up_to = len(b)
 105         data = self.read(up_to)
 106         bytes_read = len(data)
 107         b[:bytes_read] = data
 108         return bytes_read
 109
 110     def isatty(self):
 111         try:
 112             return _original_os.isatty(self.fileno())
 113         except OSError as e:
 114             raise IOError(*e.args)
 115
 116     def _trampoline(self, fd, read=False, write=False, timeout=None, timeout_exc=None):
 117         if self._closed:
 118             # Don't trampoline if we're already closed.
 119             raise IOClosed()
 120         try:
 121             return trampoline(fd, read=read, write=write, timeout=timeout,
 122                               timeout_exc=timeout_exc,
 123                               mark_as_closed=self._mark_as_closed)
 124         except IOClosed:
 125             # Our fileno has been obsoleted. Defang ourselves to
 126             # prevent spurious closes.
 127             self._mark_as_closed()
 128             raise
 129
 130     def _mark_as_closed(self):
 131         """ Mark this socket as being closed """
 132         self._closed = True
 133
 134     def write(self, data):
 135         view = memoryview(data)
 136         datalen = len(data)
 137         offset = 0
 138         while offset < datalen:
 139             try:
 140                 written = _original_os.write(self._fileno, view[offset:])
 141             except OSError as e:
 142                 if get_errno(e) not in SOCKET_BLOCKING:
 143                     raise IOError(*e.args)
 144                 trampoline(self, write=True)
 145             else:
 146                 offset += written
 147         return offset
 148
 149     def close(self):
 150         if not self._closed:
 151             self._closed = True
 152             _original_os.close(self._fileno)
 153         notify_close(self._fileno)
 154         for method in [
 155                 'fileno', 'flush', 'isatty', 'next', 'read', 'readinto',
 156                 'readline', 'readlines', 'seek', 'tell', 'truncate',
 157                 'write', 'xreadlines', '__iter__', '__next__', 'writelines']:
 158             setattr(self, method, _operation_on_closed_file)
 159
 160     def truncate(self, size=-1):
 161         if size == -1:
 162             size = self.tell()
 163         try:
 164             rv = _original_os.ftruncate(self._fileno, size)
 165         except OSError as e:
 166             raise IOError(*e.args)
 167         else:
 168             self.seek(size)  # move position&clear buffer
 169             return rv
 170
 171     def seek(self, offset, whence=_original_os.SEEK_SET):
 172         try:
 173             return _original_os.lseek(self._fileno, offset, whence)
 174         except OSError as e:
 175             raise IOError(*e.args)
 176
 177     def __enter__(self):
 178         return self
 179
 180     def __exit__(self, *args):
 181         self.close()
 182
 183
 184 _open_environment = dict(globals())
 185 _open_environment.update(dict(
 186     BufferedRandom=_OriginalBufferedRandom,
 187     BufferedWriter=_OriginalBufferedWriter,
 188     BufferedReader=_OriginalBufferedReader,
 189     TextIOWrapper=_OriginalTextIOWrapper,
 190     FileIO=GreenFileIO,
 191     os=_original_os,
 192 ))
 193
 194 _open = FunctionType(
 195     six.get_function_code(_original_pyio.open),
 196     _open_environment,
 197 )
 198
 199
 200 def GreenPipe(name, mode="r", buffering=-1, encoding=None, errors=None,
 201               newline=None, closefd=True, opener=None):
 202     try:
 203         fileno = name.fileno()
 204     except AttributeError:
 205         pass
 206     else:
 207         fileno = _original_os.dup(fileno)
 208         name.close()
 209         name = fileno
 210
 211     return _open(name, mode, buffering, encoding, errors, newline, closefd, opener)
 212
 213 GreenPipe.__doc__ = greenpipe_doc