from rpython.rlib.signature import signature from rpython.rlib import types from pypy.interpreter.error import OperationError, oefmt from pypy.interpreter.typedef import ( TypeDef, GetSetProperty, generic_new_descr, interp_attrproperty_w) from pypy.interpreter.gateway import interp2app, unwrap_spec, WrappedDefault from pypy.interpreter.buffer import SimpleView from rpython.rlib.buffer import ByteBuffer, RawByteBuffer, SubBuffer from rpython.rlib.rstring import StringBuilder from rpython.rlib.rarithmetic import r_longlong, intmask from rpython.rlib import rposix from rpython.tool.sourcetools import func_renamer from rpython.rlib.objectmodel import import_from_mixin, try_inline, keepalive_until_here from rpython.rtyper.lltypesystem import lltype, rffi from pypy.module._io.interp_iobase import ( W_IOBase, DEFAULT_BUFFER_SIZE, convert_size, trap_eintr, check_readable_w, check_writable_w, check_seekable_w) from rpython.rlib import rthread STATE_ZERO, STATE_OK, STATE_DETACHED = range(3) def make_write_blocking_error(space, written): # XXX CPython reads 'errno' here. I *think* it doesn't make sense, # because we might reach this point after calling a write() method # that may be overridden by the user, if that method returns None. # In that case what we get is a potentially nonsense errno. But # we'll use get_saved_errno() anyway, and hope (like CPython does) # that we're getting a reasonable value at this point. w_value = space.call_function( space.w_BlockingIOError, space.newint(rposix.get_saved_errno()), space.newtext("write could not complete without blocking"), space.newint(written)) return OperationError(space.w_BlockingIOError, w_value) class TryLock(object): "A Lock that raises RuntimeError when acquired twice by the same thread" def __init__(self, space): ## XXX cannot free a Lock? ## if self.lock: ## self.lock.free() self.lock = space.allocate_lock() self.owner = 0 self.operr = oefmt(space.w_RuntimeError, "reentrant call") def __enter__(self): if not self.lock.acquire(False): if self.owner == rthread.get_ident(): raise self.operr self.lock.acquire(True) self.owner = rthread.get_ident() def __exit__(self,*args): self.owner = 0 self.lock.release() class BlockingIOError(Exception): pass class W_BufferedIOBase(W_IOBase): def _check_init(self, space): raise NotImplementedError def read_w(self, space, w_size=None): """Read and return up to n bytes. If the argument is omitted, None, or negative, reads and returns all data until EOF. If the argument is positive, and the underlying raw stream is not 'interactive', multiple raw reads may be issued to satisfy the byte count (unless EOF is reached first). But for interactive raw streams (as well as sockets and pipes), at most one raw read will be issued, and a short result does not imply that EOF is imminent. Returns an empty bytes object on EOF. Returns None if the underlying raw stream was open in non-blocking mode and no data is available at the moment.""" self._unsupportedoperation(space, "read") def read1_w(self, space, w_size=None): """Read and return up to n bytes, with at most one read() call to the underlying raw stream. A short result does not imply that EOF is imminent. Returns an empty bytes object on EOF.""" self._unsupportedoperation(space, "read1") def write_w(self, space, w_data): """Write the given buffer to the IO stream. Returns the number of bytes written, which is always the length of b in bytes. Raises BlockingIOError if the buffer is full and the underlying raw stream cannot accept more data at the moment.""" self._unsupportedoperation(space, "write") def detach_w(self, space): """Disconnect this buffer from its underlying raw stream and return it. After the raw stream has been detached, the buffer is in an unusable state.""" self._unsupportedoperation(space, "detach") def readinto_w(self, space, w_buffer): return self._readinto(space, w_buffer, read_once=False) def readinto1_w(self, space, w_buffer): return self._readinto(space, w_buffer, read_once=True) def _readinto(self, space, w_buffer, read_once): rwbuffer = space.writebuf_w(w_buffer) length = rwbuffer.getlength() methodname = "read1" if read_once else "read" w_data = space.call_method(self, methodname, space.newint(length)) if not space.isinstance_w(w_data, space.w_bytes): raise oefmt(space.w_TypeError, "%s() should return bytes", methodname) data = space.bytes_w(w_data) if len(data) > length: raise oefmt(space.w_ValueError, "%s() returned too much data: " "%d bytes requested, %d returned", methodname, length, len(data)) self.output_slice(space, rwbuffer, 0, data) return space.newint(len(data)) W_BufferedIOBase.typedef = TypeDef( '_io._BufferedIOBase', W_IOBase.typedef, __doc__="""Base class for buffered IO objects. The main difference with RawIOBase is that the read() method supports omitting the size argument, and does not have a default implementation that defers to readinto(). In addition, read(), readinto() and write() may raise BlockingIOError if the underlying raw stream is in non-blocking mode and not ready; unlike their raw counterparts, they will never return None. A typical implementation should not inherit from a RawIOBase implementation, but wrap one. """, __new__ = generic_new_descr(W_BufferedIOBase), read = interp2app(W_BufferedIOBase.read_w), read1 = interp2app(W_BufferedIOBase.read1_w), write = interp2app(W_BufferedIOBase.write_w), detach = interp2app(W_BufferedIOBase.detach_w), readinto = interp2app(W_BufferedIOBase.readinto_w), readinto1 = interp2app(W_BufferedIOBase.readinto1_w), ) class BufferedMixin: def __init__(self, space): W_IOBase.__init__(self, space) self.state = STATE_ZERO self.buffer = None self.abs_pos = 0 # Absolute position inside the raw stream (-1 if # unknown). self.pos = 0 # Current logical position in the buffer self.raw_pos = 0 # Position of the raw stream in the buffer. self.read_end = -1 # Just after the last buffered byte in the buffer, # or -1 if the buffer isn't ready for reading self.write_pos = 0 # Just after the last byte actually written self.write_end = -1 # Just after the last byte waiting to be written, # or -1 if the buffer isn't ready for writing. self.lock = None self.readable = False self.writable = False self._fast_closed_check = False def _reader_reset_buf(self): self.read_end = -1 def _writer_reset_buf(self): self.write_pos = 0 self.write_end = -1 def _make_buffer(self, space, size): if space.config.translation.split_gc_address_space: # When using split GC address space, it is not possible to get the # raw address of a GC buffer. Therefore we use a buffer backed by # raw memory. return RawByteBuffer(size) else: # TODO: test whether using the raw buffer is faster return ByteBuffer(size) def _init(self, space): if self.buffer_size <= 0: raise oefmt(space.w_ValueError, "buffer size must be strictly positive") self.buffer = self._make_buffer(space, self.buffer_size) self.lock = TryLock(space) try: self._raw_tell(space) except OperationError: pass def _check_init(self, space): if self.state == STATE_ZERO: raise oefmt(space.w_ValueError, "I/O operation on uninitialized object") elif self.state == STATE_DETACHED: raise oefmt(space.w_ValueError, "raw stream has been detached") @try_inline def _check_closed(self, space, message=None): from pypy.module._io.interp_fileio import W_FileIO if self._fast_closed_check: w_raw = self.w_raw assert isinstance(w_raw, W_FileIO) if w_raw.fd >= 0: return self._check_init(space) W_IOBase._check_closed(self, space, message) def _raw_tell(self, space): from pypy.module._io.interp_fileio import W_FileIO if self._fast_closed_check: w_raw = self.w_raw assert isinstance(w_raw, W_FileIO) pos = r_longlong(w_raw._raw_tell(space)) else: w_pos = space.call_method(self.w_raw, "tell") pos = space.r_longlong_w(w_pos) if pos < 0: raise oefmt(space.w_IOError, "raw stream returned invalid position") self.abs_pos = pos return pos def closed_get_w(self, space): self._check_init(space) return space.getattr(self.w_raw, space.newtext("closed")) def name_get_w(self, space): self._check_init(space) return space.getattr(self.w_raw, space.newtext("name")) def mode_get_w(self, space): self._check_init(space) return space.getattr(self.w_raw, space.newtext("mode")) def readable_w(self, space): self._check_init(space) return space.call_method(self.w_raw, "readable") def writable_w(self, space): self._check_init(space) return space.call_method(self.w_raw, "writable") def seekable_w(self, space): self._check_init(space) return space.call_method(self.w_raw, "seekable") def isatty_w(self, space): self._check_init(space) return space.call_method(self.w_raw, "isatty") def repr_w(self, space): typename = space.type(self).name try: w_name = space.getattr(self, space.newtext("name")) except OperationError as e: if not e.match(space, space.w_Exception): raise return space.newtext("<%s>" % (typename,)) else: name_repr = space.text_w(space.repr(w_name)) return space.newtext("<%s name=%s>" % (typename, name_repr)) # ______________________________________________ @signature(types.any(), returns=types.int()) def _readahead(self): if self.readable and self.read_end != -1: available = self.read_end - self.pos assert available >= 0 return available return 0 def _raw_offset(self): if self.raw_pos >= 0 and ( (self.readable and self.read_end != -1) or (self.writable and self.write_end != -1)): return self.raw_pos - self.pos return 0 def tell_w(self, space): self._check_init(space) raw_tell = self._raw_tell(space) raw_offset = self._raw_offset() if raw_tell < raw_offset: return space.newint(0) return space.newint(raw_tell - raw_offset) @unwrap_spec(pos=r_longlong, whence=int) def seek_w(self, space, pos, whence=0): self._check_closed(space, "seek of closed file") if whence not in (0, 1, 2): raise oefmt(space.w_ValueError, "whence must be between 0 and 2, not %d", whence) check_seekable_w(space, self.w_raw) if whence != 2 and self.readable: # Check if seeking leaves us inside the current buffer, so as to # return quickly if possible. Also, we needn't take the lock in # this fast path. if self.abs_pos == -1: self._raw_tell(space) current = self.abs_pos available = self._readahead() if available > 0: if whence == 0: offset = pos - (current - self._raw_offset()) else: offset = pos if -self.pos <= offset <= available: newpos = self.pos + int(offset) assert newpos >= 0 self.pos = newpos res = current - available + offset if res < 0: return space.newint(0) return space.newint(res) # Fallback: invoke raw seek() method and clear buffer with self.lock: if self.writable: self._writer_flush_unlocked(space) if whence == 1: pos -= self._raw_offset() n = self._raw_seek(space, pos, whence) self.raw_pos = -1 if self.readable: self._reader_reset_buf() return space.newint(n) def _raw_seek(self, space, pos, whence): w_pos = space.call_method(self.w_raw, "seek", space.newint(pos), space.newint(whence)) pos = space.r_longlong_w(w_pos) if pos < 0: raise oefmt(space.w_IOError, "Raw stream returned invalid position") self.abs_pos = pos return pos def _closed(self, space): return space.is_true(space.getattr(self.w_raw, space.newtext("closed"))) def close_w(self, space): self._check_init(space) with self.lock: if self._closed(space): return flush_operr = None try: space.call_method(self, "flush") except OperationError as e: flush_operr = e raise finally: with self.lock: try: space.call_method(self.w_raw, "close") except OperationError as e: if flush_operr: e.chain_exceptions(space, flush_operr) raise self.buffer = None # free buffer memory self.maybe_unregister_rpython_finalizer_io(space) def _dealloc_warn_w(self, space, w_source): if self.w_raw: space.call_method(self.w_raw, "_dealloc_warn", w_source) else: raise oefmt(space.w_RuntimeError, "calling _dealloc_warn' on a closed or detached reader") def simple_flush_w(self, space): self._check_init(space) return space.call_method(self.w_raw, "flush") def _writer_flush_unlocked(self, space): if self.write_end == -1 or self.write_pos == self.write_end: self._writer_reset_buf() return # First, rewind rewind = self._raw_offset() + (self.pos - self.write_pos) if rewind != 0: self._raw_seek(space, -rewind, 1) self.raw_pos -= rewind written = 0 while self.write_pos < self.write_end: try: n = self._raw_write(space, self.write_pos, self.write_end) except BlockingIOError: raise make_write_blocking_error(space, 0) self.write_pos += n self.raw_pos = self.write_pos written += n # Partial writes can return successfully when interrupted by a # signal (see write(2)). We must run signal handlers before # blocking another time, possibly indefinitely. space.getexecutioncontext().checksignals() self._writer_reset_buf() def _write(self, space, data): w_data = space.newbytes(data) while True: try: w_written = space.call_method(self.w_raw, "write", w_data) except OperationError as e: if trap_eintr(space, e): continue # try again raise else: break if space.is_w(w_written, space.w_None): # Non-blocking stream would have blocked. raise BlockingIOError() written = space.getindex_w(w_written, space.w_IOError) if not 0 <= written <= len(data): raise oefmt(space.w_IOError, "raw write() returned invalid length") if self.abs_pos != -1: self.abs_pos += written return written def _raw_write(self, space, start, end): return self._write(space, self.buffer[start:end]) def detach_w(self, space): self._check_init(space) space.call_method(self, "flush") w_raw = self.w_raw self.w_raw = None self.state = STATE_DETACHED self._fast_closed_check = False return w_raw def fileno_w(self, space): self._check_init(space) return space.call_method(self.w_raw, "fileno") @unwrap_spec(w_size = WrappedDefault(None)) def truncate_w(self, space, w_size): self._check_init(space) self._check_closed(space, "truncate of closed file") if not self.writable: self._unsupportedoperation(space, "truncate") with self.lock: self._flush_and_rewind_unlocked(space) # invalidate cached position self.abs_pos = -1 return space.call_method(self.w_raw, "truncate", w_size) # ________________________________________________________________ # Read methods def read_w(self, space, w_size=None): self._check_closed(space, "read of closed file") size = convert_size(space, w_size) if size == -1: # read until the end of stream with self.lock: return self._read_all(space) elif size >= 0: res = self._read_fast(size) if res is None: with self.lock: res = self._read_generic(space, size) else: raise oefmt(space.w_ValueError, "read length must be positive or -1") if res is None: return space.w_None return space.newbytes(res) @unwrap_spec(size=int) def peek_w(self, space, size=0): self._check_closed(space, "peek of closed file") with self.lock: if self.writable: self._flush_and_rewind_unlocked(space) # Constraints: # 1. we don't want to advance the file position. # 2. we don't want to lose block alignment, so we can't shift the # buffer to make some place. # Therefore, we either return `have` bytes (if > 0), or a full # buffer. have = self._readahead() if have > 0: data = self.buffer[self.pos:self.pos+have] return space.newbytes(data) # Fill the buffer from the raw stream, and copy it to the result self._reader_reset_buf() try: size = self._fill_buffer(space) except BlockingIOError: size = 0 self.pos = 0 data = self.buffer[0:size] return space.newbytes(data) @unwrap_spec(size=int) def read1_w(self, space, size=-1): self._check_closed(space, "read of closed file") if size < 0: size = self.buffer_size if size == 0: return space.newbytes("") with self.lock: # Return up to n bytes. If at least one byte is buffered, we only # return buffered bytes. Otherwise, we do one raw read. # XXX: this mimicks the io.py implementation but is probably # wrong. If we need to read from the raw stream, then we could # actually read all `n` bytes asked by the caller (and possibly # more, so as to fill our buffer for the next reads). have = self._readahead() if have == 0: if self.writable: self._flush_and_rewind_unlocked(space) # Fill the buffer from the raw stream self._reader_reset_buf() self.pos = 0 try: have = self._fill_buffer(space) except BlockingIOError: have = 0 if size > have: size = have endpos = self.pos + size data = self.buffer[self.pos:endpos] self.pos = endpos return space.newbytes(data) def _read_all(self, space): "Read all the file, don't update the cache" # Must run with the lock held! builder = StringBuilder() # First copy what we have in the current buffer current_size = self._readahead() data = None if current_size: data = self.buffer[self.pos:self.pos + current_size] builder.append(data) self.pos += current_size # We're going past the buffer's bounds, flush it if self.writable: self._flush_and_rewind_unlocked(space) self._reader_reset_buf() while True: # Read until EOF or until read() would block w_data = space.call_method(self.w_raw, "read") if space.is_w(w_data, space.w_None): if current_size == 0: return w_data break data = space.bytes_w(w_data) size = len(data) if size == 0: break builder.append(data) current_size += size if self.abs_pos != -1: self.abs_pos += size return space.newbytes(builder.build()) def _raw_read(self, space, buffer, start, length): from pypy.module._io.interp_fileio import W_FileIO assert buffer is not None length = intmask(length) start = intmask(start) w_raw = self.w_raw target_address = lltype.nullptr(rffi.CCHARP.TO) if type(w_raw) is W_FileIO: try: raw_address = buffer.get_raw_address() except ValueError: pass else: target_address = rffi.ptradd(raw_address, start) if target_address: assert type(w_raw) is W_FileIO w_size = w_raw._readinto_raw(space, target_address, length) keepalive_until_here(buffer) else: w_view = SimpleView(SubBuffer(buffer, start, length), w_obj=self).wrap(space) while True: try: w_size = space.call_method(self.w_raw, "readinto", w_view) except OperationError as e: if trap_eintr(space, e): continue # try again raise else: break if space.is_w(w_size, space.w_None): raise BlockingIOError() size = space.int_w(w_size) if size < 0 or size > length: raise oefmt(space.w_IOError, "raw readinto() returned invalid length %d (should " "have been between 0 and %d)", size, length) if self.abs_pos != -1: self.abs_pos += size return size def _fill_buffer(self, space): self._check_closed(space) start = self.read_end if start == -1: start = 0 length = self.buffer_size - start size = self._raw_read(space, self.buffer, start, length) if size > 0: self.read_end = self.raw_pos = start + size return size def _read_generic(self, space, n): """Generic read function: read from the stream until enough bytes are read, or until an EOF occurs or until read() would block.""" # Must run with the lock held! current_size = self._readahead() if n <= current_size: return self._read_fast(n) result_buffer = self._make_buffer(space, n) remaining = n written = 0 if current_size: self.output_slice(space, result_buffer, written, self.buffer[self.pos:self.pos + current_size]) remaining -= current_size written += current_size self.pos += current_size # Flush the write buffer if necessary if self.writable: self._flush_and_rewind_unlocked(space) self._reader_reset_buf() # Read whole blocks, and don't buffer them while remaining > 0: r = self.buffer_size * (remaining // self.buffer_size) if r == 0: break try: size = self._raw_read(space, result_buffer, written, r) except BlockingIOError: if written == 0: return None size = 0 if size == 0: return result_buffer[0:written] remaining -= size written += size self.pos = 0 self.raw_pos = 0 self.read_end = 0 while remaining > 0 and self.read_end < self.buffer_size: try: size = self._fill_buffer(space) except BlockingIOError: # EOF or read() would block if written == 0: return None size = 0 if size == 0: break if remaining > 0: if size > remaining: size = remaining self.output_slice(space, result_buffer, written, self.buffer[self.pos:self.pos + size]) self.pos += size written += size remaining -= size return result_buffer[0:written] def _read_fast(self, n): """Read n bytes from the buffer if it can, otherwise return None. This function is simple enough that it can run unlocked.""" current_size = self._readahead() if n <= current_size: endpos = self.pos + n res = self.buffer[self.pos:endpos] self.pos = endpos return res return None def readline_w(self, space, w_limit=None): self._check_closed(space, "readline of closed file") limit = convert_size(space, w_limit) # First, try to find a line in the buffer. This can run # unlocked because the calls to the C API are simple enough # that they can't trigger any thread switch. have = self._readahead() if limit >= 0 and have > limit: have = limit buffer = self.buffer if isinstance(buffer, ByteBuffer): # hack at the internals of self.buffer, otherwise this loop is # extremely slow for pos in range(self.pos, self.pos+have): if buffer.data[pos] == '\n': break else: pos = -1 else: for pos in range(self.pos, self.pos+have): if buffer[pos] == '\n': break else: pos = -1 if pos >= 0: w_res = space.newbytes(self.buffer[self.pos:pos+1]) self.pos = pos + 1 return w_res if have == limit: w_res = space.newbytes(self.buffer[self.pos:self.pos+have]) self.pos += have return w_res written = 0 with self.lock: # Now we try to get some more from the raw stream chunks = [] if have > 0: chunks.append(self.buffer[self.pos:self.pos+have]) written += have self.pos += have if limit >= 0: limit -= have if self.writable: self._flush_and_rewind_unlocked(space) while True: self._reader_reset_buf() try: have = self._fill_buffer(space) except BlockingIOError: have = 0 if have == 0: break if limit >= 0 and have > limit: have = limit pos = 0 found = False while pos < have: c = self.buffer.getitem(pos) pos += 1 if c == '\n': self.pos = pos found = True break chunks.append(self.buffer[0:pos]) if found: break if have == limit: self.pos = have break written += have if limit >= 0: limit -= have return space.newbytes(''.join(chunks)) # ____________________________________________________ # Write methods def _adjust_position(self, new_pos): assert new_pos >= 0 self.pos = new_pos if self.readable and self.read_end != -1 and self.read_end < new_pos: self.read_end = self.pos def write_w(self, space, w_data): self._check_init(space) data = space.charbuf_w(w_data) size = len(data) with self.lock: self._check_closed(space, "write to closed file") if (not (self.readable and self.read_end != -1) and not (self.writable and self.write_end != -1)): self.pos = 0 self.raw_pos = 0 available = self.buffer_size - self.pos # Fast path: the data to write can be fully buffered if size <= available: for i in range(size): self.buffer[self.pos + i] = data[i] if self.write_end == -1 or self.write_pos > self.pos: self.write_pos = self.pos self._adjust_position(self.pos + size) if self.pos > self.write_end: self.write_end = self.pos return space.newint(size) # First write the current buffer try: self._writer_flush_unlocked(space) except OperationError as e: if not e.match(space, space.w_BlockingIOError): raise if self.readable: self._reader_reset_buf() # Make some place by shifting the buffer for i in range(self.write_pos, self.write_end): self.buffer.setitem(i - self.write_pos, self.buffer.getitem(i)) self.write_end -= self.write_pos self.raw_pos -= self.write_pos newpos = self.pos - self.write_pos assert newpos >= 0 self.pos = newpos self.write_pos = 0 available = self.buffer_size - self.write_end assert available >= 0 if size <= available: # Everything can be buffered for i in range(size): self.buffer[self.write_end + i] = data[i] self.write_end += size self.pos += size return space.newint(size) # Buffer as much as possible for i in range(available): self.buffer[self.write_end + i] = data[i] self.write_end += available self.pos += available # Modifying the existing exception will will change # e.characters_written but not e.args[2]. Therefore # we just replace with a new error. raise make_write_blocking_error(space, available) # Adjust the raw stream position if it is away from the logical # stream position. This happens if the read buffer has been filled # but not modified (and therefore _bufferedwriter_flush_unlocked() # didn't rewind the raw stream by itself). offset = self._raw_offset() if offset: self._raw_seek(space, -offset, 1) self.raw_pos -= offset # Then write buf itself. At this point the buffer has been emptied remaining = size written = 0 while remaining > self.buffer_size: try: n = self._write(space, data[written:]) except BlockingIOError: # Write failed because raw file is non-blocking if remaining > self.buffer_size: # Can't buffer everything, still buffer as much as # possible for i in range(self.buffer_size): self.buffer[i] = data[written + i] self.raw_pos = 0 self._adjust_position(self.buffer_size) self.write_end = self.buffer_size written += self.buffer_size raise make_write_blocking_error(space, written) break written += n remaining -= n # Partial writes can return successfully when interrupted by a # signal (see write(2)). We must run signal handlers before # blocking another time, possibly indefinitely. space.getexecutioncontext().checksignals() if self.readable: self._reader_reset_buf() if remaining > 0: for i in range(remaining): self.buffer[i] = data[written + i] written += remaining self.write_pos = 0 self.write_end = remaining self._adjust_position(remaining) self.raw_pos = 0 return space.newint(written) def flush_w(self, space): self._check_closed(space, "flush of closed file") with self.lock: self._flush_and_rewind_unlocked(space) def _flush_and_rewind_unlocked(self, space): self._writer_flush_unlocked(space) if self.readable: # Rewind the raw stream so that its position corresponds to # the current logical position. try: self._raw_seek(space, -self._raw_offset(), 1) finally: self._reader_reset_buf() class BufferedReaderMixin(object): import_from_mixin(BufferedMixin) def readinto_w(self, space, w_buffer): return self._readinto(space, w_buffer, read_once=False) def readinto1_w(self, space, w_buffer): return self._readinto(space, w_buffer, read_once=True) def _readinto(self, space, w_buffer, read_once): self._check_init(space) self._check_closed(space, "readinto of closed file") rwbuffer = space.writebuf_w(w_buffer) length = rwbuffer.getlength() with self.lock: have = self._readahead() if have >= length: self.output_slice(space, rwbuffer, 0, self.buffer[self.pos:self.pos + length]) self.pos += length return space.newint(length) written = 0 if have > 0: self.output_slice(space, rwbuffer, 0, self.buffer[self.pos:self.read_end]) written = have while written < length: if self.writable: self._flush_and_rewind_unlocked(space) self._reader_reset_buf() self.pos = 0 if written + len(self.buffer) < length: try: got = self._raw_read( space, rwbuffer, written, length - written) written += got except BlockingIOError: got = 0 if got == 0: break elif read_once and written: break else: try: have = self._fill_buffer(space) except BlockingIOError: have = 0 if have == 0: break endpos = min(have, length - written) assert endpos >= 0 self.output_slice(space, rwbuffer, written, self.buffer[0:endpos]) written += endpos self.pos = endpos if read_once: break return space.newint(written) class W_BufferedReader(W_BufferedIOBase): import_from_mixin(BufferedReaderMixin) @unwrap_spec(buffer_size=int) def descr_init(self, space, w_raw, buffer_size=DEFAULT_BUFFER_SIZE): from pypy.module._io.interp_fileio import W_FileIO self.state = STATE_ZERO check_readable_w(space, w_raw) self.w_raw = w_raw self.buffer_size = buffer_size self.readable = True self._init(space) self._reader_reset_buf() self.state = STATE_OK self._fast_closed_check = (type(self) is W_BufferedReader and type(w_raw) is W_FileIO) W_BufferedReader.typedef = TypeDef( '_io.BufferedReader', W_BufferedIOBase.typedef, __new__ = generic_new_descr(W_BufferedReader), __init__ = interp2app(W_BufferedReader.descr_init), __getstate__ = interp2app(W_BufferedReader.getstate_w), read = interp2app(W_BufferedReader.read_w), peek = interp2app(W_BufferedReader.peek_w), read1 = interp2app(W_BufferedReader.read1_w), readinto = interp2app(W_BufferedReader.readinto_w), readinto1 = interp2app(W_BufferedReader.readinto1_w), raw = interp_attrproperty_w("w_raw", cls=W_BufferedReader), readline = interp2app(W_BufferedReader.readline_w), # from the mixin class __repr__ = interp2app(W_BufferedReader.repr_w), readable = interp2app(W_BufferedReader.readable_w), seekable = interp2app(W_BufferedReader.seekable_w), seek = interp2app(W_BufferedReader.seek_w), tell = interp2app(W_BufferedReader.tell_w), close = interp2app(W_BufferedReader.close_w), flush = interp2app(W_BufferedReader.simple_flush_w), # Not flush_w! detach = interp2app(W_BufferedReader.detach_w), truncate = interp2app(W_BufferedReader.truncate_w), fileno = interp2app(W_BufferedReader.fileno_w), isatty = interp2app(W_BufferedReader.isatty_w), _dealloc_warn = interp2app(W_BufferedReader._dealloc_warn_w), closed = GetSetProperty(W_BufferedReader.closed_get_w), name = GetSetProperty(W_BufferedReader.name_get_w), mode = GetSetProperty(W_BufferedReader.mode_get_w), ) class W_BufferedWriter(W_BufferedIOBase): import_from_mixin(BufferedMixin) @unwrap_spec(buffer_size=int) def descr_init(self, space, w_raw, buffer_size=DEFAULT_BUFFER_SIZE): from pypy.module._io.interp_fileio import W_FileIO self.state = STATE_ZERO check_writable_w(space, w_raw) self.w_raw = w_raw self.buffer_size = buffer_size self.writable = True self._init(space) self._writer_reset_buf() self.state = STATE_OK self._fast_closed_check = (type(self) is W_BufferedWriter and type(w_raw) is W_FileIO) W_BufferedWriter.typedef = TypeDef( '_io.BufferedWriter', W_BufferedIOBase.typedef, __new__ = generic_new_descr(W_BufferedWriter), __init__ = interp2app(W_BufferedWriter.descr_init), __getstate__ = interp2app(W_BufferedWriter.getstate_w), write = interp2app(W_BufferedWriter.write_w), flush = interp2app(W_BufferedWriter.flush_w), raw = interp_attrproperty_w("w_raw", cls=W_BufferedWriter), # from the mixin class __repr__ = interp2app(W_BufferedWriter.repr_w), writable = interp2app(W_BufferedWriter.writable_w), seekable = interp2app(W_BufferedWriter.seekable_w), seek = interp2app(W_BufferedWriter.seek_w), tell = interp2app(W_BufferedWriter.tell_w), close = interp2app(W_BufferedWriter.close_w), fileno = interp2app(W_BufferedWriter.fileno_w), isatty = interp2app(W_BufferedWriter.isatty_w), detach = interp2app(W_BufferedWriter.detach_w), truncate = interp2app(W_BufferedWriter.truncate_w), _dealloc_warn = interp2app(W_BufferedWriter._dealloc_warn_w), closed = GetSetProperty(W_BufferedWriter.closed_get_w), name = GetSetProperty(W_BufferedWriter.name_get_w), mode = GetSetProperty(W_BufferedWriter.mode_get_w), ) def make_forwarding_method(method, writer=False, reader=False): @func_renamer(method + '_w') def method_w(self, space, __args__): if writer: if self.w_writer is None: raise oefmt(space.w_ValueError, "I/O operation on uninitialized object") w_meth = space.getattr(self.w_writer, space.newtext(method)) w_result = space.call_args(w_meth, __args__) if reader: if self.w_reader is None: raise oefmt(space.w_ValueError, "I/O operation on uninitialized object") w_meth = space.getattr(self.w_reader, space.newtext(method)) w_result = space.call_args(w_meth, __args__) return w_result return method_w class W_BufferedRWPair(W_BufferedIOBase): w_reader = None w_writer = None @unwrap_spec(buffer_size=int) def descr_init(self, space, w_reader, w_writer, buffer_size=DEFAULT_BUFFER_SIZE): try: self.w_reader = W_BufferedReader(space) self.w_reader.descr_init(space, w_reader, buffer_size) self.w_writer = W_BufferedWriter(space) self.w_writer.descr_init(space, w_writer, buffer_size) except Exception: self.w_reader = None self.w_writer = None raise # forward to reader for method in ['read', 'peek', 'read1', 'readinto', 'readable']: locals()[method + '_w'] = make_forwarding_method( method, reader=True) # forward to writer for method in ['write', 'flush', 'writable']: locals()[method + '_w'] = make_forwarding_method( method, writer=True) # forward to both def close_w(self, space, __args__): if self.w_writer is None: raise oefmt(space.w_ValueError, "I/O operation on uninitialized object") w_meth = space.getattr(self.w_writer, space.newtext("close")) try: space.call_args(w_meth, __args__) except OperationError as e: pass else: e = None if self.w_reader is None: raise oefmt(space.w_ValueError, "I/O operation on uninitialized object") w_meth = space.getattr(self.w_reader, space.newtext("close")) try: space.call_args(w_meth, __args__) except OperationError as e2: if e: e2.chain_exceptions(space, e) e = e2 if e: raise e def needs_finalizer(self): # self.w_writer and self.w_reader have their own finalizer return type(self) is not W_BufferedRWPair def isatty_w(self, space): if space.is_true(space.call_method(self.w_writer, "isatty")): return space.w_True return space.call_method(self.w_reader, "isatty") def closed_get_w(self, space): return space.getattr(self.w_writer, space.newtext("closed")) methods = dict((method, interp2app(getattr(W_BufferedRWPair, method + '_w'))) for method in ['read', 'peek', 'read1', 'readinto', 'readable', 'write', 'flush', 'writable', 'close', 'isatty']) W_BufferedRWPair.typedef = TypeDef( '_io.BufferedRWPair', W_BufferedIOBase.typedef, __new__ = generic_new_descr(W_BufferedRWPair), __init__ = interp2app(W_BufferedRWPair.descr_init), __getstate__ = interp2app(W_BufferedRWPair.getstate_w), closed = GetSetProperty(W_BufferedRWPair.closed_get_w), **methods ) class W_BufferedRandom(W_BufferedIOBase): import_from_mixin(BufferedReaderMixin) @unwrap_spec(buffer_size=int) def descr_init(self, space, w_raw, buffer_size=DEFAULT_BUFFER_SIZE): from pypy.module._io.interp_fileio import W_FileIO self.state = STATE_ZERO check_readable_w(space, w_raw) check_writable_w(space, w_raw) check_seekable_w(space, w_raw) self.w_raw = w_raw self.buffer_size = buffer_size self.readable = self.writable = True self._init(space) self._reader_reset_buf() self._writer_reset_buf() self.pos = 0 self.state = STATE_OK self._fast_closed_check = (type(self) is W_BufferedWriter and type(w_raw) is W_FileIO) W_BufferedRandom.typedef = TypeDef( '_io.BufferedRandom', W_BufferedIOBase.typedef, __new__ = generic_new_descr(W_BufferedRandom), __init__ = interp2app(W_BufferedRandom.descr_init), __getstate__ = interp2app(W_BufferedRandom.getstate_w), read = interp2app(W_BufferedRandom.read_w), peek = interp2app(W_BufferedRandom.peek_w), read1 = interp2app(W_BufferedRandom.read1_w), readline = interp2app(W_BufferedRandom.readline_w), readinto = interp2app(W_BufferedRandom.readinto_w), readinto1 = interp2app(W_BufferedRandom.readinto1_w), write = interp2app(W_BufferedRandom.write_w), flush = interp2app(W_BufferedRandom.flush_w), raw = interp_attrproperty_w("w_raw", cls=W_BufferedRandom), # from the mixin class __repr__ = interp2app(W_BufferedRandom.repr_w), readable = interp2app(W_BufferedRandom.readable_w), writable = interp2app(W_BufferedRandom.writable_w), seekable = interp2app(W_BufferedRandom.seekable_w), seek = interp2app(W_BufferedRandom.seek_w), tell = interp2app(W_BufferedRandom.tell_w), close = interp2app(W_BufferedRandom.close_w), detach = interp2app(W_BufferedRandom.detach_w), truncate = interp2app(W_BufferedRandom.truncate_w), fileno = interp2app(W_BufferedRandom.fileno_w), isatty = interp2app(W_BufferedRandom.isatty_w), _dealloc_warn = interp2app(W_BufferedRandom._dealloc_warn_w), closed = GetSetProperty(W_BufferedRandom.closed_get_w), name = GetSetProperty(W_BufferedRandom.name_get_w), mode = GetSetProperty(W_BufferedRandom.mode_get_w), )