remoting/protocol/channel_multiplexer.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "remoting/protocol/channel_multiplexer.h"
   6
   7 #include <string.h>
   8
   9 #include "base/bind.h"
  10 #include "base/callback.h"
  11 #include "base/location.h"
  12 #include "base/single_thread_task_runner.h"
  13 #include "base/stl_util.h"
  14 #include "base/thread_task_runner_handle.h"
  15 #include "net/base/net_errors.h"
  16 #include "net/socket/stream_socket.h"
  17 #include "remoting/protocol/message_serialization.h"
  18
  19 namespace remoting {
  20 namespace protocol {
  21
  22 namespace {
  23 const int kChannelIdUnknown = -1;
  24 const int kMaxPacketSize = 1024;
  25
  26 class PendingPacket {
  27  public:
  28   PendingPacket(scoped_ptr<MultiplexPacket> packet,
  29                 const base::Closure& done_task)
  30       : packet(packet.Pass()),
  31         done_task(done_task),
  32         pos(0U) {
  33   }
  34   ~PendingPacket() {
  35     done_task.Run();
  36   }
  37
  38   bool is_empty() { return pos >= packet->data().size(); }
  39
  40   int Read(char* buffer, size_t size) {
  41     size = std::min(size, packet->data().size() - pos);
  42     memcpy(buffer, packet->data().data() + pos, size);
  43     pos += size;
  44     return size;
  45   }
  46
  47  private:
  48   scoped_ptr<MultiplexPacket> packet;
  49   base::Closure done_task;
  50   size_t pos;
  51
  52   DISALLOW_COPY_AND_ASSIGN(PendingPacket);
  53 };
  54
  55 }  // namespace
  56
  57 const char ChannelMultiplexer::kMuxChannelName[] = "mux";
  58
  59 struct ChannelMultiplexer::PendingChannel {
  60   PendingChannel(const std::string& name,
  61                  const ChannelCreatedCallback& callback)
  62       : name(name), callback(callback) {
  63   }
  64   std::string name;
  65   ChannelCreatedCallback callback;
  66 };
  67
  68 class ChannelMultiplexer::MuxChannel {
  69  public:
  70   MuxChannel(ChannelMultiplexer* multiplexer, const std::string& name,
  71              int send_id);
  72   ~MuxChannel();
  73
  74   const std::string& name() { return name_; }
  75   int receive_id() { return receive_id_; }
  76   void set_receive_id(int id) { receive_id_ = id; }
  77
  78   // Called by ChannelMultiplexer.
  79   scoped_ptr<net::StreamSocket> CreateSocket();
  80   void OnIncomingPacket(scoped_ptr<MultiplexPacket> packet,
  81                         const base::Closure& done_task);
  82   void OnWriteFailed();
  83
  84   // Called by MuxSocket.
  85   void OnSocketDestroyed();
  86   bool DoWrite(scoped_ptr<MultiplexPacket> packet,
  87                const base::Closure& done_task);
  88   int DoRead(net::IOBuffer* buffer, int buffer_len);
  89
  90  private:
  91   ChannelMultiplexer* multiplexer_;
  92   std::string name_;
  93   int send_id_;
  94   bool id_sent_;
  95   int receive_id_;
  96   MuxSocket* socket_;
  97   std::list<PendingPacket*> pending_packets_;
  98
  99   DISALLOW_COPY_AND_ASSIGN(MuxChannel);
 100 };
 101
 102 class ChannelMultiplexer::MuxSocket : public net::StreamSocket,
 103                                       public base::NonThreadSafe,
 104                                       public base::SupportsWeakPtr<MuxSocket> {
 105  public:
 106   MuxSocket(MuxChannel* channel);
 107   ~MuxSocket() override;
 108
 109   void OnWriteComplete();
 110   void OnWriteFailed();
 111   void OnPacketReceived();
 112
 113   // net::StreamSocket interface.
 114   int Read(net::IOBuffer* buffer,
 115            int buffer_len,
 116            const net::CompletionCallback& callback) override;
 117   int Write(net::IOBuffer* buffer,
 118             int buffer_len,
 119             const net::CompletionCallback& callback) override;
 120
 121   int SetReceiveBufferSize(int32 size) override {
 122     NOTIMPLEMENTED();
 123     return net::ERR_NOT_IMPLEMENTED;
 124   }
 125   int SetSendBufferSize(int32 size) override {
 126     NOTIMPLEMENTED();
 127     return net::ERR_NOT_IMPLEMENTED;
 128   }
 129
 130   int Connect(const net::CompletionCallback& callback) override {
 131     NOTIMPLEMENTED();
 132     return net::ERR_NOT_IMPLEMENTED;
 133   }
 134   void Disconnect() override { NOTIMPLEMENTED(); }
 135   bool IsConnected() const override {
 136     NOTIMPLEMENTED();
 137     return true;
 138   }
 139   bool IsConnectedAndIdle() const override {
 140     NOTIMPLEMENTED();
 141     return false;
 142   }
 143   int GetPeerAddress(net::IPEndPoint* address) const override {
 144     NOTIMPLEMENTED();
 145     return net::ERR_NOT_IMPLEMENTED;
 146   }
 147   int GetLocalAddress(net::IPEndPoint* address) const override {
 148     NOTIMPLEMENTED();
 149     return net::ERR_NOT_IMPLEMENTED;
 150   }
 151   const net::BoundNetLog& NetLog() const override {
 152     NOTIMPLEMENTED();
 153     return net_log_;
 154   }
 155   void SetSubresourceSpeculation() override { NOTIMPLEMENTED(); }
 156   void SetOmniboxSpeculation() override { NOTIMPLEMENTED(); }
 157   bool WasEverUsed() const override { return true; }
 158   bool UsingTCPFastOpen() const override { return false; }
 159   bool WasNpnNegotiated() const override { return false; }
 160   net::NextProto GetNegotiatedProtocol() const override {
 161     return net::kProtoUnknown;
 162   }
 163   bool GetSSLInfo(net::SSLInfo* ssl_info) override {
 164     NOTIMPLEMENTED();
 165     return false;
 166   }
 167
 168  private:
 169   MuxChannel* channel_;
 170
 171   net::CompletionCallback read_callback_;
 172   scoped_refptr<net::IOBuffer> read_buffer_;
 173   int read_buffer_size_;
 174
 175   bool write_pending_;
 176   int write_result_;
 177   net::CompletionCallback write_callback_;
 178
 179   net::BoundNetLog net_log_;
 180
 181   DISALLOW_COPY_AND_ASSIGN(MuxSocket);
 182 };
 183
 184
 185 ChannelMultiplexer::MuxChannel::MuxChannel(
 186     ChannelMultiplexer* multiplexer,
 187     const std::string& name,
 188     int send_id)
 189     : multiplexer_(multiplexer),
 190       name_(name),
 191       send_id_(send_id),
 192       id_sent_(false),
 193       receive_id_(kChannelIdUnknown),
 194       socket_(nullptr) {
 195 }
 196
 197 ChannelMultiplexer::MuxChannel::~MuxChannel() {
 198   // Socket must be destroyed before the channel.
 199   DCHECK(!socket_);
 200   STLDeleteElements(&pending_packets_);
 201 }
 202
 203 scoped_ptr<net::StreamSocket> ChannelMultiplexer::MuxChannel::CreateSocket() {
 204   DCHECK(!socket_);  // Can't create more than one socket per channel.
 205   scoped_ptr<MuxSocket> result(new MuxSocket(this));
 206   socket_ = result.get();
 207   return result.Pass();
 208 }
 209
 210 void ChannelMultiplexer::MuxChannel::OnIncomingPacket(
 211     scoped_ptr<MultiplexPacket> packet,
 212     const base::Closure& done_task) {
 213   DCHECK_EQ(packet->channel_id(), receive_id_);
 214   if (packet->data().size() > 0) {
 215     pending_packets_.push_back(new PendingPacket(packet.Pass(), done_task));
 216     if (socket_) {
 217       // Notify the socket that we have more data.
 218       socket_->OnPacketReceived();
 219     }
 220   }
 221 }
 222
 223 void ChannelMultiplexer::MuxChannel::OnWriteFailed() {
 224   if (socket_)
 225     socket_->OnWriteFailed();
 226 }
 227
 228 void ChannelMultiplexer::MuxChannel::OnSocketDestroyed() {
 229   DCHECK(socket_);
 230   socket_ = nullptr;
 231 }
 232
 233 bool ChannelMultiplexer::MuxChannel::DoWrite(
 234     scoped_ptr<MultiplexPacket> packet,
 235     const base::Closure& done_task) {
 236   packet->set_channel_id(send_id_);
 237   if (!id_sent_) {
 238     packet->set_channel_name(name_);
 239     id_sent_ = true;
 240   }
 241   return multiplexer_->DoWrite(packet.Pass(), done_task);
 242 }
 243
 244 int ChannelMultiplexer::MuxChannel::DoRead(net::IOBuffer* buffer,
 245                                            int buffer_len) {
 246   int pos = 0;
 247   while (buffer_len > 0 && !pending_packets_.empty()) {
 248     DCHECK(!pending_packets_.front()->is_empty());
 249     int result = pending_packets_.front()->Read(
 250         buffer->data() + pos, buffer_len);
 251     DCHECK_LE(result, buffer_len);
 252     pos += result;
 253     buffer_len -= pos;
 254     if (pending_packets_.front()->is_empty()) {
 255       delete pending_packets_.front();
 256       pending_packets_.erase(pending_packets_.begin());
 257     }
 258   }
 259   return pos;
 260 }
 261
 262 ChannelMultiplexer::MuxSocket::MuxSocket(MuxChannel* channel)
 263     : channel_(channel),
 264       read_buffer_size_(0),
 265       write_pending_(false),
 266       write_result_(0) {
 267 }
 268
 269 ChannelMultiplexer::MuxSocket::~MuxSocket() {
 270   channel_->OnSocketDestroyed();
 271 }
 272
 273 int ChannelMultiplexer::MuxSocket::Read(
 274     net::IOBuffer* buffer, int buffer_len,
 275     const net::CompletionCallback& callback) {
 276   DCHECK(CalledOnValidThread());
 277   DCHECK(read_callback_.is_null());
 278
 279   int result = channel_->DoRead(buffer, buffer_len);
 280   if (result == 0) {
 281     read_buffer_ = buffer;
 282     read_buffer_size_ = buffer_len;
 283     read_callback_ = callback;
 284     return net::ERR_IO_PENDING;
 285   }
 286   return result;
 287 }
 288
 289 int ChannelMultiplexer::MuxSocket::Write(
 290     net::IOBuffer* buffer, int buffer_len,
 291     const net::CompletionCallback& callback) {
 292   DCHECK(CalledOnValidThread());
 293
 294   scoped_ptr<MultiplexPacket> packet(new MultiplexPacket());
 295   size_t size = std::min(kMaxPacketSize, buffer_len);
 296   packet->mutable_data()->assign(buffer->data(), size);
 297
 298   write_pending_ = true;
 299   bool result = channel_->DoWrite(packet.Pass(), base::Bind(
 300       &ChannelMultiplexer::MuxSocket::OnWriteComplete, AsWeakPtr()));
 301
 302   if (!result) {
 303     // Cannot complete the write, e.g. if the connection has been terminated.
 304     return net::ERR_FAILED;
 305   }
 306
 307   // OnWriteComplete() might be called above synchronously.
 308   if (write_pending_) {
 309     DCHECK(write_callback_.is_null());
 310     write_callback_ = callback;
 311     write_result_ = size;
 312     return net::ERR_IO_PENDING;
 313   }
 314
 315   return size;
 316 }
 317
 318 void ChannelMultiplexer::MuxSocket::OnWriteComplete() {
 319   write_pending_ = false;
 320   if (!write_callback_.is_null()) {
 321     net::CompletionCallback cb;
 322     std::swap(cb, write_callback_);
 323     cb.Run(write_result_);
 324   }
 325 }
 326
 327 void ChannelMultiplexer::MuxSocket::OnWriteFailed() {
 328   if (!write_callback_.is_null()) {
 329     net::CompletionCallback cb;
 330     std::swap(cb, write_callback_);
 331     cb.Run(net::ERR_FAILED);
 332   }
 333 }
 334
 335 void ChannelMultiplexer::MuxSocket::OnPacketReceived() {
 336   if (!read_callback_.is_null()) {
 337     int result = channel_->DoRead(read_buffer_.get(), read_buffer_size_);
 338     read_buffer_ = nullptr;
 339     DCHECK_GT(result, 0);
 340     net::CompletionCallback cb;
 341     std::swap(cb, read_callback_);
 342     cb.Run(result);
 343   }
 344 }
 345
 346 ChannelMultiplexer::ChannelMultiplexer(StreamChannelFactory* factory,
 347                                        const std::string& base_channel_name)
 348     : base_channel_factory_(factory),
 349       base_channel_name_(base_channel_name),
 350       next_channel_id_(0),
 351       parser_(base::Bind(&ChannelMultiplexer::OnIncomingPacket,
 352                          base::Unretained(this)),
 353               &reader_),
 354       weak_factory_(this) {
 355 }
 356
 357 ChannelMultiplexer::~ChannelMultiplexer() {
 358   DCHECK(pending_channels_.empty());
 359   STLDeleteValues(&channels_);
 360
 361   // Cancel creation of the base channel if it hasn't finished.
 362   if (base_channel_factory_)
 363     base_channel_factory_->CancelChannelCreation(base_channel_name_);
 364 }
 365
 366 void ChannelMultiplexer::CreateChannel(const std::string& name,
 367                                        const ChannelCreatedCallback& callback) {
 368   if (base_channel_.get()) {
 369     // Already have |base_channel_|. Create new multiplexed channel
 370     // synchronously.
 371     callback.Run(GetOrCreateChannel(name)->CreateSocket());
 372   } else if (!base_channel_.get() && !base_channel_factory_) {
 373     // Fail synchronously if we failed to create |base_channel_|.
 374     callback.Run(nullptr);
 375   } else {
 376     // Still waiting for the |base_channel_|.
 377     pending_channels_.push_back(PendingChannel(name, callback));
 378
 379     // If this is the first multiplexed channel then create the base channel.
 380     if (pending_channels_.size() == 1U) {
 381       base_channel_factory_->CreateChannel(
 382           base_channel_name_,
 383           base::Bind(&ChannelMultiplexer::OnBaseChannelReady,
 384                      base::Unretained(this)));
 385     }
 386   }
 387 }
 388
 389 void ChannelMultiplexer::CancelChannelCreation(const std::string& name) {
 390   for (std::list<PendingChannel>::iterator it = pending_channels_.begin();
 391        it != pending_channels_.end(); ++it) {
 392     if (it->name == name) {
 393       pending_channels_.erase(it);
 394       return;
 395     }
 396   }
 397 }
 398
 399 void ChannelMultiplexer::OnBaseChannelReady(
 400     scoped_ptr<net::StreamSocket> socket) {
 401   base_channel_factory_ = nullptr;
 402   base_channel_ = socket.Pass();
 403
 404   if (base_channel_.get()) {
 405     // Initialize reader and writer.
 406     reader_.StartReading(base_channel_.get());
 407     writer_.Init(base_channel_.get(),
 408                  base::Bind(&ChannelMultiplexer::OnWriteFailed,
 409                             base::Unretained(this)));
 410   }
 411
 412   DoCreatePendingChannels();
 413 }
 414
 415 void ChannelMultiplexer::DoCreatePendingChannels() {
 416   if (pending_channels_.empty())
 417     return;
 418
 419   // Every time this function is called it connects a single channel and posts a
 420   // separate task to connect other channels. This is necessary because the
 421   // callback may destroy the multiplexer or somehow else modify
 422   // |pending_channels_| list (e.g. call CancelChannelCreation()).
 423   base::ThreadTaskRunnerHandle::Get()->PostTask(
 424       FROM_HERE, base::Bind(&ChannelMultiplexer::DoCreatePendingChannels,
 425                             weak_factory_.GetWeakPtr()));
 426
 427   PendingChannel c = pending_channels_.front();
 428   pending_channels_.erase(pending_channels_.begin());
 429   scoped_ptr<net::StreamSocket> socket;
 430   if (base_channel_.get())
 431     socket = GetOrCreateChannel(c.name)->CreateSocket();
 432   c.callback.Run(socket.Pass());
 433 }
 434
 435 ChannelMultiplexer::MuxChannel* ChannelMultiplexer::GetOrCreateChannel(
 436     const std::string& name) {
 437   // Check if we already have a channel with the requested name.
 438   std::map<std::string, MuxChannel*>::iterator it = channels_.find(name);
 439   if (it != channels_.end())
 440     return it->second;
 441
 442   // Create a new channel if we haven't found existing one.
 443   MuxChannel* channel = new MuxChannel(this, name, next_channel_id_);
 444   ++next_channel_id_;
 445   channels_[channel->name()] = channel;
 446   return channel;
 447 }
 448
 449
 450 void ChannelMultiplexer::OnWriteFailed(int error) {
 451   for (std::map<std::string, MuxChannel*>::iterator it = channels_.begin();
 452        it != channels_.end(); ++it) {
 453     base::ThreadTaskRunnerHandle::Get()->PostTask(
 454         FROM_HERE, base::Bind(&ChannelMultiplexer::NotifyWriteFailed,
 455                               weak_factory_.GetWeakPtr(), it->second->name()));
 456   }
 457 }
 458
 459 void ChannelMultiplexer::NotifyWriteFailed(const std::string& name) {
 460   std::map<std::string, MuxChannel*>::iterator it = channels_.find(name);
 461   if (it != channels_.end()) {
 462     it->second->OnWriteFailed();
 463   }
 464 }
 465
 466 void ChannelMultiplexer::OnIncomingPacket(scoped_ptr<MultiplexPacket> packet,
 467                                           const base::Closure& done_task) {
 468   DCHECK(packet->has_channel_id());
 469   if (!packet->has_channel_id()) {
 470     LOG(ERROR) << "Received packet without channel_id.";
 471     done_task.Run();
 472     return;
 473   }
 474
 475   int receive_id = packet->channel_id();
 476   MuxChannel* channel = nullptr;
 477   std::map<int, MuxChannel*>::iterator it =
 478       channels_by_receive_id_.find(receive_id);
 479   if (it != channels_by_receive_id_.end()) {
 480     channel = it->second;
 481   } else {
 482     // This is a new |channel_id| we haven't seen before. Look it up by name.
 483     if (!packet->has_channel_name()) {
 484       LOG(ERROR) << "Received packet with unknown channel_id and "
 485           "without channel_name.";
 486       done_task.Run();
 487       return;
 488     }
 489     channel = GetOrCreateChannel(packet->channel_name());
 490     channel->set_receive_id(receive_id);
 491     channels_by_receive_id_[receive_id] = channel;
 492   }
 493
 494   channel->OnIncomingPacket(packet.Pass(), done_task);
 495 }
 496
 497 bool ChannelMultiplexer::DoWrite(scoped_ptr<MultiplexPacket> packet,
 498                                  const base::Closure& done_task) {
 499   return writer_.Write(SerializeAndFrameMessage(*packet), done_task);
 500 }
 501
 502 }  // namespace protocol
 503 }  // namespace remoting