Extract meta data about open sockets into separate class.

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,
                                                    bool shared: false}) {
@@ skipping 262 matching lines @@
   static const int PROTOCOL_IPV4 = 1 << 0;
   static const int PROTOCOL_IPV6 = 1 << 1;
 
   static const int NORMAL_TOKEN_BATCH_SIZE = 8;
   static const int LISTENING_TOKEN_BATCH_SIZE = 2;
 
   static const Duration _RETRY_DURATION = const Duration(milliseconds: 250);
   static const Duration _RETRY_DURATION_LOOPBACK =
       const Duration(milliseconds: 25);
 
-  // Use default Map so we keep order.
-  static Map<int, _NativeSocket> _sockets = new Map<int, _NativeSocket>();
-
   // Socket close state
   bool isClosed = false;
   bool isClosing = false;
   bool isClosedRead = false;
   bool closedReadEventSent = false;
   bool isClosedWrite = false;
   Completer closeCompleter = new Completer.sync();
 
   // Handlers and receive port for socket events from the event handler.
   final List eventHandlers = new List(EVENT_COUNT + 1);
@@ skipping 14 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;
+
+  static bool connectedResourceHandler = false;
+  _ReadWriteResourceInfo resourceInfo;
 
   // 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])
@@ skipping 97 matching lines @@
               connectNext();
             } else {
               // Query the local port, for error messages.
               socket.port;
               // Set up timer for when we should retry the next address
               // (if any).
               var duration = address.isLoopback ?
                   _RETRY_DURATION_LOOPBACK :
                   _RETRY_DURATION;
               var timer = new Timer(duration, connectNext);
+              setupResourceInfo(socket);

[Søren Gjesse, 2015/09/02 09:07:49]

Shouldn't this move down to just after timer.cancel() below?

What is happening with connectNext is that if a destination have more than one
IP address as the result of a DNS lookup each of these are tired with a timeout,
and the first that actually reports as being connected will close any other open
ones in the "connecting" Map.

[ricow1, 2015/09/02 10:45:42]

If I do that, and there is an error, I do a socket.close(), which will trigger
the resource info to be removed (but we can't remove it, because it is not
there)

+
               connecting[socket] = timer;
               // Setup handlers for receiving the first write event which
               // indicate that the socket is fully connected.
               socket.setHandlers(
                   write: () {
                     timer.cancel();
                     socket.setListening(read: false, write: false);
                     completer.complete(socket);
                     connecting.remove(socket);
                     connecting.forEach((s, t) {
@@ skipping 31 matching lines @@
               .then((list) {
                 if (list.length == 0) {
                   throw createError(response, "Failed host lookup: '$host'");
                 }
                 return list[0];
               });
         })
         .then((address) {
           var socket = new _NativeSocket.listen();
           socket.localAddress = address;
-
           var result = socket.nativeCreateBindListen(address._in_addr,
                                                      port,
                                                      backlog,
                                                      v6Only,
                                                      shared);
           if (result is OSError) {
             throw new SocketException("Failed to create server socket",
                                       osError: result,
                                       address: address,
                                       port: port);
           }
           if (port != 0) socket.localPort = port;
+          setupResourceInfo(socket);
           socket.connectToEventHandler();
           return socket;
         });
   }
 
+  static void setupResourceInfo(_NativeSocket socket) {
+    socket.resourceInfo = new _SocketResourceInfo(socket);
+  }
+
   static Future<_NativeSocket> bindDatagram(
       host, int port, bool reuseAddress) {
     return new Future.value(host)
         .then((host) {
           if (host is _InternetAddress) return host;
           return lookup(host)
               .then((list) {
                 if (list.length == 0) {
                   throw createError(response, "Failed host lookup: '$host'");
                 }
                 return list[0];
               });
         })
         .then((address) {
           var socket = new _NativeSocket.datagram(address);
           var result = socket.nativeCreateBindDatagram(
               address._in_addr, port, reuseAddress);
           if (result is OSError) {
             throw new SocketException("Failed to create datagram socket",
                                       osError: result,
                                       address: address,
                                       port: port);
           }
           if (port != 0) socket.localPort = port;
+          setupResourceInfo(socket);
           return socket;
         });
   }
 
   _NativeSocket.datagram(this.localAddress)
     : typeFlags = TYPE_NORMAL_SOCKET | TYPE_UDP_SOCKET;
 
   _NativeSocket.normal() : typeFlags = TYPE_NORMAL_SOCKET | TYPE_TCP_SOCKET;
 
   _NativeSocket.listen() : typeFlags = TYPE_LISTENING_SOCKET | TYPE_TCP_SOCKET {
@@ skipping 21 matching lines @@
     if (isClosing || isClosed) return null;
     len = min(available, len == null ? available : len);
     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;
+      if (!isInternal && !isPipe) {

[ricow1, 2015/09/01 12:06:24]

These "if (!isInternal && !isPipe) {" will go away when I add support for pipes
and internal uses. The fields used is already in the superclass

[Søren Gjesse, 2015/09/02 09:07:49]

Please add a TODO then.

Maybe check if resourceInfo is non-null instead (in the other places as well).

[ricow1, 2015/09/02 10:45:42]

So the reason I did not want to do that is that I want to actually catch cases
where we did not setup resouceInfo, but where not internal or pipe (i.e., we
could get into a situation where we had an open socket without a resourceInfo
object, which we then would not see in the observatory). I can change this to:
if (resourceInfo != null) {
  resourceInfo.readCount++;
  resourceInfo.lastRead = timestamp;
} else {
  assert(isInternal or isPipe);
}
but that is more verbose. I have no strong feelings towards one or the other,
but would like to at least keep the coverage, I will let you decide. Added TODO
here and the other places.

[Søren Gjesse, 2015/09/02 12:24:49]

I like the more verbose version with the assert.

[ricow1, 2015/09/02 12:45:50]

Done (instead I put the assert before)

+        resourceInfo.totalRead += result.length;
+      }
     }
-    readCount++;
-    lastRead = timestamp;
+    if (!isInternal && !isPipe) {
+      resourceInfo.readCount++;
+      resourceInfo.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) {
       // Read the next available. Available is only for the next datagram, not
       // the sum of all datagrams pending, so we need to call after each
       // receive. If available becomes > 0, the _NativeSocket will continue to
       // emit read events.
       available = nativeAvailable();
-      totalRead += result.data.length;
+      if (!isInternal && !isPipe) {
+        resourceInfo.totalRead += result.data.length;
+      }
     }
-    readCount++;
-    lastRead = timestamp;
+    if (!isInternal && !isPipe) {
+      resourceInfo.readCount++;
+      resourceInfo.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 18 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;
-    writeCount++;
-    lastWrite = timestamp;
+    if (!isInternal && !isPipe) {
+      resourceInfo.totalWritten += result;
+      resourceInfo.writeCount++;
+      resourceInfo.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;
+    if (!isInternal && !isPipe) {
+      resourceInfo.totalWritten += result;
+      resourceInfo.writeCount++;
+      resourceInfo.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.localAddress = address;
-    totalRead += 1;
-    lastRead = timestamp;
+    setupResourceInfo(socket);
+    if (!isInternal && !isPipe) {
+      resourceInfo.totalRead += 1;
+      resourceInfo.lastRead = timestamp;
+    }
     return socket;
   }
 
   int get port {
     if (localPort != 0) return localPort;
     if (isClosing || isClosed) throw const SocketException.closed();
     var result = nativeGetPort();
     if (result is OSError) throw result;
     return localPort = result;
   }
@@ skipping 92 matching lines @@
             available = nativeAvailable();
             issueReadEvent();
             continue;
           }
         }
 
         var handler = eventHandlers[i];
         if (i == DESTROYED_EVENT) {
           assert(isClosing);
           assert(!isClosed);
+          if (!isPipe && !isInternal) {
+            _SocketResourceInfo.SocketClosed(resourceInfo);
+          }
           isClosed = true;
           closeCompleter.complete();
           disconnectFromEventHandler();
           if (handler != null) handler();
           continue;
         }
 
         if (i == ERROR_EVENT) {
           if (!isClosing) {
             reportError(nativeGetError(), "");
@@ skipping 97 matching lines @@
     int fullData = (typeFlags & TYPE_TYPE_MASK) | data;
     assert(!isClosing);
     connectToEventHandler();
     _EventHandler._sendData(this, eventPort.sendPort, fullData);
   }
 
   void connectToEventHandler() {
     assert(!isClosed);
     if (eventPort == null) {
       eventPort = new RawReceivePort(multiplex);
-      _sockets[_serviceId] = this;
+    }
+    if (!connectedResourceHandler) {

[ricow1, 2015/09/01 12:06:25]

I am not sure this is the correct location for this, if you have a better place
let me know.

[Søren Gjesse, 2015/09/02 12:24:49]

I think this is fine for now.

+      registerExtension('__getOpenSockets',
+                        _SocketResourceInfo.getOpenSockets);
+      registerExtension('__getSocketByID',
+                        _SocketResourceInfo.getSocketInfoMapByID);
+
+      connectedResourceHandler = true;
     }
   }
 
   void disconnectFromEventHandler() {
     assert(eventPort != null);
     eventPort.close();
     eventPort = null;
-    _sockets.remove(_serviceId);
     // Now that we don't track this Socket anymore, we can clear the owner
     // field.
     owner = 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;
   }
 
@@ skipping 84 matching lines @@
   void leaveMulticast(InternetAddress addr, NetworkInterface interface) {
     var interfaceAddr = multicastAddress(addr, interface);
     var interfaceIndex = interface == null ? 0 : interface.index;
     var result = nativeLeaveMulticast(
         addr._in_addr,
         interfaceAddr == null ? null : interfaceAddr._in_addr,
         interfaceIndex);
     if (result is OSError) throw result;
   }
 
-  String get _serviceTypePath => 'io/sockets';
-  String get _serviceTypeName => 'Socket';
-
-  String _JSONKind() {
-    return isListening ? "Listening" :
-           isPipe ? "Pipe" :
-           isInternal ? "Internal" : "Normal";
-  }
-
-  Map _toJSONPipe(bool ref) {
-    var name = 'Anonymous Pipe';
-    var r = {
-      'id': _servicePath,
-      'type': _serviceType(ref),
-      'name': name,
-      'user_name': name,
-      'kind': _JSONKind(),
-    };
-    if (ref) {
-      return r;
-    }
-    r['readClosed'] = isClosedRead;
-    r['writeClosed'] = isClosedWrite;
-    r['closing'] = isClosing;
-    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),
-      'name': name,
-      'user_name': name,
-      'kind': _JSONKind(),
-    };
-    if (ref) {
-      return r;
-    }
-    var protocol = isTcp ? "TCP" : isUdp ? "UDP" : null;
-    var localAddress;
-    var localPort;
-    var rAddress;
-    var rPort;
-    try {
-      localAddress = address.address;
-    } catch (e) { }
-    try {
-      localPort = port;
-    } catch (e) { }
-    try {
-      rAddress = this.remoteAddress.address;
-    } catch (e) { }
-    try {
-      rPort = remotePort;
-    } catch (e) { }
-    r['localAddress'] = localAddress;
-    r['localPort'] = localPort;
-    r['remoteAddress'] = rAddress;
-    r['remotePort'] = rPort;
-    r['protocol'] = protocol;
-    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 _toJSON(bool ref) {
-    var map;
-    if (isPipe) {
-      map =  _toJSONPipe(ref);
-    } else if (isInternal) {
-      map = _toJSONInternal(ref);
-    } else {
-      map = _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)
       native "Socket_SendTo";
@@ skipping 132 matching lines @@
       _referencePort.listen((sendPort) {
         sendPort.send(
           [_socket.address,
            _socket.port,
            _v6Only]);
       });
     }
     return new _RawServerSocketReference(_referencePort.sendPort);
   }
 
-  Map _toJSON(bool ref) => _socket._toJSON(ref);
-
   void set _owner(owner) { _socket.owner = owner; }
 }
 
 
 class _RawServerSocketReference implements RawServerSocketReference {
   final SendPort _sendPort;
 
   _RawServerSocketReference(this._sendPort);
 
   Future<RawServerSocket> create() {
@@ skipping 157 matching lines @@
   }
 
   void _onSubscriptionStateChange() {
     if (_controller.hasListener) {
       _resume();
     } else {
       _socket.close();
     }
   }
 
-  Map _toJSON(bool ref) => _socket._toJSON(ref);
   void set _owner(owner) { _socket.owner = owner; }
 }
 
 
 patch class ServerSocket {
   /* patch */ static Future<ServerSocket> bind(address,
                                                int port,
                                                {int backlog: 0,
                                                 bool v6Only: false,
                                                 bool shared: false}) {
@@ skipping 42 matching lines @@
   int get port => _socket.port;
 
   InternetAddress get address => _socket.address;
 
   Future close() => _socket.close().then((_) => this);
 
   ServerSocketReference get reference {
     return new _ServerSocketReference(_socket.reference);
   }
 
-  Map _toJSON(bool ref) => _socket._toJSON(ref);
-
   void set _owner(owner) { _socket._owner = owner; }
 }
 
 
 patch class Socket {
   /* patch */ static Future<Socket> connect(host, int port, {sourceAddress}) {
     return RawSocket.connect(host, port, sourceAddress: sourceAddress).then(
         (socket) => new _Socket(socket));
   }
 }
@@ skipping 293 matching lines @@
     if (_detachReady != null) {
       _detachReady.complete(null);
     } else {
       if (_raw != null) {
         _raw.shutdown(SocketDirection.SEND);
         _disableWriteEvent();
       }
     }
   }
 
-  Map _toJSON(bool ref) => _raw._toJSON(ref);
   void set _owner(owner) { _raw._owner = owner; }
 }
 
 
 patch class RawDatagramSocket {
   /* patch */ static Future<RawDatagramSocket> bind(
       host, int port, {bool reuseAddress: true}) {
     return _RawDatagramSocket.bind(host, port, reuseAddress);
   }
 }
@@ skipping 135 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();