Buffer data usage in the DataUseAggregator.

This CL adds simple buffering of data usage, in order to notify
observers less often and make it possible to amortize data usage.

Data usage amortization on Android will be implemented in a later CL.

BUG=518051

Review URL: https://codereview.chromium.org/1396223002

Cr-Commit-Position: refs/heads/master@{#354160}

components/data_usage/core/data_use.cc

@@ -21,4 +21,10 @@ DataUse::DataUse(const GURL& url,
       tx_bytes(tx_bytes),
       rx_bytes(rx_bytes) {}
 
+bool DataUse::CanCombineWith(const DataUse& other) const {
+  return url == other.url && request_time == other.request_time &&
+         first_party_for_cookies == other.first_party_for_cookies &&
+         tab_id == other.tab_id && connection_type == other.connection_type;
+}
+
 }  // namespace data_usage

components/data_usage/core/data_use.h

@@ -22,6 +22,9 @@ struct DataUse {
           int64_t tx_bytes,
           int64_t rx_bytes);
 
+  // Returns true if |this| and |other| are identical except for byte counts.
+  bool CanCombineWith(const DataUse& other) const;
+
   GURL url;
   // The time when the request that is associated with these bytes was started.
   base::Time request_time;

components/data_usage/core/data_use_aggregator.cc

@@ -4,12 +4,19 @@
 
 #include "components/data_usage/core/data_use_aggregator.h"
 
+#include "base/bind.h"
+#include "base/message_loop/message_loop.h"
+#include "base/single_thread_task_runner.h"
 #include "net/base/network_change_notifier.h"
 #include "net/url_request/url_request.h"
 
 namespace data_usage {
 
-DataUseAggregator::DataUseAggregator() {}
+DataUseAggregator::DataUseAggregator()
+    : off_the_record_tx_bytes_since_last_flush_(0),
+      off_the_record_rx_bytes_since_last_flush_(0),
+      is_flush_pending_(false),
+      weak_ptr_factory_(this) {}
 
 DataUseAggregator::~DataUseAggregator() {}
 
@@ -29,29 +36,55 @@ void DataUseAggregator::ReportDataUse(const net::URLRequest& request,
                                       int64_t rx_bytes) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
-  // TODO(sclittle): Once actual buffering/aggregation is being done, consider
-  // combining consecutive data use entries from the same request.
   DataUse data_use(request.url(), request.request_time(),
                    request.first_party_for_cookies(), tab_id,
                    net::NetworkChangeNotifier::GetConnectionType(), tx_bytes,
                    rx_bytes);
 
-  // TODO(sclittle): Buffer and amortize data use on supported platforms.
-  NotifyDataUse(std::vector<DataUse>(1, data_use));
+  // As an optimization, attempt to combine the newly reported data use with the
+  // most recent buffered data use, if the annotations on the data use are the
+  // same.
+  if (!buffered_data_use_.empty() &&
+      buffered_data_use_.back().CanCombineWith(data_use)) {
+    buffered_data_use_.back().tx_bytes += tx_bytes;
+    buffered_data_use_.back().rx_bytes += rx_bytes;
+  } else {
+    buffered_data_use_.push_back(data_use);
+  }
+
+  if (!is_flush_pending_) {
+    // Post a flush operation to happen in the future, so that the
+    // DataUseAggregator has a chance to batch together some data use before
+    // notifying observers.
+    base::MessageLoop::current()->task_runner()->PostTask(
+        FROM_HERE,
+        base::Bind(&DataUseAggregator::FlushBufferedDataUse, GetWeakPtr()));
+    is_flush_pending_ = true;
+  }
 }
 
 void DataUseAggregator::ReportOffTheRecordDataUse(int64_t tx_bytes,
                                                   int64_t rx_bytes) {
   DCHECK(thread_checker_.CalledOnValidThread());
-  // TODO(sclittle): Once data usage amortization is implemented, keep track of
-  // the off-the-record bytes so that they can be taken out of the amortized
-  // data usage calculations if applicable.
+  off_the_record_tx_bytes_since_last_flush_ += tx_bytes;
+  off_the_record_rx_bytes_since_last_flush_ += rx_bytes;
+}
+
+base::WeakPtr<DataUseAggregator> DataUseAggregator::GetWeakPtr() {
+  return weak_ptr_factory_.GetWeakPtr();
 }
 
-void DataUseAggregator::NotifyDataUse(
-    const std::vector<DataUse>& data_use_sequence) {
+void DataUseAggregator::FlushBufferedDataUse() {
   DCHECK(thread_checker_.CalledOnValidThread());
-  FOR_EACH_OBSERVER(Observer, observer_list_, OnDataUse(data_use_sequence));
+
+  // TODO(sclittle): Amortize data use on supported platforms before notifying
+  // observers.
+  FOR_EACH_OBSERVER(Observer, observer_list_, OnDataUse(buffered_data_use_));
+
+  buffered_data_use_.clear();
+  off_the_record_tx_bytes_since_last_flush_ = 0;
+  off_the_record_rx_bytes_since_last_flush_ = 0;
+  is_flush_pending_ = false;
 }
 
 }  // namespace data_usage

components/data_usage/core/data_use_aggregator.h

@@ -10,6 +10,7 @@
 #include <vector>
 
 #include "base/macros.h"
+#include "base/memory/weak_ptr.h"
 #include "base/observer_list.h"
 #include "base/threading/thread_checker.h"
 #include "components/data_usage/core/data_use.h"
@@ -50,12 +51,29 @@ class DataUseAggregator {
   // Virtual for testing.
   virtual void ReportOffTheRecordDataUse(int64_t tx_bytes, int64_t rx_bytes);
 
+  base::WeakPtr<DataUseAggregator> GetWeakPtr();
+
  private:
-  void NotifyDataUse(const std::vector<DataUse>& data_use_sequence);
+  // Flush any buffered data use and notify observers.
+  void FlushBufferedDataUse();
 
   base::ThreadChecker thread_checker_;
   base::ObserverList<Observer> observer_list_;
 
+  // Buffer of unreported data use.
+  std::vector<DataUse> buffered_data_use_;
+
+  // The total amount of off-the-record data usage that has happened since the
+  // last time the buffer was flushed.
+  int64_t off_the_record_tx_bytes_since_last_flush_;
+  int64_t off_the_record_rx_bytes_since_last_flush_;
+
+  // Indicates if a FlushBufferedDataUse() callback has been posted to run later
+  // on the IO thread.
+  bool is_flush_pending_;
+
+  base::WeakPtrFactory<DataUseAggregator> weak_ptr_factory_;
+
   DISALLOW_COPY_AND_ASSIGN(DataUseAggregator);
 };