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 {
 
+// SocketWatcherDelegate receives socket performance events from different
+// SocketWatchers, and passes them to the NetworkQualityEstimator on the
+// provided task runner. SocketWatcherDelegate is thread safe.
+class NetworkQualityEstimator::SocketWatcherDelegate {

[Ryan Sleevi, 2016/04/04 17:30:09]

Yeah, I don't think you should need this abstraction (for the reason explained
in the header). It should be fine to have the NQE::SW directly use/call into NQE
via WeakPtr.

[tbansal1, 2016/04/04 18:35:46]

Is it okay if I keep this? The delegate interface looks cleaner.

[Ryan Sleevi, 2016/04/04 18:53:37]

Can you explain why you feel that way? I'd like to understand to make sure I'm
not missing something.

The downside is you're adding an additional level of indirection, except that
indirection is only relevant to your implementation (this .cc), so it doesn't by
you any abstractions or decoupling.

It's sort of like having a

void A() {
  PostTask(B())
}

void B() {
  // stuff
}


void DoSomething() {
  A();
}


That seems like it makes it harder to read (now you have to read A to know that
it'll call B)

Alternatively, it could be rewritten as

void B() {
  // Do stuff;
}

void DoSomething(
  PostTask(B());
}

Which seems slightly clearer.

if there was more logic between the two, I agree, the abstraction might be
helpful - but it just seems to make it harder to know "what will happen" in the
NQE::SocketWatcher here

[tbansal1, 2016/04/04 22:07:03]

I have removed the SocketDelegate. I just thought it was probably incorrect if
SocketFactory was creating copies of NQE's weak pointer. 

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

As long as the copying of NQE's WeakPtr only happens in
network_quality_estimator.cc, that's generally OK.

That's generally how/where you should use a WeakPtr - that is, it should never
'escape' the implementation details. Having it passed/copied between multiple
classes is fine, so long as those classes all live in the same .cc file, and
there's no "public" way (e.g. in the headers via public/protected methods) to
get access to those weak ptrs.

Does that help clarify?

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

Yes. Thanks for the detailed reply.

+ public:
+  SocketWatcherDelegate(
+      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) {
+    DCHECK(task_runner_);
+  }
+
+  ~SocketWatcherDelegate() {}
+
+  void OnUpdatedRTTAvailable(SocketPerformanceWatcherFactory::Protocol protocol,
+                             const base::TimeDelta& rtt) {
+    task_runner_->PostTask(
+        FROM_HERE, base::Bind(&NetworkQualityEstimator::OnUpdatedRTTAvailable,
+                              network_quality_estimator_, protocol, rtt));
+  }
+
+ private:
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+
+  base::WeakPtr<NetworkQualityEstimator> network_quality_estimator_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketWatcherDelegate);
+};
+
+// SocketWatcher implements SocketPerformanceWatcher, and notifies
+// SocketWatcherDelegate of various socket performance events. SocketWatcher is
+// not thread-safe.
+class NetworkQualityEstimator::SocketWatcher : public SocketPerformanceWatcher {
+ public:
+  //  |delegate| is guaranteed to be non-null during SocketWatcher's lifetime.
+  SocketWatcher(SocketPerformanceWatcherFactory::Protocol protocol,
+                SocketWatcherDelegate* delegate)
+      : protocol_(protocol), delegate_(delegate) {
+    DCHECK(delegate_);
+  }
+
+  ~SocketWatcher() {}
+
+  // SocketPerformanceWatcher implementation:
+  void OnUpdatedRTTAvailable(const base::TimeDelta& rtt) override {
+    DCHECK(thread_checker_.CalledOnValidThread());
+
+    delegate_->OnUpdatedRTTAvailable(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_;
+
+  SocketWatcherDelegate* delegate_;
+
+  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:
+  //  |delegate| is guaranteed to be non-null during SocketWatcherFactory's
+  //  lifetime.
+  explicit SocketWatcherFactory(SocketWatcherDelegate* delegate)
+      : delegate_(delegate) {
+    DCHECK(delegate_);
+  }
+
+  ~SocketWatcherFactory() override {}
+
+  // SocketPerformanceWatcherFactory implementation:
+  scoped_ptr<SocketPerformanceWatcher> CreateSocketPerformanceWatcher(
+      const Protocol protocol) override {
+    return scoped_ptr<SocketPerformanceWatcher>(
+        new SocketWatcher(protocol, delegate_));
+  }
+
+ private:
+  SocketWatcherDelegate* delegate_;
+
+  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();
+
+  delegate_.reset(new SocketWatcherDelegate(base::ThreadTaskRunnerHandle::Get(),
+                                            weak_ptr_factory_.GetWeakPtr()));
+
+  watcher_factory_.reset(new SocketWatcherFactory(delegate_.get()));
 }
 
 // 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