Fix ChannelMultiplexer to properly handle base channel creation failure.

remoting/protocol/channel_multiplexer.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.
 
 #include "remoting/protocol/channel_multiplexer.h"
 
 #include <string.h>
 
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/location.h"
+#include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
+#include "base/thread_task_runner_handle.h"
 #include "net/base/net_errors.h"
 #include "net/socket/stream_socket.h"
 #include "remoting/protocol/util.h"
 
 namespace remoting {
 namespace protocol {
 
 namespace {
 const int kChannelIdUnknown = -1;
 const int kMaxPacketSize = 1024;
@@ skipping 334 matching lines @@
     std::swap(cb, read_callback_);
     cb.Run(result);
   }
 }
 
 ChannelMultiplexer::ChannelMultiplexer(ChannelFactory* factory,
                                        const std::string& base_channel_name)
     : base_channel_factory_(factory),
       base_channel_name_(base_channel_name),
       next_channel_id_(0),
-      destroyed_flag_(NULL) {
+      ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {
 }
 
 ChannelMultiplexer::~ChannelMultiplexer() {
   DCHECK(pending_channels_.empty());
   STLDeleteValues(&channels_);
 
   // Cancel creation of the base channel if it hasn't finished.
   if (base_channel_factory_)
     base_channel_factory_->CancelChannelCreation(base_channel_name_);
-
-  if (destroyed_flag_)
-    *destroyed_flag_ = true;
 }
 
 void ChannelMultiplexer::CreateStreamChannel(
     const std::string& name,
     const StreamChannelCallback& callback) {
   if (base_channel_.get()) {
     // Already have |base_channel_|. Create new multiplexed channel
     // synchronously.
     callback.Run(GetOrCreateChannel(name)->CreateSocket());
   } else if (!base_channel_.get() && !base_channel_factory_) {
@@ skipping 28 matching lines @@
       return;
     }
   }
 }
 
 void ChannelMultiplexer::OnBaseChannelReady(
     scoped_ptr<net::StreamSocket> socket) {
   base_channel_factory_ = NULL;
   base_channel_ = socket.Pass();
 
-  if (!base_channel_.get()) {
-    // Notify all callers that we can't create any channels.
-    for (std::list<PendingChannel>::iterator it = pending_channels_.begin();
-         it != pending_channels_.end(); ++it) {
-      it->callback.Run(scoped_ptr<net::StreamSocket>());
-    }
-    pending_channels_.clear();
-    return;
+  if (base_channel_.get()) {
+    // Initialize reader and writer.
+    reader_.Init(base_channel_.get(),
+                 base::Bind(&ChannelMultiplexer::OnIncomingPacket,
+                            base::Unretained(this)));
+    writer_.Init(base_channel_.get(),
+                 base::Bind(&ChannelMultiplexer::OnWriteFailed,
+                            base::Unretained(this)));
   }
 
-  // Initialize reader and writer.
-  reader_.Init(base_channel_.get(),
-               base::Bind(&ChannelMultiplexer::OnIncomingPacket,
-                          base::Unretained(this)));
-  writer_.Init(base_channel_.get(),
-               base::Bind(&ChannelMultiplexer::OnWriteFailed,
-                          base::Unretained(this)));
+  DoCreatePendingChannels();
+}
 
-  // Now create all pending channels.
-  for (std::list<PendingChannel>::iterator it = pending_channels_.begin();
-       it != pending_channels_.end(); ++it) {
-    it->callback.Run(GetOrCreateChannel(it->name)->CreateSocket());
-  }
-  pending_channels_.clear();
+void ChannelMultiplexer::DoCreatePendingChannels() {
+  if (pending_channels_.empty())
+    return;
+
+  // Every time this function is called it connects a single channel and posts a

[Wez, 2012/09/25 23:34:15]

typo: Remove the "a"

[Sergey Ulanov, 2012/09/26 00:46:45]

replaced "tasks" with "task" instead. 

+  // separate tasks to connect other channels. This is necessary because the
+  // callback may destroy the multiplexer or somehow else modify
+  // |pending_channels_| list (e.g. call CancelChannelCreation()).
+  base::ThreadTaskRunnerHandle::Get()->PostTask(
+      FROM_HERE, base::Bind(&ChannelMultiplexer::DoCreatePendingChannels,
+                            weak_factory_.GetWeakPtr()));
+
+  PendingChannel c = pending_channels_.front();
+  pending_channels_.erase(pending_channels_.begin());
+  scoped_ptr<net::StreamSocket> socket;
+  if (base_channel_.get())
+    socket = GetOrCreateChannel(c.name)->CreateSocket();
+  c.callback.Run(socket.Pass());
 }
 
+
 ChannelMultiplexer::MuxChannel* ChannelMultiplexer::GetOrCreateChannel(
     const std::string& name) {
   // Check if we already have a channel with the requested name.
   std::map<std::string, MuxChannel*>::iterator it = channels_.find(name);
   if (it != channels_.end())
     return it->second;
 
   // Create a new channel if we haven't found existing one.
   MuxChannel* channel = new MuxChannel(this, name, next_channel_id_);
   ++next_channel_id_;
   channels_[channel->name()] = channel;
   return channel;
 }
 
 
 void ChannelMultiplexer::OnWriteFailed(int error) {
-  bool destroyed = false;
-  destroyed_flag_ = &destroyed;
   for (std::map<std::string, MuxChannel*>::iterator it = channels_.begin();
        it != channels_.end(); ++it) {
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE, base::Bind(&ChannelMultiplexer::NotifyWriteFailed,
+                              weak_factory_.GetWeakPtr(), it->second->name()));
+  }
+}
+
+void ChannelMultiplexer::NotifyWriteFailed(const std::string& name) {
+  std::map<std::string, MuxChannel*>::iterator it = channels_.find(name);
+  if (it != channels_.end()) {
     it->second->OnWriteFailed();
-    if (destroyed)
-      return;
   }
-  destroyed_flag_ = NULL;
 }
 
 void ChannelMultiplexer::OnIncomingPacket(scoped_ptr<MultiplexPacket> packet,
                                           const base::Closure& done_task) {
   if (!packet->has_channel_id()) {
     LOG(ERROR) << "Received packet without channel_id.";
     done_task.Run();
     return;
   }
 
@@ skipping 19 matching lines @@
   channel->OnIncomingPacket(packet.Pass(), done_task);
 }
 
 bool ChannelMultiplexer::DoWrite(scoped_ptr<MultiplexPacket> packet,
                                  const base::Closure& done_task) {
   return writer_.Write(SerializeAndFrameMessage(*packet), done_task);
 }
 
 }  // namespace protocol
 }  // namespace remoting