Port WebSockets to Mojo IPC

content/browser/websocket/websocket_manager.cc

Index: content/browser/websocket/websocket_manager.cc
diff --git a/content/browser/websocket/websocket_manager.cc b/content/browser/websocket/websocket_manager.cc
new file mode 100644
index 0000000000000000000000000000000000000000..d33079b6b7dd878c93322fb7d4d16894b501e7bf
--- /dev/null
+++ b/content/browser/websocket/websocket_manager.cc
@@ -0,0 +1,201 @@
+// Copyright 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/websocket/websocket_manager.h"
+
+#include <stddef.h>
+
+#include <algorithm>
+#include <set>
+#include <string>
+#include <vector>
+
+#include "base/callback.h"
+#include "base/logging.h"
+#include "base/numerics/safe_conversions.h"
+#include "base/rand_util.h"
+#include "base/stl_util.h"
+#include "content/browser/blob_storage/chrome_blob_storage_context.h"
+#include "content/browser/child_process_security_policy_impl.h"
+#include "content/browser/websocket/websocket_impl.h"
+#include "content/public/browser/browser_thread.h"
+#include "net/url_request/url_request_context_getter.h"
+
+namespace content {
+
+namespace {
+
+// Max number of pending connections per WebSocketManager used for per-renderer
+// WebSocket throttling.
+const int kMaxPendingWebSocketConnections = 255;
+
+}  // namespace
+
+class WebSocketManager::IOThreadState : public WebSocketImpl::Delegate {
+ public:
+  IOThreadState(
+      int process_id,
+      StoragePartition* storage_partition,
+      const scoped_refptr<ChromeBlobStorageContext>& blob_storage_context)
+      : process_id_(process_id),
+        storage_partition_(storage_partition),
+        blob_storage_context_(blob_storage_context),
+        num_pending_connections_(0),
+        num_current_succeeded_connections_(0),
+        num_previous_succeeded_connections_(0),
+        num_current_failed_connections_(0),
+        num_previous_failed_connections_(0) {}
+
+  ~IOThreadState() override {
+    for (auto impl : impls_) {
+      impl->GoAway();
+      delete impl;
+    }
+  }
+
+  void CreateWebSocket(int render_frame_id, mojom::WebSocketRequest request) {
+    if (num_pending_connections_ >= kMaxPendingWebSocketConnections) {
+      // XXX
+#if 0
+      if (!Send(new WebSocketMsg_NotifyFailure(
+              routing_id,
+              "Error in connection establishment: "
+              "net::ERR_INSUFFICIENT_RESOURCES"))) {
+        DVLOG(1) << "Sending of message type "
+                 << "WebSocketMsg_NotifyFailure failed.";
+      }
+#endif
+      return;
+    }
+
+    // We keep all WebSocketImpls alive until we either observe a connection
+    // error (the client dropped its connection) or we need to shutdown.
+
+    impls_.insert(new WebSocketImpl(this,
+                                    std::move(request),
+                                    render_frame_id,
+                                    CalculateDelay()));
+    ++num_pending_connections_;
+
+    if (!throttling_period_timer_.IsRunning()) {
+      throttling_period_timer_.Start(
+          FROM_HERE,
+          base::TimeDelta::FromMinutes(2),
+          this,
+          &IOThreadState::ThrottlingPeriodTimerCallback);
+    }
+  }
+
+ private:
+  // Calculate delay as described in the per-renderer WebSocket throttling
+  // design doc:
+  // https://docs.google.com/document/d/1aw2oN5PKfk-1gLnBrlv1OwLA8K3-ykM2ckwX2lubTg4/edit?usp=sharing
+  base::TimeDelta CalculateDelay() const {
+    int64_t f = num_previous_failed_connections_ +
+                num_current_failed_connections_;
+    int64_t s = num_previous_succeeded_connections_ +
+                num_current_succeeded_connections_;
+    int p = num_pending_connections_;
+    return base::TimeDelta::FromMilliseconds(
+        base::RandInt(1000, 5000) *
+        (1 << std::min(p + f / (s + 1), INT64_C(16))) / 65536);
+  }
+
+  void ThrottlingPeriodTimerCallback() {
+    num_previous_failed_connections_ = num_current_failed_connections_;
+    num_current_failed_connections_ = 0;
+
+    num_previous_succeeded_connections_ = num_current_succeeded_connections_;
+    num_current_succeeded_connections_ = 0;
+
+    if (num_pending_connections_ == 0 &&
+        num_previous_failed_connections_ == 0 &&
+        num_previous_succeeded_connections_ == 0) {
+      throttling_period_timer_.Stop();
+    }
+  }
+
+  // WebSocketImpl::Delegate methods:
+
+  int GetClientProcessId() override {
+    return process_id_;
+  }
+
+  StoragePartition* GetStoragePartition() override {
+    return storage_partition_;
+  }
+
+  storage::BlobStorageContext* GetBlobStorageContext() override {
+    return blob_storage_context_->context();
+  }
+
+  void OnReceivedResponseFromServer(WebSocketImpl* impl) override {
+    // XXX
+    impl->OnHandshakeSucceeded();
+    --num_pending_connections_;
+    DCHECK_GE(num_pending_connections_, 0);
+    ++num_current_succeeded_connections_;
+  }
+
+  void OnLostConnectionToClient(WebSocketImpl* impl) override {
+    if (!impl->handshake_succeeded()) {
+      // Update throttling counters (failure).
+      --num_pending_connections_;
+      DCHECK_GE(num_pending_connections_, 0);
+      ++num_current_failed_connections_;
+    }
+    impls_.erase(impl);
+    delete impl;
+  }
+
+  int process_id_;
+  StoragePartition* storage_partition_;
+  scoped_refptr<ChromeBlobStorageContext> blob_storage_context_;
+
+  std::set<WebSocketImpl*> impls_;
+
+  // Timer and counters for per-renderer WebSocket throttling.
+  base::RepeatingTimer throttling_period_timer_;
+
+  // The current number of pending connections.
+  int num_pending_connections_;
+
+  // The number of handshakes that failed in the current and previous time
+  // period.
+  int64_t num_current_succeeded_connections_;
+  int64_t num_previous_succeeded_connections_;
+
+  // The number of handshakes that succeeded in the current and previous time
+  // period.
+  int64_t num_current_failed_connections_;
+  int64_t num_previous_failed_connections_;
+};
+
+WebSocketManager::WebSocketManager(
+    int process_id,
+    StoragePartition* storage_partition,
+    const scoped_refptr<ChromeBlobStorageContext>& blob_storage_context)
+    : io_thread_state_(new IOThreadState(process_id,
+                                         storage_partition,
+                                         blob_storage_context)) {}
+
+WebSocketManager::~WebSocketManager() {
+  BrowserThread::DeleteSoon(BrowserThread::IO, FROM_HERE, io_thread_state_);
+}
+
+void WebSocketManager::CreateWebSocket(
+    int render_frame_id,
+    mojom::WebSocketRequest request) {
+  DCHECK(!BrowserThread::CurrentlyOn(BrowserThread::IO));
+
+  BrowserThread::PostTask(
+      BrowserThread::IO,
+      FROM_HERE,
+      base::Bind(&IOThreadState::CreateWebSocket,
+                 base::Unretained(io_thread_state_),
+                 render_frame_id,
+                 base::Passed(&request)));
+}
+
+}  // namespace content