python-eventlet/eventlet/greenio/py2.py

   1 import errno
   2 import os
   3
   4 from eventlet.greenio.base import (
   5     _operation_on_closed_file,
   6     greenpipe_doc,
   7     set_nonblocking,
   8     socket,
   9     SOCKET_BLOCKING,
  10 )
  11 from eventlet.hubs import trampoline, notify_close, notify_opened, IOClosed
  12 from eventlet.support import get_errno, six
  13
  14 __all__ = ['_fileobject', 'GreenPipe']
  15
  16 _fileobject = socket._fileobject
  17
  18
  19 class GreenPipe(_fileobject):
  20
  21     __doc__ = greenpipe_doc
  22
  23     def __init__(self, f, mode='r', bufsize=-1):
  24         if not isinstance(f, six.string_types + (int, file)):
  25             raise TypeError('f(ile) should be int, str, unicode or file, not %r' % f)
  26
  27         if isinstance(f, six.string_types):
  28             f = open(f, mode, 0)
  29
  30         if isinstance(f, int):
  31             fileno = f
  32             self._name = "<fd:%d>" % fileno
  33         else:
  34             fileno = os.dup(f.fileno())
  35             self._name = f.name
  36             if f.mode != mode:
  37                 raise ValueError('file.mode %r does not match mode parameter %r' % (f.mode, mode))
  38             self._name = f.name
  39             f.close()
  40
  41         super(GreenPipe, self).__init__(_SocketDuckForFd(fileno), mode, bufsize)
  42         set_nonblocking(self)
  43         self.softspace = 0
  44
  45     @property
  46     def name(self):
  47         return self._name
  48
  49     def __repr__(self):
  50         return "<%s %s %r, mode %r at 0x%x>" % (
  51             self.closed and 'closed' or 'open',
  52             self.__class__.__name__,
  53             self.name,
  54             self.mode,
  55             (id(self) < 0) and (sys.maxint + id(self)) or id(self))
  56
  57     def close(self):
  58         super(GreenPipe, self).close()
  59         for method in [
  60                 'fileno', 'flush', 'isatty', 'next', 'read', 'readinto',
  61                 'readline', 'readlines', 'seek', 'tell', 'truncate',
  62                 'write', 'xreadlines', '__iter__', '__next__', 'writelines']:
  63             setattr(self, method, _operation_on_closed_file)
  64
  65     def __enter__(self):
  66         return self
  67
  68     def __exit__(self, *args):
  69         self.close()
  70
  71     def _get_readahead_len(self):
  72         return len(self._rbuf.getvalue())
  73
  74     def _clear_readahead_buf(self):
  75         len = self._get_readahead_len()
  76         if len > 0:
  77             self.read(len)
  78
  79     def tell(self):
  80         self.flush()
  81         try:
  82             return os.lseek(self.fileno(), 0, 1) - self._get_readahead_len()
  83         except OSError as e:
  84             raise IOError(*e.args)
  85
  86     def seek(self, offset, whence=0):
  87         self.flush()
  88         if whence == 1 and offset == 0:  # tell synonym
  89             return self.tell()
  90         if whence == 1:  # adjust offset by what is read ahead
  91             offset -= self._get_readahead_len()
  92         try:
  93             rv = os.lseek(self.fileno(), offset, whence)
  94         except OSError as e:
  95             raise IOError(*e.args)
  96         else:
  97             self._clear_readahead_buf()
  98             return rv
  99
 100     if getattr(file, "truncate", None):  # not all OSes implement truncate
 101         def truncate(self, size=-1):
 102             self.flush()
 103             if size == -1:
 104                 size = self.tell()
 105             try:
 106                 rv = os.ftruncate(self.fileno(), size)
 107             except OSError as e:
 108                 raise IOError(*e.args)
 109             else:
 110                 self.seek(size)  # move position&clear buffer
 111                 return rv
 112
 113     def isatty(self):
 114         try:
 115             return os.isatty(self.fileno())
 116         except OSError as e:
 117             raise IOError(*e.args)
 118
 119
 120 class _SocketDuckForFd(object):
 121     """Class implementing all socket method used by _fileobject
 122     in cooperative manner using low level os I/O calls.
 123     """
 124     _refcount = 0
 125
 126     def __init__(self, fileno):
 127         self._fileno = fileno
 128         notify_opened(fileno)
 129         self._closed = False
 130
 131     def _trampoline(self, fd, read=False, write=False, timeout=None, timeout_exc=None):
 132         if self._closed:
 133             # Don't trampoline if we're already closed.
 134             raise IOClosed()
 135         try:
 136             return trampoline(fd, read=read, write=write, timeout=timeout,
 137                               timeout_exc=timeout_exc,
 138                               mark_as_closed=self._mark_as_closed)
 139         except IOClosed:
 140             # Our fileno has been obsoleted. Defang ourselves to
 141             # prevent spurious closes.
 142             self._mark_as_closed()
 143             raise
 144
 145     def _mark_as_closed(self):
 146         current = self._closed
 147         self._closed = True
 148         return current
 149
 150     @property
 151     def _sock(self):
 152         return self
 153
 154     def fileno(self):
 155         return self._fileno
 156
 157     def recv(self, buflen):
 158         while True:
 159             try:
 160                 data = os.read(self._fileno, buflen)
 161                 return data
 162             except OSError as e:
 163                 if get_errno(e) not in SOCKET_BLOCKING:
 164                     raise IOError(*e.args)
 165             self._trampoline(self, read=True)
 166
 167     def recv_into(self, buf, nbytes=0, flags=0):
 168         if nbytes == 0:
 169             nbytes = len(buf)
 170         data = self.recv(nbytes)
 171         buf[:nbytes] = data
 172         return len(data)
 173
 174     def send(self, data):
 175         while True:
 176             try:
 177                 return os.write(self._fileno, data)
 178             except OSError as e:
 179                 if get_errno(e) not in SOCKET_BLOCKING:
 180                     raise IOError(*e.args)
 181                 else:
 182                     trampoline(self, write=True)
 183
 184     def sendall(self, data):
 185         len_data = len(data)
 186         os_write = os.write
 187         fileno = self._fileno
 188         try:
 189             total_sent = os_write(fileno, data)
 190         except OSError as e:
 191             if get_errno(e) != errno.EAGAIN:
 192                 raise IOError(*e.args)
 193             total_sent = 0
 194         while total_sent < len_data:
 195             self._trampoline(self, write=True)
 196             try:
 197                 total_sent += os_write(fileno, data[total_sent:])
 198             except OSError as e:
 199                 if get_errno(e) != errno. EAGAIN:
 200                     raise IOError(*e.args)
 201
 202     def __del__(self):
 203         self._close()
 204
 205     def _close(self):
 206         was_closed = self._mark_as_closed()
 207         if was_closed:
 208             return
 209         notify_close(self._fileno)
 210         try:
 211             os.close(self._fileno)
 212         except:
 213             # os.close may fail if __init__ didn't complete
 214             # (i.e file dscriptor passed to popen was invalid
 215             pass
 216
 217     def __repr__(self):
 218         return "%s:%d" % (self.__class__.__name__, self._fileno)
 219
 220     def _reuse(self):
 221         self._refcount += 1
 222
 223     def _drop(self):
 224         self._refcount -= 1
 225         if self._refcount == 0:
 226             self._close()