Only use one timer to handle HttpServer::idleTimeout.

sdk/lib/io/http_impl.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.
 
 part of dart.io;
 
 const int _HEADERS_BUFFER_SIZE = 8 * 1024;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
@@ skipping 1884 matching lines @@
   static const _CLOSING = 2;
   static const _DETACHED = 3;
 
   int _state = _IDLE;
 
   final Socket _socket;
   final _HttpServer _httpServer;
   final _HttpParser _httpParser;
   StreamSubscription _subscription;
   Timer _idleTimer;
   final Uint8List _headersBuffer = new Uint8List(_HEADERS_BUFFER_SIZE);

[Søren Gjesse, 2014/02/24 14:14:35]

Why the empty lines?

[Anders Johnsen, 2014/02/24 14:35:10]

Done.

 
+
+  bool _idleMark = false;
+
   Future _streamFuture;
 
   _HttpConnection(this._socket, this._httpServer)
       : _httpParser = new _HttpParser.requestParser() {
-    _startTimeout();
     _httpParser.listenToStream(_socket);
     _subscription = _httpParser.listen(
         (incoming) {
-          _stopTimeout();
+          _httpServer._markActive(this);
           // If the incoming was closed, close the connection.
           incoming.dataDone.then((closing) {
             if (closing) destroy();
           });
           // Only handle one incoming request at the time. Keep the
           // stream paused until the request has been send.
           _subscription.pause();
           _state = _ACTIVE;
           var outgoing = new _HttpOutgoing(_socket);
           var response = new _HttpResponse(incoming.uri,
                                            incoming.headers.protocolVersion,
                                            outgoing,
                                            _httpServer.serverHeader);
           var request = new _HttpRequest(response, incoming, _httpServer, this);
           _streamFuture = outgoing.done
               .then((_) {
                 response.deadline = null;
                 if (_state == _DETACHED) return;
                 if (response.persistentConnection &&
                     request.persistentConnection &&
                     incoming.fullBodyRead &&
                     !_httpParser.upgrade &&
                     !_httpServer.closed) {
                   _state = _IDLE;
-                  _startTimeout();
+                  _idleMark = false;
+                  _httpServer._markIdle(this);
                   // Resume the subscription for incoming requests as the
                   // request is now processed.
                   _subscription.resume();
                 } else {
                   // Close socket, keep-alive not used or body sent before
                   // received data was handled.
                   destroy();
                 }
               }, onError: (_) {
                 destroy();
               });
           response._ignoreBody = request.method == "HEAD";
           response._httpRequest = request;
           _httpServer._handleRequest(request);
         },
         onDone: () {
           destroy();
         },
         onError: (error) {
           // Ignore failed requests that was closed before headers was received.
           destroy();
         });
   }
 
-  void _startTimeout() {
-    assert(_state == _IDLE);
-    _stopTimeout();
-    if (_httpServer.idleTimeout == null) return;
-    _idleTimer = new Timer(_httpServer.idleTimeout, () {
-      destroy();
-    });
+  void markIdle() {
+    _idleMark = true;
   }
 
-  void _stopTimeout() {
-    if (_idleTimer != null) _idleTimer.cancel();
-  }
+  bool get isMarkedIdle => _idleMark;
 
   void destroy() {
-    _stopTimeout();
     if (_state == _CLOSING || _state == _DETACHED) return;
     _state = _CLOSING;
     _socket.destroy();
     _httpServer._connectionClosed(this);
   }
 
   Future<Socket> detachSocket() {
-    _stopTimeout();
     _state = _DETACHED;
     // Remove connection from server.
     _httpServer._connectionClosed(this);
 
     _HttpDetachedIncoming detachedIncoming = _httpParser.detachIncoming();
 
     return _streamFuture.then((_) {
       return new _DetachedSocket(_socket, detachedIncoming);
     });
   }
 
   HttpConnectionInfo get connectionInfo => _HttpConnectionInfo.create(_socket);
 
   bool get _isActive => _state == _ACTIVE;
   bool get _isIdle => _state == _IDLE;
   bool get _isClosing => _state == _CLOSING;
   bool get _isDetached => _state == _DETACHED;
 }
 
 
 // HTTP server waiting for socket connections.
 class _HttpServer extends Stream<HttpRequest> implements HttpServer {
   String serverHeader;
 
-  Duration idleTimeout = const Duration(seconds: 120);
+  Duration _idleTimeout;
+  Timer _idleTimer;
 
   static Future<HttpServer> bind(address, int port, int backlog) {
     return ServerSocket.bind(address, port, backlog: backlog).then((socket) {
       return new _HttpServer._(socket, true);
     });
   }
 
   static Future<HttpServer> bindSecure(address,
                                        int port,
                                        int backlog,
                                        String certificate_name,
                                        bool requestClientCertificate) {
     return SecureServerSocket.bind(
         address,
         port,
         certificate_name,
         backlog: backlog,
         requestClientCertificate: requestClientCertificate)
         .then((socket) {
           return new _HttpServer._(socket, true);
         });
   }
 
   _HttpServer._(this._serverSocket, this._closeServer) {
     _controller = new StreamController<HttpRequest>(sync: true,
                                                     onCancel: close);
+    idleTimeout = const Duration(seconds: 120);
   }
 
   _HttpServer.listenOn(this._serverSocket) : _closeServer = false {
     _controller = new StreamController<HttpRequest>(sync: true,
                                                     onCancel: close);
+    idleTimeout = const Duration(seconds: 120);
+  }
+
+  Duration get idleTimeout => _idleTimeout;
+
+  void set idleTimeout(Duration duration) {

[Søren Gjesse, 2014/02/24 14:14:35]

Update the dartdoc comment to say that it might take up to 2xidleTimeout for the
timeout to actually happen.

[Anders Johnsen, 2014/02/24 14:35:10]

Done.

+    if (_idleTimer != null) {
+      _idleTimer.cancel();
+      _idleTimer = null;
+    }
+    _idleTimeout = duration;
+    if (_idleTimeout != null) {
+      _idleTimer = new Timer.periodic(_idleTimeout, (_) {
+        for (var idle in _idleConnections.toList()) {
+          if (idle.isMarkedIdle) {
+            idle.destroy();
+          } else {
+            idle.markIdle();
+          }
+        }
+      });
+    }
   }
 
   StreamSubscription<HttpRequest> listen(void onData(HttpRequest event),
                                          {Function onError,
                                          void onDone(),
                                          bool cancelOnError}) {
     _serverSocket.listen(
         (Socket socket) {
           socket.setOption(SocketOption.TCP_NODELAY, true);
           // Accept the client connection.
           _HttpConnection connection = new _HttpConnection(socket, this);
-          _connections.add(connection);
+          _idleConnections.add(connection);
         },
         onError: (error) {
           // Ignore HandshakeExceptions as they are bound to a single request,
           // and are not fatal for the server.
           if (error is! HandshakeException) {
             _controller.addError(error);
           }
         },
         onDone: _controller.close);
     return _controller.stream.listen(onData,
                                      onError: onError,
                                      onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
   Future close({bool force: false}) {
     closed = true;
     Future result;
     if (_serverSocket != null && _closeServer) {
       result = _serverSocket.close();
     } else {
       result = new Future.value();
     }
+    idleTimeout = null;
     if (force) {
-      for (var c in _connections.toList()) {
+      for (var c in _activeConnections.toList()) {
         c.destroy();
       }
-      assert(_connections.isEmpty);
-    } else {
-      for (var c in _connections.where((c) => c._isIdle).toList()) {
-        c.destroy();
-      }
+      assert(_activeConnections.isEmpty);
+    }
+    for (var c in _idleConnections.toList()) {
+      c.destroy();
     }
     _maybeCloseSessionManager();
     return result;
   }
 
   void _maybeCloseSessionManager() {
     if (closed &&
-        _connections.isEmpty &&
+        _idleConnections.isEmpty &&
+        _activeConnections.isEmpty &&
         _sessionManagerInstance != null) {
       _sessionManagerInstance.close();
       _sessionManagerInstance = null;
     }
   }
 
   int get port {
     if (closed) throw new HttpException("HttpServer is not bound to a socket");
     return _serverSocket.port;
   }
 
   InternetAddress get address {
     if (closed) throw new HttpException("HttpServer is not bound to a socket");
     return _serverSocket.address;
   }
 
   set sessionTimeout(int timeout) {
     _sessionManager.sessionTimeout = timeout;
   }
 
   void _handleRequest(HttpRequest request) => _controller.add(request);
 
   void _handleError(error) {
     if (!closed) _controller.addError(error);
   }
 
   void _connectionClosed(_HttpConnection connection) {
-    _connections.remove(connection);
+    // Remove itself from either idle or active connections.
+    connection.unlink();
     _maybeCloseSessionManager();
   }
 
+  void _markIdle(_HttpConnection connection) {
+    _activeConnections.remove(connection);
+    _idleConnections.add(connection);
+  }
+
+  void _markActive(_HttpConnection connection) {
+    _idleConnections.remove(connection);
+    _activeConnections.add(connection);
+  }
+
   _HttpSessionManager get _sessionManager {
     // Lazy init.
     if (_sessionManagerInstance == null) {
       _sessionManagerInstance = new _HttpSessionManager();
     }
     return _sessionManagerInstance;
   }
 
   HttpConnectionsInfo connectionsInfo() {
     HttpConnectionsInfo result = new HttpConnectionsInfo();
-    result.total = _connections.length;
-    _connections.forEach((_HttpConnection conn) {
+    result.total = _activeConnections.length + _idleConnections.length;
+    _activeConnections.forEach((_HttpConnection conn) {
       if (conn._isActive) {
         result.active++;
-      } else if (conn._isIdle) {
-        result.idle++;
       } else {
         assert(conn._isClosing);
         result.closing++;
       }
     });
+    _idleConnections.forEach((_HttpConnection conn) {
+      result.idle++;
+      assert(conn._isIdle);
+    });
     return result;
   }
 
   _HttpSessionManager _sessionManagerInstance;
 
   // Indicated if the http server has been closed.
   bool closed = false;
 
   // The server listen socket. Untyped as it can be both ServerSocket and
   // SecureServerSocket.
   final _serverSocket;
   final bool _closeServer;
 
   // Set of currently connected clients.
-  final LinkedList<_HttpConnection> _connections
+  final LinkedList<_HttpConnection> _activeConnections
+      = new LinkedList<_HttpConnection>();
+  final LinkedList<_HttpConnection> _idleConnections
       = new LinkedList<_HttpConnection>();
   StreamController<HttpRequest> _controller;
-  // TODO(ajohnsen): Use close queue?
 }
 
 
 class _ProxyConfiguration {
   static const String PROXY_PREFIX = "PROXY ";
   static const String DIRECT_PREFIX = "DIRECT";
 
   _ProxyConfiguration(String configuration) : proxies = new List<_Proxy>() {
     if (configuration == null) {
       throw new HttpException("Invalid proxy configuration $configuration");
@@ skipping 392 matching lines @@
   const _RedirectInfo(this.statusCode, this.method, this.location);
 }
 
 String _getHttpVersion() {
   var version = Platform.version;
   // Only include major and minor version numbers.
   int index = version.indexOf('.', version.indexOf('.') + 1);
   version = version.substring(0, index);
   return 'Dart/$version (dart:io)';
 }