Make FileInputStream and FileOutputStream actually asynchronous.

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(RandomAccessFile this._file, int this._length) {
-    _streamMarkedClosed = true;
-    _checkScheduleCallbacks();
+  _FileInputStream(String name) {
+    _file = new File(name);
+    _data = [];
+    _position = 0;
+    _file.errorHandler = (String s) {
+      if (_clientErrorHandler != null) {
+        _clientErrorHandler();
+      }
+    };
+    _file.open();
+    _file.openHandler = (openedFile) {
+      _readDataFromFile(openedFile);
+    };
   }
 
   _FileInputStream.fromStdio(int fd) {
     assert(fd == 0);
     _file = _File._openStdioSync(fd);
-    _length = _file.lengthSync();
-    _streamMarkedClosed = true;
-    _checkScheduleCallbacks();
+    _data = [];
+    _position = 0;
+    _readDataFromFile(_file);
+  }
+
+  void _readDataFromFile(RandomAccessFile openedFile) {
+    openedFile.errorHandler = (String s) {
+      if (_clientErrorHandler != null) {
+        _clientErrorHandler();
+      }
+    };
+    openedFile.length();

[Søren Gjesse, 2012/02/28 07:40:25]

What is length when the file is stdin?

[Søren Gjesse, 2012/02/28 08:10:26]

This is correct. I forgot that the constructor _FileInputStream.fromStdio is
used when stdin is redirected from a file and the file descriptor is for an open
file - not a pipe.

+    openedFile.lengthHandler = (length) {
+      var contents = new ByteArray(length);
+      if (length != 0) {
+        openedFile.readList(contents, 0, length);
+        openedFile.readListHandler = (read) {
+          if (read != length) {
+            if (_clientErrorHandler != null) {
+              _clientErrorHandler();
+            }
+            _streamMarkedClosed = true;
+            _checkScheduleCallbacks();
+          } else {
+            _data = contents;
+            _streamMarkedClosed = true;
+            _checkScheduleCallbacks();
+          }
+          openedFile.close();
+        };
+      } else {
+        _streamMarkedClosed = true;
+        _checkScheduleCallbacks();
+        openedFile.close();
+      }
+    };
   }
 
   int available() {
-    return _closed ? 0 : _length - _file.positionSync();
+    return _closed ? 0 : _data.length - _position;
   }
 
   void pipe(OutputStream output, [bool close = true]) {
     _pipe(this, output, close: close);
   }
 
   List<int> _read(int bytesToRead) {
     ByteArray result = new ByteArray(bytesToRead);
-    int bytesRead = _file.readListSync(result, 0, bytesToRead);
-    if (bytesRead < bytesToRead) {
-      ByteArray buffer = new ByteArray(bytesRead);
-      buffer.setRange(0, bytesRead, result);
-      result = buffer;
-    }
+    result.setRange(0, bytesToRead, _data, _position);
+    _position += bytesToRead;
     _checkScheduleCallbacks();
     return result;
   }
 
   int _readInto(List<int> buffer, int offset, int len) {
-    int result = _file.readListSync(buffer, offset, len);
+    buffer.setRange(offset, len, _data, _position);
+    _position += len;
     _checkScheduleCallbacks();
-    return result;
+    return len;
   }
 
   void _close() {
     if (_closed) return;
-    _file.closeSync();
     _closed = true;
   }
 
-  RandomAccessFile _file;
-  int _length;
+  File _file;
+  List<int> _data;
+  int _position;
   bool _closed = false;
 }
 
 
 class _FileOutputStream implements OutputStream {
-  _FileOutputStream(this._file);
+  _FileOutputStream(String name, int mode) {
+    _pendingOperations = new List<List<int>>();
+    var f = new File(name);
+    f.open(mode);
+    f.openHandler = (openedFile) {
+      _file = openedFile;
+      _setupFileHandlers();
+      _processPendingOperations();
+    };
+    f.errorHandler = (e) {
+      if (_errorHandler != null) _errorHandler();
+    };
+  }
 
   _FileOutputStream.fromStdio(int fd) {
     assert(1 <= fd && fd <= 2);
     _file = _File._openStdioSync(fd);
+    _setupFileHandlers();
+  }
+
+
+  void _setupFileHandlers() {
+    _file.errorHandler = (e) {
+      if (_errorHandler != null) _errorHandler();
+    };
+    _file.noPendingWriteHandler = () {
+      if (!_streamMarkedClosed && _noPendingWriteHandler != null) {
+        _noPendingWriteHandler();
+      }
+    };
+    _file.closeHandler = () {
+      if (_closeHandler != null) _closeHandler();
+    };
   }
 
   bool write(List<int> buffer, [bool copyBuffer = false]) {
-    bool result = _write(buffer, 0, buffer.length);
-    if (result) {
-      _checkScheduleCallbacks();
+    var data = buffer;
+    if (copyBuffer) {
+      var length = buffer.length;
+      data = new ByteArray(length);
+      data.setRange(0, length, buffer, 0);
     }
-    return result;
+    if (_file == null) {
+      _pendingOperations.add(data);
+    } else {
+      _write(data, 0, data.length);
+    }
+    return false;

[Søren Gjesse, 2012/02/28 07:40:25]

We should probably revisit whether write and writeFrom should have boolean
return type.

[Mads Ager (google), 2012/02/28 08:21:00]

Agreed, we probably should.

   }
 
   bool writeFrom(List<int> buffer, [int offset = 0, int len]) {
-    bool result = _write(
-        buffer, offset, (len == null) ? buffer.length - offset : len);
-    if (result) {
-      _checkScheduleCallbacks();
+    // A copy is required by the interface.
+    var length = buffer.length - offset;
+    if (len != null) {
+      if (len > length) throw new IndexOutOfRangeException(len);
+      length = len;
     }
-    return result;
+    var copy = new ByteArray(length);
+    copy.setRange(0, length, buffer, offset);
+    return write(copy);
   }
 
   void close() {
-    if (_scheduledNoPendingWriteCallback != null) {
-      _scheduledNoPendingWriteCallback.cancel();
-    }
-    if (!_streamMarkedClosed) {
-      _file.closeSync();
-      _streamMarkedClosed = true;
-      _checkScheduleCallbacks();
+    if (_file == null) {
+      _pendingOperations.add(null);
+    } else {
+      if (!_streamMarkedClosed) {
+        _file.close();
+        _streamMarkedClosed = true;
+      }
     }
   }
 
   void set noPendingWriteHandler(void callback()) {
     _noPendingWriteHandler = callback;
-    _checkScheduleCallbacks();
   }
 
   void set closeHandler(void callback()) {
     _closeHandler = callback;
-    _checkScheduleCallbacks();
   }
 
   void set errorHandler(void callback()) {
-    // TODO(sgjesse): How to handle this?
+    _errorHandler = callback;
   }
 
-  bool _write(List<int> buffer, int offset, int len) {
-    int bytesWritten = _file.writeListSync(buffer, offset, len);
-    if (bytesWritten == len) {
-      return true;
-    } else {
-      throw "FileOutputStream: write error";
-    }
+  void _processPendingOperations() {
+    _pendingOperations.forEach((buffer) {
+      (buffer != null) ? write(buffer) : close();
+    });
+    _pendingOperations = null;
   }
 
-  void _checkScheduleCallbacks() {
-    void issueNoPendingWriteCallback(Timer timer) {
-      _scheduledNoPendingWriteCallback = null;
-      if (_noPendingWriteHandler !== null) {
-        _noPendingWriteHandler();
-        _checkScheduleCallbacks();
-      }
-    }
-
-    void issueCloseCallback(Timer timer) {
-      if (_closeHandler !== null) _closeHandler();
-    }
-
-    // Schedule no pending write callbacks if the stream is not yet
-    // closed and close callback if it is closing.
-    if (!_closeCallbackCalled) {
-      if (_scheduledNoPendingWriteCallback == null) {
-        _scheduledNoPendingWriteCallback =
-            new Timer(issueNoPendingWriteCallback, 0);
-      }
-      if (_streamMarkedClosed && _scheduledCloseCallback == null) {
-        _scheduledCloseCallback = new Timer(issueCloseCallback, 0);
-      }
-    }
+  void _write(List<int> buffer, int offset, int len) {
+    _file.writeList(buffer, offset, len);
   }
 
   RandomAccessFile _file;
 
   // When this is set to true the stream is marked closed. When a
   // stream is marked closed no more data can be written.
   bool _streamMarkedClosed = false;
 
   // When this is set to true the close callback has been called and
   // the stream is fully closed.
   bool _closeCallbackCalled = false;
 
-  Timer _scheduledNoPendingWriteCallback;
-  Timer _scheduledCloseCallback;
+  // List of pending writes that were issued before the underlying
+  // file was successfully opened.
+  List<List<int>> _pendingOperations;
+
   Function _noPendingWriteHandler;
   Function _closeHandler;
+  Function _errorHandler;
 }
 
 
 // Helper class containing static file helper methods.
 class _FileUtils {
   static final kExistsRequest = 0;
   static final kCreateRequest = 1;
   static final kDeleteRequest = 2;
   static final kOpenRequest = 3;
   static final kFullPathRequest = 4;
@@ skipping 288 matching lines @@
       throw new FileIOException(
           "Mixed use of synchronous and asynchronous API");
     }
     String result = _FileUtils.checkedFullPath(_name);
     if (result == null) {
       throw new FileIOException("fullPath failed");
     }
     return result;
   }
 
   void openInputStream() {

[Søren Gjesse, 2012/02/28 07:40:25]

So you kept the Sync/non Sync versions for now?

[Mads Ager (google), 2012/02/28 08:21:00]

Yes, I will do a follow-up change today to get rid of that again. :)

     _asyncUsed = true;
     // Create a new file object to handle the opening of the file for
     // creating an input stream. Currently the file input stream uses
     // synchronous calls on the opened file so we need to open it
     // synchronously.
-    File file = new File(this._name);
-    file.errorHandler = (String error) {
-      if (_errorHandler != null) _errorHandler(error);
-    };
-    RandomAccessFile openedFile = file.openSync();
-    InputStream stream =
-        new _FileInputStream(openedFile, openedFile.lengthSync());
-    new Timer(
-        (Timer ignore) {
-          if (_inputStreamHandler != null) _inputStreamHandler(stream);
-        }, 0);
+    new Timer((t) {
+      if (_inputStreamHandler != null) {
+        _inputStreamHandler(new _FileInputStream(_name));
+      }
+    }, 0);
   }
 
   InputStream openInputStreamSync() {
-    if (_asyncUsed) {
-      throw new FileIOException(
-          "Mixed use of synchronous and asynchronous API");
-    }
-    RandomAccessFile openedFile = openSync();
-    return new _FileInputStream(openedFile, openedFile.lengthSync());
+    return new _FileInputStream(_name);
   }
 
   void openOutputStream([FileMode mode = FileMode.WRITE]) {
     _asyncUsed = true;
     if (mode != FileMode.WRITE &&
         mode != FileMode.APPEND) {
       throw new FileIOException(
           "Wrong FileMode. Use FileMode.WRITE or FileMode.APPEND");
     }
-    // Create a new file object to handle the opening of the file for
-    // creating an input stream. Currently the file input stream uses
-    // synchronous calls on the opened file so we need to open it
-    // synchronously.
-    File file = new File(this._name);
-    file.errorHandler = (String error) {
-      if (_errorHandler != null) _errorHandler(error);
-    };
-    RandomAccessFile openedFile = file.openSync(mode);
-    OutputStream stream = new _FileOutputStream(openedFile);
+    OutputStream stream = new _FileOutputStream(_name, mode);
     new Timer(
         (Timer ignore) {
           if (_outputStreamHandler != null) _outputStreamHandler(stream);
         }, 0);
   }
 
   OutputStream openOutputStreamSync([FileMode mode = FileMode.WRITE]) {
     if (_asyncUsed) {
       throw new FileIOException(
           "Mixed use of synchronous and asynchronous API");
     }
     if (mode != FileMode.WRITE &&
         mode != FileMode.APPEND) {
       throw new FileIOException(
           "Wrong FileMode. Use FileMode.WRITE or FileMode.APPEND");
     }
-    RandomAccessFile openedFile = openSync(mode);
-    return new _FileOutputStream(openedFile);
+    return new _FileOutputStream(_name, mode);
   }
 
   void readAsBytes() {
     _asyncUsed = true;
     var chunks = new _BufferList();
-    openInputStream();
-    inputStreamHandler = (inputStream) {
-      inputStream.closeHandler = () {
-        if (_readAsBytesHandler != null) {
-          _readAsBytesHandler(chunks.readBytes(chunks.length));
-        }
-      };
-      inputStream.dataHandler = () {
-        var chunk = inputStream.read();
-        chunks.add(chunk);
-      };
-      inputStream.errorHandler = () {
-        if (_errorHandler != null) {
-          _errorHandler("Failed to read file as bytes: $_name");
-        }
-      };
+    var stream = openInputStreamSync();
+    stream.closeHandler = () {
+      if (_readAsBytesHandler != null) {
+        _readAsBytesHandler(chunks.readBytes(chunks.length));
+      }
+    };
+    stream.dataHandler = () {
+      var chunk = stream.read();
+      chunks.add(chunk);
+    };
+    stream.errorHandler = () {
+      if (_errorHandler != null) {
+        _errorHandler("Failed to read file as bytes: $_name");
+      }
     };
   }
 
   List<int> readAsBytesSync() {
     if (_asyncUsed) {
       throw new FileIOException(
           "Mixed use of synchronous and asynchronous API");
     }
     var opened = openSync();
     var length = opened.lengthSync();
     var result = new ByteArray(length);
     var read = opened.readListSync(result, 0, length);
     if (read != length) {
       throw new FileIOException("Failed reading file as bytes: $_name");
     }
+    opened.close();
     return result;
   }
 
   _StringDecoder _getDecoder(encoding) {
     if (encoding == "UTF-8") {
       return new _UTF8Decoder();
     } else if (encoding == "ISO-8859-1") {
       return new _Latin1Decoder();
     } else if (encoding == "ASCII") {
       return new _AsciiDecoder();
@@ skipping 526 matching lines @@
   void set readByteHandler(void handler(int byte)) {
     _readByteHandler = handler;
   }
 
   void set readListHandler(void handler(int read)) {
     _readListHandler = handler;
   }
 
   void set noPendingWriteHandler(void handler()) {
     _noPendingWriteHandler = handler;
+    if (_pendingWrites == 0) {
+      _noPendingWriteTimer = new Timer((t) {
+        if (_noPendingWriteHandler != null) _noPendingWriteHandler();
+      }, 0);
+    }
   }
 
   void set positionHandler(void handler(int pos)) {
     _positionHandler = handler;
   }
 
   void set setPositionHandler(void handler()) {
     _setPositionHandler = handler;
   }
 
   void set truncateHandler(void handler()) {
     _truncateHandler = handler;
   }
 
   void set lengthHandler(void handler(int length)) {
     _lengthHandler = handler;
   }
 
   void set flushHandler(void handler()) {
     _flushHandler = handler;
   }
 
   void _ensureFileService() {
     if (_fileService == null) {
       _fileService = _FileUtils.newServicePort();
     }
   }
 
-  void _writeEnqueued() => _pendingWrites++;
+  void _writeEnqueued() {
+    _pendingWrites++;
+    if (_noPendingWriteTimer != null) {
+      _noPendingWriteTimer.cancel();
+      _noPendingWriteTimer = null;
+    }
+  }
 
   void  _writeCompleted() {
     _pendingWrites--;
     if (_pendingWrites == 0 && _noPendingWriteHandler != null) {
       _noPendingWriteHandler();
     }
   }
 
 
   String _name;
   int _id;
   bool _asyncUsed;
   int _pendingWrites = 0;
 
   SendPort _fileService;
 
+  Timer _noPendingWriteTimer;
+
   Function _closeHandler;
   Function _readByteHandler;
   Function _readListHandler;
   Function _noPendingWriteHandler;
   Function _positionHandler;
   Function _setPositionHandler;
   Function _truncateHandler;
   Function _lengthHandler;
   Function _flushHandler;
   Function _errorHandler;
 }