Make event-handlers edge-triggered and move socket-state to Dart.

runtime/bin/socket_patch.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.
 
 patch class RawServerSocket {
   /* patch */ static Future<RawServerSocket> bind(address,
                                                   int port,
                                                   {int backlog: 0,
                                                    bool v6Only: false}) {
     return _RawServerSocket.bind(address, port, backlog, v6Only);
@@ skipping 238 matching lines @@
   static const int PROTOCOL_IPV6 = 1 << 1;
 
   // Socket close state
   bool isClosed = false;
   bool isClosing = false;
   bool isClosedRead = false;
   bool isClosedWrite = false;
   Completer closeCompleter = new Completer.sync();
 
   // Handlers and receive port for socket events from the event handler.
-  int eventMask = 0;
-  List eventHandlers;
+  final List eventHandlers = new List(EVENT_COUNT + 1);
   RawReceivePort eventPort;
 
-  // Indicates if native interrupts can be activated.
-  bool canActivateEvents = true;
-
   // The type flags for this socket.
   final int typeFlags;
 
   // Holds the port of the socket, 0 if not known.
   int localPort = 0;
 
   // Holds the address used to connect or bind the socket.
   InternetAddress address;
 
+  int available = 0;
+
+  bool sendReadEvents = false;
+  bool readEventIssued = false;
+
+  bool sendWriteEvents = false;
+  bool writeEventIssued = false;
+  bool writeAvailable = false;
+
   static Future<List<InternetAddress>> lookup(
       String host, {InternetAddressType type: InternetAddressType.ANY}) {
     return _IOService.dispatch(_SOCKET_LOOKUP, [host, type._value])
         .then((response) {
           if (isErrorResponse(response)) {
             throw createError(response, "Failed host lookup: '$host'");
           } else {
             return response.skip(1).map((result) {
               var type = new InternetAddressType._from(result[0]);
               return new _InternetAddress(result[1], host, result[2]);
@@ skipping 134 matching lines @@
             throw new SocketException("Failed to create datagram socket",
                                       osError: result,
                                       address: address,
                                       port: port);
           }
           if (port != 0) socket.localPort = port;
           return socket;
         });
   }
 
-  _NativeSocket.datagram(this.address)
-      : typeFlags = TYPE_NORMAL_SOCKET {
-    eventHandlers = new List(EVENT_COUNT + 1);
-  }
+  _NativeSocket.datagram(this.address) : typeFlags = TYPE_NORMAL_SOCKET;
 
-  _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET {
-    eventHandlers = new List(EVENT_COUNT + 1);
-  }
+  _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET;
 
-  _NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET {
-    eventHandlers = new List(EVENT_COUNT + 1);
-  }
+  _NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET;
 
-  _NativeSocket.pipe() : typeFlags = TYPE_PIPE {
-    eventHandlers = new List(EVENT_COUNT + 1);
-  }
+  _NativeSocket.pipe() : typeFlags = TYPE_PIPE;
 
   _NativeSocket.watch(int id) : typeFlags = TYPE_NORMAL_SOCKET {
-    eventHandlers = new List(EVENT_COUNT + 1);
     isClosedWrite = true;
     nativeSetSocketId(id);
   }
 
-  int available() {
-    if (isClosing || isClosed) return 0;
-    var result = nativeAvailable();
-    if (result is OSError) {
-      reportError(result, "Available failed");
-      return 0;
-    } else {
-      return result;
-    }
-  }
-
   List<int> read(int len) {
     if (len != null && len <= 0) {
       throw new ArgumentError("Illegal length $len");
     }
     if (isClosing || isClosed) return null;
-    var result = nativeRead(len == null ? -1 : len);
+    var result = nativeRead(min(available, len == null ? available : len));
     if (result is OSError) {
       reportError(result, "Read failed");
       return null;
     }
+    if (result != null) available -= result.length;
     return result;
   }
 
   Datagram receive() {
     if (isClosing || isClosed) return null;
     var result = nativeRecvFrom();
     if (result is OSError) {
       reportError(result, "Receive failed");
       return null;
     }
+    if (result != null) {
+      if (Platform.isMacOS) {
+        // Mac includes the header size, so we need to query the actual
+        // available.
+        available = nativeAvailable();
+      } else {
+        available -= result.data.length;
+      }
+    }
     return result;
   }
 
   int write(List<int> buffer, int offset, int bytes) {
     if (buffer is! List) throw new ArgumentError();
     if (offset == null) offset = 0;
     if (bytes == null) {
       if (offset > buffer.length) {
         throw new RangeError.value(offset);
       }
@@ skipping 10 matching lines @@
     if (isClosing || isClosed) return 0;
     if (bytes == 0) return 0;
     _BufferAndStart bufferAndStart =
         _ensureFastAndSerializableByteData(buffer, offset, offset + bytes);
     var result =
         nativeWrite(bufferAndStart.buffer, bufferAndStart.start, bytes);
     if (result is OSError) {
       scheduleMicrotask(() => reportError(result, "Write failed"));
       result = 0;
     }
+    if (result >= 0 && result < bytes) {
+      writeAvailable = false;
+    }
+    if (result < 0) result = -result;

[Søren Gjesse, 2014/02/18 14:28:00]

If the negative result is needed please explain it in a comment.

[Anders Johnsen, 2014/02/19 09:43:03]

Done.

     return result;
   }
 
   int send(List<int> buffer, int offset, int bytes,
            InternetAddress address, int port) {
     if (isClosing || isClosed) return 0;
     _BufferAndStart bufferAndStart =
         _ensureFastAndSerializableByteData(
             buffer, offset, bytes);
     var result = nativeSendTo(
@@ skipping 24 matching lines @@
   int get remotePort {
     return nativeGetRemotePeer()[1];
   }
 
   InternetAddress get remoteAddress {
     var result = nativeGetRemotePeer()[0];
     var type = new InternetAddressType._from(result[0]);
     return new _InternetAddress(result[1], null, result[2]);
   }
 
+  void issueReadEvent() {
+    if (readEventIssued) return;
+    readEventIssued = true;
+    void issue() {
+      readEventIssued = false;
+      if (isClosing) return;
+      if (!sendReadEvents) return;
+      if (available == 0) {
+        if (isClosedRead) {
+          if (isClosedWrite) close();
+          var handler = eventHandlers[CLOSED_EVENT];
+          if (handler == null) return;
+          handler();
+        }
+        return;
+      }
+      var handler = eventHandlers[READ_EVENT];
+      if (handler == null) return;
+      readEventIssued = true;
+      handler();
+      scheduleMicrotask(issue);
+    }
+    scheduleMicrotask(issue);
+  }
+
+  void issueWriteEvent({bool delayed: true}) {
+    if (writeEventIssued) return;
+    if (!writeAvailable) return;
+    void issue() {
+      writeEventIssued = false;
+      if (!writeAvailable) return;
+      if (isClosing) return;
+      if (!sendWriteEvents) return;
+      sendWriteEvents = false;
+      var handler = eventHandlers[WRITE_EVENT];
+      if (handler == null) return;
+      handler();
+    }
+    if (delayed) {
+      writeEventIssued = true;
+      scheduleMicrotask(issue);
+    } else {
+      issue();
+    }
+  }
+
   // Multiplexes socket events to the socket handlers.
   void multiplex(int events) {
-    canActivateEvents = false;
     for (int i = FIRST_EVENT; i <= LAST_EVENT; i++) {
       if (((events & (1 << i)) != 0)) {
         if ((i == CLOSED_EVENT || i == READ_EVENT) && isClosedRead) continue;
+        if (isClosing && i != DESTROYED_EVENT) continue;
         if (i == CLOSED_EVENT &&
             typeFlags != TYPE_LISTENING_SOCKET &&
             !isClosing &&
             !isClosed) {
           isClosedRead = true;
+          issueReadEvent();
+          continue;
+        }
+
+        if (i == WRITE_EVENT) {
+          writeAvailable = true;
+          issueWriteEvent(delayed: false);
+          continue;
+        }
+
+        if (i == READ_EVENT &&
+            typeFlags != TYPE_LISTENING_SOCKET) {
+          var avail = nativeAvailable();
+          if (avail is int) {
+            available = avail;
+          } else {
+            // Available failed. Mark socket as having data, to ensure read
+            // events, and thus reporting of this error.
+            available = 1;
+          }
+          issueReadEvent();

[Søren Gjesse, 2014/02/18 14:28:00]

Why is there no "delayed: false" for this as well?

[Anders Johnsen, 2014/02/19 09:43:03]

It's always delayed. If it wasn't I was having trouble with order of read vs
close events.

+          continue;
         }
 
         var handler = eventHandlers[i];
         if (i == DESTROYED_EVENT) {
           assert(!isClosed);
           isClosed = true;
           closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
           continue;
         }
-        assert(handler != null);
-        if (i == WRITE_EVENT) {
-          // If the event was disabled before we had a chance to fire the event,
-          // discard it. If we register again, we'll get a new one.
-          if ((eventMask & (1 << i)) == 0) continue;
-          // Unregister the out handler before executing it. There is
-          // no need to notify the eventhandler as handlers are
-          // disabled while the event is handled.
-          eventMask &= ~(1 << i);
-        }
 
-        // Don't call the in handler if there is no data available
-        // after all.
-        if (i == READ_EVENT &&
-            typeFlags != TYPE_LISTENING_SOCKET &&
-            available() == 0) {
-          continue;
-        }
         if (i == ERROR_EVENT) {
           if (!isClosing) {
             reportError(nativeGetError(), "");
           }
         } else if (!isClosed) {
           // If the connection is closed right after it's accepted, there's a
           // chance the close-handler is not set.
           if (handler != null) handler();
         }
       }
     }
-    if (isClosedRead && isClosedWrite) close();
-    canActivateEvents = true;
-    activateHandlers();
   }
 
   void setHandlers({read, write, error, closed, destroyed}) {
     eventHandlers[READ_EVENT] = read;
     eventHandlers[WRITE_EVENT] = write;
     eventHandlers[ERROR_EVENT] = error;
     eventHandlers[CLOSED_EVENT] = closed;
     eventHandlers[DESTROYED_EVENT] = destroyed;
   }
 
   void setListening({read: true, write: true}) {
-    eventMask = (1 << CLOSED_EVENT) | (1 << ERROR_EVENT);
-    if (read) eventMask |= (1 << READ_EVENT);
-    if (write) eventMask |= (1 << WRITE_EVENT);
-    activateHandlers();
-  }
-
-
-  void activateHandlers() {
-    if (canActivateEvents && !isClosing && !isClosed) {
-      if ((eventMask & ((1 << READ_EVENT) | (1 << WRITE_EVENT))) == 0) {
-        // If we don't listen for either read or write, disconnect as we won't
-        // get close and error events anyway.
-        if (eventPort != null) disconnectFromEventHandler();
-      } else {
-        int data = eventMask;
-        if (isClosedRead) data &= ~(1 << READ_EVENT);
-        if (isClosedWrite) data &= ~(1 << WRITE_EVENT);
-        data |= typeFlags;
-        sendToEventHandler(data);
-      }
+    sendReadEvents = read;
+    sendWriteEvents = write;
+    if (read) issueReadEvent();
+    if (write) issueWriteEvent();
+    if (eventPort == null) {
+      int flags = typeFlags;
+      if (!isClosedRead) flags |= 1 << READ_EVENT;
+      if (!isClosedWrite) flags |= 1 << WRITE_EVENT;
+      sendToEventHandler(flags);
     }
   }
 
   Future close() {
     if (!isClosing && !isClosed) {
       sendToEventHandler(1 << CLOSE_COMMAND);
       isClosing = true;
     }
     return closeCompleter.future;
   }
@@ skipping 14 matching lines @@
           throw new ArgumentError(direction);
       }
     }
   }
 
   void shutdownWrite() {
     if (!isClosing && !isClosed) {
       if (isClosedRead) {
         close();
       } else {
-        bool connected = eventPort != null;
         sendToEventHandler(1 << SHUTDOWN_WRITE_COMMAND);
-        if (!connected) disconnectFromEventHandler();
       }
       isClosedWrite = true;
     }
   }
 
   void shutdownRead() {
     if (!isClosing && !isClosed) {
       if (isClosedWrite) {
         close();
       } else {
-        bool connected = eventPort != null;
         sendToEventHandler(1 << SHUTDOWN_READ_COMMAND);
-        if (!connected) disconnectFromEventHandler();
       }
       isClosedRead = true;
     }
   }
 
   void sendToEventHandler(int data) {
+    assert(!isClosed);
     connectToEventHandler();
-    assert(!isClosed);
     _EventHandler._sendData(this, eventPort, data);
   }
 
   void connectToEventHandler() {
     if (eventPort == null) {
       eventPort = new RawReceivePort(multiplex);
     }
   }
 
   void disconnectFromEventHandler() {
-    if (eventPort != null) {
-      eventPort.close();
-      eventPort = null;
-    }
+    assert(eventPort != null);
+    eventPort.close();
+    eventPort = null;
   }
 
   // Check whether this is an error response from a native port call.
   static bool isErrorResponse(response) {
     return response is List && response[0] != _SUCCESS_RESPONSE;
   }
 
   // Create the appropriate error/exception from different returned
   // error objects.
   static createError(error,
@@ skipping 137 matching lines @@
 
   _RawServerSocket(this._socket) {
     var zone = Zone.current;
     _controller = new StreamController(sync: true,
         onListen: _onSubscriptionStateChange,
         onCancel: _onSubscriptionStateChange,
         onPause: _onPauseStateChange,
         onResume: _onPauseStateChange);
     _socket.setHandlers(
       read: zone.bindCallback(() {
-        var socket = _socket.accept();
-        if (socket != null) _controller.add(new _RawSocket(socket));
+        do {
+          var socket = _socket.accept();
+          if (socket == null) return;
+          _controller.add(new _RawSocket(socket));
+        } while (!_controller.isPaused);
       }),
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
         _controller.close();
       }),
       destroyed: _controller.close
     );
   }
 
   StreamSubscription<RawSocket> listen(void onData(RawSocket event),
@@ skipping 102 matching lines @@
                                             {Function onError,
                                              void onDone(),
                                              bool cancelOnError}) {
     return _controller.stream.listen(
         onData,
         onError: onError,
         onDone: onDone,
         cancelOnError: cancelOnError);
   }
 
-  int available() => _socket.available();
+  int available() => _socket.available;
 
   List<int> read([int len]) {
     if (_isMacOSTerminalInput) {
       var available = this.available();
       if (available == 0) return null;
       var data = _socket.read(len);
       if (data == null || data.length < available) {
         // Reading less than available from a Mac OS terminal indicate Ctrl-D.
         // This is interpreted as read closed.
         scheduleMicrotask(() => _controller.add(RawSocketEvent.READ_CLOSED));
@@ skipping 544 matching lines @@
 
 Datagram _makeDatagram(List<int> data,
                        String address,
                        List<int> in_addr,
                        int port) {
   return new Datagram(
       data,
       new _InternetAddress(address, null, in_addr),
       port);
 }