Add SocketWatcherFactory as a helper class to NQE

net/base/network_quality_estimator.cc

 // Copyright 2015 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 "net/base/network_quality_estimator.h"
 
 #include <float.h>
 #include <algorithm>
 #include <cmath>
 #include <limits>
 #include <utility>
 #include <vector>
 
 #include "base/logging.h"
 #include "base/metrics/histogram.h"
 #include "base/metrics/histogram_base.h"
 #include "base/strings/string_number_conversions.h"
+#include "base/thread_task_runner_handle.h"
 #include "build/build_config.h"
 #include "net/base/load_flags.h"
 #include "net/base/load_timing_info.h"
 #include "net/base/network_interfaces.h"
+#include "net/base/socket_performance_watcher.h"
 #include "net/base/url_util.h"
 #include "net/url_request/url_request.h"
 #include "url/gurl.h"
 
 #if defined(OS_ANDROID)
 #include "net/android/network_library.h"
 #endif  // OS_ANDROID
 
 namespace {
 
@@ skipping 78 matching lines @@
   const char prefix[] = "NQE.";
   return base::Histogram::FactoryGet(
       prefix + statistic_name + GetNameForConnectionType(type), kLowerLimit,
       max_limit, kBucketCount, base::HistogramBase::kUmaTargetedHistogramFlag);
 }
 
 }  // namespace
 
 namespace net {
 
+// SocketWatcher implements SocketPerformanceWatcher, and notifies
+// NetworkQualityEstimator of various socket performance events. SocketWatcher
+// is not thread-safe.
+class NetworkQualityEstimator::SocketWatcher : public SocketPerformanceWatcher {

[bengr, 2016/04/07 00:20:05]

Any chance I could convince you to move this to another file?

[Ryan Sleevi, 2016/04/07 00:27:53]

I specifically asked for this. I don't feel you can reasonably move to another
file w/o adding another (unnecessary) layer of indirection, and I'm not sure I
see the value.

[tbansal1, 2016/04/07 01:43:36]

Acknowledged.

+ public:
+  SocketWatcher(
+      SocketPerformanceWatcherFactory::Protocol protocol,

[Ryan Sleevi, 2016/04/04 22:34:04]

DESIGN NOTE (Future): As/if you change SocketPerformanceWatcher to be
TCPSocket::PerformanceWatcher (or delegate, whatever), I don't think it'll make
sense to need to track the |protocol| here, because
TCPSocket::PerformanceWatchers will always be Protocol::TCP. And if you have a
QuicConnection::PerformanceWatcher (which again, is really about whether or not
the TCPSocket and QUIC abstraction share the same pre/post conditions/api shape,
or whether they need distinct API shapes, such as to handle things like QUIC
connection migrations, so probably so?), the factory creating those wouldn't
need to track protocol either, because they'd always be QUIC watchers.

But you could still have both factories call NQE::OnUpdatedRTTAvailable - since
NQE only cares about the aggregated, and not per-destination IPs.

[tbansal1, 2016/04/04 23:46:32]

OK, I will write down a quick design note to see which approach makes sense.
Should we move the SPWF and SPW or not to TcpClientSocket. Or should we only
move SPW.  If we move both, then we need to pass 2 factories down the stack with
HttpNetworkSession::Params which might be wasteful given that the two factories
are not doing anything really different.

+      scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+      const base::WeakPtr<NetworkQualityEstimator>& network_quality_estimator)
+      : protocol_(protocol),
+        task_runner_(std::move(task_runner)),
+        network_quality_estimator_(network_quality_estimator) {}
+
+  ~SocketWatcher() override {}
+
+  // SocketPerformanceWatcher implementation:
+  void OnUpdatedRTTAvailable(const base::TimeDelta& rtt) override {
+    DCHECK(thread_checker_.CalledOnValidThread());
+
+    task_runner_->PostTask(
+        FROM_HERE, base::Bind(&NetworkQualityEstimator::OnUpdatedRTTAvailable,
+                              network_quality_estimator_, protocol_, rtt));
+  }
+
+  bool ShouldNotifyUpdatedRTT() const override {
+    DCHECK(thread_checker_.CalledOnValidThread());
+
+    return true;
+  }
+
+  void Reset() override { DCHECK(thread_checker_.CalledOnValidThread()); }
+
+ private:
+  // Transport layer protocol used by the socket that |this| is watching.
+  const SocketPerformanceWatcherFactory::Protocol protocol_;

[bengr, 2016/04/07 00:20:05]

I agree with Ryan's design note above. 

[tbansal1, 2016/04/07 01:43:36]

I believe the eventual decision was to keep 1 factory interface, and 1 watcher
interface (see the design note that I shared on April 4 titled "SPW and SPWF").

+
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+
+  base::WeakPtr<NetworkQualityEstimator> network_quality_estimator_;
+
+  base::ThreadChecker thread_checker_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketWatcher);
+};
+
+// SocketWatcherFactory implements SocketPerformanceWatcherFactory, and is
+// owned by NetworkQualityEstimator. SocketWatcherFactory is thread safe.
+class NetworkQualityEstimator::SocketWatcherFactory
+    : public SocketPerformanceWatcherFactory {
+ public:
+  SocketWatcherFactory(
+      scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+      const base::WeakPtr<NetworkQualityEstimator>& network_quality_estimator)
+      : task_runner_(std::move(task_runner)),
+        network_quality_estimator_(network_quality_estimator) {}
+
+  ~SocketWatcherFactory() override {}
+
+  // SocketPerformanceWatcherFactory implementation:
+  scoped_ptr<SocketPerformanceWatcher> CreateSocketPerformanceWatcher(
+      const Protocol protocol) override {
+    return scoped_ptr<SocketPerformanceWatcher>(
+        new SocketWatcher(protocol, task_runner_, network_quality_estimator_));
+  }
+
+ private:
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+
+  base::WeakPtr<NetworkQualityEstimator> network_quality_estimator_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketWatcherFactory);
+};
+
 const int32_t NetworkQualityEstimator::kInvalidThroughput = 0;
 
 NetworkQualityEstimator::NetworkQualityEstimator(
     scoped_ptr<ExternalEstimateProvider> external_estimates_provider,
     const std::map<std::string, std::string>& variation_params)
     : NetworkQualityEstimator(std::move(external_estimates_provider),
                               variation_params,
                               false,
                               false) {}
 
 NetworkQualityEstimator::NetworkQualityEstimator(
     scoped_ptr<ExternalEstimateProvider> external_estimates_provider,
     const std::map<std::string, std::string>& variation_params,
     bool allow_local_host_requests_for_tests,
     bool allow_smaller_responses_for_tests)
     : allow_localhost_requests_(allow_local_host_requests_for_tests),
       allow_small_responses_(allow_smaller_responses_for_tests),
       last_connection_change_(base::TimeTicks::Now()),
       current_network_id_(
           NetworkID(NetworkChangeNotifier::ConnectionType::CONNECTION_UNKNOWN,
                     std::string())),
       downstream_throughput_kbps_observations_(
           GetWeightMultiplierPerSecond(variation_params)),
       rtt_observations_(GetWeightMultiplierPerSecond(variation_params)),
-      external_estimate_provider_(std::move(external_estimates_provider)) {
+      external_estimate_provider_(std::move(external_estimates_provider)),
+      weak_ptr_factory_(this) {
   static_assert(kMinRequestDurationMicroseconds > 0,
                 "Minimum request duration must be > 0");
   static_assert(kDefaultHalfLifeSeconds > 0,
                 "Default half life duration must be > 0");
   static_assert(kMaximumNetworkQualityCacheSize > 0,
                 "Size of the network quality cache must be > 0");
   // This limit should not be increased unless the logic for removing the
   // oldest cache entry is rewritten to use a doubly-linked-list LRU queue.
   static_assert(kMaximumNetworkQualityCacheSize <= 10,
                 "Size of the network quality cache must <= 10");
 
   ObtainOperatingParams(variation_params);
   NetworkChangeNotifier::AddConnectionTypeObserver(this);
   if (external_estimate_provider_) {
     RecordExternalEstimateProviderMetrics(
         EXTERNAL_ESTIMATE_PROVIDER_STATUS_AVAILABLE);
     external_estimate_provider_->SetUpdatedEstimateDelegate(this);
     QueryExternalEstimateProvider();
   } else {
     RecordExternalEstimateProviderMetrics(
         EXTERNAL_ESTIMATE_PROVIDER_STATUS_NOT_AVAILABLE);
   }
   current_network_id_ = GetCurrentNetworkID();
   AddDefaultEstimates();
+
+  watcher_factory_.reset(new SocketWatcherFactory(
+      base::ThreadTaskRunnerHandle::Get(), weak_ptr_factory_.GetWeakPtr()));
 }
 
 // static
 const base::TimeDelta NetworkQualityEstimator::InvalidRTT() {
   return base::TimeDelta::Max();
 }
 
 void NetworkQualityEstimator::ObtainOperatingParams(
     const std::map<std::string, std::string>& variation_params) {
   DCHECK(thread_checker_.CalledOnValidThread());
@@ skipping 190 matching lines @@
   DCHECK(thread_checker_.CalledOnValidThread());
   throughput_observer_list_.AddObserver(throughput_observer);
 }
 
 void NetworkQualityEstimator::RemoveThroughputObserver(
     ThroughputObserver* throughput_observer) {
   DCHECK(thread_checker_.CalledOnValidThread());
   throughput_observer_list_.RemoveObserver(throughput_observer);
 }
 
+SocketPerformanceWatcherFactory*
+NetworkQualityEstimator::GetSocketPerformanceWatcherFactory() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  return watcher_factory_.get();
+}
+
 void NetworkQualityEstimator::RecordRTTUMA(int32_t estimated_value_msec,
                                            int32_t actual_value_msec) const {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   // Record the difference between the actual and the estimated value.
   if (estimated_value_msec >= actual_value_msec) {
     base::HistogramBase* difference_rtt =
         GetHistogram("DifferenceRTTEstimatedAndActual.",
                      current_network_id_.type, 10 * 1000);  // 10 seconds
     difference_rtt->Add(estimated_value_msec - actual_value_msec);
@@ skipping 508 matching lines @@
   }
   DCHECK_LT(cached_network_qualities_.size(),
             static_cast<size_t>(kMaximumNetworkQualityCacheSize));
 
   cached_network_qualities_.insert(std::make_pair(
       current_network_id_, CachedNetworkQuality(network_quality)));
   DCHECK_LE(cached_network_qualities_.size(),
             static_cast<size_t>(kMaximumNetworkQualityCacheSize));
 }
 
-scoped_ptr<SocketPerformanceWatcher>
-NetworkQualityEstimator::CreateSocketPerformanceWatcher(
-    const Protocol protocol) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-
-  return scoped_ptr<SocketPerformanceWatcher>(
-      new SocketPerformanceWatcher(protocol, this));
-}
-
 void NetworkQualityEstimator::OnUpdatedRTTAvailable(
-    const Protocol protocol,
+    SocketPerformanceWatcherFactory::Protocol protocol,
     const base::TimeDelta& rtt) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   switch (protocol) {
-    case PROTOCOL_TCP:
+    case SocketPerformanceWatcherFactory::PROTOCOL_TCP:
       NotifyObserversOfRTT(RttObservation(rtt, base::TimeTicks::Now(), TCP));
       return;
-    case PROTOCOL_QUIC:
+    case SocketPerformanceWatcherFactory::PROTOCOL_QUIC:
       NotifyObserversOfRTT(RttObservation(rtt, base::TimeTicks::Now(), QUIC));
       return;
     default:
       NOTREACHED();
   }
 }
 
 void NetworkQualityEstimator::NotifyObserversOfRTT(
     const RttObservation& observation) {
   FOR_EACH_OBSERVER(
@@ skipping 50 matching lines @@
 
 NetworkQualityEstimator::NetworkQuality&
     NetworkQualityEstimator::NetworkQuality::
     operator=(const NetworkQuality& other) {
   rtt_ = other.rtt_;
   downstream_throughput_kbps_ = other.downstream_throughput_kbps_;
   return *this;
 }
 
 }  // namespace net