sync: abort active HTTP requests on shutdown.

chrome/browser/sync/engine/net/server_connection_manager.cc

 // Copyright (c) 2011 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 "chrome/browser/sync/engine/net/server_connection_manager.h"
 
 #include <errno.h>
 
 #include <ostream>
 #include <string>
 #include <vector>
 
+#include "base/auto_reset.h"
 #include "base/command_line.h"
 #include "build/build_config.h"
 #include "chrome/browser/sync/engine/net/url_translator.h"
 #include "chrome/browser/sync/engine/syncer.h"
 #include "chrome/browser/sync/engine/syncproto.h"
 #include "chrome/browser/sync/protocol/sync.pb.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
 #include "chrome/common/net/http_return.h"
 #include "googleurl/src/gurl.h"
 
@@ skipping 28 matching lines @@
     ENUM_CASE(SYNC_AUTH_ERROR);
     ENUM_CASE(SERVER_CONNECTION_OK);
     ENUM_CASE(RETRY);
   }
   NOTREACHED();
   return "";
 }
 
 #undef ENUM_CASE
 
-bool ServerConnectionManager::Post::ReadBufferResponse(
+ServerConnectionManager::Connection::Connection(
+    ServerConnectionManager* scm) : scm_(scm), timing_info_(0) {
+}
+
+ServerConnectionManager::Connection::~Connection() {
+}
+
+bool ServerConnectionManager::Connection::ReadBufferResponse(
     string* buffer_out,
     HttpResponse* response,
     bool require_response) {
   if (RC_REQUEST_OK != response->response_code) {
     response->server_status = HttpResponse::SYNC_SERVER_ERROR;
     return false;
   }
 
   if (require_response && (1 > response->content_length))
     return false;
 
   const int64 bytes_read = ReadResponse(buffer_out,
       static_cast<int>(response->content_length));
   if (bytes_read != response->content_length) {
     response->server_status = HttpResponse::IO_ERROR;
     return false;
   }
   return true;
 }
 
-bool ServerConnectionManager::Post::ReadDownloadResponse(
+bool ServerConnectionManager::Connection::ReadDownloadResponse(
     HttpResponse* response,
     string* buffer_out) {
   const int64 bytes_read = ReadResponse(buffer_out,
       static_cast<int>(response->content_length));
 
   if (bytes_read != response->content_length) {
     LOG(ERROR) << "Mismatched content lengths, server claimed " <<
         response->content_length << ", but sent " << bytes_read;
     response->server_status = HttpResponse::IO_ERROR;
     return false;
   }
   return true;
 }
 
 namespace {
 
 string StripTrailingSlash(const string& s) {
   int stripped_end_pos = s.size();
   if (s.at(stripped_end_pos - 1) == '/') {
     stripped_end_pos = stripped_end_pos - 1;
   }
 
   return s.substr(0, stripped_end_pos);
 }
 
 }  // namespace
 
 // TODO(chron): Use a GURL instead of string concatenation.
-string ServerConnectionManager::Post::MakeConnectionURL(
+string ServerConnectionManager::Connection::MakeConnectionURL(
     const string& sync_server,
     const string& path,
     bool use_ssl) const {
   string connection_url = (use_ssl ? "https://" : "http://");
   connection_url += sync_server;
   connection_url = StripTrailingSlash(connection_url);
   connection_url += path;
 
   return connection_url;
 }
 
-int ServerConnectionManager::Post::ReadResponse(string* out_buffer,
-                                                int length) {
+int ServerConnectionManager::Connection::ReadResponse(string* out_buffer,
+                                                      int length) {
   int bytes_read = buffer_.length();
   CHECK(length <= bytes_read);
   out_buffer->assign(buffer_);
   return bytes_read;
 }
 
 ScopedServerStatusWatcher::ScopedServerStatusWatcher(
     ServerConnectionManager* conn_mgr, HttpResponse* response)
     : conn_mgr_(conn_mgr),
       response_(response),
@@ skipping 18 matching lines @@
     bool use_ssl,
     const string& user_agent)
     : sync_server_(server),
       sync_server_port_(port),
       user_agent_(user_agent),
       use_ssl_(use_ssl),
       proto_sync_path_(kSyncServerSyncPath),
       get_time_path_(kSyncServerGetTimePath),
       error_count_(0),
       server_status_(HttpResponse::NONE),
-      server_reachable_(false) {
+      server_reachable_(false),
+      terminated_(false),
+      active_connection_(false) {
 }
 
 ServerConnectionManager::~ServerConnectionManager() {
 }
 
 void ServerConnectionManager::NotifyStatusChanged() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   FOR_EACH_OBSERVER(ServerConnectionEventListener, listeners_,
      OnServerConnectionEvent(
          ServerConnectionEvent(server_status_, server_reachable_)));
 }
 
 bool ServerConnectionManager::PostBufferWithCachedAuth(
     const PostBufferParams* params, ScopedServerStatusWatcher* watcher) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   string path =
       MakeSyncServerPath(proto_sync_path(), MakeSyncQueryString(client_id_));
   return PostBufferToPath(params, path, auth_token(), watcher);
 }
 
 bool ServerConnectionManager::PostBufferToPath(const PostBufferParams* params,
     const string& path, const string& auth_token,
     ScopedServerStatusWatcher* watcher) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(watcher != NULL);
 
   if (auth_token.empty()) {
     params->response->server_status = HttpResponse::SYNC_AUTH_ERROR;
     return false;
   }
 
-  scoped_ptr<Post> post(MakePost());
-  post->set_timing_info(params->timing_info);
-  bool ok = post->Init(path.c_str(), auth_token, params->buffer_in,
-                       params->response);
+  // The AutoReset is to temporarily bind active_connection_ to the local
+  // |post|. It's done this way to protect against a race between a call
+  // to Abort and assignment of active_connection_.

[akalin, 2011/08/31 07:20:39]

I don't get it.  What race?  This still seems racy, since the AutoReset goes out
of scope without the lock being held.

[tim (not reviewing), 2011/08/31 14:46:17]

We have to make sure that if terminate is called, we either have an
active_connection_ set and we're using it in here in an aborted state, or we
have not yet created it so we can bail out before we create.  The AutoReset was
so that I didn't have to add a Clear / reset... call at every early return point
from this function, which you'd have to do if you had a ClearActiveConnection. 
You could maybe just call that after Init, but we still interact with the post
after that point already so that seems weird.

An alternative is wiring up a call to Clear... in the dtor for Connection.  I
can look into that.

+  scoped_ptr<Connection> post;
+  scoped_ptr<AutoReset<Connection*> > active_connection_resetter;
+  {

[akalin, 2011/08/31 07:20:39]

I suggest making MakeConnection() (or some non-virtual wrapper function):

- Hold the lock
- Check the terminated flag, and return NULL if terminated
- Otherwise, create a post and set active connection to it
- Return the post

also have a ClearActiveConnection() helper function that:
- holds the lock
- Clears active_connection_

Then you can just do:

scoped_ptr<Connection> post(MakeActiveConnection());

if (!post.get()) {
  ...
  return false;
}

bool ok = post->Init(...);

...

+    base::AutoLock lock(terminate_connection_lock_);
+    if (terminated_) {
+      params->response->server_status = HttpResponse::CONNECTION_UNAVAILABLE;
+      return false;
+    }
+
+    DCHECK(!active_connection_);
+    post.reset(MakeConnection());
+    active_connection_resetter.reset(
+        new AutoReset<Connection*>(&active_connection_, post.get()));
+    post->set_timing_info(params->timing_info);
+  }
+
+  // Note that |post| may be aborted by now, which will just cause Init to fail
+  // with CONNECTION_UNAVAILABLE.
+  bool ok = active_connection_->Init(

[akalin, 2011/08/31 07:20:39]

clearer to do post->Init() here, I think

+      path.c_str(), auth_token, params->buffer_in, params->response);
 
   if (params->response->server_status == HttpResponse::SYNC_AUTH_ERROR) {
     InvalidateAndClearAuthToken();
   }
 
   if (!ok || RC_REQUEST_OK != params->response->response_code) {
     IncrementErrorCount();
     return false;
   }
 
   if (post->ReadBufferResponse(params->buffer_out, params->response, true)) {
     params->response->server_status = HttpResponse::SERVER_CONNECTION_OK;
     server_reachable_ = true;
     return true;
   }
   return false;
 }
 
 bool ServerConnectionManager::CheckTime(int32* out_time) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   // Verify that the server really is reachable by checking the time. We need
   // to do this because of wifi interstitials that intercept messages from the
   // client and return HTTP OK instead of a redirect.
   HttpResponse response;
   ScopedServerStatusWatcher watcher(this, &response);
   string post_body = "command=get_time";
 
   for (int i = 0 ; i < 3;  i++) {
-    scoped_ptr<Post> post(MakePost());
+    scoped_ptr<Connection> post;

[akalin, 2011/08/31 07:20:39]

similarly as above, you can just do:

scoped_ptr<Connection> post(MakeActiveConnection());
if (!post.get()) {
  break;
}

bool ok = post->Init(...);

+    scoped_ptr<AutoReset<Connection*> > active_connection_resetter;
+    {
+      base::AutoLock lock(terminate_connection_lock_);
+      if (terminated_)
+        break;
+      DCHECK(!active_connection_);
+      post.reset(MakeConnection());
+      active_connection_resetter.reset(
+          new AutoReset<Connection*>(&active_connection_, post.get()));
+    }
 
     // Note that the server's get_time path doesn't require authentication.
     string get_time_path =
         MakeSyncServerPath(kSyncServerGetTimePath, post_body);
     VLOG(1) << "Requesting get_time from:" << get_time_path;
 
     string blank_post_body;
     bool ok = post->Init(get_time_path.c_str(), blank_post_body,
         blank_post_body, &response);
     if (!ok) {
@@ skipping 12 matching lines @@
     }
     *out_time = atoi(time_response.c_str());
     VLOG(1) << "Server was reachable.";
     return true;
   }
   IncrementErrorCount();
   return false;
 }
 
 bool ServerConnectionManager::IsServerReachable() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   int32 time;
   return CheckTime(&time);
 }
 
 bool ServerConnectionManager::IsUserAuthenticated() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   return IsGoodReplyFromServer(server_status_);
 }
 
 bool ServerConnectionManager::CheckServerReachable() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   const bool server_is_reachable = IsServerReachable();
   if (server_reachable_ != server_is_reachable) {
     server_reachable_ = server_is_reachable;
     NotifyStatusChanged();
   }
   return server_is_reachable;
 }
 
 bool ServerConnectionManager::IncrementErrorCount() {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   error_count_++;
 
   if (error_count_ > kMaxConnectionErrorsBeforeReset) {
     error_count_ = 0;
 
     if (!IsServerReachable()) {
       LOG(WARNING) << "Too many connection failures, server is not reachable. "
                    << "Resetting connections.";
     } else {
       LOG(WARNING) << "Multiple connection failures while server is reachable.";
     }
     return false;
   }
 
   return true;
 }
 
 void ServerConnectionManager::SetServerParameters(const string& server_url,
                                                   int port,
                                                   bool use_ssl) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   sync_server_ = server_url;
   sync_server_port_ = port;
   use_ssl_ = use_ssl;
 }
 
 // Returns the current server parameters in server_url and port.
 void ServerConnectionManager::GetServerParameters(string* server_url,
                                                   int* port,
                                                   bool* use_ssl) const {
   if (server_url != NULL)
@@ skipping 14 matching lines @@
     return std::string();
   // We just want the hostname, so we don't need to switch on use_ssl.
   server_url = "http://" + server_url;
   GURL gurl(server_url);
   DCHECK(gurl.is_valid()) << gurl;
   return gurl.host();
 }
 
 void ServerConnectionManager::AddListener(
     ServerConnectionEventListener* listener) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   listeners_.AddObserver(listener);
 }
 
 void ServerConnectionManager::RemoveListener(
     ServerConnectionEventListener* listener) {
-  DCHECK(CalledOnValidThread());
+  DCHECK(thread_checker_.CalledOnValidThread());
   listeners_.RemoveObserver(listener);
 }
 
-ServerConnectionManager::Post* ServerConnectionManager::MakePost() {
+ServerConnectionManager::Connection* ServerConnectionManager::MakeConnection()
+{
   return NULL;  // For testing.
 }
 
+void ServerConnectionManager::TerminateAllIO() {
+  base::AutoLock lock(terminate_connection_lock_);
+  terminated_ = true;
+  if (active_connection_)
+    active_connection_->Abort();
+
+  // Sever our ties to this connection object. Note that it still may exist,
+  // since we don't own it, but it has been neutered.
+  active_connection_ = NULL;
+}
+
 bool FillMessageWithShareDetails(sync_pb::ClientToServerMessage* csm,
                                  syncable::DirectoryManager* manager,
                                  const std::string& share) {
   syncable::ScopedDirLookup dir(manager, share);
   if (!dir.good()) {
     VLOG(1) << "Dir lookup failed";
     return false;
   }
   string birthday = dir->store_birthday();
   if (!birthday.empty())
     csm->set_store_birthday(birthday);
   csm->set_share(share);
   return true;
 }
 
 std::ostream& operator << (std::ostream& s, const struct HttpResponse& hr) {
   s << " Response Code (bogus on error): " << hr.response_code;
   s << " Content-Length (bogus on error): " << hr.content_length;
   s << " Server Status: " << hr.server_status;
   return s;
 }
 
 }  // namespace browser_sync