Reporting of policy errors via host-offline-reason: part 3

remoting/host/remoting_me2me_host.cc

 // Copyright (c) 2012 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.
 //
 // This file implements a standalone host process for Me2Me.
 
 #include <string>
 
 #include "base/at_exit.h"
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/command_line.h"
 #include "base/debug/alias.h"
 #include "base/files/file_path.h"
 #include "base/files/file_util.h"
 #include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
 #include "base/message_loop/message_loop.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/string_util.h"
 #include "base/strings/utf_string_conversions.h"
 #include "base/threading/thread.h"
 #include "build/build_config.h"
 #include "crypto/nss_util.h"
 #include "ipc/ipc_channel.h"
 #include "ipc/ipc_channel_proxy.h"
 #include "ipc/ipc_listener.h"
 #include "media/base/media.h"
-#include "net/base/network_change_notifier.h"
 #include "net/socket/client_socket_factory.h"
 #include "net/socket/ssl_server_socket.h"
 #include "net/url_request/url_fetcher.h"
 #include "remoting/base/auto_thread_task_runner.h"
 #include "remoting/base/breakpad.h"
 #include "remoting/base/constants.h"
 #include "remoting/base/logging.h"
 #include "remoting/base/rsa_key_pair.h"
 #include "remoting/base/service_urls.h"
 #include "remoting/base/util.h"
 #include "remoting/host/branding.h"
 #include "remoting/host/chromoting_host.h"
 #include "remoting/host/chromoting_host_context.h"
 #include "remoting/host/chromoting_messages.h"
 #include "remoting/host/config_file_watcher.h"
 #include "remoting/host/config_watcher.h"
 #include "remoting/host/desktop_environment.h"
 #include "remoting/host/desktop_session_connector.h"
-#include "remoting/host/dns_blackhole_checker.h"
-#include "remoting/host/heartbeat_sender.h"
 #include "remoting/host/host_change_notification_listener.h"
 #include "remoting/host/host_config.h"
 #include "remoting/host/host_event_logger.h"
 #include "remoting/host/host_exit_codes.h"
 #include "remoting/host/host_main.h"
 #include "remoting/host/host_status_logger.h"
 #include "remoting/host/ipc_constants.h"
 #include "remoting/host/ipc_desktop_environment.h"
 #include "remoting/host/ipc_host_event_logger.h"
 #include "remoting/host/logging.h"
 #include "remoting/host/me2me_desktop_environment.h"
+#include "remoting/host/minimum_heartbeat_supporter.h"
 #include "remoting/host/pairing_registry_delegate.h"
 #include "remoting/host/policy_hack/policy_watcher.h"
 #include "remoting/host/session_manager_factory.h"
-#include "remoting/host/signaling_connector.h"
 #include "remoting/host/single_window_desktop_environment.h"
 #include "remoting/host/token_validator_factory_impl.h"
 #include "remoting/host/usage_stats_consent.h"
 #include "remoting/host/username.h"
 #include "remoting/host/video_frame_recorder_host_extension.h"
 #include "remoting/protocol/me2me_host_authenticator_factory.h"
 #include "remoting/protocol/network_settings.h"
 #include "remoting/protocol/pairing_registry.h"
 #include "remoting/protocol/token_validator.h"
 #include "remoting/signaling/xmpp_signal_strategy.h"
@@ skipping 158 matching lines @@
   bool OnUsernamePolicyUpdate(base::DictionaryValue* policies);
   bool OnNatPolicyUpdate(base::DictionaryValue* policies);
   bool OnRelayPolicyUpdate(base::DictionaryValue* policies);
   bool OnUdpPortPolicyUpdate(base::DictionaryValue* policies);
   bool OnCurtainPolicyUpdate(base::DictionaryValue* policies);
   bool OnHostTalkGadgetPrefixPolicyUpdate(base::DictionaryValue* policies);
   bool OnHostTokenUrlPolicyUpdate(base::DictionaryValue* policies);
   bool OnPairingPolicyUpdate(base::DictionaryValue* policies);
   bool OnGnubbyAuthPolicyUpdate(base::DictionaryValue* policies);
 
+  scoped_refptr<MinimumHeartbeatSupporter> CreateMinimumHeartbeatSupporter();
+
   void StartHost();
 
   void OnHeartbeatSuccessful();
   void OnUnknownHostIdError();
 
   void OnAuthFailed();
 
   void RestartHost();
 
   // Stops the host and shuts down the process with the specified |exit_code|.
   void ShutdownHost(HostExitCodes exit_code);
 
-  void ScheduleHostShutdown();
-
   void ShutdownOnNetworkThread();
 
   void OnPolicyWatcherShutdown();
 
   // Crashes the process in response to a daemon's request. The daemon passes
   // the location of the code that detected the fatal error resulted in this
   // request.
   void OnCrash(const std::string& function_name,
                const std::string& file_name,
                const int& line_number);
 
   scoped_ptr<ChromotingHostContext> context_;
 
-  // Created on the UI thread but used from the network thread.
-  scoped_ptr<net::NetworkChangeNotifier> network_change_notifier_;
-
   // Accessed on the UI thread.
   scoped_ptr<IPC::ChannelProxy> daemon_channel_;
 
   // XMPP server/remoting bot configuration (initialized from the command line).
   XmppSignalStrategy::XmppServerConfig xmpp_server_config_;
   std::string directory_bot_jid_;
 
   // Created on the UI thread but used from the network thread.
   base::FilePath host_config_path_;
   std::string host_config_;
@@ skipping 29 matching lines @@
   ThirdPartyAuthConfig third_party_auth_config_;
   bool enable_gnubby_auth_;
 
   // Boolean to change flow, where ncessary, if we're
   // capturing a window instead of the entire desktop.
   bool enable_window_capture_;
 
   // Used to specify which window to stream, if enabled.
   webrtc::WindowId window_id_;
 
-  scoped_ptr<OAuthTokenGetter> oauth_token_getter_;
-  scoped_ptr<XmppSignalStrategy> signal_strategy_;
-  scoped_ptr<SignalingConnector> signaling_connector_;
-  scoped_ptr<HeartbeatSender> heartbeat_sender_;
+  scoped_refptr<MinimumHeartbeatSupporter> minimum_heartbeat_supporter_;
   scoped_ptr<HostChangeNotificationListener> host_change_notification_listener_;
   scoped_ptr<HostStatusLogger> host_status_logger_;
   scoped_ptr<HostEventLogger> host_event_logger_;
 
   scoped_ptr<ChromotingHost> host_;
 
   // Used to keep this HostProcess alive until it is shutdown.
   scoped_refptr<HostProcess> self_;
 
 #if defined(REMOTING_MULTI_PROCESS)
   DesktopSessionConnector* desktop_session_connector_;
 #endif  // defined(REMOTING_MULTI_PROCESS)
 
   int* exit_code_out_;
   bool signal_parent_;
 
   scoped_ptr<PairingRegistry::Delegate> pairing_registry_delegate_;
+
+  // Factory of weak pointers to |this| that will to be referenced from the
+  // network thread (weak pointers are constrained to a single thread).
+  base::WeakPtrFactory<HostProcess> weak_factory_for_network_task_runner_;

[Wez, 2014/12/09 00:51:58]

This should just be |weak_factory_| and the comment needn't describe the
threading restriction on WeakPtrFactory, since that's inherent in what WeakPtrs
are.

[Łukasz Anforowicz, 2014/12/09 18:47:07]

Maybe.  This is what I started with, but I wasn't explicitly destroying the
factory, meaning that the thread asserts in the factory were triggered.  When
fixing this and adding explicit destroying of the factory, I changed its name,
so reader of ShutdownOnNetworkThread can more easily understand why this
weak_factory is being destroyed in ShutdownOnNetworkThread and not elsewhere.

 };

[Wez, 2014/12/09 00:51:58]

While you're here this should have DISALLOW_COPY_AND_ASSIGN

[Łukasz Anforowicz, 2014/12/09 18:47:07]

Ok.

 
 HostProcess::HostProcess(scoped_ptr<ChromotingHostContext> context,
                          int* exit_code_out)
     : context_(context.Pass()),
       state_(HOST_INITIALIZING),
       use_service_account_(false),
       enable_vp9_(false),
       frame_recorder_buffer_size_(0),
       host_username_match_required_(false),
       allow_nat_traversal_(true),
       allow_relay_(true),
       min_udp_port_(0),
       max_udp_port_(0),
       allow_pairing_(true),
       curtain_required_(false),
       enable_gnubby_auth_(false),
       enable_window_capture_(false),
       window_id_(0),
 #if defined(REMOTING_MULTI_PROCESS)
       desktop_session_connector_(NULL),
 #endif  // defined(REMOTING_MULTI_PROCESS)
       self_(this),
       exit_code_out_(exit_code_out),
-      signal_parent_(false) {
+      signal_parent_(false),
+      weak_factory_for_network_task_runner_(this) {
   StartOnUiThread();
 }
 
 HostProcess::~HostProcess() {
   // Verify that UI components have been torn down.
   DCHECK(!config_watcher_);
   DCHECK(!daemon_channel_);
   DCHECK(!desktop_environment_factory_);
 
   // We might be getting deleted on one of the threads the |host_context| owns,
@@ skipping 361 matching lines @@
 
   context_->network_task_runner()->PostTask(
       FROM_HERE,
       base::Bind(&HostProcess::StartOnNetworkThread, this));
 }
 
 void HostProcess::ShutdownOnUiThread() {
   DCHECK(context_->ui_task_runner()->BelongsToCurrentThread());
 
   // Tear down resources that need to be torn down on the UI thread.
-  network_change_notifier_.reset();
   daemon_channel_.reset();
   desktop_environment_factory_.reset();
 
   // It is now safe for the HostProcess to be deleted.
   self_ = NULL;
 
 #if defined(OS_LINUX)
   // Cause the global AudioPipeReader to be freed, otherwise the audio
   // thread will remain in-use and prevent the process from exiting.
   // TODO(wez): DesktopEnvironmentFactory should own the pipe reader.
   // See crbug.com/161373 and crbug.com/104544.
   AudioCapturerLinux::InitializePipeReader(NULL, base::FilePath());
 #endif
 }
 
-// Overridden from HeartbeatSender::Listener
 void HostProcess::OnUnknownHostIdError() {
   LOG(ERROR) << "Host ID not found.";
   ShutdownHost(kInvalidHostIdExitCode);
 }
 
 void HostProcess::OnHeartbeatSuccessful() {
   HOST_LOG << "Host ready to receive connections.";
 #if defined(OS_POSIX)
   if (signal_parent_) {
     kill(getppid(), SIGUSR1);
@@ skipping 66 matching lines @@
   if (!config.GetString(kXmppLoginConfigPath, &xmpp_server_config_.username) ||
       !(config.GetString(kXmppAuthTokenConfigPath,
                          &xmpp_server_config_.auth_token) ||
         config.GetString(kOAuthRefreshTokenConfigPath,
                          &oauth_refresh_token_))) {
     LOG(ERROR) << "XMPP credentials are not defined in the config.";
     return false;
   }
 
   if (!oauth_refresh_token_.empty()) {
-    // SignalingConnector is responsible for getting OAuth token.
+    // SignalingConnector (inside MinimumHeartbeatSupporter) is responsible for
+    // getting OAuth token.
     xmpp_server_config_.auth_token = "";
     xmpp_server_config_.auth_service = "oauth2";
   } else if (!config.GetString(kXmppAuthServiceConfigPath,
                                &xmpp_server_config_.auth_service)) {
     // For the me2me host, we default to ClientLogin token for chromiumsync
     // because earlier versions of the host had no HTTP stack with which to
     // request an OAuth2 access token.
     xmpp_server_config_.auth_service = kChromotingTokenDefaultServiceName;
   }
 
@@ skipping 359 matching lines @@
   } else {
     HOST_LOG << "Policy disables gnubby auth.";
   }
 
   if (desktop_environment_factory_)
     desktop_environment_factory_->SetEnableGnubbyAuth(enable_gnubby_auth_);
 
   return true;
 }
 
+scoped_refptr<MinimumHeartbeatSupporter>
+HostProcess::CreateMinimumHeartbeatSupporter() {
+  DCHECK(!host_id_.empty());  // Assumming |ApplyConfig| has already been run.
+  return MinimumHeartbeatSupporter::Create(
+      base::Bind(&HostProcess::OnHeartbeatSuccessful,
+                 weak_factory_for_network_task_runner_.GetWeakPtr()),
+      base::Bind(&HostProcess::OnUnknownHostIdError,
+                 weak_factory_for_network_task_runner_.GetWeakPtr()),
+      base::Bind(&HostProcess::OnAuthFailed,
+                 weak_factory_for_network_task_runner_.GetWeakPtr()),

[Wez, 2014/12/09 00:51:58]

Why are we passing three Callbacks here rather than encapsulating the
functionality as an interface that HostProcess can then implement?

[Łukasz Anforowicz, 2014/12/09 18:47:07]

We rely on WeakPtrs translating the callbacks into no-ops when the HostProcess
is no longer alive.  Doing this via interface is possible, but I think not as
clean (i.e. MinimumHeartbeatSupporter cannot just pass through the callbacks to
other objects, but has to have a flag indicating whether the listener is still
alive + code to forward the callbacks/listeners from encapsulated objects - look
at patchset#1 to see how it would look like [i.e. see SetListener method on
MinimumHeartbeatSupporter + forwarding methods in implementation of
MinimumHeartbeatSupporter class];  I think I prefer current code to patchset#1).

+      *context_, xmpp_server_config_, talkgadget_prefix_, host_id_, key_pair_,
+      directory_bot_jid_, oauth_refresh_token_, use_service_account_);
+}
+
 void HostProcess::StartHost() {
   DCHECK(context_->network_task_runner()->BelongsToCurrentThread());
   DCHECK(!host_);
-  DCHECK(!signal_strategy_.get());
+  DCHECK(!minimum_heartbeat_supporter_.get());

[Wez, 2014/12/09 00:51:58]

nit: Remove the .get()

[Łukasz Anforowicz, 2014/12/09 18:47:07]

Done.

+
   DCHECK(state_ == HOST_INITIALIZING || state_ == HOST_STOPPING_TO_RESTART ||
          state_ == HOST_STOPPED) << state_;
   state_ = HOST_STARTED;
 
-  signal_strategy_.reset(
-      new XmppSignalStrategy(net::ClientSocketFactory::GetDefaultFactory(),
-                             context_->url_request_context_getter(),
-                             xmpp_server_config_));
-
-  scoped_ptr<DnsBlackholeChecker> dns_blackhole_checker(
-      new DnsBlackholeChecker(context_->url_request_context_getter(),
-                              talkgadget_prefix_));
-
-  // Create a NetworkChangeNotifier for use by the signaling connector.
-  network_change_notifier_.reset(net::NetworkChangeNotifier::Create());
-
-  signaling_connector_.reset(new SignalingConnector(
-      signal_strategy_.get(),
-      dns_blackhole_checker.Pass(),
-      base::Bind(&HostProcess::OnAuthFailed, this)));
-
-  if (!oauth_refresh_token_.empty()) {
-    scoped_ptr<OAuthTokenGetter::OAuthCredentials> oauth_credentials;
-    oauth_credentials.reset(
-        new OAuthTokenGetter::OAuthCredentials(
-            xmpp_server_config_.username, oauth_refresh_token_,
-            use_service_account_));
-
-    oauth_token_getter_.reset(new OAuthTokenGetter(
-        oauth_credentials.Pass(), context_->url_request_context_getter(),
-        false));
-
-    signaling_connector_->EnableOAuth(oauth_token_getter_.get());
-  }
+  minimum_heartbeat_supporter_ = CreateMinimumHeartbeatSupporter();
 
   uint32 network_flags = 0;
   if (allow_nat_traversal_) {
     network_flags = NetworkSettings::NAT_TRAVERSAL_STUN |
                     NetworkSettings::NAT_TRAVERSAL_OUTGOING;
     if (allow_relay_)
       network_flags |= NetworkSettings::NAT_TRAVERSAL_RELAY;
   }
 
   NetworkSettings network_settings(network_flags);
 
   if (min_udp_port_ && max_udp_port_) {
     network_settings.min_port = min_udp_port_;
     network_settings.max_port = max_udp_port_;
   } else if (!allow_nat_traversal_) {
     // For legacy reasons we have to restrict the port range to a set of default
     // values when nat traversal is disabled, even if the port range was not
     // set in policy.
     network_settings.min_port = NetworkSettings::kDefaultMinPort;
     network_settings.max_port = NetworkSettings::kDefaultMaxPort;
   }
 
   host_.reset(new ChromotingHost(
-      signal_strategy_.get(),
+      minimum_heartbeat_supporter_->signal_strategy(),
       desktop_environment_factory_.get(),
-      CreateHostSessionManager(signal_strategy_.get(), network_settings,
+      CreateHostSessionManager(minimum_heartbeat_supporter_->signal_strategy(),
+                               network_settings,
                                context_->url_request_context_getter()),
-      context_->audio_task_runner(),
-      context_->input_task_runner(),
+      context_->audio_task_runner(), context_->input_task_runner(),
       context_->video_capture_task_runner(),
-      context_->video_encode_task_runner(),
-      context_->network_task_runner(),
+      context_->video_encode_task_runner(), context_->network_task_runner(),
       context_->ui_task_runner()));
 
   if (enable_vp9_) {
     scoped_ptr<protocol::CandidateSessionConfig> config =
         host_->protocol_config()->Clone();
     config->EnableVideoCodec(protocol::ChannelConfig::CODEC_VP9);
     host_->set_protocol_config(config.Pass());
   }
 
   if (frame_recorder_buffer_size_ > 0) {
     scoped_ptr<VideoFrameRecorderHostExtension> frame_recorder_extension(
         new VideoFrameRecorderHostExtension());
     frame_recorder_extension->SetMaxContentBytes(frame_recorder_buffer_size_);
     host_->AddExtension(frame_recorder_extension.Pass());
   }
 
   // TODO(simonmorris): Get the maximum session duration from a policy.
 #if defined(OS_LINUX)
   host_->SetMaximumSessionDuration(base::TimeDelta::FromHours(20));
 #endif
 
-  heartbeat_sender_.reset(new HeartbeatSender(
-      base::Bind(&HostProcess::OnHeartbeatSuccessful, base::Unretained(this)),
-      base::Bind(&HostProcess::OnUnknownHostIdError, base::Unretained(this)),
-      host_id_, signal_strategy_.get(), key_pair_, directory_bot_jid_));
+  host_change_notification_listener_.reset(new HostChangeNotificationListener(
+      this, host_id_, minimum_heartbeat_supporter_->signal_strategy(),
+      directory_bot_jid_));
 
-  host_change_notification_listener_.reset(new HostChangeNotificationListener(
-      this, host_id_, signal_strategy_.get(), directory_bot_jid_));
-
-  host_status_logger_.reset(
-      new HostStatusLogger(host_->AsWeakPtr(), ServerLogEntry::ME2ME,
-                           signal_strategy_.get(), directory_bot_jid_));
+  host_status_logger_.reset(new HostStatusLogger(
+      host_->AsWeakPtr(), ServerLogEntry::ME2ME,
+      minimum_heartbeat_supporter_->signal_strategy(), directory_bot_jid_));
 
   // Set up reporting the host status notifications.
 #if defined(REMOTING_MULTI_PROCESS)
   host_event_logger_.reset(
       new IpcHostEventLogger(host_->AsWeakPtr(), daemon_channel_.get()));
 #else  // !defined(REMOTING_MULTI_PROCESS)
   host_event_logger_ =
       HostEventLogger::Create(host_->AsWeakPtr(), kApplicationName);
 #endif  // !defined(REMOTING_MULTI_PROCESS)
 
@@ skipping 16 matching lines @@
 }
 
 void HostProcess::ShutdownHost(HostExitCodes exit_code) {
   DCHECK(context_->network_task_runner()->BelongsToCurrentThread());
 
   *exit_code_out_ = exit_code;
 
   switch (state_) {
     case HOST_INITIALIZING:
       state_ = HOST_STOPPING;
+      {
+        DCHECK(!minimum_heartbeat_supporter_.get());
+        scoped_refptr<MinimumHeartbeatSupporter> tmpHeartbeatSupporter =
+            CreateMinimumHeartbeatSupporter();
+        tmpHeartbeatSupporter->SendHostOfflineReason(
+            ExitCodeToString(exit_code),
+            base::TimeDelta::FromSeconds(10) /* timeout */);
+      }
       ShutdownOnNetworkThread();
       break;
 
     case HOST_STARTED:
       state_ = HOST_STOPPING;
-      heartbeat_sender_->SetHostOfflineReason(
-          ExitCodeToString(exit_code), base::Bind(base::DoNothing));
-      ScheduleHostShutdown();
+      minimum_heartbeat_supporter_->SendHostOfflineReason(
+          ExitCodeToString(exit_code),
+          base::TimeDelta::FromSeconds(10) /* timeout */);
+      ShutdownOnNetworkThread();
       break;
 
     case HOST_STOPPING_TO_RESTART:
       state_ = HOST_STOPPING;
       break;
 
     case HOST_STOPPING:
     case HOST_STOPPED:
       // Host is already stopped or being stopped. No action is required.
       break;
   }
 }
 
-// TODO(weitaosu): shut down the host once we get an ACK for the offline status
-//                  XMPP message.
-void HostProcess::ScheduleHostShutdown() {
-  // Delay the shutdown by 2 second to allow SendOfflineStatus to complete.
-  context_->network_task_runner()->PostDelayedTask(
-      FROM_HERE,
-      base::Bind(&HostProcess::ShutdownOnNetworkThread, base::Unretained(this)),
-      base::TimeDelta::FromSeconds(2));
-}
-
 void HostProcess::ShutdownOnNetworkThread() {
   DCHECK(context_->network_task_runner()->BelongsToCurrentThread());
 
+  // Need to invalidate WeakPtrs on the same thread where they are dereferenced.

[Wez, 2014/12/09 00:51:58]

nit: This is always a requirement for invalidating them, not an implementation
limitation, so not really worth documenting here.

+  weak_factory_for_network_task_runner_.InvalidateWeakPtrs();
+
   host_.reset();
   host_event_logger_.reset();
   host_status_logger_.reset();
-  heartbeat_sender_.reset();
+  minimum_heartbeat_supporter_ = nullptr;

[Wez, 2014/12/09 00:51:58]

Why are you NULLing this here?

   host_change_notification_listener_.reset();
-  signaling_connector_.reset();
-  oauth_token_getter_.reset();
-  signal_strategy_.reset();
-  network_change_notifier_.reset();
 
   if (state_ == HOST_STOPPING_TO_RESTART) {
     StartHost();
   } else if (state_ == HOST_STOPPING) {
     state_ = HOST_STOPPED;
 
     if (policy_watcher_.get()) {
       policy_watcher_->StopWatching(
           base::Bind(&HostProcess::OnPolicyWatcherShutdown, this));
     } else {
@@ skipping 58 matching lines @@
   int exit_code = kSuccessExitCode;
   new HostProcess(context.Pass(), &exit_code);
 
   // Run the main (also UI) message loop until the host no longer needs it.
   message_loop.Run();
 
   return exit_code;
 }
 
 }  // namespace remoting