Prototype of I/O statistics for Observatory

runtime/bin/socket_patch.dart

Index: runtime/bin/socket_patch.dart
diff --git a/runtime/bin/socket_patch.dart b/runtime/bin/socket_patch.dart
index 56f04d16ac83f5e20daea12c422a37d3ca1feb92..95666a52bf39619cf33ff1599234f5e137571f90 100644
--- a/runtime/bin/socket_patch.dart
+++ b/runtime/bin/socket_patch.dart
@@ -205,6 +205,64 @@ class _NetworkInterface implements NetworkInterface {
 }
 
 
+// Statics information for the observatory.
+class _SocketsObservatory {
+  static int socketCount = 0;
+  static Set sockets = new Set();
+
+  static add(_NativeSocket socket) {
+    sockets.add(socket);
+    socketCount++;
+  }
+
+  static remove(_NativeSicket socket) {
+    sockets.remove(socket);
+    socketCount--;
+  }
+
+  static String toJSON() {
+    var sb = new StringBuffer();
+    sb.write('{"type":"io","sockets":[');
+    sockets.forEach((s) {

[Cutch, 2014/02/13 22:42:09]

You probably should use the dart:convert library instead of serializing the JSON
yourself.

[Søren Gjesse, 2014/02/14 12:10:13]

Absolutely - this was just a first hack.

+      var type = s.typeFlags == _NativeSocket.TYPE_LISTENING_SOCKET ? "LISTENING" : "NORMAL";
+      var localAddress;
+      var localPort;
+      var remoteAddress;
+      var remotePort;
+      try {
+        localAddress = s.address.address;
+      } catch (e) {
+        localAddress = "UNKNOWN";
+      }
+      try {
+        localPort = s.port;
+      } catch (e) {
+        localPort = "UNKNOWN";
+      }
+      try {
+        remoteAddress = s.remoteAddress.address;
+      } catch (e) {
+        remoteAddress = "UNKNOWN";
+      }
+      try {
+        remotePort = s.remotePort;
+      } catch (e) {
+        remotePort = "UNKNOWN";
+      }
+      sb.write('{"type","$type"},');
+      sb.write('{"localHost","$localAddress"},');
+      sb.write('{"localPort","$localPort"},');
+      sb.write('{"remoteAddress","$remoteAddress"},');
+      sb.write('{"remotePort","$remotePort"},');
+      sb.write('{"bytesRead","${s.bytesRead}"},');
+      sb.write('{"bytesWritten","${s.bytesWritten}"},');

[Cutch, 2014/02/13 22:42:09]

What about rates over the last second, 5 seconds, minute, ?

[Søren Gjesse, 2014/02/14 12:10:13]

Added rate calculation using a timer. We might have to find a better way, and we
should only enable it when the VM service is used.

+    });
+    sb.write(']}');
+    return sb.toString();
+  }
+}
+
+
 // The _NativeSocket class encapsulates an OS socket.
 class _NativeSocket extends NativeFieldWrapperClass1 {
   // Bit flags used when communicating between the eventhandler and
@@ -272,6 +330,10 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
   // Holds the address used to connect or bind the socket.
   InternetAddress address;
 
+  // Statistics.
+  int bytesRead = 0;
+  int bytesWritten = 0;
+
   static Future<List<InternetAddress>> lookup(
       String host, {InternetAddressType type: InternetAddressType.ANY}) {
     return _IOService.dispatch(_SOCKET_LOOKUP, [host, type._value])
@@ -426,27 +488,31 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
         });
   }
 
-  _NativeSocket.datagram(this.address)
-      : typeFlags = TYPE_NORMAL_SOCKET {
+  _NativeSocket.datagram(this.address) : typeFlags = TYPE_NORMAL_SOCKET {
     eventHandlers = new List(EVENT_COUNT + 1);
+    _SocketsObservatory.add(this);
   }
 
   _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET {
     eventHandlers = new List(EVENT_COUNT + 1);
+    _SocketsObservatory.add(this);
   }
 
   _NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET {
     eventHandlers = new List(EVENT_COUNT + 1);
+    _SocketsObservatory.add(this);
   }
 
   _NativeSocket.pipe() : typeFlags = TYPE_PIPE {
     eventHandlers = new List(EVENT_COUNT + 1);
+    _SocketsObservatory.add(this);
   }
 
   _NativeSocket.watch(int id) : typeFlags = TYPE_NORMAL_SOCKET {
     eventHandlers = new List(EVENT_COUNT + 1);
     isClosedWrite = true;
     nativeSetSocketId(id);
+    _SocketsObservatory.add(this);
   }
 
   int available() {
@@ -470,6 +536,7 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
       reportError(result, "Read failed");
       return null;
     }
+    bytesRead += result.length;
     return result;
   }
 
@@ -510,6 +577,7 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
       scheduleMicrotask(() => reportError(result, "Write failed"));
       result = 0;
     }
+    bytesWritten += result;
     return result;
   }
 
@@ -574,6 +642,7 @@ class _NativeSocket extends NativeFieldWrapperClass1 {
           closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
+          _SocketsObservatory.remove(this);
           continue;
         }
         assert(handler != null);
@@ -869,7 +938,9 @@ class _RawServerSocket extends Stream<RawSocket>
     _socket.setHandlers(
       read: zone.bindCallback(() {
         var socket = _socket.accept();
-        if (socket != null) _controller.add(new _RawSocket(socket));
+        if (socket != null) {
+          _controller.add(new _RawSocket(socket));
+        }
       }),
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
@@ -1558,3 +1629,5 @@ Datagram _makeDatagram(List<int> data,
       new _InternetAddress(address, null, in_addr),
       port);
 }
+
+String _test() => _SocketsObservatory.toJSON();