Add reprioritization to TransportConnectJobs

This change makes it possible to reprioritize socket pool ConnectJobs
to match their corresponding Requests. For TransportConnectJobs that are
reprioritized while undergoing host resolution, the corresponding host
resolver request is also reprioritized to match the connect job.

Bug: 166689
Change-Id: Icaed423c6f363fa15e9e379aa135a3b98330d1a1
Reviewed-on: https://chromium-review.googlesource.com/c/1320210
Commit-Queue: Lily Chen <chlily@chromium.org>
Reviewed-by: Matt Menke <mmenke@chromium.org>
Cr-Commit-Position: refs/heads/master@{#606129}

net/dns/mock_host_resolver.cc

@@ -68,6 +68,7 @@ class MockHostResolverBase::RequestImpl
       : request_host_(request_host),
         parameters_(optional_parameters ? optional_parameters.value()
                                         : ResolveHostParameters()),
+        priority_(parameters_.initial_priority),
         host_resolver_flags_(ParametersToHostResolverFlags(parameters_)),
         id_(0),
         resolver_(resolver),
@@ -138,6 +139,10 @@ class MockHostResolverBase::RequestImpl
 
   size_t id() { return id_; }
 
+  RequestPriority priority() const { return priority_; }
+
+  void set_priority(RequestPriority priority) { priority_ = priority; }
+
   void set_id(size_t id) {
     DCHECK_GT(id, 0u);
     DCHECK_EQ(0u, id_);
@@ -150,6 +155,7 @@ class MockHostResolverBase::RequestImpl
  private:
   const HostPortPair request_host_;
   const ResolveHostParameters parameters_;
+  RequestPriority priority_;
   int host_resolver_flags_;
 
   base::Optional<AddressList> address_results_;
@@ -176,7 +182,9 @@ class MockHostResolverBase::LegacyRequestImpl : public HostResolver::Request {
 
   ~LegacyRequestImpl() override {}
 
-  void ChangeRequestPriority(RequestPriority priority) override {}
+  void ChangeRequestPriority(RequestPriority priority) override {
+    inner_request_->set_priority(priority);
+  }
 
   int Start() {
     return inner_request_->Start(base::BindOnce(
@@ -289,12 +297,6 @@ int MockHostResolverBase::ResolveStaleFromCache(
   return rv;
 }
 
-void MockHostResolverBase::DetachRequest(size_t id) {
-  auto it = requests_.find(id);
-  CHECK(it != requests_.end());
-  requests_.erase(it);
-}
-
 HostCache* MockHostResolverBase::GetHostCache() {
   return cache_.get();
 }
@@ -319,6 +321,41 @@ void MockHostResolverBase::ResolveAllPending() {
   }
 }
 
+void MockHostResolverBase::ResolveNow(size_t id) {
+  auto it = requests_.find(id);
+  if (it == requests_.end())
+    return;  // was canceled
+
+  RequestImpl* req = it->second;
+  requests_.erase(it);
+
+  AddressList addresses;
+  int error = ResolveProc(
+      req->request_host(),
+      DnsQueryTypeToAddressFamily(req->parameters().dns_query_type),
+      req->host_resolver_flags(), req->parameters().source, &addresses);
+  if (error == OK && !req->parameters().is_speculative)
+    req->set_address_results(addresses);
+  req->OnAsyncCompleted(id, error);
+}
+
+void MockHostResolverBase::DetachRequest(size_t id) {
+  auto it = requests_.find(id);
+  CHECK(it != requests_.end());
+  requests_.erase(it);
+}
+
+MockHostResolverBase::RequestImpl* MockHostResolverBase::request(size_t id) {
+  RequestMap::iterator request = requests_.find(id);
+  DCHECK(request != requests_.end());
+  return (*request).second;
+}
+
+RequestPriority MockHostResolverBase::request_priority(size_t id) {
+  DCHECK(request(id));
+  return request(id)->priority();
+}
+
 // start id from 1 to distinguish from NULL RequestHandle
 MockHostResolverBase::MockHostResolverBase(bool use_caching)
     : last_request_priority_(DEFAULT_PRIORITY),
@@ -448,24 +485,6 @@ int MockHostResolverBase::ResolveProc(const HostPortPair& host,
   return rv;
 }
 
-void MockHostResolverBase::ResolveNow(size_t id) {
-  auto it = requests_.find(id);
-  if (it == requests_.end())
-    return;  // was canceled
-
-  RequestImpl* req = it->second;
-  requests_.erase(it);
-
-  AddressList addresses;
-  int error = ResolveProc(
-      req->request_host(),
-      DnsQueryTypeToAddressFamily(req->parameters().dns_query_type),
-      req->host_resolver_flags(), req->parameters().source, &addresses);
-  if (error == OK && !req->parameters().is_speculative)
-    req->set_address_results(addresses);
-  req->OnAsyncCompleted(id, error);
-}
-
 //-----------------------------------------------------------------------------
 
 RuleBasedHostResolverProc::Rule::Rule(ResolverType resolver_type,

net/dns/mock_host_resolver.h

@@ -130,17 +130,30 @@ class MockHostResolverBase
                  HostCache::EntryStaleness* stale_out) const override;
   void SetDnsConfigOverrides(const DnsConfigOverrides& overrides) override {}
 
-  // Detach cancelled request.
-  void DetachRequest(size_t id);
+  // Returns true if there are pending requests that can be resolved by invoking
+  // ResolveAllPending().
+  bool has_pending_requests() const { return !requests_.empty(); }
 
   // Resolves all pending requests. It is only valid to invoke this if
   // set_ondemand_mode was set before. The requests are resolved asynchronously,
   // after this call returns.
   void ResolveAllPending();
 
-  // Returns true if there are pending requests that can be resolved by invoking
-  // ResolveAllPending().
-  bool has_pending_requests() const { return !requests_.empty(); }
+  // Each request is assigned an ID when started and stored with the resolver
+  // for async resolution, starting with 1. IDs are not reused. Once a request
+  // completes, it is destroyed, and can no longer be accessed.
+
+  // Resolve request stored in |requests_|. Pass rv to callback.
+  void ResolveNow(size_t id);
+
+  // Detach cancelled request.
+  void DetachRequest(size_t id);
+
+  // Returns the request with the given id.
+  RequestImpl* request(size_t id);
+
+  // Returns the priority of the request with the given id.
+  RequestPriority request_priority(size_t id);
 
   // The number of times that Resolve() has been called.
   size_t num_resolve() const {
@@ -188,8 +201,6 @@ class MockHostResolverBase
                   HostResolverFlags flags,
                   HostResolverSource source,
                   AddressList* addresses);
-  // Resolve request stored in |requests_|. Pass rv to callback.
-  void ResolveNow(size_t id);
 
   RequestPriority last_request_priority_;
   bool synchronous_mode_;

net/socket/client_socket_pool_base.cc

@@ -68,6 +68,11 @@ std::unique_ptr<StreamSocket> ConnectJob::PassSocket() {
   return std::move(socket_);
 }
 
+void ConnectJob::ChangePriority(RequestPriority priority) {
+  set_priority(priority);
+  ChangePriorityInternal(priority);
+}
+
 int ConnectJob::Connect() {
   if (!timeout_duration_.is_zero())
     timer_.Start(FROM_HERE, timeout_duration_, this, &ConnectJob::OnTimeout);
@@ -159,7 +164,8 @@ void ClientSocketPoolBaseHelper::Request::AssignJob(ConnectJob* job) {
   DCHECK(job);
   DCHECK(!job_);
   job_ = job;
-  // TODO(chlily): Make job priority match request priority.
+  if (job_->priority() != priority_)
+    job_->ChangePriority(priority_);
 }
 
 ConnectJob* ClientSocketPoolBaseHelper::Request::ReleaseJob() {

net/socket/client_socket_pool_base.h

@@ -98,12 +98,15 @@ class NET_EXPORT_PRIVATE ConnectJob {
   // Accessors
   const std::string& group_name() const { return group_name_; }
   const NetLogWithSource& net_log() { return net_log_; }
+  RequestPriority priority() const { return priority_; }
 
   // Releases ownership of the underlying socket to the caller.
   // Returns the released socket, or NULL if there was a connection
   // error.
   std::unique_ptr<StreamSocket> PassSocket();
 
+  void ChangePriority(RequestPriority priority);
+
   // Begins connecting the socket.  Returns OK on success, ERR_IO_PENDING if it
   // cannot complete synchronously without blocking, or another net error code
   // on error.  In asynchronous completion, the ConnectJob will notify
@@ -126,7 +129,7 @@ class NET_EXPORT_PRIVATE ConnectJob {
   const NetLogWithSource& net_log() const { return net_log_; }
 
  protected:
-  RequestPriority priority() const { return priority_; }
+  void set_priority(RequestPriority priority) { priority_ = priority; }
   const SocketTag& socket_tag() const { return socket_tag_; }
   ClientSocketPool::RespectLimits respect_limits() const {
     return respect_limits_;
@@ -142,6 +145,8 @@ class NET_EXPORT_PRIVATE ConnectJob {
  private:
   virtual int ConnectInternal() = 0;
 
+  virtual void ChangePriorityInternal(RequestPriority priority) {}
+
   void LogConnectStart();
   void LogConnectCompletion(int net_error);
 
@@ -150,8 +155,7 @@ class NET_EXPORT_PRIVATE ConnectJob {
 
   const std::string group_name_;
   const base::TimeDelta timeout_duration_;
-  // TODO(akalin): Support reprioritization.
-  const RequestPriority priority_;
+  RequestPriority priority_;
   const SocketTag socket_tag_;
   const ClientSocketPool::RespectLimits respect_limits_;
   // Timer to abort jobs that take too long.

net/socket/transport_client_socket_pool.cc

@@ -245,6 +245,7 @@ int TransportConnectJob::DoLoop(int result) {
 
   return rv;
 }
+
 int TransportConnectJob::DoResolveHost() {
   next_state_ = STATE_RESOLVE_HOST_COMPLETE;
   connect_timing_.dns_start = base::TimeTicks::Now();
@@ -439,6 +440,14 @@ int TransportConnectJob::ConnectInternal() {
   return DoLoop(OK);
 }
 
+void TransportConnectJob::ChangePriorityInternal(RequestPriority priority) {
+  if (next_state_ == STATE_RESOLVE_HOST_COMPLETE) {
+    DCHECK(request_);
+    // Change the request priority in the host resolver.
+    request_->ChangeRequestPriority(priority);
+  }
+}
+
 void TransportConnectJob::CopyConnectionAttemptsFromSockets() {
   if (transport_socket_)
     transport_socket_->GetConnectionAttempts(&connection_attempts_);

net/socket/transport_client_socket_pool.h

@@ -154,6 +154,10 @@ class NET_EXPORT_PRIVATE TransportConnectJob : public ConnectJob {
   // Otherwise, it returns a net error code.
   int ConnectInternal() override;
 
+  // If host resolution is currently underway, change the priority of the host
+  // resolver request.
+  void ChangePriorityInternal(RequestPriority priority) override;
+
   void CopyConnectionAttemptsFromSockets();
 
   scoped_refptr<TransportSocketParams> params_;
@@ -234,7 +238,7 @@ class NET_EXPORT_PRIVATE TransportClientSocketPool : public ClientSocketPool {
       bool include_nested_pools) const override;
   base::TimeDelta ConnectionTimeout() const override;
 
-  // HigherLayeredPool implementation.
+  // LowerLayeredPool implementation.
   bool IsStalled() const override;
   void AddHigherLayeredPool(HigherLayeredPool* higher_pool) override;
   void RemoveHigherLayeredPool(HigherLayeredPool* higher_pool) override;

net/socket/transport_client_socket_pool_unittest.cc

@@ -215,6 +215,231 @@ TEST_F(TransportClientSocketPoolTest, SetResolvePriorityOnInit) {
   }
 }
 
+TEST_F(TransportClientSocketPoolTest, ReprioritizeRequests) {
+  host_resolver_->set_ondemand_mode(true);
+
+  TestCompletionCallback callback1;
+  ClientSocketHandle handle1;
+  int rv1 = handle1.Init("a", params_, LOW, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback1.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv1, IsError(ERR_IO_PENDING));
+
+  TestCompletionCallback callback2;
+  ClientSocketHandle handle2;
+  int rv2 = handle2.Init("a", params_, HIGHEST, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback2.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv2, IsError(ERR_IO_PENDING));
+
+  TestCompletionCallback callback3;
+  ClientSocketHandle handle3;
+  int rv3 = handle3.Init("a", params_, LOWEST, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback3.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv3, IsError(ERR_IO_PENDING));
+
+  TestCompletionCallback callback4;
+  ClientSocketHandle handle4;
+  int rv4 = handle4.Init("a", params_, MEDIUM, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback4.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv4, IsError(ERR_IO_PENDING));
+
+  TestCompletionCallback callback5;
+  ClientSocketHandle handle5;
+  int rv5 = handle5.Init("a", params_, HIGHEST, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback5.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv5, IsError(ERR_IO_PENDING));
+
+  TestCompletionCallback callback6;
+  ClientSocketHandle handle6;
+  int rv6 = handle6.Init("a", params_, LOW, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback6.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv6, IsError(ERR_IO_PENDING));
+
+  // New jobs are created for each of the first 6 requests with the
+  // corresponding priority.
+  //
+  // Queue of pending requests:
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    2      2   HIGHEST
+  //    5      5   HIGHEST
+  //    4      4   MEDIUM
+  //    1      1   LOW
+  //    6      6   LOW
+  //    3      3   LOWEST
+  EXPECT_EQ(LOW, host_resolver_->request_priority(1));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(2));
+  EXPECT_EQ(LOWEST, host_resolver_->request_priority(3));
+  EXPECT_EQ(MEDIUM, host_resolver_->request_priority(4));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(5));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(6));
+
+  // Inserting a highest-priority request steals the job from the lowest
+  // priority request and reprioritizes it to match the new request.
+  TestCompletionCallback callback7;
+  ClientSocketHandle handle7;
+  int rv7 = handle7.Init("a", params_, HIGHEST, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback7.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv7, IsError(ERR_IO_PENDING));
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    2      2   HIGHEST
+  //    5      5   HIGHEST
+  //    7      3   HIGHEST
+  //    4      4   MEDIUM
+  //    1      1   LOW
+  //    6      6   LOW
+  //    3          LOWEST
+  EXPECT_EQ(LOW, host_resolver_->request_priority(1));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(2));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(3));  // reprioritized
+  EXPECT_EQ(MEDIUM, host_resolver_->request_priority(4));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(5));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(6));
+
+  TestCompletionCallback callback8;
+  ClientSocketHandle handle8;
+  int rv8 = handle8.Init("a", params_, HIGHEST, SocketTag(),
+                         ClientSocketPool::RespectLimits::ENABLED,
+                         callback8.callback(), &pool_, NetLogWithSource());
+  EXPECT_THAT(rv8, IsError(ERR_IO_PENDING));
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    2      2   HIGHEST
+  //    5      5   HIGHEST
+  //    7      3   HIGHEST
+  //    8      6   HIGHEST
+  //    4      4   MEDIUM
+  //    1      1   LOW
+  //    6          LOW
+  //    3          LOWEST
+  EXPECT_EQ(LOW, host_resolver_->request_priority(1));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(2));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(3));
+  EXPECT_EQ(MEDIUM, host_resolver_->request_priority(4));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(5));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(6));  // reprioritized
+
+  // A request completes, then the socket is returned to the socket pool and
+  // goes to the highest remaining request. The job from the highest request
+  // should then be reassigned to the first request without a job.
+  host_resolver_->ResolveNow(2);
+  EXPECT_THAT(callback2.WaitForResult(), IsOk());
+  EXPECT_TRUE(handle2.is_initialized());
+  EXPECT_TRUE(handle2.socket());
+  handle2.Reset();
+  EXPECT_THAT(callback5.WaitForResult(), IsOk());
+  EXPECT_TRUE(handle5.is_initialized());
+  EXPECT_TRUE(handle5.socket());
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    7      3   HIGHEST
+  //    8      6   HIGHEST
+  //    4      4   MEDIUM
+  //    1      1   LOW
+  //    6      5   LOW
+  //    3          LOWEST
+  EXPECT_EQ(LOW, host_resolver_->request_priority(1));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(3));
+  EXPECT_EQ(MEDIUM, host_resolver_->request_priority(4));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(5));  // reprioritized
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(6));
+
+  // Cancelling a request with a job reassigns the job to a lower request.
+  handle7.Reset();
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    8      6   HIGHEST
+  //    4      4   MEDIUM
+  //    1      1   LOW
+  //    6      5   LOW
+  //    3      3   LOWEST
+  EXPECT_EQ(LOW, host_resolver_->request_priority(1));
+  EXPECT_EQ(LOWEST, host_resolver_->request_priority(3));  // reprioritized
+  EXPECT_EQ(MEDIUM, host_resolver_->request_priority(4));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(5));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(6));
+
+  // Reprioritizing a request changes its job's priority.
+  pool_.SetPriority("a", &handle4, LOWEST);
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    8      6   HIGHEST
+  //    1      1   LOW
+  //    6      5   LOW
+  //    3      3   LOWEST
+  //    4      4   LOWEST
+  EXPECT_EQ(LOW, host_resolver_->request_priority(1));
+  EXPECT_EQ(LOWEST, host_resolver_->request_priority(3));
+  EXPECT_EQ(LOWEST, host_resolver_->request_priority(4));  // reprioritized
+  EXPECT_EQ(LOW, host_resolver_->request_priority(5));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(6));
+
+  pool_.SetPriority("a", &handle3, MEDIUM);
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    8      6   HIGHEST
+  //    3      3   MEDIUM
+  //    1      1   LOW
+  //    6      5   LOW
+  //    4      4   LOWEST
+  EXPECT_EQ(LOW, host_resolver_->request_priority(1));
+  EXPECT_EQ(MEDIUM, host_resolver_->request_priority(3));  // reprioritized
+  EXPECT_EQ(LOWEST, host_resolver_->request_priority(4));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(5));
+  EXPECT_EQ(HIGHEST, host_resolver_->request_priority(6));
+
+  // Host resolution finishes for a lower-down request. The highest request
+  // should get the socket and its job should be reassigned to the lower
+  // request.
+  host_resolver_->ResolveNow(1);
+  EXPECT_THAT(callback8.WaitForResult(), IsOk());
+  EXPECT_TRUE(handle8.is_initialized());
+  EXPECT_TRUE(handle8.socket());
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    3      3   MEDIUM
+  //    1      6   LOW
+  //    6      5   LOW
+  //    4      4   LOWEST
+  EXPECT_EQ(MEDIUM, host_resolver_->request_priority(3));
+  EXPECT_EQ(LOWEST, host_resolver_->request_priority(4));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(5));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(6));  // reprioritized
+
+  // Host resolution finishes for the highest request. Nothing gets
+  // reprioritized.
+  host_resolver_->ResolveNow(3);
+  EXPECT_THAT(callback3.WaitForResult(), IsOk());
+  EXPECT_TRUE(handle3.is_initialized());
+  EXPECT_TRUE(handle3.socket());
+  // Request  Job  Priority
+  // =======  ===  ========
+  //    1      6   LOW
+  //    6      5   LOW
+  //    4      4   LOWEST
+  EXPECT_EQ(LOWEST, host_resolver_->request_priority(4));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(5));
+  EXPECT_EQ(LOW, host_resolver_->request_priority(6));
+
+  host_resolver_->ResolveAllPending();
+  EXPECT_THAT(callback1.WaitForResult(), IsOk());
+  EXPECT_TRUE(handle1.is_initialized());
+  EXPECT_TRUE(handle1.socket());
+  EXPECT_THAT(callback4.WaitForResult(), IsOk());
+  EXPECT_TRUE(handle4.is_initialized());
+  EXPECT_TRUE(handle4.socket());
+  EXPECT_THAT(callback6.WaitForResult(), IsOk());
+  EXPECT_TRUE(handle6.is_initialized());
+  EXPECT_TRUE(handle6.socket());
+}
+
 TEST_F(TransportClientSocketPoolTest, InitHostResolutionFailure) {
   host_resolver_->rules()->AddSimulatedFailure("unresolvable.host.name");
   TestCompletionCallback callback;