Add read/write stats to Socket and RandomAccessFile.

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 189 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(_NativeSocket socket) {
-    _SocketStat stats = sockets.remove(socket);
-    assert(stats != null);
-    socketCount--;
-    if (socketCount == 0) stopTimer();
-  }
-
-  static update(_) {
-    sockets.forEach((socket, stat) {
-      stat.update(socket);
-    });
-  }
-
-  static startTimer() {
-    if (timer != null) return;
-    // TODO(sgjesse): Enable the rate timer.
-    // 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({'type': 'Socket', 'kind': kind, 'protocol': protocol,
-                   '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 NativeFieldWrapperClass1 can not be used with a mixin, due to missing
 // implicit constructor.
 class _NativeSocketNativeWrapper extends NativeFieldWrapperClass1 {}
 
 
 // The _NativeSocket class encapsulates an OS socket.
 class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
   // 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
@@ skipping 80 matching lines @@
 
   int tokens = 0;
 
   bool sendReadEvents = false;
   bool readEventIssued = false;
 
   bool sendWriteEvents = false;
   bool writeEventIssued = false;
   bool writeAvailable = false;
 
+  static final Stopwatch sw = new Stopwatch()..start();
   // Statistics.
   int totalRead = 0;
   int totalWritten = 0;
+  int readCount = 0;
+  int writeCount = 0;
+  double lastRead;
+  double lastWrite;
 
   // The owner object is the object that the Socket is being used by, e.g.
   // a HttpServer, a WebSocket connection, a process pipe, etc.
   Object owner;
 
+  static double get timestamp => sw.elapsedMicroseconds / 1000000.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 184 matching lines @@
     if (len == 0) return null;
     var result = nativeRead(len);
     if (result is OSError) {
       reportError(result, "Read failed");
       return null;
     }
     if (result != null) {
       available -= result.length;
       totalRead += result.length;
     }
+    readCount++;
+    lastRead = timestamp;
     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;
       }
+      totalRead += result.data.length;
     }
+    readCount++;
+    lastRead = timestamp;
     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 19 matching lines @@
     }
     // 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;
+    writeCount++;
+    lastWrite = timestamp;
     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;
     }
+    totalWritten += result;
+    writeCount++;
+    lastWrite = timestamp;
     return result;
   }
 
   _NativeSocket accept() {
     // Don't issue accept if we're closing.
     if (isClosing || isClosed) return null;
     assert(available > 0);
     available--;
     tokens++;
     returnTokens(LISTENING_TOKEN_BATCH_SIZE);
     var socket = new _NativeSocket.normal();
     if (nativeAccept(socket) != true) return null;
     socket.localPort = localPort;
     socket.address = address;
     totalRead += 1;
+    lastRead = timestamp;
     return socket;
   }
 
   int get port {
     if (localPort != 0) return localPort;
     var result = nativeGetPort();
     if (result is OSError) throw result;
     return localPort = result;
   }
 
@@ skipping 339 matching lines @@
       'name': name,
       'user_name': name,
       'kind': _JSONKind(),
     };
     if (ref) {
       return r;
     }
     r['readClosed'] = isClosedRead;
     r['writeClosed'] = isClosedWrite;
     r['closing'] = isClosing;
-    r['listening'] = isListening;
     r['fd'] = nativeGetSocketId();
     if (owner != null) {
       r['owner'] = owner._toJSON(true);
+    }
+    return r;
+  }
+
+  Map _toJSONInternal(bool ref) {
+    var name = 'Internal';
+    var r = {
+      'id': _servicePath,
+      'type': _serviceType(ref),
+      'name': name,
+      'user_name': name,
+      'kind': _JSONKind(),
+    };
+    if (ref) {
+      return r;
+    }
+    r['closing'] = isClosing;
+    r['fd'] = nativeGetSocketId();
+    if (owner != null) {
+      r['owner'] = owner._toJSON(true);
     }
     return r;
   }
 
   Map _toJSONNetwork(bool ref) {
     var name = '${address.host}:$port';
     if (isTcp && !isListening) name += " <-> ${remoteAddress.host}:$remotePort";
     var r = {
       'id': _servicePath,
       'type': _serviceType(ref),
@@ skipping 31 matching lines @@
     r['closing'] = isClosing;
     r['listening'] = isListening;
     r['fd'] = nativeGetSocketId();
     if (owner != null) {
       r['owner'] = owner._toJSON(true);
     }
     return r;
   }
 
   Map _toJSON(bool ref) {
+    var map;
     if (isPipe) {
-      return _toJSONPipe(ref);
+      map =  _toJSONPipe(ref);
+    } else if (isInternal) {
+      map = _toJSONInternal(ref);
+    } else {
+      map = _toJSONNetwork(ref);
     }
-    return _toJSONNetwork(ref);
+    if (!ref) {
+      map['available'] = available;
+      map['totalRead'] = totalRead;
+      map['totalWritten'] = totalWritten;
+      map['readCount'] = totalWritten;
+      map['writeCount'] = writeCount;
+      map['lastRead'] = lastRead;
+      map['lastWrite'] = lastWrite;
+    }
+    return map;
   }
 
   void nativeSetSocketId(int id) native "Socket_SetSocketId";
   nativeAvailable() native "Socket_Available";
   nativeRead(int len) native "Socket_Read";
   nativeRecvFrom() native "Socket_RecvFrom";
   nativeWrite(List<int> buffer, int offset, int bytes)
       native "Socket_WriteList";
   nativeSendTo(List<int> buffer, int offset, int bytes,
                List<int> address, int port)
@@ skipping 808 matching lines @@
                        String address,
                        List<int> in_addr,
                        int port) {
   return new Datagram(
       data,
       new _InternetAddress(address, null, in_addr),
       port);
 }
 
 String _socketsStats() => _SocketsObservatory.toJSON();