dart:io | Enable multithreaded secure networking encryption.

sdk/lib/io/secure_socket.dart

 // Copyright (c) 2013, 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.
 
 part of dart.io;
 
 /**
  * A high-level class for communicating securely over a TCP socket, using
  * TLS and SSL. The [SecureSocket] exposes both a [Stream] and an
  * [IOSink] interface, making it ideal for using together with
@@ skipping 85 matching lines @@
    * future completes the [SecureSocket] has completed the TLS
    * handshake. Using this function requires that the other end of the
    * connection is going to start the TLS handshake.
    *
    * If the [socket] already has a subscription, this subscription
    * will no longer receive and events. In most cases calling
    * [:pause:] on this subscription before starting TLS handshake is
    * the right thing to do.
    *
    * If some of the data of the TLS handshake has already been read
-   * from the socket this data can be passed in the [carryOverData]
+   * from the socket this data can be passed in the [bufferedData]
    * parameter. This data will be processed before any other data
    * available on the socket.
    *
    * See [SecureServerSocket.bind] for more information on the
    * arguments.
    *
    */
   static Future<SecureSocket> secureServer(
       Socket socket,
       String certificateName,
-      {List<int> carryOverData,
+      {List<int> bufferedData,
        bool requestClientCertificate: false,
        bool requireClientCertificate: false}) {
     var completer = new Completer();
     (socket as dynamic)._detachRaw()
         .then((detachedRaw) {
           return RawSecureSocket.secureServer(
             detachedRaw[0],
             certificateName,
             subscription: detachedRaw[1],
-            carryOverData: carryOverData,
+            bufferedData: bufferedData,
             requestClientCertificate: requestClientCertificate,
             requireClientCertificate: requireClientCertificate);
           })
         .then((raw) {
           completer.complete(new SecureSocket._(raw));
         });
     return completer.future;
   }
 
   /**
@@ skipping 139 matching lines @@
    * future completes the [RawSecureSocket] has completed the TLS
    * handshake. Using this function requires that the other end of the
    * connection is going to start the TLS handshake.
    *
    * If the [socket] already has a subscription, pass the existing
    * subscription in the [subscription] parameter. The secure socket
    * will take over the subscription and process any subsequent
    * events.
    *
    * If some of the data of the TLS handshake has already been read
-   * from the socket this data can be passed in the [carryOverData]
+   * from the socket this data can be passed in the [bufferedData]
    * parameter. This data will be processed before any other data
    * available on the socket.
    *
    * See [RawSecureServerSocket.bind] for more information on the
    * arguments.
    *
    */
   static Future<RawSecureSocket> secureServer(
       RawSocket socket,
       String certificateName,
       {StreamSubscription subscription,
-       List<int> carryOverData,
+       List<int> bufferedData,

[Anders Johnsen, 2013/06/13 13:14:17]

Can you factor out this change?

        bool requestClientCertificate: false,
        bool requireClientCertificate: false}) {
     socket.readEventsEnabled = false;
     socket.writeEventsEnabled = false;
     return _RawSecureSocket.connect(
         socket.address,
         socket.remotePort,
         certificateName,
         is_server: true,
         socket: socket,
         subscription: subscription,
-        carryOverData: carryOverData,
+        bufferedData: bufferedData,
         requestClientCertificate: requestClientCertificate,
         requireClientCertificate: requireClientCertificate);
   }
 
   /**
    * Get the peer certificate for a connected RawSecureSocket.  If this
    * RawSecureSocket is the server end of a secure socket connection,
    * [peerCertificate] will return the client certificate, or null, if no
    * client certificate was received.  If it is the client end,
    * [peerCertificate] will return the server's certificate.
@@ skipping 14 matching lines @@
   final String subject;
   final String issuer;
   final DateTime startValidity;
   final DateTime endValidity;
 }
 
 
 class _RawSecureSocket extends Stream<RawSocketEvent>
                        implements RawSecureSocket {
   // Status states
-  static final int NOT_CONNECTED = 200;
   static final int HANDSHAKE = 201;
   static final int CONNECTED = 202;
   static final int CLOSED = 203;
 
   // Buffer identifiers.
   // These must agree with those in the native C++ implementation.
   static final int READ_PLAINTEXT = 0;
   static final int WRITE_PLAINTEXT = 1;
   static final int READ_ENCRYPTED = 2;
   static final int WRITE_ENCRYPTED = 3;
   static final int NUM_BUFFERS = 4;
 
+  // Is a buffer identifier for an encrypted buffer?
+  static bool isEncrypted(int identifier) => identifier >= READ_ENCRYPTED;
+
   RawSocket _socket;
   final Completer<_RawSecureSocket> _handshakeComplete =
       new Completer<_RawSecureSocket>();
   StreamController<RawSocketEvent> _controller;
   Stream<RawSocketEvent> _stream;
   StreamSubscription<RawSocketEvent> _socketSubscription;
-  List<int> _carryOverData;
-  int _carryOverDataIndex = 0;
+  List<int> _bufferedData;
+  int _bufferedDataIndex = 0;
   final InternetAddress address;
   final bool is_server;
   final String certificateName;
   final bool requestClientCertificate;
   final bool requireClientCertificate;
   final bool sendClientCertificate;
   final Function onBadCertificate;
 
-  var _status = NOT_CONNECTED;
+  var _status = HANDSHAKE;
   bool _writeEventsEnabled = true;
   bool _readEventsEnabled = true;
+  int _pauseCount = 0;
+  bool _pendingReadEvent = false;
   bool _socketClosedRead = false;  // The network socket is closed for reading.
   bool _socketClosedWrite = false;  // The network socket is closed for writing.
   bool _closedRead = false;  // The secure socket has fired an onClosed event.
   bool _closedWrite = false;  // The secure socket has been closed for writing.
-  bool _filterReadEmpty = true;  // There is no buffered data to read.
-  bool _filterWriteEmpty = true;  // There is no buffered data to be written.
+  bool _readEmpty = true;  // There is no buffered data to read.
+  bool _writeEmpty = true;  // There is no buffered data to be written.
   bool _connectPending = false;
+  bool _filterPending = false;
+  bool _filterActive = false;
+
   _SecureFilter _secureFilter = new _SecureFilter();
+  int _filterPointer;
+  SendPort _filterService;
 
   static Future<_RawSecureSocket> connect(
       host,
       int requestedPort,
       String certificateName,
       {bool is_server,
        RawSocket socket,
        StreamSubscription subscription,
-       List<int> carryOverData,
+       List<int> bufferedData,
        bool requestClientCertificate: false,
        bool requireClientCertificate: false,
        bool sendClientCertificate: false,
        bool onBadCertificate(X509Certificate certificate)}) {
     var future;
     if (host is String) {
       future = InternetAddress.lookup(host).then((addrs) => addrs.first);
     } else {
       future = new Future.value(host);
     }
     return future.then((addr) {
      return new _RawSecureSocket(addr,
                                  requestedPort,
                                  certificateName,
                                  is_server,
                                  socket,
                                  subscription,
-                                 carryOverData,
+                                 bufferedData,
                                  requestClientCertificate,
                                  requireClientCertificate,
                                  sendClientCertificate,
                                  onBadCertificate)
          ._handshakeComplete.future;
     });
   }
 
   _RawSecureSocket(
       InternetAddress this.address,
       int requestedPort,
       String this.certificateName,
       bool this.is_server,
       RawSocket socket,
       StreamSubscription this._socketSubscription,
-      List<int> this._carryOverData,
+      List<int> this._bufferedData,
       bool this.requestClientCertificate,
       bool this.requireClientCertificate,
       bool this.sendClientCertificate,
       bool this.onBadCertificate(X509Certificate certificate)) {
     _controller = new StreamController<RawSocketEvent>(
         sync: true,
         onListen: _onSubscriptionStateChange,
         onPause: _onPauseStateChange,
         onResume: _onPauseStateChange,
         onCancel: _onSubscriptionStateChange);
     _stream = _controller.stream;
     // Throw an ArgumentError if any field is invalid.  After this, all
     // errors will be reported through the future or the stream.
     _verifyFields();
     _secureFilter.init();
-    if (_carryOverData != null) _readFromCarryOver();
+    _filterPointer = _secureFilter._pointer();
     _secureFilter.registerHandshakeCompleteCallback(
         _secureHandshakeCompleteHandler);
     if (onBadCertificate != null) {
       _secureFilter.registerBadCertificateCallback(onBadCertificate);
     }
     var futureSocket;
     if (socket == null) {
       futureSocket = RawSocket.connect(address, requestedPort);
     } else {
       futureSocket = new Future.value(socket);
@@ skipping 15 matching lines @@
       }
       _connectPending = true;
       _secureFilter.connect(address.host,
                             port,
                             is_server,
                             certificateName,
                             requestClientCertificate ||
                                 requireClientCertificate,
                             requireClientCertificate,
                             sendClientCertificate);
-      _status = HANDSHAKE;
       _secureHandshake();
     })
     .catchError((error) {
       _handshakeComplete.completeError(error);
       _close();
     });
   }
 
   StreamSubscription listen(void onData(RawSocketEvent data),
                             {void onError(error),
                              void onDone(),
                              bool cancelOnError}) {
-    if (_writeEventsEnabled) {
-      _writeEventsEnabled = false;
-      _controller.add(RawSocketEvent.WRITE);
-    }
+    _sendWriteEvent();
     return _stream.listen(onData,
                           onError: onError,
                           onDone: onDone,
                           cancelOnError: cancelOnError);
   }
 
   void _verifyFields() {
     assert(is_server is bool);
     assert(_socket == null || _socket is RawSocket);
     if (address is! InternetAddress) {
@@ skipping 21 matching lines @@
    }
 
   int get port => _socket.port;
 
   String get remoteHost => _socket.remoteHost;
 
   int get remotePort => _socket.remotePort;
 
   int available() {
     if (_status != CONNECTED) return 0;
-    _readEncryptedData();
     return _secureFilter.buffers[READ_PLAINTEXT].length;
   }
 
   void close() {
     shutdown(SocketDirection.BOTH);
   }
 
   void _close() {
     _closedWrite = true;
     _closedRead = true;
     if (_socket != null) {
       _socket.close();
     }
     _socketClosedWrite = true;
     _socketClosedRead = true;
-    if (_secureFilter != null) {
+    if (!_filterActive && _secureFilter != null) {
       _secureFilter.destroy();
       _secureFilter = null;
     }
     if (_socketSubscription != null) {
       _socketSubscription.cancel();
     }
     _controller.close();
     _status = CLOSED;
   }
 
   void shutdown(SocketDirection direction) {
     if (direction == SocketDirection.SEND ||
         direction == SocketDirection.BOTH) {
       _closedWrite = true;
-      _writeEncryptedData();
-      if (_filterWriteEmpty) {
+      if (_writeEmpty) {
         _socket.shutdown(SocketDirection.SEND);
         _socketClosedWrite = true;
         if (_closedRead) {
           _close();
         }
       }
     }
     if (direction == SocketDirection.RECEIVE ||
         direction == SocketDirection.BOTH) {
       _closedRead = true;
       _socketClosedRead = true;
       _socket.shutdown(SocketDirection.RECEIVE);
       if (_socketClosedWrite) {
         _close();
       }
     }
   }
 
   bool get writeEventsEnabled => _writeEventsEnabled;
 
   void set writeEventsEnabled(bool value) {
-    if (value &&
-        _controller.hasListener &&
-        _secureFilter != null &&
-        _secureFilter.buffers[WRITE_PLAINTEXT].free > 0) {
-      Timer.run(() => _controller.add(RawSocketEvent.WRITE));
-    } else {
-      _writeEventsEnabled = value;
+    _writeEventsEnabled = value;
+    if (value) {
+      Timer.run(() => _sendWriteEvent());
     }
   }
 
   bool get readEventsEnabled => _readEventsEnabled;
 
   void set readEventsEnabled(bool value) {
-    _readEventsEnabled = value;
-    if (value &&
-        ((_secureFilter != null &&
-          _secureFilter.buffers[READ_PLAINTEXT].length > 0) ||
-         _socketClosedRead)) {
-      // We might not have no underlying socket to set off read events.
-      Timer.run(_readHandler);
-    }
+      _readEventsEnabled = value;
+      _scheduleReadEvent();
   }
 
-  List<int> read([int len]) {
+  List<int> read([int length]) {
     if (_closedRead) {
       throw new SocketException("Reading from a closed socket");
     }
     if (_status != CONNECTED) {
       return null;
     }
     var buffer = _secureFilter.buffers[READ_PLAINTEXT];
-    _readEncryptedData();
-    int toRead = buffer.length;
-    if (len != null) {
-      if (len is! int || len < 0) {
+    if (length == null) {
+      length = buffer.length;
+    } else if (length is! int || length < 0) {
         throw new ArgumentError(
-            "Invalid len parameter in SecureSocket.read (len: $len)");
-      }
-      if (len < toRead) {
-        toRead = len;
-      }
+            "Invalid length parameter in SecureSocket.read (length: $length)");
+    } else {
+      length = min(length, buffer.length);
     }
-    List<int> result = (toRead == 0) ? null :
-        buffer.data.sublist(buffer.start, buffer.start + toRead);
-    buffer.advanceStart(toRead);
-
-    // Set up a read event if the filter still has data.
-    if (!_filterReadEmpty) {
-      Timer.run(_readHandler);
+    if (length == 0) return null;
+    List<int> result = new Uint8List(length);
+    int bytesRead = 0;
+    while (bytesRead < length) {  // Loop over 2 data ranges in circular buffer.
+      int toRead = buffer.linearLength;
+      result.setRange(bytesRead,
+                      bytesRead + toRead,
+                      buffer.data,
+                      buffer.start);
+      buffer.advanceStart(toRead);
+      bytesRead += toRead;
     }
-
-    if (_socketClosedRead) {  // An onClose event is pending.
-      // _closedRead is false, since we are in a read  call.
-      if (!_filterReadEmpty) {
-        // _filterReadEmpty may be out of date since read empties
-        // the plaintext buffer after calling _readEncryptedData.
-        // TODO(whesse): Fix this as part of fixing read.
-        _readEncryptedData();
-      }
-      if (_filterReadEmpty) {
-        // This can't be an else clause: the value of _filterReadEmpty changes.
-        // This must be asynchronous, because we are in a read call.
-        Timer.run(_closeHandler);
-      }
-    }
-
+    _runFilter();
     return result;
   }
 
-  // Write the data to the socket, and flush it as much as possible
-  // until it would block.  If the write would block, _writeEncryptedData sets
-  // up handlers to flush the pipeline when possible.
+  // Write the data to the socket, and schedule the filter to encrypt it.
   int write(List<int> data, [int offset, int bytes]) {
     if (_closedWrite) {
       _controller.addError(new SocketException("Writing to a closed socket"));
       return 0;
     }
     if (_status != CONNECTED) return 0;
 
     if (offset == null) offset = 0;
     if (bytes == null) bytes = data.length - offset;
 
     var buffer = _secureFilter.buffers[WRITE_PLAINTEXT];
     if (bytes > buffer.free) {
       bytes = buffer.free;
     }
-    if (bytes > 0) {
-      int startIndex = buffer.start + buffer.length;
-      buffer.data.setRange(startIndex, startIndex + bytes, data, offset);
-      buffer.length += bytes;
+    int written = 0;
+    int toWrite = min(bytes, buffer.linearFree);
+    while (toWrite > 0) {  // The buffer may contain two linear free ranges.
+      _writeEmpty = false;
+      buffer.data.setRange(buffer.end, buffer.end + toWrite, data, offset);
+      buffer.advanceEnd(toWrite);
+      offset += toWrite;
+      written += toWrite;
+      toWrite = min(bytes - written, buffer.linearFree);
     }
-    _writeEncryptedData();  // Tries to flush all pipeline stages.
-    return bytes;
+    _runFilter();
+    return written;
   }
 
   X509Certificate get peerCertificate => _secureFilter.peerCertificate;
 
   bool setOption(SocketOption option, bool enabled) {
     if (_socket == null) return false;
     return _socket.setOption(option, enabled);
   }
 
-  void _writeHandler() {
-    if (_status == CLOSED) return;
-    _writeEncryptedData();
-    if (_filterWriteEmpty && _closedWrite && !_socketClosedWrite) {
-      // Close _socket for write, by calling shutdown(), to avoid cloning the
-      // socket closing code in shutdown().
-      shutdown(SocketDirection.SEND);
-    }
-    if (_status == HANDSHAKE) {
-      try {
-        _secureHandshake();
-      } catch (e) { _reportError(e, "RawSecureSocket error"); }
-    } else if (_status == CONNECTED &&
-               _controller.hasListener &&
-               _writeEventsEnabled &&
-               _secureFilter.buffers[WRITE_PLAINTEXT].free > 0) {
-      // Reset the one-shot handler.
-      _writeEventsEnabled = false;
-      _controller.add(RawSocketEvent.WRITE);
-    }
-  }
-
   void _eventDispatcher(RawSocketEvent event) {
     if (event == RawSocketEvent.READ) {
       _readHandler();
     } else if (event == RawSocketEvent.WRITE) {
       _writeHandler();
     } else if (event == RawSocketEvent.READ_CLOSED) {
       _closeHandler();
     }
   }
 
-  void _readFromCarryOver() {
-    assert(_carryOverData != null);
-    var encrypted = _secureFilter.buffers[READ_ENCRYPTED];
-    var bytes = _carryOverData.length - _carryOverDataIndex;
-    int startIndex = encrypted.start + encrypted.length;
-    encrypted.data.setRange(startIndex,
-                            startIndex + bytes,
-                            _carryOverData,
-                            _carryOverDataIndex);
-    encrypted.length += bytes;
-    _carryOverDataIndex += bytes;
-    if (_carryOverData.length == _carryOverDataIndex) {
-      _carryOverData = null;
-    }
-  }
-
   void _readHandler() {
-    if (_status == CLOSED) {
-      return;
-    } else if (_status == HANDSHAKE) {
-      try {
-        _secureHandshake();
-        if (_status != HANDSHAKE) _readHandler();
-      } catch (e) { _reportError(e, "RawSecureSocket error"); }
-    } else {
-      if (_status != CONNECTED) {
-        // Cannot happen.
-        throw new SocketException("Internal SocketIO Error");
-      }
-      try {
-        _readEncryptedData();
-      } catch (e) { _reportError(e, "RawSecureSocket error"); }
-      if (!_filterReadEmpty) {
-        if (_readEventsEnabled) {
-          if (_secureFilter.buffers[READ_PLAINTEXT].length > 0) {
-            _controller.add(RawSocketEvent.READ);
-          }
-          if (_socketClosedRead) {
-            // Keep firing read events until we are paused or buffer is empty.
-            Timer.run(_readHandler);
-          }
-        }
-      } else if (_socketClosedRead) {
-        _closeHandler();
-      }
-    }
+    _readSocket();
+    _runFilter();
+  }
+
+  void _writeHandler() {
+    _writeSocket();
+    _runFilter();
   }
 
   void _doneHandler() {
-    if (_filterReadEmpty) {
+    if (_readEmpty) {
       _close();
     }
   }
 
   void _errorHandler(e) {
     _reportError(e, 'Error on underlying RawSocket');
   }
 
   void _reportError(e, String message) {
     // TODO(whesse): Call _reportError from all internal functions that throw.
     if (e is SocketException) {
       e = new SocketException('$message (${e.message})', e.osError);
     } else if (e is OSError) {
       e = new SocketException(message, e);
     } else {
       e = new SocketException('$message (${e.toString()})', null);
     }
     if (_connectPending) {
       _handshakeComplete.completeError(e);
     } else {
       _controller.addError(e);
     }
     _close();
   }
 
   void _closeHandler() {
     if  (_status == CONNECTED) {
       if (_closedRead) return;
       _socketClosedRead = true;
-      if (_filterReadEmpty) {
+      if (_readEmpty) {
         _closedRead = true;
         _controller.add(RawSocketEvent.READ_CLOSED);
         if (_socketClosedWrite) {
           _close();
         }
+      } else {
+        _runFilter();
       }
     } else if (_status == HANDSHAKE) {
+      _socketClosedRead = true;
+      if (_readEmpty) {
       _reportError(
           new SocketException('Connection terminated during handshake'),
-          'handshake error');
+          'RawSecureSocket error');
+      } else {
+        _secureHandshake();
+      }
     }
   }
 
   void _secureHandshake() {
-    _readEncryptedData();
-    _secureFilter.handshake();
-    _writeEncryptedData();
+    try {
+      _secureFilter.handshake();
+      _writeEmpty = false;
+      _readSocket();
+      _writeSocket();
+      _runFilter();
+    } catch (e) {
+      _reportError(e, "RawSecureSocket error");
+    }
   }
 
   void _secureHandshakeCompleteHandler() {
     _status = CONNECTED;
     if (_connectPending) {
       _connectPending = false;
-      // If we complete the future synchronously, user code will run here,
-      // and modify the state of the RawSecureSocket.  For example, it
-      // could close the socket, and set _filter to null.
+      // We don't want user code to run synchronously in this callback.
       Timer.run(() => _handshakeComplete.complete(this));
     }
   }
 
   void _onPauseStateChange() {
+    if (_controller.isPaused) {
+      _pauseCount++;
+    } else {
+      _pauseCount--;
+      if (_pauseCount == 0) {
+        _scheduleReadEvent();
+        _sendWriteEvent();  // Can send event synchronously.
+      }
+    }
+
     if (!_socketClosedRead || !_socketClosedWrite) {
       if (_controller.isPaused) {
         _socketSubscription.pause();
       } else {
         _socketSubscription.resume();
       }
     }
   }
 
   void _onSubscriptionStateChange() {
     if (_controller.hasListener) {
       // TODO(ajohnsen): Do something here?
     }
   }
 
-  void _readEncryptedData() {
-    // Read from the socket, and push it through the filter as far as
-    // possible.
+  void _readSocket() {
+    if (_status == CLOSED) return;
     var encrypted = _secureFilter.buffers[READ_ENCRYPTED];
-    var plaintext = _secureFilter.buffers[READ_PLAINTEXT];
-    bool progress = true;
-    while (progress) {
-      progress = false;
-      // Do not try to read plaintext from the filter while handshaking.
-      if ((_status == CONNECTED) && plaintext.free > 0) {
-        int bytes = _secureFilter.processBuffer(READ_PLAINTEXT);
-        if (bytes > 0) {
-          plaintext.length += bytes;
-          progress = true;
-        }
-      }
-      if (encrypted.length > 0) {
-        int bytes = _secureFilter.processBuffer(READ_ENCRYPTED);
-        if (bytes > 0) {
-          encrypted.advanceStart(bytes);
-          progress = true;
-        }
-      }
-      if (!_socketClosedRead && encrypted.free > 0) {
-        if (_carryOverData != null) {
-          _readFromCarryOver();
-          progress = true;
-        } else {
-          List<int> data = _socket.read(encrypted.free);
-          if (data != null) {
-            int bytes = data.length;
-            int startIndex = encrypted.start + encrypted.length;
-            encrypted.data.setRange(startIndex, startIndex + bytes, data);
-            encrypted.length += bytes;
-            progress = true;
+    while (true) {  // Loop over both linear segments in the buffer.
+      var encrypted_free = encrypted.linearFree;
+      if (encrypted_free == 0) break;
+      List<int> data = readSocketOrBufferedData(encrypted_free);
+      if (data == null) break;
+      int bytes = data.length;
+      encrypted.data.setRange(encrypted.end, encrypted.end + bytes, data);
+      encrypted.advanceEnd(bytes);
+      _readEmpty = false;
+    }
+  }
+
+  void _runFilter() {
+    _filterPending = true;
+    _tryFilter();
+  }
+
+  void _tryFilter() {
+    if (_status == CLOSED) return;
+    if (_filterPending && !_filterActive) {
+      _filterActive = true;
+      _filterPending = false;
+      _pushAllFilterStages().then((progress) {
+        _filterActive = false;
+        if (_status == CLOSED) {
+          _secureFilter.destroy();
+          _secureFilter = null;
+          return;
+        }
+        if (_writeEmpty && _closedWrite && !_socketClosedWrite) {
+          // Checks for and handles all cases of partially closed sockets.
+          shutdown(SocketDirection.SEND);
+          if (_status == CLOSED) return;
+        }
+        if (_readEmpty && _socketClosedRead && !_closedRead) {
+          if (_status == HANDSHAKE) {
+            _secureFilter.handshake();
+            if (_status == HANDSHAKE) {
+              _reportError(
+                  new SocketException('Connection terminated during handshake'),
+                  'RawSecureSocket error');
+            }
           }
-        }
-      }
-    }
-    // If there is any data in any stages of the filter, there should
-    // be data in the plaintext buffer after this process.
-    // TODO(whesse): Verify that this is true, and there can be no
-    // partial encrypted block stuck in the secureFilter.
-    _filterReadEmpty = (plaintext.length == 0);
-  }
-
-  void _writeEncryptedData() {
+          _closeHandler();
+        }
+        if (_status == CLOSED) return;
+        if (progress) {
+          _filterPending = true;
+          _readSocket();
+          _writeSocket();
+          if (_status == HANDSHAKE) {
+            _secureHandshake();
+          }
+        }
+        _tryFilter();
+      });
+    }
+  }
+
+  List<int> readSocketOrBufferedData(int bytes) {
+    if (_bufferedData != null) {
+      if (bytes > _bufferedData.length - _bufferedDataIndex) {
+        bytes = _bufferedData.length - _bufferedDataIndex;
+      }
+      var result = _bufferedData.sublist(_bufferedDataIndex,
+                                          _bufferedDataIndex + bytes);
+      _bufferedDataIndex += bytes;
+      if (_bufferedData.length == _bufferedDataIndex) {
+        _bufferedData = null;
+      }
+      return result;
+    } else if (!_socketClosedRead) {
+      try {
+        return _socket.read(bytes);
+      } catch (e) {
+        _reportError(e, "RawSecureSocket error reading encrypted socket");
+        return null;
+      }
+    } else {
+      return null;
+    }
+  }
+
+  void _writeSocket() {
     if (_socketClosedWrite) return;
     var encrypted = _secureFilter.buffers[WRITE_ENCRYPTED];
-    var plaintext = _secureFilter.buffers[WRITE_PLAINTEXT];
-    while (true) {
-      if (encrypted.length > 0) {
-        // Write from the filter to the socket.
-        int bytes = _socket.write(encrypted.data,
-                                  encrypted.start,
-                                  encrypted.length);
-        encrypted.advanceStart(bytes);
-        if (encrypted.length > 0) {
+    while (true) {  // Loop over both linear ranges in circular buffer.
+      // Get the length of the first data segment in circular buffer.
+      var encrypted = _secureFilter.buffers[WRITE_ENCRYPTED];
+      var encrypted_length = encrypted.linearLength;
+      if (encrypted_length == 0) break;
+      int bytes = _socket.write(encrypted.data,
+                                encrypted.start,
+                                encrypted_length);
+      encrypted.advanceStart(bytes);
+      if (bytes < encrypted_length) {
           // The socket has blocked while we have data to write.
           // We must be notified when it becomes unblocked.
           _socket.writeEventsEnabled = true;
-          _filterWriteEmpty = false;
           break;
-        }
-      } else {
-        var plaintext = _secureFilter.buffers[WRITE_PLAINTEXT];
-        if (plaintext.length > 0) {
-           int plaintext_bytes = _secureFilter.processBuffer(WRITE_PLAINTEXT);
-           plaintext.advanceStart(plaintext_bytes);
-        }
-        int bytes = _secureFilter.processBuffer(WRITE_ENCRYPTED);
-        if (bytes <= 0) {
-          // We know the WRITE_ENCRYPTED buffer is empty, and the
-          // filter wrote zero bytes to it, so the filter must be empty.
-          // Also, the WRITE_PLAINTEXT buffer must have been empty, or
-          // it would have written to the filter.
-          // TODO(whesse): Verify that the filter works this way.
-          _filterWriteEmpty = true;
-          break;
-        }
-        encrypted.length += bytes;
-      }
-    }
+      }
+    }
+  }
+
+  // If a read event should be sent, add it to the controller.
+  _scheduleReadEvent() {
+    if (!_pendingReadEvent &&
+        _readEventsEnabled &&
+        _pauseCount == 0 &&
+        _secureFilter != null &&
+        !_secureFilter.buffers[READ_PLAINTEXT].isEmpty) {
+      _pendingReadEvent = true;
+      Timer.run(_sendReadEvent);
+    }
+  }
+
+  _sendReadEvent() {
+    _pendingReadEvent = false;
+    if (_readEventsEnabled &&
+        _pauseCount == 0 &&
+        _secureFilter != null &&
+        !_secureFilter.buffers[READ_PLAINTEXT].isEmpty) {
+      _controller.add(RawSocketEvent.READ);
+      _scheduleReadEvent();
+    }
+  }
+
+  // If a write event should be sent, add it to the controller.
+  _sendWriteEvent() {
+    if (!_closedWrite &&
+        _writeEventsEnabled &&
+        _pauseCount == 0 &&
+        _secureFilter != null &&
+        _secureFilter.buffers[WRITE_PLAINTEXT].free > 0) {
+      _writeEventsEnabled = false;
+      _controller.add(RawSocketEvent.WRITE);
+    }
+  }
+
+  Future<bool> _pushAllFilterStages() {
+    if (_filterService == null) {
+      _filterService = _SecureFilter._newServicePort();
+    }
+    List args = [_filterPointer, _status != CONNECTED];
+    var bufs = _secureFilter.buffers;
+    for (var i = 0; i < NUM_BUFFERS; ++i) {
+      args.add(bufs[i].start);
+      args.add(bufs[i].end);
+    }
+
+    return _filterService.call(args).then((positions) {
+      // Compute writeEmpty as "write plaintext buffer and write encrypted
+      // buffer were empty when we started and are empty now".
+      _writeEmpty = bufs[WRITE_PLAINTEXT].isEmpty &&
+          positions[2 * WRITE_ENCRYPTED + 2] ==
+          positions[2 * WRITE_ENCRYPTED + 3];
+      // If we were in handshake when this started, _writeEmpty may be false
+      // because the handshake wrote data after we checked.
+      if (positions[1]) _writeEmpty = false;
+
+      // Compute readEmpty as "both read buffers were empty when we started
+      // and are empty now".
+      _readEmpty = bufs[READ_ENCRYPTED].isEmpty &&
+          positions[2 * READ_PLAINTEXT + 2] ==
+          positions[2 * READ_PLAINTEXT + 3];
+
+      bool progress = false;
+      ExternalBuffer buffer = bufs[WRITE_PLAINTEXT];
+      int new_start = positions[2 * WRITE_PLAINTEXT + 2];
+      if (new_start != buffer.start) {
+        progress = true;
+        // If WRITE_PLAINTEXT goes from full to not full:
+        if (buffer.free == 0) {
+          buffer.start = new_start;
+          _sendWriteEvent();
+        } else {
+          buffer.start = new_start;
+        }
+      }
+      buffer = bufs[READ_ENCRYPTED];
+      new_start = positions[2 * READ_ENCRYPTED + 2];
+      if (new_start != buffer.start) {
+        progress = true;
+        // If READ_ENCRYPTED goes from full to not full:
+        if (buffer.free == 0) {
+          buffer.start = new_start;
+          _readSocket();
+        } else {
+          buffer.start = new_start;
+        }
+      }
+      buffer = bufs[WRITE_ENCRYPTED];
+      int new_end = positions[2 * WRITE_ENCRYPTED + 3];
+      if (new_end != buffer.end) {
+        progress = true;
+        // If WRITE_ENCRYPTED goes from empty to non-empty:
+        if (buffer.length == 0) {
+          buffer.end = new_end;
+          _writeSocket();
+        } else {
+          buffer.end = new_end;
+        }
+      }
+      buffer = bufs[READ_PLAINTEXT];
+      new_end = positions[2 * READ_PLAINTEXT  + 3];
+      if (new_end != buffer.end) {
+        progress = true;
+        // If READ_PLAINTEXT goes from empty to non-empty:
+        if (buffer.length == 0) {
+          buffer.end = new_end;
+          _scheduleReadEvent();
+        } else {
+          buffer.end = new_end;
+        }
+      }
+      return progress;
+    });
   }
 }
 
 
+/**
+ * A circular buffer backed by an external byte array.  Accessed from
+ * both C++ and Dart code in an unsynchronized way, with one reading
+ * and one writing.  All updates to start and end are done by Dart code.
+ */
 class _ExternalBuffer {
-  // Performance is improved if a full buffer of plaintext fits
-  // in the encrypted buffer, when encrypted.
-  static final int SIZE = 8 * 1024;
-  static final int ENCRYPTED_SIZE = 10 * 1024;
-  _ExternalBuffer() : start = 0, length = 0;
-
-  // TODO(whesse): Consider making this a circular buffer.  Only if it helps.
-  void advanceStart(int numBytes) {
-    start += numBytes;
-    length -= numBytes;
-    if (length == 0) {
-      start = 0;
-    }
-  }
-
-  int get free => data.length - (start + length);
+  _ExternalBuffer(this.size) : start = 0, end = 0;
+
+  void advanceStart(int bytes) {
+    assert(start > end || start + bytes <= end);
+    start += bytes;
+    if (start >= size) {
+      start -= size;
+      assert(start <= end);
+      assert(start < size);
+    }
+  }
+
+  void advanceEnd(int bytes) {
+    assert(start <= end || start > end + bytes);
+    end += bytes;
+    if (end >= size) {
+      end -= size;
+      assert(end < start);
+      assert(end < size);
+    }
+  }
+
+  bool get isEmpty => end == start;
+
+  int get length {
+    if (start > end) return size + end - start;
+    return end - start;
+  }
+
+  int get linearLength {
+    if (start > end) return size - start;
+    return end - start;
+  }
+
+  int get free {
+    if (start > end) return start - end - 1;
+    return size + start - end - 1;
+  }
+
+  int get linearFree {
+    if (start > end) return start - end - 1;
+    if (start == 0) return size - end - 1;
+    return size - end;
+  }
+
 
   List data;  // This will be a ExternalByteArray, backed by C allocated data.
   int start;
-  int length;
+  int end;
+  int size;
 }
 
 
 abstract class _SecureFilter {
   external factory _SecureFilter();
 
+  external static SendPort _newServicePort();
+
   void connect(String hostName,
                int port,
                bool is_server,
                String certificateName,
                bool requestClientCertificate,
                bool requireClientCertificate,
                bool sendClientCertificate);
   void destroy();
   void handshake();
   void init();
   X509Certificate get peerCertificate;
   int processBuffer(int bufferIndex);
   void registerBadCertificateCallback(Function callback);
   void registerHandshakeCompleteCallback(Function handshakeCompleteHandler);
+  int _pointer();
 
   List<_ExternalBuffer> get buffers;
 }