mojo/edk/system/raw_channel_posix.cc

   1 // Copyright 2013 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 "mojo/edk/system/raw_channel.h"
   6
   7 #include <errno.h>
   8 #include <sys/uio.h>
   9 #include <unistd.h>
  10
  11 #include <algorithm>
  12 #include <deque>
  13
  14 #include "base/bind.h"
  15 #include "base/location.h"
  16 #include "base/logging.h"
  17 #include "base/macros.h"
  18 #include "base/memory/scoped_ptr.h"
  19 #include "base/memory/weak_ptr.h"
  20 #include "base/message_loop/message_loop.h"
  21 #include "base/synchronization/lock.h"
  22 #include "mojo/edk/embedder/platform_channel_utils_posix.h"
  23 #include "mojo/edk/embedder/platform_handle.h"
  24 #include "mojo/edk/embedder/platform_handle_vector.h"
  25 #include "mojo/edk/system/transport_data.h"
  26
  27 namespace mojo {
  28 namespace system {
  29
  30 namespace {
  31
  32 class RawChannelPosix : public RawChannel,
  33                         public base::MessageLoopForIO::Watcher {
  34  public:
  35   explicit RawChannelPosix(embedder::ScopedPlatformHandle handle);
  36   ~RawChannelPosix() override;
  37
  38   // |RawChannel| public methods:
  39   size_t GetSerializedPlatformHandleSize() const override;
  40
  41  private:
  42   // |RawChannel| protected methods:
  43   // Actually override this so that we can send multiple messages with (only)
  44   // FDs if necessary.
  45   void EnqueueMessageNoLock(scoped_ptr<MessageInTransit> message) override;
  46   // Override this to handle those extra FD-only messages.
  47   bool OnReadMessageForRawChannel(
  48       const MessageInTransit::View& message_view) override;
  49   IOResult Read(size_t* bytes_read) override;
  50   IOResult ScheduleRead() override;
  51   embedder::ScopedPlatformHandleVectorPtr GetReadPlatformHandles(
  52       size_t num_platform_handles,
  53       const void* platform_handle_table) override;
  54   IOResult WriteNoLock(size_t* platform_handles_written,
  55                        size_t* bytes_written) override;
  56   IOResult ScheduleWriteNoLock() override;
  57   bool OnInit() override;
  58   void OnShutdownNoLock(scoped_ptr<ReadBuffer> read_buffer,
  59                         scoped_ptr<WriteBuffer> write_buffer) override;
  60
  61   // |base::MessageLoopForIO::Watcher| implementation:
  62   void OnFileCanReadWithoutBlocking(int fd) override;
  63   void OnFileCanWriteWithoutBlocking(int fd) override;
  64
  65   // Implements most of |Read()| (except for a bit of clean-up):
  66   IOResult ReadImpl(size_t* bytes_read);
  67
  68   // Watches for |fd_| to become writable. Must be called on the I/O thread.
  69   void WaitToWrite();
  70
  71   embedder::ScopedPlatformHandle fd_;
  72
  73   // The following members are only used on the I/O thread:
  74   scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> read_watcher_;
  75   scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> write_watcher_;
  76
  77   bool pending_read_;
  78
  79   std::deque<embedder::PlatformHandle> read_platform_handles_;
  80
  81   // The following members are used on multiple threads and protected by
  82   // |write_lock()|:
  83   bool pending_write_;
  84
  85   // This is used for posting tasks from write threads to the I/O thread. It
  86   // must only be accessed under |write_lock_|. The weak pointers it produces
  87   // are only used/invalidated on the I/O thread.
  88   base::WeakPtrFactory<RawChannelPosix> weak_ptr_factory_;
  89
  90   DISALLOW_COPY_AND_ASSIGN(RawChannelPosix);
  91 };
  92
  93 RawChannelPosix::RawChannelPosix(embedder::ScopedPlatformHandle handle)
  94     : fd_(handle.Pass()),
  95       pending_read_(false),
  96       pending_write_(false),
  97       weak_ptr_factory_(this) {
  98   DCHECK(fd_.is_valid());
  99 }
 100
 101 RawChannelPosix::~RawChannelPosix() {
 102   DCHECK(!pending_read_);
 103   DCHECK(!pending_write_);
 104
 105   // No need to take the |write_lock()| here -- if there are still weak pointers
 106   // outstanding, then we're hosed anyway (since we wouldn't be able to
 107   // invalidate them cleanly, since we might not be on the I/O thread).
 108   DCHECK(!weak_ptr_factory_.HasWeakPtrs());
 109
 110   // These must have been shut down/destroyed on the I/O thread.
 111   DCHECK(!read_watcher_);
 112   DCHECK(!write_watcher_);
 113
 114   embedder::CloseAllPlatformHandles(&read_platform_handles_);
 115 }
 116
 117 size_t RawChannelPosix::GetSerializedPlatformHandleSize() const {
 118   // We don't actually need any space on POSIX (since we just send FDs).
 119   return 0;
 120 }
 121
 122 void RawChannelPosix::EnqueueMessageNoLock(
 123     scoped_ptr<MessageInTransit> message) {
 124   if (message->transport_data()) {
 125     embedder::PlatformHandleVector* const platform_handles =
 126         message->transport_data()->platform_handles();
 127     if (platform_handles &&
 128         platform_handles->size() > embedder::kPlatformChannelMaxNumHandles) {
 129       // We can't attach all the FDs to a single message, so we have to "split"
 130       // the message. Send as many control messages as needed first with FDs
 131       // attached (and no data).
 132       size_t i = 0;
 133       for (; platform_handles->size() - i >
 134                  embedder::kPlatformChannelMaxNumHandles;
 135            i += embedder::kPlatformChannelMaxNumHandles) {
 136         scoped_ptr<MessageInTransit> fd_message(new MessageInTransit(
 137             MessageInTransit::kTypeRawChannel,
 138             MessageInTransit::kSubtypeRawChannelPosixExtraPlatformHandles, 0,
 139             nullptr));
 140         embedder::ScopedPlatformHandleVectorPtr fds(
 141             new embedder::PlatformHandleVector(
 142                 platform_handles->begin() + i,
 143                 platform_handles->begin() + i +
 144                     embedder::kPlatformChannelMaxNumHandles));
 145         fd_message->SetTransportData(
 146             make_scoped_ptr(new TransportData(fds.Pass())));
 147         RawChannel::EnqueueMessageNoLock(fd_message.Pass());
 148       }
 149
 150       // Remove the handles that we "moved" into the other messages.
 151       platform_handles->erase(platform_handles->begin(),
 152                               platform_handles->begin() + i);
 153     }
 154   }
 155
 156   RawChannel::EnqueueMessageNoLock(message.Pass());
 157 }
 158
 159 bool RawChannelPosix::OnReadMessageForRawChannel(
 160     const MessageInTransit::View& message_view) {
 161   DCHECK_EQ(message_view.type(), MessageInTransit::kTypeRawChannel);
 162
 163   if (message_view.subtype() ==
 164       MessageInTransit::kSubtypeRawChannelPosixExtraPlatformHandles) {
 165     // We don't need to do anything. |RawChannel| won't extract the platform
 166     // handles, and they'll be accumulated in |Read()|.
 167     return true;
 168   }
 169
 170   return RawChannel::OnReadMessageForRawChannel(message_view);
 171 }
 172
 173 RawChannel::IOResult RawChannelPosix::Read(size_t* bytes_read) {
 174   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 175   DCHECK(!pending_read_);
 176
 177   IOResult rv = ReadImpl(bytes_read);
 178   if (rv != IO_SUCCEEDED && rv != IO_PENDING) {
 179     // Make sure that |OnFileCanReadWithoutBlocking()| won't be called again.
 180     read_watcher_.reset();
 181   }
 182   return rv;
 183 }
 184
 185 RawChannel::IOResult RawChannelPosix::ScheduleRead() {
 186   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 187   DCHECK(!pending_read_);
 188
 189   pending_read_ = true;
 190
 191   return IO_PENDING;
 192 }
 193
 194 embedder::ScopedPlatformHandleVectorPtr RawChannelPosix::GetReadPlatformHandles(
 195     size_t num_platform_handles,
 196     const void* /*platform_handle_table*/) {
 197   DCHECK_GT(num_platform_handles, 0u);
 198
 199   if (read_platform_handles_.size() < num_platform_handles) {
 200     embedder::CloseAllPlatformHandles(&read_platform_handles_);
 201     read_platform_handles_.clear();
 202     return embedder::ScopedPlatformHandleVectorPtr();
 203   }
 204
 205   embedder::ScopedPlatformHandleVectorPtr rv(
 206       new embedder::PlatformHandleVector(num_platform_handles));
 207   rv->assign(read_platform_handles_.begin(),
 208              read_platform_handles_.begin() + num_platform_handles);
 209   read_platform_handles_.erase(
 210       read_platform_handles_.begin(),
 211       read_platform_handles_.begin() + num_platform_handles);
 212   return rv.Pass();
 213 }
 214
 215 RawChannel::IOResult RawChannelPosix::WriteNoLock(
 216     size_t* platform_handles_written,
 217     size_t* bytes_written) {
 218   write_lock().AssertAcquired();
 219
 220   DCHECK(!pending_write_);
 221
 222   size_t num_platform_handles = 0;
 223   ssize_t write_result;
 224   if (write_buffer_no_lock()->HavePlatformHandlesToSend()) {
 225     embedder::PlatformHandle* platform_handles;
 226     void* serialization_data;  // Actually unused.
 227     write_buffer_no_lock()->GetPlatformHandlesToSend(
 228         &num_platform_handles, &platform_handles, &serialization_data);
 229     DCHECK_GT(num_platform_handles, 0u);
 230     DCHECK_LE(num_platform_handles, embedder::kPlatformChannelMaxNumHandles);
 231     DCHECK(platform_handles);
 232
 233     // TODO(vtl): Reduce code duplication. (This is duplicated from below.)
 234     std::vector<WriteBuffer::Buffer> buffers;
 235     write_buffer_no_lock()->GetBuffers(&buffers);
 236     DCHECK(!buffers.empty());
 237     const size_t kMaxBufferCount = 10;
 238     iovec iov[kMaxBufferCount];
 239     size_t buffer_count = std::min(buffers.size(), kMaxBufferCount);
 240     for (size_t i = 0; i < buffer_count; ++i) {
 241       iov[i].iov_base = const_cast<char*>(buffers[i].addr);
 242       iov[i].iov_len = buffers[i].size;
 243     }
 244
 245     write_result = embedder::PlatformChannelSendmsgWithHandles(
 246         fd_.get(), iov, buffer_count, platform_handles, num_platform_handles);
 247     for (size_t i = 0; i < num_platform_handles; i++)
 248       platform_handles[i].CloseIfNecessary();
 249   } else {
 250     std::vector<WriteBuffer::Buffer> buffers;
 251     write_buffer_no_lock()->GetBuffers(&buffers);
 252     DCHECK(!buffers.empty());
 253
 254     if (buffers.size() == 1) {
 255       write_result = embedder::PlatformChannelWrite(fd_.get(), buffers[0].addr,
 256                                                     buffers[0].size);
 257     } else {
 258       const size_t kMaxBufferCount = 10;
 259       iovec iov[kMaxBufferCount];
 260       size_t buffer_count = std::min(buffers.size(), kMaxBufferCount);
 261       for (size_t i = 0; i < buffer_count; ++i) {
 262         iov[i].iov_base = const_cast<char*>(buffers[i].addr);
 263         iov[i].iov_len = buffers[i].size;
 264       }
 265
 266       write_result =
 267           embedder::PlatformChannelWritev(fd_.get(), iov, buffer_count);
 268     }
 269   }
 270
 271   if (write_result >= 0) {
 272     *platform_handles_written = num_platform_handles;
 273     *bytes_written = static_cast<size_t>(write_result);
 274     return IO_SUCCEEDED;
 275   }
 276
 277   if (errno == EPIPE)
 278     return IO_FAILED_SHUTDOWN;
 279
 280   if (errno != EAGAIN && errno != EWOULDBLOCK) {
 281     PLOG(WARNING) << "sendmsg/write/writev";
 282     return IO_FAILED_UNKNOWN;
 283   }
 284
 285   return ScheduleWriteNoLock();
 286 }
 287
 288 RawChannel::IOResult RawChannelPosix::ScheduleWriteNoLock() {
 289   write_lock().AssertAcquired();
 290
 291   DCHECK(!pending_write_);
 292
 293   // Set up to wait for the FD to become writable.
 294   // If we're not on the I/O thread, we have to post a task to do this.
 295   if (base::MessageLoop::current() != message_loop_for_io()) {
 296     message_loop_for_io()->PostTask(FROM_HERE,
 297                                     base::Bind(&RawChannelPosix::WaitToWrite,
 298                                                weak_ptr_factory_.GetWeakPtr()));
 299     pending_write_ = true;
 300     return IO_PENDING;
 301   }
 302
 303   if (message_loop_for_io()->WatchFileDescriptor(
 304           fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE,
 305           write_watcher_.get(), this)) {
 306     pending_write_ = true;
 307     return IO_PENDING;
 308   }
 309
 310   return IO_FAILED_UNKNOWN;
 311 }
 312
 313 bool RawChannelPosix::OnInit() {
 314   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 315
 316   DCHECK(!read_watcher_);
 317   read_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher());
 318   DCHECK(!write_watcher_);
 319   write_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher());
 320
 321   if (!message_loop_for_io()->WatchFileDescriptor(
 322           fd_.get().fd, true, base::MessageLoopForIO::WATCH_READ,
 323           read_watcher_.get(), this)) {
 324     // TODO(vtl): I'm not sure |WatchFileDescriptor()| actually fails cleanly
 325     // (in the sense of returning the message loop's state to what it was before
 326     // it was called).
 327     read_watcher_.reset();
 328     write_watcher_.reset();
 329     return false;
 330   }
 331
 332   return true;
 333 }
 334
 335 void RawChannelPosix::OnShutdownNoLock(
 336     scoped_ptr<ReadBuffer> /*read_buffer*/,
 337     scoped_ptr<WriteBuffer> /*write_buffer*/) {
 338   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 339   write_lock().AssertAcquired();
 340
 341   read_watcher_.reset();   // This will stop watching (if necessary).
 342   write_watcher_.reset();  // This will stop watching (if necessary).
 343
 344   pending_read_ = false;
 345   pending_write_ = false;
 346
 347   DCHECK(fd_.is_valid());
 348   fd_.reset();
 349
 350   weak_ptr_factory_.InvalidateWeakPtrs();
 351 }
 352
 353 void RawChannelPosix::OnFileCanReadWithoutBlocking(int fd) {
 354   DCHECK_EQ(fd, fd_.get().fd);
 355   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 356
 357   if (!pending_read_) {
 358     NOTREACHED();
 359     return;
 360   }
 361
 362   pending_read_ = false;
 363   size_t bytes_read = 0;
 364   IOResult io_result = Read(&bytes_read);
 365   if (io_result != IO_PENDING)
 366     OnReadCompleted(io_result, bytes_read);
 367
 368   // On failure, |read_watcher_| must have been reset; on success,
 369   // we assume that |OnReadCompleted()| always schedules another read.
 370   // Otherwise, we could end up spinning -- getting
 371   // |OnFileCanReadWithoutBlocking()| again and again but not doing any actual
 372   // read.
 373   // TODO(yzshen): An alternative is to stop watching if RawChannel doesn't
 374   // schedule a new read. But that code won't be reached under the current
 375   // RawChannel implementation.
 376   DCHECK(!read_watcher_ || pending_read_);
 377 }
 378
 379 void RawChannelPosix::OnFileCanWriteWithoutBlocking(int fd) {
 380   DCHECK_EQ(fd, fd_.get().fd);
 381   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 382
 383   IOResult io_result;
 384   size_t platform_handles_written = 0;
 385   size_t bytes_written = 0;
 386   {
 387     base::AutoLock locker(write_lock());
 388
 389     DCHECK(pending_write_);
 390
 391     pending_write_ = false;
 392     io_result = WriteNoLock(&platform_handles_written, &bytes_written);
 393   }
 394
 395   if (io_result != IO_PENDING)
 396     OnWriteCompleted(io_result, platform_handles_written, bytes_written);
 397 }
 398
 399 RawChannel::IOResult RawChannelPosix::ReadImpl(size_t* bytes_read) {
 400   char* buffer = nullptr;
 401   size_t bytes_to_read = 0;
 402   read_buffer()->GetBuffer(&buffer, &bytes_to_read);
 403
 404   size_t old_num_platform_handles = read_platform_handles_.size();
 405   ssize_t read_result = embedder::PlatformChannelRecvmsg(
 406       fd_.get(), buffer, bytes_to_read, &read_platform_handles_);
 407   if (read_platform_handles_.size() > old_num_platform_handles) {
 408     DCHECK_LE(read_platform_handles_.size() - old_num_platform_handles,
 409               embedder::kPlatformChannelMaxNumHandles);
 410
 411     // We should never accumulate more than |TransportData::kMaxPlatformHandles
 412     // + embedder::kPlatformChannelMaxNumHandles| handles. (The latter part is
 413     // possible because we could have accumulated all the handles for a message,
 414     // then received the message data plus the first set of handles for the next
 415     // message in the subsequent |recvmsg()|.)
 416     if (read_platform_handles_.size() >
 417         (TransportData::GetMaxPlatformHandles() +
 418          embedder::kPlatformChannelMaxNumHandles)) {
 419       LOG(ERROR) << "Received too many platform handles";
 420       embedder::CloseAllPlatformHandles(&read_platform_handles_);
 421       read_platform_handles_.clear();
 422       return IO_FAILED_UNKNOWN;
 423     }
 424   }
 425
 426   if (read_result > 0) {
 427     *bytes_read = static_cast<size_t>(read_result);
 428     return IO_SUCCEEDED;
 429   }
 430
 431   // |read_result == 0| means "end of file".
 432   if (read_result == 0)
 433     return IO_FAILED_SHUTDOWN;
 434
 435   if (errno == EAGAIN || errno == EWOULDBLOCK)
 436     return ScheduleRead();
 437
 438   if (errno == ECONNRESET)
 439     return IO_FAILED_BROKEN;
 440
 441   PLOG(WARNING) << "recvmsg";
 442   return IO_FAILED_UNKNOWN;
 443 }
 444
 445 void RawChannelPosix::WaitToWrite() {
 446   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
 447
 448   DCHECK(write_watcher_);
 449
 450   if (!message_loop_for_io()->WatchFileDescriptor(
 451           fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE,
 452           write_watcher_.get(), this)) {
 453     {
 454       base::AutoLock locker(write_lock());
 455
 456       DCHECK(pending_write_);
 457       pending_write_ = false;
 458     }
 459     OnWriteCompleted(IO_FAILED_UNKNOWN, 0, 0);
 460   }
 461 }
 462
 463 }  // namespace
 464
 465 // -----------------------------------------------------------------------------
 466
 467 // Static factory method declared in raw_channel.h.
 468 // static
 469 scoped_ptr<RawChannel> RawChannel::Create(
 470     embedder::ScopedPlatformHandle handle) {
 471   return make_scoped_ptr(new RawChannelPosix(handle.Pass()));
 472 }
 473
 474 }  // namespace system
 475 }  // namespace mojo