Add buffering to File.openInputStream, so that the entire file is not read in at once.

runtime/bin/file_impl.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 class _FileInputStream extends _BaseDataInputStream implements InputStream {
-  _FileInputStream(String name) {
+  _FileInputStream(String name)
+      : _data = [],
+        _position = 0,
+        _filePosition = 0 {
     var file = new File(name);
-    _data = [];
-    _position = 0;
-    var chained = file.open(FileMode.READ).chain((openedFile) {
-      return _readDataFromFile(openedFile);
+    var future = file.open(FileMode.READ);
+    future.handleException((e) {
+      _reportError(e);
+      return true;
+      });

[Søren Gjesse, 2012/06/07 12:17:13]

Indentation.

[Bill Hesse, 2012/06/07 13:15:08]

Done.

+    future.then(_setupOpenedFile);
+  }
+
+  _FileInputStream.fromStdio(int fd)
+      : _data = [],
+        _position = 0,
+        _filePosition = 0 {
+    assert(fd == 0);
+    _setupOpenedFile(_File._openStdioSync(fd));
+  }
+
+  void _setupOpenedFile(RandomAccessFile openedFile) {
+    _openedFile = openedFile;
+    var chained = _openedFile.length().chain((len) {
+      _fileLength = len;
+      return _fillBuffer();
     });
     chained.handleException((e) {
       _reportError(e);
       return true;
     });
+    chained.then((ignored) => _checkScheduleCallbacks());
   }
 
-  _FileInputStream.fromStdio(int fd) {
-    assert(fd == 0);
-    var file = _File._openStdioSync(fd);
-    _data = [];
-    _position = 0;
-    _readDataFromFile(file).handleException((e) {
-      _reportError(e);
-      return true;
+  Future<int> _fillBuffer() {
+    Expect.equals(_position, _data.length);
+    // Expect.isTrue(_filePosition < _fileLength);

[Søren Gjesse, 2012/06/07 12:17:13]

Code in comments.

[Bill Hesse, 2012/06/07 13:15:08]

Done.

+
+    int size = Math.min(_bufferLength, _fileLength - _filePosition);
+    if (_data.length != size) {
+      _data = new Uint8List(size);
+    }
+    var future =  _openedFile.readList(_data, 0, _data.length);

[Bill Hesse, 2012/06/07 13:15:08]

tab?

+    future.transform((read) {

[Bill Hesse, 2012/06/07 13:15:08]

future = future.transform

Fixed.

+      _filePosition += read;
+      if (read != _data.length) {
+        _data.removeRange(read, _data.length - read);
+      }
+      _position = 0;
+
+      if (_fileLength == _filePosition) {
+        _streamMarkedClosed = true;
+        _openedFile.close();
+      }
+      return read;
     });
-  }
-
-  Future<RandomAccessFile> _closeAfterRead(RandomAccessFile openedFile) {
-    return openedFile.close().transform((ignore) {
-      _streamMarkedClosed = true;
-      _checkScheduleCallbacks();
-      return openedFile;
-    });
-  }
-
-  Future<RandomAccessFile> _readDataFromFile(RandomAccessFile openedFile) {
-    return openedFile.length().chain((length) {
-      var contents = new Uint8List(length);
-      if (length != 0) {
-        return openedFile.readList(contents, 0, length).chain((read) {
-          if (read != length) {
-            throw new FileIOException(
-                'Failed reading file contents in FileInputStream');
-          } else {
-            _data = contents;
-          }
-          return _closeAfterRead(openedFile);
-        });
-      } else {
-        return _closeAfterRead(openedFile);
-      }
-    });
+    return future;
   }
 
   int available() {
-    return _closed ? 0 : _data.length - _position;
+    return closed ? 0 : _data.length - _position;
   }
 
   void pipe(OutputStream output, [bool close = true]) {
     _pipe(this, output, close: close);
   }
 
   List<int> _read(int bytesToRead) {
     List<int> result = new Uint8List(bytesToRead);

[Søren Gjesse, 2012/06/07 12:17:13]

Maybe we can avoid the copying here if reading the exact content of _data - the
_data can be returned.

[Bill Hesse, 2012/06/07 13:15:08]

Done.

     result.setRange(0, bytesToRead, _data, _position);
     _position += bytesToRead;
-    _checkScheduleCallbacks();
+    if (_position == _data.length && !_streamMarkedClosed) {
+      _fillBuffer().then((ignored) {
+        _checkScheduleCallbacks();
+      });
+    } else {
+      _checkScheduleCallbacks();
+    }
     return result;
   }
 
   int _readInto(List<int> buffer, int offset, int len) {
     buffer.setRange(offset, len, _data, _position);
     _position += len;

[Søren Gjesse, 2012/06/07 12:17:13]

Maybe refactor the duplicate (here and above in _read) code for filling the
buffer.

[Bill Hesse, 2012/06/07 13:15:08]

Done.

-    _checkScheduleCallbacks();
+    if (_position == _data.length && !_streamMarkedClosed) {
+      _fillBuffer().then((ignored) {
+        _checkScheduleCallbacks();
+      });
+    } else {
+      _checkScheduleCallbacks();
+    }
     return len;
   }
 
   void _close() {
-    if (_closed) return;
-    _closed = true;
+    _streamMarkedClosed = true;
+    _data = [];
+    _position = 0;
+    if (!_openedFile.closed) {
+      _openedFile.close();
+    }
   }
 
+  static final int _bufferLength = 64 * 1024;
+
+  RandomAccessFile _openedFile;
   List<int> _data;
   int _position;
-  bool _closed = false;
+  int _filePosition;
+  int _fileLength;
 }
 
 
 class _FileOutputStream extends _BaseOutputStream implements OutputStream {
   _FileOutputStream(String name, FileMode mode) {
     _pendingOperations = new List<List<int>>();
     var f = new File(name);
     var openFuture = f.open(mode);
     openFuture.then((openedFile) {
       _file = openedFile;
@@ skipping 580 matching lines @@
 
   SendPort _fileService;
 }
 
 
 class _RandomAccessFile extends _FileBase implements RandomAccessFile {
   _RandomAccessFile(int this._id, String this._name);
 
   Future<RandomAccessFile> close() {
     Completer<RandomAccessFile> completer = new Completer<RandomAccessFile>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     _ensureFileService();
     List request = new List(2);
     request[0] = _FileUtils.CLOSE_REQUEST;
     request[1] = _id;
     // Set the id_ to 0 (NULL) to ensure the no more async requests
-    // can be issues for this file.
+    // can be issued for this file.
     _id = 0;
     return _fileService.call(request).transform((result) {
       if (result != -1) {
         _id = result;
         return this;
       } else {
         throw new FileIOException("Cannot close file '$_name'");
       }
     });
   }
 
   void closeSync() {
     var id = _FileUtils.close(_id);
     if (id == -1) {
       throw new FileIOException("Cannot close file '$_name'");
     }
     _id = id;
   }
 
   Future<int> readByte() {
     _ensureFileService();
     Completer<int> completer = new Completer<int>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(2);
     request[0] = _FileUtils.READ_BYTE_REQUEST;
     request[1] = _id;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "readByte failed for file '$_name'");
       }
       return response;
     });
@@ skipping 14 matching lines @@
     if (buffer is !List || offset is !int || bytes is !int) {
       // Complete asynchronously so the user has a chance to setup
       // handlers without getting exceptions when registering the
       // then handler.
       new Timer(0, (t) {
         completer.completeException(new FileIOException(
             "Invalid arguments to readList for file '$_name'"));
       });
       return completer.future;
     };
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(3);
     request[0] = _FileUtils.READ_LIST_REQUEST;
     request[1] = _id;
     request[2] = bytes;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "readList failed for file '$_name'");
       }
       var read = response[1];
@@ skipping 29 matching lines @@
     if (value is !int) {
       // Complete asynchronously so the user has a chance to setup
       // handlers without getting exceptions when registering the
       // then handler.
       new Timer(0, (t) {
           completer.completeException(new FileIOException(
               "Invalid argument to writeByte for file '$_name'"));
       });
       return completer.future;
     }
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(3);
     request[0] = _FileUtils.WRITE_BYTE_REQUEST;
     request[1] = _id;
     request[2] = value;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "writeByte failed for file '$_name'");
       }
       return this;
@@ skipping 20 matching lines @@
     if (buffer is !List || offset is !int || bytes is !int) {
       // Complete asynchronously so the user has a chance to setup
       // handlers without getting exceptions when registering the
       // then handler.
       new Timer(0, (t) {
           completer.completeException(new FileIOException(
           "Invalid arguments to writeList for file '$_name'"));
       });
       return completer.future;
     }
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
 
     List result;
     try {
       result =
           _FileUtils.ensureFastAndSerializableBuffer(buffer, offset, bytes);
     } catch (var e) {
       // Complete asynchronously so the user has a chance to setup
       // handlers without getting exceptions when registering the
       // then handler.
       new Timer(0, (t) => completer.completeException(e));
@@ skipping 33 matching lines @@
     if (result is OSError) {
       throw new FileIOException("writeList failed for file '$_name'", result);
     }
     return result;
   }
 
   Future<RandomAccessFile> writeString(String string,
                                        [Encoding encoding = Encoding.UTF_8]) {
     _ensureFileService();
     Completer<RandomAccessFile> completer = new Completer<RandomAccessFile>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(3);
     request[0] = _FileUtils.WRITE_STRING_REQUEST;
     request[1] = _id;
     request[2] = string;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "writeString failed for file '$_name'");
       }
       return this;
     });
   }
 
   int writeStringSync(String string, [Encoding encoding = Encoding.UTF_8]) {
     _checkNotClosed();
     var result = _FileUtils.checkedWriteString(_id, string);
     if (result is OSError) {
       throw new FileIOException("writeString failed for file '$_name'");
     }
     return result;
   }
 
   Future<int> position() {
     _ensureFileService();
     Completer<int> completer = new Completer<int>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(2);
     request[0] = _FileUtils.POSITION_REQUEST;
     request[1] = _id;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "position failed for file '$_name'");
       }
       return response;
     });
   }
 
   int positionSync() {
     _checkNotClosed();
     var result = _FileUtils.position(_id);
     if (result is OSError) {
       throw new FileIOException("position failed for file '$_name'", result);
     }
     return result;
   }
 
   Future<RandomAccessFile> setPosition(int position) {
     _ensureFileService();
     Completer<RandomAccessFile> completer = new Completer<RandomAccessFile>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(3);
     request[0] = _FileUtils.SET_POSITION_REQUEST;
     request[1] = _id;
     request[2] = position;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "setPosition failed for file '$_name'");
       }
       return this;
     });
   }
 
   void setPositionSync(int position) {
     _checkNotClosed();
     var result = _FileUtils.setPosition(_id, position);
     if (result is OSError) {
       throw new FileIOException("setPosition failed for file '$_name'", result);
     }
   }
 
   Future<RandomAccessFile> truncate(int length) {
     _ensureFileService();
     Completer<RandomAccessFile> completer = new Completer<RandomAccessFile>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(3);
     request[0] = _FileUtils.TRUNCATE_REQUEST;
     request[1] = _id;
     request[2] = length;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "truncate failed for file '$_name'");
       }
       return this;
     });
   }
 
   void truncateSync(int length) {
     _checkNotClosed();
     var result = _FileUtils.truncate(_id, length);
     if (result is OSError) {
       throw new FileIOException("truncate failed for file '$_name'", result);
     }
   }
 
   Future<int> length() {
     _ensureFileService();
     Completer<int> completer = new Completer<int>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(2);
     request[0] = _FileUtils.LENGTH_REQUEST;
     request[1] = _id;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "length failed for file '$_name'");
       }
       return response;
     });
   }
 
   int lengthSync() {
     _checkNotClosed();
     var result = _FileUtils.length(_id);
     if (result is OSError) {
       throw new FileIOException("length failed for file '$_name'", result);
     }
     return result;
   }
 
   Future<RandomAccessFile> flush() {
     _ensureFileService();
     Completer<RandomAccessFile> completer = new Completer<RandomAccessFile>();
-    if (_isClosed) return _completeWithClosedException(completer);
+    if (closed) return _completeWithClosedException(completer);
     List request = new List(2);
     request[0] = _FileUtils.FLUSH_REQUEST;
     request[1] = _id;
     return _fileService.call(request).transform((response) {
       if (_isErrorResponse(response)) {
         throw _exceptionFromResponse(response,
                                      "flush failed for file '$_name'");
       }
       return this;
     });
   }
 
   void flushSync() {
     _checkNotClosed();
     var result = _FileUtils.flush(_id);
     if (result is OSError) {
       throw new FileIOException("flush failed for file '$_name'", result);
     }
   }
 
   String get name() => _name;
 
   void _ensureFileService() {
     if (_fileService == null) {
       _fileService = _FileUtils.newServicePort();
     }
   }
 
-  bool get _isClosed() => _id == 0;
+  bool get closed() => _id == 0;
 
   void _checkNotClosed() {
-    if (_isClosed) {
+    if (closed) {
       throw new FileIOException("File closed '$_name'");
     }
   }
 
   Future _completeWithClosedException(Completer completer) {
     new Timer(0, (t) {
       completer.completeException(
           new FileIOException("File closed '$_name'"));
     });
     return completer.future;
   }
 
   final String _name;
   int _id;
 
   SendPort _fileService;
 }