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)";
   }
 }
 
 
+// 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
   // 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 47 matching lines @@
 
   // 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;
 
+  // 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])
         .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 {
     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() {
     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);
     if (result is OSError) {
       reportError(result, "Read failed");
       return null;
     }
+    bytesRead += result.length;
     return result;
   }
 
   Datagram receive() {
     if (isClosing || isClosed) return null;
     var result = nativeRecvFrom();
     if (result is OSError) {
       reportError(result, "Receive failed");
       return null;
     }
@@ skipping 20 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;
     }
+    bytesWritten += 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(
@@ skipping 44 matching lines @@
           isClosedRead = true;
         }
 
         var handler = eventHandlers[i];
         if (i == DESTROYED_EVENT) {
           assert(!isClosed);
           isClosed = true;
           closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
+          _SocketsObservatory.remove(this);
           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.
@@ skipping 275 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));
+        if (socket != null) {
+          _controller.add(new _RawSocket(socket));
+        }
       }),
       error: zone.bindUnaryCallback((e) {
         _controller.addError(e);
         _controller.close();
       }),
       destroyed: _controller.close
     );
   }
 
   StreamSubscription<RawSocket> listen(void onData(RawSocket event),
@@ skipping 668 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 _test() => _SocketsObservatory.toJSON();