Prototype of I/O statistics for Observatory

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 187 matching lines @@
   final List<InternetAddress> addresses = [];
 
   _NetworkInterface(this.name, this.index);
 
   String toString() {
     return "NetworkInterface('$name', $addresses)";
   }
 }
 
 
+class _Rate {
+  final int buckets;
+  final data;
+  int lastValue = 0;
+  int nextBucket = 0;
+
+  _Rate(int buckets) : buckets = buckets, data = new List.filled(buckets, 0);
+
+  void update(int value) {
+    data[nextBucket] = value - lastValue;
+    lastValue = value;
+    nextBucket = (nextBucket + 1) % buckets;
+  }
+
+  int get rate {
+    int sum = data.fold(0, (prev, element) => prev + element);
+    return sum ~/ buckets;
+  }
+}
+
+// Statics information for the observatory.
+class _SocketStat {
+  _Rate readRate = new _Rate(5);
+  _Rate writeRate = new _Rate(5);
+
+  void update(_NativeSocket socket) {
+    readRate.update(socket.totalRead);
+    writeRate.update(socket.totalWritten);
+  }
+}
+
+class _SocketsObservatory {
+  static int socketCount = 0;
+  static Map sockets = new Map<_NativeSocket, _SocketStat>();
+  static Timer timer;
+
+  static add(_NativeSocket socket) {
+    if (socketCount == 0) startTimer();
+    sockets[socket] = new _SocketStat();
+    socketCount++;
+  }
+
+  static remove(_NativeSicket socket) {
+    sockets.remove(socket);

[Anders Johnsen, 2014/02/19 16:14:34]

assert it is in the map.

[Søren Gjesse, 2014/03/06 11:13:43]

Done.

+    socketCount--;
+    if (socketCount == 0) stopTimer();
+  }
+
+  static update(_) {
+    sockets.forEach((socket, stat) {
+      stat.update(socket);
+    });
+  }
+
+  static startTimer() {
+    if (timer != null) return;
+    timer = new Timer.periodic(new Duration(seconds: 1), update);
+  }
+
+  static stopTimer() {
+    if (timer == null) return;
+    timer.cancel();
+    timer = null;
+  }
+
+  static String generateResponse() {
+    var response = new Map();
+    response['type'] = 'SocketList';
+    var members = new List();
+    response['members'] = members;
+    sockets.forEach((socket, stat) {
+      var kind =
+          socket.isListening ? "LISTENING" :
+          socket.isPipe ? "PIPE" :
+          socket.isInternal ? "INTERNAL" : "NORMAL";
+      var protocol =
+          socket.isTcp ? "tcp" :
+          socket.isUdp ? "udp" : "";
+      var localAddress;
+      var localPort;
+      var remoteAddress;
+      var remotePort;
+      try {
+        localAddress = socket.address.address;
+      } catch (e) {
+        localAddress = "n/a";
+      }
+      try {
+        localPort = socket.port;
+      } catch (e) {
+        localPort = "n/a";
+      }
+      try {
+        remoteAddress = socket.remoteAddress.address;
+      } catch (e) {
+        remoteAddress = "n/a";
+      }
+      try {
+        remotePort = socket.remotePort;
+      } catch (e) {
+        remotePort = "n/a";
+      }
+      members.add({'kind': kind, 'protocol': protocol,

[Cutch, 2014/02/19 16:05:13]

Need a 'type': 'Socket' in this map.

[Søren Gjesse, 2014/03/06 11:13:43]

Done.

+                   'localAddress': localAddress, 'localPort': localPort,
+                   'remoteAddress': remoteAddress, 'remotePort': remotePort,
+                   'totalRead': socket.totalRead,
+                   'totalWritten': socket.totalWritten,
+                   'readPerSec': stat.readRate.rate,
+                   'writePerSec': stat.writeRate.rate});
+    });
+    return JSON.encode(response);;
+  }
+
+  static String toJSON() {
+    try {
+      return generateResponse();
+    } catch (e, s) {
+      return '{"type":"Error","text":"$e","stacktrace":"$s"}';
+    }
+  }
+}
+
+
 // The _NativeSocket class encapsulates an OS socket.
 class _NativeSocket extends NativeFieldWrapperClass1 {
   // Bit flags used when communicating between the eventhandler and
   // dart code. The EVENT flags are used to indicate events of
   // interest when sending a message from dart code to the
   // eventhandler. When receiving a message from the eventhandler the
   // EVENT flags indicate the events that actually happened. The
   // COMMAND flags are used to send commands from dart to the
   // eventhandler. COMMAND flags are never received from the
   // eventhandler. Additional flags are used to communicate other
@@ skipping 13 matching lines @@
   static const int FIRST_COMMAND = CLOSE_COMMAND;
   static const int LAST_COMMAND = SHUTDOWN_WRITE_COMMAND;
 
   // Type flag send to the eventhandler providing additional
   // information on the type of the file descriptor.
   static const int LISTENING_SOCKET = 16;
   static const int PIPE_SOCKET = 17;
   static const int TYPE_NORMAL_SOCKET = 0;
   static const int TYPE_LISTENING_SOCKET = 1 << LISTENING_SOCKET;
   static const int TYPE_PIPE = 1 << PIPE_SOCKET;
+  static const int TYPE_TYPE_MASK = TYPE_LISTENING_SOCKET | PIPE_SOCKET;
+
+  // Protocol flags.
+  static const int TCP_SOCKET = 18;
+  static const int UDP_SOCKET = 19;
+  static const int INTERNAL_SOCKET = 20;
+  static const int TYPE_TCP_SOCKET = 1 << TCP_SOCKET;
+  static const int TYPE_UDP_SOCKET = 1 << UDP_SOCKET;
+  static const int TYPE_INTERNAL_SOCKET = 1 << INTERNAL_SOCKET;
+  static const int TYPE_PROTOCOL_MASK =
+      TYPE_TCP_SOCKET | TYPE_UDP_SOCKET | TYPE_INTERNAL_SOCKET;
+
 
   // Native port messages.
   static const HOST_NAME_LOOKUP = 0;
   static const LIST_INTERFACES = 1;
   static const REVERSE_LOOKUP = 2;
 
   // Protocol flags.
   static const int PROTOCOL_IPV4 = 1 << 0;
   static const int PROTOCOL_IPV6 = 1 << 1;
 
@@ skipping 19 matching lines @@
 
   int available = 0;
 
   bool sendReadEvents = false;
   bool readEventIssued = false;
 
   bool sendWriteEvents = false;
   bool writeEventIssued = false;
   bool writeAvailable = false;
 
+  // Statistics.
+  int totalRead = 0;
+  int totalWritten = 0;
+
   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;
+  _NativeSocket.datagram(this.address)
+      : typeFlags = TYPE_NORMAL_SOCKET | TYPE_UDP_SOCKET {
+    _SocketsObservatory.add(this);

[Anders Johnsen, 2014/02/19 16:14:34]

Move to setListening, where eventPort == null.

[Søren Gjesse, 2014/03/06 11:13:43]

Done.

+  }
 
-  _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET;
+  _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET | TYPE_TCP_SOCKET {
+    _SocketsObservatory.add(this);
+  }
 
-  _NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET;
+  _NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET | TYPE_TCP_SOCKET {
+    _SocketsObservatory.add(this);
+  }
 
-  _NativeSocket.pipe() : typeFlags = TYPE_PIPE;
+  _NativeSocket.pipe() : typeFlags = TYPE_PIPE {
+    _SocketsObservatory.add(this);
+  }
 
-  _NativeSocket.watch(int id) : typeFlags = TYPE_NORMAL_SOCKET {
+  _NativeSocket.watch(int id)
+      : typeFlags = TYPE_NORMAL_SOCKET | TYPE_INTERNAL_SOCKET {
     isClosedWrite = true;
     nativeSetSocketId(id);
+    _SocketsObservatory.add(this);
   }
 
+  bool get isListening => (typeFlags & TYPE_LISTENING_SOCKET) != 0;
+  bool get isPipe => (typeFlags & TYPE_PIPE) != 0;
+  bool get isInternal => (typeFlags & TYPE_INTERNAL_SOCKET) != 0;
+  bool get isTcp => (typeFlags & TYPE_TCP_SOCKET) != 0;
+  bool get isUdp => (typeFlags & TYPE_UDP_SOCKET) != 0;
+
   List<int> read(int len) {
     if (len != null && len <= 0) {
       throw new ArgumentError("Illegal length $len");
     }
     if (isClosing || isClosed) return null;
     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;
+    totalRead += result.length;

[Anders Johnsen, 2014/02/19 16:14:34]

put in '!= null'.

[Søren Gjesse, 2014/03/06 11:13:43]

Done.

     return result;
   }
 
   Datagram receive() {
     if (isClosing || isClosed) return null;
     var result = nativeRecvFrom();
     if (result is OSError) {
       reportError(result, "Receive failed");
       return null;
     }
@@ skipping 37 matching lines @@
       result = 0;
     }
     // The result may be negative, if we forced a short write for testing
     // purpose. In such case, don't mark writeAvailable as false, as we don't
     // know if we'll receive an event. It's better to just retry.
     if (result >= 0 && result < bytes) {
       writeAvailable = false;
     }
     // Negate the result, as stated above.
     if (result < 0) result = -result;
+    totalWritten += result;
     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(
         bufferAndStart.buffer, bufferAndStart.start, bytes,
         address._in_addr, port);
     if (result is OSError) {
       scheduleMicrotask(() => reportError(result, "Send failed"));
       result = 0;
     }
     return result;
   }
 
   _NativeSocket accept() {
     // Don't issue accept if we're closing.
     if (isClosing || isClosed) return null;
     var socket = new _NativeSocket.normal();
-    if (nativeAccept(socket) != true) return null;
+    if (nativeAccept(socket) != true) {
+      _SocketsObservatory.remove(socket);

[Anders Johnsen, 2014/02/19 16:14:34]

This can be removed with above change.

[Søren Gjesse, 2014/03/06 11:13:43]

Done.

+      return null;
+    }
     socket.localPort = localPort;
     socket.address = address;
+    totalRead += 1;
     return socket;
   }
 
   int get port {
     if (localPort != 0) return localPort;
     return localPort = nativeGetPort();
   }
 
   int get remotePort {
     return nativeGetRemotePeer()[1];
@@ skipping 51 matching lines @@
     }
   }
 
   // Multiplexes socket events to the socket handlers.
   void multiplex(int events) {
     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 &&
+            !isListening  &&
             !isClosing &&
             !isClosed) {
           isClosedRead = true;
           issueReadEvent();
           continue;
         }
 
         if (i == WRITE_EVENT) {
           writeAvailable = true;
           issueWriteEvent(delayed: false);
           continue;
         }
 
-        if (i == READ_EVENT &&
-            typeFlags != TYPE_LISTENING_SOCKET) {
+        if (i == READ_EVENT && !isListening) {
           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();
           continue;
         }
 
         var handler = eventHandlers[i];
         if (i == DESTROYED_EVENT) {
           assert(!isClosed);
           isClosed = true;
           closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
+          _SocketsObservatory.remove(this);
           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.
@@ skipping 10 matching lines @@
     eventHandlers[CLOSED_EVENT] = closed;
     eventHandlers[DESTROYED_EVENT] = destroyed;
   }
 
   void setListening({read: true, write: true}) {
     sendReadEvents = read;
     sendWriteEvents = write;
     if (read) issueReadEvent();
     if (write) issueWriteEvent();
     if (eventPort == null) {
-      int flags = typeFlags;
+      int flags = typeFlags & TYPE_TYPE_MASK;
       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;
@@ skipping 899 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);
 }
+
+String _socketsStats() => _SocketsObservatory.toJSON();