| OLD | NEW |
| 1 // Copyright 2013 The Chromium Authors. All rights reserved. | 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 | 2 // Use of this source code is governed by a BSD-style license that can be |
| 3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
| 4 | 4 |
| 5 #include "mojo/system/raw_channel.h" | 5 #include "mojo/system/raw_channel.h" |
| 6 | 6 |
| 7 #include <errno.h> | 7 #include <errno.h> |
| 8 #include <string.h> | |
| 9 #include <sys/uio.h> | 8 #include <sys/uio.h> |
| 10 #include <unistd.h> | 9 #include <unistd.h> |
| 11 | 10 |
| 12 #include <algorithm> | 11 #include <algorithm> |
| 13 #include <deque> | |
| 14 #include <vector> | |
| 15 | 12 |
| 16 #include "base/basictypes.h" | 13 #include "base/basictypes.h" |
| 17 #include "base/bind.h" | 14 #include "base/bind.h" |
| 18 #include "base/compiler_specific.h" | 15 #include "base/compiler_specific.h" |
| 19 #include "base/location.h" | 16 #include "base/location.h" |
| 20 #include "base/logging.h" | 17 #include "base/logging.h" |
| 21 #include "base/memory/scoped_ptr.h" | 18 #include "base/memory/scoped_ptr.h" |
| 22 #include "base/memory/weak_ptr.h" | 19 #include "base/memory/weak_ptr.h" |
| 23 #include "base/message_loop/message_loop.h" | 20 #include "base/message_loop/message_loop.h" |
| 24 #include "base/posix/eintr_wrapper.h" | 21 #include "base/posix/eintr_wrapper.h" |
| 25 #include "base/stl_util.h" | |
| 26 #include "base/synchronization/lock.h" | 22 #include "base/synchronization/lock.h" |
| 27 #include "mojo/system/embedder/platform_handle.h" | 23 #include "mojo/system/embedder/platform_handle.h" |
| 28 #include "mojo/system/message_in_transit.h" | |
| 29 | 24 |
| 30 namespace mojo { | 25 namespace mojo { |
| 31 namespace system { | 26 namespace system { |
| 32 | 27 |
| 33 namespace { | 28 namespace { |
| 34 | 29 |
| 35 const size_t kReadSize = 4096; | |
| 36 | |
| 37 class RawChannelPosix : public RawChannel, | 30 class RawChannelPosix : public RawChannel, |
| 38 public base::MessageLoopForIO::Watcher { | 31 public base::MessageLoopForIO::Watcher { |
| 39 public: | 32 public: |
| 40 RawChannelPosix(embedder::ScopedPlatformHandle handle, | 33 RawChannelPosix(embedder::ScopedPlatformHandle handle, |
| 41 Delegate* delegate, | 34 Delegate* delegate, |
| 42 base::MessageLoopForIO* message_loop_for_io); | 35 base::MessageLoopForIO* message_loop_for_io); |
| 43 virtual ~RawChannelPosix(); | 36 virtual ~RawChannelPosix(); |
| 44 | 37 |
| 38 private: |
| 45 // |RawChannel| implementation: | 39 // |RawChannel| implementation: |
| 46 virtual bool Init() OVERRIDE; | 40 virtual IOResult Read(size_t* bytes_read) OVERRIDE; |
| 47 virtual void Shutdown() OVERRIDE; | 41 virtual IOResult ScheduleRead() OVERRIDE; |
| 48 virtual bool WriteMessage(scoped_ptr<MessageInTransit> message) OVERRIDE; | 42 virtual IOResult WriteNoLock(size_t* bytes_written) OVERRIDE; |
| 43 virtual IOResult ScheduleWriteNoLock() OVERRIDE; |
| 44 virtual bool OnInit() OVERRIDE; |
| 45 virtual void OnShutdownNoLock( |
| 46 scoped_ptr<ReadBuffer> read_buffer, |
| 47 scoped_ptr<WriteBuffer> write_buffer) OVERRIDE; |
| 49 | 48 |
| 50 private: | |
| 51 // |base::MessageLoopForIO::Watcher| implementation: | 49 // |base::MessageLoopForIO::Watcher| implementation: |
| 52 virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE; | 50 virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE; |
| 53 virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE; | 51 virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE; |
| 54 | 52 |
| 55 // Watches for |fd_| to become writable. Must be called on the I/O thread. | 53 // Watches for |fd_| to become writable. Must be called on the I/O thread. |
| 56 void WaitToWrite(); | 54 void WaitToWrite(); |
| 57 | 55 |
| 58 // Calls |delegate()->OnFatalError(fatal_error)|. Must be called on the I/O | |
| 59 // thread WITHOUT |write_lock_| held. | |
| 60 void CallOnFatalError(Delegate::FatalError fatal_error); | |
| 61 | |
| 62 // Writes the message at the front of |write_message_queue_|, starting at | |
| 63 // |write_message_offset_|. It removes and destroys if the write completes and | |
| 64 // otherwise updates |write_message_offset_|. Returns true on success. Must be | |
| 65 // called under |write_lock_|. | |
| 66 bool WriteFrontMessageNoLock(); | |
| 67 | |
| 68 // Cancels all pending writes and destroys the contents of | |
| 69 // |write_message_queue_|. Should only be called if |write_stopped_| is false; | |
| 70 // sets |write_stopped_| to true. Must be called under |write_lock_|. | |
| 71 void CancelPendingWritesNoLock(); | |
| 72 | |
| 73 embedder::ScopedPlatformHandle fd_; | 56 embedder::ScopedPlatformHandle fd_; |
| 74 | 57 |
| 75 // Only used on the I/O thread: | 58 // The following members are only used on the I/O thread: |
| 76 scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> read_watcher_; | 59 scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> read_watcher_; |
| 77 scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> write_watcher_; | 60 scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> write_watcher_; |
| 78 | 61 |
| 79 // We store data from |read()|s in |read_buffer_|. The start of |read_buffer_| | 62 bool pending_read_; |
| 80 // is always aligned with a message boundary (we will copy memory to ensure | |
| 81 // this), but |read_buffer_| may be larger than the actual number of bytes we | |
| 82 // have. | |
| 83 std::vector<char> read_buffer_; | |
| 84 size_t read_buffer_num_valid_bytes_; | |
| 85 | 63 |
| 86 base::Lock write_lock_; // Protects the following members. | 64 // The following members are used on multiple threads and protected by |
| 87 bool write_stopped_; | 65 // |write_lock()|: |
| 88 // TODO(vtl): When C++11 is available, switch this to a deque of | 66 bool pending_write_; |
| 89 // |scoped_ptr|/|unique_ptr|s. | 67 |
| 90 std::deque<MessageInTransit*> write_message_queue_; | |
| 91 size_t write_message_offset_; | |
| 92 // This is used for posting tasks from write threads to the I/O thread. It | 68 // This is used for posting tasks from write threads to the I/O thread. It |
| 93 // must only be accessed under |write_lock_|. The weak pointers it produces | 69 // must only be accessed under |write_lock_|. The weak pointers it produces |
| 94 // are only used/invalidated on the I/O thread. | 70 // are only used/invalidated on the I/O thread. |
| 95 base::WeakPtrFactory<RawChannelPosix> weak_ptr_factory_; | 71 base::WeakPtrFactory<RawChannelPosix> weak_ptr_factory_; |
| 96 | 72 |
| 97 DISALLOW_COPY_AND_ASSIGN(RawChannelPosix); | 73 DISALLOW_COPY_AND_ASSIGN(RawChannelPosix); |
| 98 }; | 74 }; |
| 99 | 75 |
| 100 RawChannelPosix::RawChannelPosix(embedder::ScopedPlatformHandle handle, | 76 RawChannelPosix::RawChannelPosix(embedder::ScopedPlatformHandle handle, |
| 101 Delegate* delegate, | 77 Delegate* delegate, |
| 102 base::MessageLoopForIO* message_loop_for_io) | 78 base::MessageLoopForIO* message_loop_for_io) |
| 103 : RawChannel(delegate, message_loop_for_io), | 79 : RawChannel(delegate, message_loop_for_io), |
| 104 fd_(handle.Pass()), | 80 fd_(handle.Pass()), |
| 105 read_buffer_num_valid_bytes_(0), | 81 pending_read_(false), |
| 106 write_stopped_(false), | 82 pending_write_(false), |
| 107 write_message_offset_(0), | |
| 108 weak_ptr_factory_(this) { | 83 weak_ptr_factory_(this) { |
| 109 CHECK_EQ(RawChannel::message_loop_for_io()->type(), | |
| 110 base::MessageLoop::TYPE_IO); | |
| 111 DCHECK(fd_.is_valid()); | 84 DCHECK(fd_.is_valid()); |
| 112 } | 85 } |
| 113 | 86 |
| 114 RawChannelPosix::~RawChannelPosix() { | 87 RawChannelPosix::~RawChannelPosix() { |
| 115 DCHECK(write_stopped_); | 88 DCHECK(!pending_read_); |
| 116 DCHECK(!fd_.is_valid()); | 89 DCHECK(!pending_write_); |
| 117 | 90 |
| 118 // No need to take the |write_lock_| here -- if there are still weak pointers | 91 // No need to take the |write_lock()| here -- if there are still weak pointers |
| 119 // outstanding, then we're hosed anyway (since we wouldn't be able to | 92 // outstanding, then we're hosed anyway (since we wouldn't be able to |
| 120 // invalidate them cleanly, since we might not be on the I/O thread). | 93 // invalidate them cleanly, since we might not be on the I/O thread). |
| 121 DCHECK(!weak_ptr_factory_.HasWeakPtrs()); | 94 DCHECK(!weak_ptr_factory_.HasWeakPtrs()); |
| 122 | 95 |
| 123 // These must have been shut down/destroyed on the I/O thread. | 96 // These must have been shut down/destroyed on the I/O thread. |
| 124 DCHECK(!read_watcher_.get()); | 97 DCHECK(!read_watcher_.get()); |
| 125 DCHECK(!write_watcher_.get()); | 98 DCHECK(!write_watcher_.get()); |
| 126 } | 99 } |
| 127 | 100 |
| 128 bool RawChannelPosix::Init() { | 101 RawChannel::IOResult RawChannelPosix::Read(size_t* bytes_read) { |
| 102 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); |
| 103 DCHECK(!pending_read_); |
| 104 |
| 105 char* buffer = NULL; |
| 106 size_t bytes_to_read = 0; |
| 107 read_buffer()->GetBuffer(&buffer, &bytes_to_read); |
| 108 |
| 109 ssize_t read_result = HANDLE_EINTR(read(fd_.get().fd, buffer, bytes_to_read)); |
| 110 |
| 111 if (read_result >= 0) { |
| 112 *bytes_read = static_cast<size_t>(read_result); |
| 113 return IO_SUCCEEDED; |
| 114 } |
| 115 |
| 116 if (errno != EAGAIN && errno != EWOULDBLOCK) { |
| 117 PLOG(ERROR) << "read"; |
| 118 |
| 119 // Make sure that |OnFileCanReadWithoutBlocking()| won't be called again. |
| 120 read_watcher_.reset(); |
| 121 |
| 122 return IO_FAILED; |
| 123 } |
| 124 |
| 125 return ScheduleRead(); |
| 126 } |
| 127 |
| 128 RawChannel::IOResult RawChannelPosix::ScheduleRead() { |
| 129 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); |
| 130 DCHECK(!pending_read_); |
| 131 |
| 132 pending_read_ = true; |
| 133 |
| 134 return IO_PENDING; |
| 135 } |
| 136 |
| 137 RawChannel::IOResult RawChannelPosix::WriteNoLock(size_t* bytes_written) { |
| 138 write_lock().AssertAcquired(); |
| 139 |
| 140 DCHECK(!pending_write_); |
| 141 |
| 142 std::vector<WriteBuffer::Buffer> buffers; |
| 143 write_buffer_no_lock()->GetBuffers(&buffers); |
| 144 DCHECK(!buffers.empty()); |
| 145 |
| 146 ssize_t write_result = -1; |
| 147 if (buffers.size() == 1) { |
| 148 write_result = HANDLE_EINTR( |
| 149 write(fd_.get().fd, buffers[0].addr, buffers[0].size)); |
| 150 } else { |
| 151 // Note that using |writev()| is measurably slower than using |write()| -- |
| 152 // at least in a microbenchmark -- but much faster than using multiple |
| 153 // |write()|s. |
| 154 const size_t kMaxBufferCount = 10; |
| 155 iovec iov[kMaxBufferCount]; |
| 156 size_t buffer_count = std::min(buffers.size(), kMaxBufferCount); |
| 157 |
| 158 for (size_t i = 0; i < buffer_count; ++i) { |
| 159 iov[i].iov_base = const_cast<char*>(buffers[i].addr); |
| 160 iov[i].iov_len = buffers[i].size; |
| 161 } |
| 162 |
| 163 write_result = HANDLE_EINTR(writev(fd_.get().fd, iov, buffer_count)); |
| 164 } |
| 165 |
| 166 if (write_result >= 0) { |
| 167 *bytes_written = static_cast<size_t>(write_result); |
| 168 return IO_SUCCEEDED; |
| 169 } |
| 170 |
| 171 if (errno != EAGAIN && errno != EWOULDBLOCK) { |
| 172 PLOG(ERROR) << "write of size " |
| 173 << write_buffer_no_lock()->GetTotalBytesToWrite(); |
| 174 return IO_FAILED; |
| 175 } |
| 176 |
| 177 return ScheduleWriteNoLock(); |
| 178 } |
| 179 |
| 180 RawChannel::IOResult RawChannelPosix::ScheduleWriteNoLock() { |
| 181 write_lock().AssertAcquired(); |
| 182 |
| 183 DCHECK(!pending_write_); |
| 184 |
| 185 // Set up to wait for the FD to become writable. |
| 186 // If we're not on the I/O thread, we have to post a task to do this. |
| 187 if (base::MessageLoop::current() != message_loop_for_io()) { |
| 188 message_loop_for_io()->PostTask( |
| 189 FROM_HERE, |
| 190 base::Bind(&RawChannelPosix::WaitToWrite, |
| 191 weak_ptr_factory_.GetWeakPtr())); |
| 192 pending_write_ = true; |
| 193 return IO_PENDING; |
| 194 } |
| 195 |
| 196 if (message_loop_for_io()->WatchFileDescriptor( |
| 197 fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE, |
| 198 write_watcher_.get(), this)) { |
| 199 pending_write_ = true; |
| 200 return IO_PENDING; |
| 201 } |
| 202 |
| 203 return IO_FAILED; |
| 204 } |
| 205 |
| 206 bool RawChannelPosix::OnInit() { |
| 129 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); | 207 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); |
| 130 | 208 |
| 131 DCHECK(!read_watcher_.get()); | 209 DCHECK(!read_watcher_.get()); |
| 132 read_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher()); | 210 read_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher()); |
| 133 DCHECK(!write_watcher_.get()); | 211 DCHECK(!write_watcher_.get()); |
| 134 write_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher()); | 212 write_watcher_.reset(new base::MessageLoopForIO::FileDescriptorWatcher()); |
| 135 | 213 |
| 136 // No need to take the lock. No one should be using us yet. | |
| 137 DCHECK(write_message_queue_.empty()); | |
| 138 | |
| 139 if (!message_loop_for_io()->WatchFileDescriptor(fd_.get().fd, true, | 214 if (!message_loop_for_io()->WatchFileDescriptor(fd_.get().fd, true, |
| 140 base::MessageLoopForIO::WATCH_READ, read_watcher_.get(), this)) { | 215 base::MessageLoopForIO::WATCH_READ, read_watcher_.get(), this)) { |
| 141 // TODO(vtl): I'm not sure |WatchFileDescriptor()| actually fails cleanly | 216 // TODO(vtl): I'm not sure |WatchFileDescriptor()| actually fails cleanly |
| 142 // (in the sense of returning the message loop's state to what it was before | 217 // (in the sense of returning the message loop's state to what it was before |
| 143 // it was called). | 218 // it was called). |
| 144 read_watcher_.reset(); | 219 read_watcher_.reset(); |
| 145 write_watcher_.reset(); | 220 write_watcher_.reset(); |
| 146 return false; | 221 return false; |
| 147 } | 222 } |
| 148 | 223 |
| 149 return true; | 224 return true; |
| 150 } | 225 } |
| 151 | 226 |
| 152 void RawChannelPosix::Shutdown() { | 227 void RawChannelPosix::OnShutdownNoLock( |
| 153 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); | 228 scoped_ptr<ReadBuffer> /*read_buffer*/, |
| 154 | 229 scoped_ptr<WriteBuffer> /*write_buffer*/) { |
| 155 base::AutoLock locker(write_lock_); | 230 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); |
| 156 if (!write_stopped_) | 231 write_lock().AssertAcquired(); |
| 157 CancelPendingWritesNoLock(); | |
| 158 | 232 |
| 159 read_watcher_.reset(); // This will stop watching (if necessary). | 233 read_watcher_.reset(); // This will stop watching (if necessary). |
| 160 write_watcher_.reset(); // This will stop watching (if necessary). | 234 write_watcher_.reset(); // This will stop watching (if necessary). |
| 161 | 235 |
| 236 pending_read_ = false; |
| 237 pending_write_ = false; |
| 238 |
| 162 DCHECK(fd_.is_valid()); | 239 DCHECK(fd_.is_valid()); |
| 163 fd_.reset(); | 240 fd_.reset(); |
| 164 | 241 |
| 165 weak_ptr_factory_.InvalidateWeakPtrs(); | 242 weak_ptr_factory_.InvalidateWeakPtrs(); |
| 166 } | 243 } |
| 167 | 244 |
| 168 // Reminder: This must be thread-safe, and takes ownership of |message|. | |
| 169 bool RawChannelPosix::WriteMessage(scoped_ptr<MessageInTransit> message) { | |
| 170 base::AutoLock locker(write_lock_); | |
| 171 if (write_stopped_) | |
| 172 return false; | |
| 173 | |
| 174 if (!write_message_queue_.empty()) { | |
| 175 write_message_queue_.push_back(message.release()); | |
| 176 return true; | |
| 177 } | |
| 178 | |
| 179 write_message_queue_.push_front(message.release()); | |
| 180 DCHECK_EQ(write_message_offset_, 0u); | |
| 181 bool result = WriteFrontMessageNoLock(); | |
| 182 DCHECK(result || write_message_queue_.empty()); | |
| 183 | |
| 184 if (!result) { | |
| 185 // Even if we're on the I/O thread, don't call |OnFatalError()| in the | |
| 186 // nested context. | |
| 187 message_loop_for_io()->PostTask( | |
| 188 FROM_HERE, | |
| 189 base::Bind(&RawChannelPosix::CallOnFatalError, | |
| 190 weak_ptr_factory_.GetWeakPtr(), | |
| 191 Delegate::FATAL_ERROR_FAILED_WRITE)); | |
| 192 } else if (!write_message_queue_.empty()) { | |
| 193 // Set up to wait for the FD to become writable. If we're not on the I/O | |
| 194 // thread, we have to post a task to do this. | |
| 195 if (base::MessageLoop::current() == message_loop_for_io()) { | |
| 196 WaitToWrite(); | |
| 197 } else { | |
| 198 message_loop_for_io()->PostTask( | |
| 199 FROM_HERE, | |
| 200 base::Bind(&RawChannelPosix::WaitToWrite, | |
| 201 weak_ptr_factory_.GetWeakPtr())); | |
| 202 } | |
| 203 } | |
| 204 | |
| 205 return result; | |
| 206 } | |
| 207 | |
| 208 void RawChannelPosix::OnFileCanReadWithoutBlocking(int fd) { | 245 void RawChannelPosix::OnFileCanReadWithoutBlocking(int fd) { |
| 209 DCHECK_EQ(fd, fd_.get().fd); | 246 DCHECK_EQ(fd, fd_.get().fd); |
| 210 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); | 247 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); |
| 211 | 248 |
| 212 bool did_dispatch_message = false; | 249 if (!pending_read_) { |
| 213 // Tracks the offset of the first undispatched message in |read_buffer_|. | 250 NOTREACHED(); |
| 214 // Currently, we copy data to ensure that this is zero at the beginning. | 251 return; |
| 215 size_t read_buffer_start = 0; | 252 } |
| 216 for (;;) { | 253 |
| 217 if (read_buffer_.size() - (read_buffer_start + read_buffer_num_valid_bytes_) | 254 pending_read_ = false; |
| 218 < kReadSize) { | 255 size_t bytes_read = 0; |
| 219 // Use power-of-2 buffer sizes. | 256 IOResult result = Read(&bytes_read); |
| 220 // TODO(vtl): Make sure the buffer doesn't get too large (and enforce the | 257 if (result != IO_PENDING) |
| 221 // maximum message size to whatever extent necessary). | 258 OnReadCompleted(result == IO_SUCCEEDED, bytes_read); |
| 222 // TODO(vtl): We may often be able to peek at the header and get the real | 259 |
| 223 // required extra space (which may be much bigger than |kReadSize|). | 260 // On failure, |read_watcher_| must have been reset; on success, |
| 224 size_t new_size = std::max(read_buffer_.size(), kReadSize); | 261 // we assume that |OnReadCompleted()| always schedules another read. |
| 225 while (new_size < | 262 // Otherwise, we could end up spinning -- getting |
| 226 read_buffer_start + read_buffer_num_valid_bytes_ + kReadSize) | 263 // |OnFileCanReadWithoutBlocking()| again and again but not doing any actual |
| 227 new_size *= 2; | 264 // read. |
| 228 | 265 // TODO(yzshen): An alternative is to stop watching if RawChannel doesn't |
| 229 // TODO(vtl): It's suboptimal to zero out the fresh memory. | 266 // schedule a new read. But that code won't be reached under the current |
| 230 read_buffer_.resize(new_size, 0); | 267 // RawChannel implementation. |
| 231 } | 268 DCHECK(!read_watcher_.get() || pending_read_); |
| 232 | |
| 233 ssize_t bytes_read = HANDLE_EINTR( | |
| 234 read(fd_.get().fd, | |
| 235 &read_buffer_[read_buffer_start + read_buffer_num_valid_bytes_], | |
| 236 kReadSize)); | |
| 237 if (bytes_read < 0) { | |
| 238 if (errno != EAGAIN && errno != EWOULDBLOCK) { | |
| 239 PLOG(ERROR) << "read"; | |
| 240 | |
| 241 // Make sure that |OnFileCanReadWithoutBlocking()| won't be called | |
| 242 // again. | |
| 243 read_watcher_.reset(); | |
| 244 | |
| 245 CallOnFatalError(Delegate::FATAL_ERROR_FAILED_READ); | |
| 246 return; | |
| 247 } | |
| 248 | |
| 249 break; | |
| 250 } | |
| 251 | |
| 252 read_buffer_num_valid_bytes_ += static_cast<size_t>(bytes_read); | |
| 253 | |
| 254 // Dispatch all the messages that we can. | |
| 255 size_t message_size; | |
| 256 // Note that we rely on short-circuit evaluation here: | |
| 257 // - |read_buffer_start| may be an invalid index into |read_buffer_| if | |
| 258 // |read_buffer_num_valid_bytes_| is zero. | |
| 259 // - |message_size| is only valid if |GetNextMessageSize()| returns true. | |
| 260 // TODO(vtl): Use |message_size| more intelligently (e.g., to request the | |
| 261 // next read). | |
| 262 while (read_buffer_num_valid_bytes_ > 0 && | |
| 263 MessageInTransit::GetNextMessageSize( | |
| 264 &read_buffer_[read_buffer_start], read_buffer_num_valid_bytes_, | |
| 265 &message_size) && | |
| 266 read_buffer_num_valid_bytes_ >= message_size) { | |
| 267 // TODO(vtl): FIXME -- replace "unowned buffer" |MessageInTransit|s with | |
| 268 // some sort of "view" abstraction. | |
| 269 MessageInTransit message(MessageInTransit::UNOWNED_BUFFER, message_size, | |
| 270 &read_buffer_[read_buffer_start]); | |
| 271 DCHECK_EQ(message.total_size(), message_size); | |
| 272 | |
| 273 // Dispatch the message. | |
| 274 delegate()->OnReadMessage(message); | |
| 275 if (!read_watcher_.get()) { | |
| 276 // |Shutdown()| was called in |OnReadMessage()|. | |
| 277 // TODO(vtl): Add test for this case. | |
| 278 return; | |
| 279 } | |
| 280 did_dispatch_message = true; | |
| 281 | |
| 282 // Update our state. | |
| 283 read_buffer_start += message_size; | |
| 284 read_buffer_num_valid_bytes_ -= message_size; | |
| 285 } | |
| 286 | |
| 287 // If we dispatched any messages, stop reading for now (and let the message | |
| 288 // loop do its thing for another round). | |
| 289 // TODO(vtl): Is this the behavior we want? (Alternatives: i. Dispatch only | |
| 290 // a single message. Risks: slower, more complex if we want to avoid lots of | |
| 291 // copying. ii. Keep reading until there's no more data and dispatch all the | |
| 292 // messages we can. Risks: starvation of other users of the message loop.) | |
| 293 if (did_dispatch_message) | |
| 294 break; | |
| 295 | |
| 296 // If we didn't max out |kReadSize|, stop reading for now. | |
| 297 if (static_cast<size_t>(bytes_read) < kReadSize) | |
| 298 break; | |
| 299 | |
| 300 // Else try to read some more.... | |
| 301 } | |
| 302 | |
| 303 // Move data back to start. | |
| 304 if (read_buffer_start > 0) { | |
| 305 if (read_buffer_num_valid_bytes_ > 0) { | |
| 306 memmove(&read_buffer_[0], &read_buffer_[read_buffer_start], | |
| 307 read_buffer_num_valid_bytes_); | |
| 308 } | |
| 309 read_buffer_start = 0; | |
| 310 } | |
| 311 } | 269 } |
| 312 | 270 |
| 313 void RawChannelPosix::OnFileCanWriteWithoutBlocking(int fd) { | 271 void RawChannelPosix::OnFileCanWriteWithoutBlocking(int fd) { |
| 314 DCHECK_EQ(fd, fd_.get().fd); | 272 DCHECK_EQ(fd, fd_.get().fd); |
| 315 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); | 273 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); |
| 316 | 274 |
| 317 bool did_fail = false; | 275 IOResult result = IO_FAILED; |
| 276 size_t bytes_written = 0; |
| 318 { | 277 { |
| 319 base::AutoLock locker(write_lock_); | 278 base::AutoLock locker(write_lock()); |
| 320 DCHECK_EQ(write_stopped_, write_message_queue_.empty()); | 279 |
| 321 | 280 DCHECK(pending_write_); |
| 322 if (write_stopped_) { | 281 |
| 323 write_watcher_.reset(); | 282 pending_write_ = false; |
| 324 return; | 283 result = WriteNoLock(&bytes_written); |
| 284 } |
| 285 |
| 286 if (result != IO_PENDING) |
| 287 OnWriteCompleted(result == IO_SUCCEEDED, bytes_written); |
| 288 } |
| 289 |
| 290 void RawChannelPosix::WaitToWrite() { |
| 291 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); |
| 292 |
| 293 DCHECK(write_watcher_.get()); |
| 294 |
| 295 if (!message_loop_for_io()->WatchFileDescriptor( |
| 296 fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE, |
| 297 write_watcher_.get(), this)) { |
| 298 { |
| 299 base::AutoLock locker(write_lock()); |
| 300 |
| 301 DCHECK(pending_write_); |
| 302 pending_write_ = false; |
| 325 } | 303 } |
| 326 | 304 OnWriteCompleted(false, 0); |
| 327 bool result = WriteFrontMessageNoLock(); | 305 } |
| 328 DCHECK(result || write_message_queue_.empty()); | |
| 329 | |
| 330 if (!result) { | |
| 331 did_fail = true; | |
| 332 write_watcher_.reset(); | |
| 333 } else if (!write_message_queue_.empty()) { | |
| 334 WaitToWrite(); | |
| 335 } | |
| 336 } | |
| 337 if (did_fail) | |
| 338 CallOnFatalError(Delegate::FATAL_ERROR_FAILED_WRITE); | |
| 339 } | |
| 340 | |
| 341 void RawChannelPosix::WaitToWrite() { | |
| 342 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); | |
| 343 | |
| 344 DCHECK(write_watcher_.get()); | |
| 345 bool result = message_loop_for_io()->WatchFileDescriptor( | |
| 346 fd_.get().fd, false, base::MessageLoopForIO::WATCH_WRITE, | |
| 347 write_watcher_.get(), this); | |
| 348 DCHECK(result); | |
| 349 } | |
| 350 | |
| 351 void RawChannelPosix::CallOnFatalError(Delegate::FatalError fatal_error) { | |
| 352 DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io()); | |
| 353 // TODO(vtl): Add a "write_lock_.AssertNotAcquired()"? | |
| 354 delegate()->OnFatalError(fatal_error); | |
| 355 } | |
| 356 | |
| 357 // TODO(vtl): This will collide with yzshen's work. This is just a hacky, | |
| 358 // minimally-invasive function that does what I want (write/resume writing a | |
| 359 // |MessageInTransit| that may consist of more than one buffer). | |
| 360 ssize_t WriteMessageInTransit(int fd, | |
| 361 MessageInTransit* message, | |
| 362 size_t offset, | |
| 363 size_t* bytes_to_write) { | |
| 364 *bytes_to_write = message->total_size() - offset; | |
| 365 if (!message->secondary_buffer_size()) { | |
| 366 // Only write from the main buffer. | |
| 367 DCHECK_LT(offset, message->main_buffer_size()); | |
| 368 DCHECK_LE(*bytes_to_write, message->main_buffer_size()); | |
| 369 return HANDLE_EINTR( | |
| 370 write(fd, | |
| 371 static_cast<const char*>(message->main_buffer()) + offset, | |
| 372 *bytes_to_write)); | |
| 373 } | |
| 374 if (offset >= message->main_buffer_size()) { | |
| 375 // Only write from the secondary buffer. | |
| 376 DCHECK_LT(offset - message->main_buffer_size(), | |
| 377 message->secondary_buffer_size()); | |
| 378 DCHECK_LE(*bytes_to_write, message->secondary_buffer_size()); | |
| 379 return HANDLE_EINTR( | |
| 380 write(fd, | |
| 381 static_cast<const char*>(message->secondary_buffer()) + | |
| 382 (offset - message->main_buffer_size()), | |
| 383 *bytes_to_write)); | |
| 384 } | |
| 385 // Write from both buffers. (Note that using |writev()| is measurably slower | |
| 386 // than using |write()| -- at least in a microbenchmark -- but much faster | |
| 387 // than using two |write()|s.) | |
| 388 DCHECK_EQ(*bytes_to_write, message->main_buffer_size() - offset + | |
| 389 message->secondary_buffer_size()); | |
| 390 struct iovec iov[2] = { | |
| 391 { const_cast<char*>( | |
| 392 static_cast<const char*>(message->main_buffer()) + offset), | |
| 393 message->main_buffer_size() - offset }, | |
| 394 { const_cast<void*>(message->secondary_buffer()), | |
| 395 message->secondary_buffer_size() } | |
| 396 }; | |
| 397 return HANDLE_EINTR(writev(fd, iov, 2)); | |
| 398 } | |
| 399 | |
| 400 bool RawChannelPosix::WriteFrontMessageNoLock() { | |
| 401 write_lock_.AssertAcquired(); | |
| 402 | |
| 403 DCHECK(!write_stopped_); | |
| 404 DCHECK(!write_message_queue_.empty()); | |
| 405 | |
| 406 MessageInTransit* message = write_message_queue_.front(); | |
| 407 DCHECK_LT(write_message_offset_, message->total_size()); | |
| 408 size_t bytes_to_write; | |
| 409 ssize_t bytes_written = WriteMessageInTransit(fd_.get().fd, | |
| 410 message, | |
| 411 write_message_offset_, | |
| 412 &bytes_to_write); | |
| 413 if (bytes_written < 0) { | |
| 414 if (errno != EAGAIN && errno != EWOULDBLOCK) { | |
| 415 PLOG(ERROR) << "write of size " << bytes_to_write; | |
| 416 CancelPendingWritesNoLock(); | |
| 417 return false; | |
| 418 } | |
| 419 | |
| 420 // We simply failed to write since we'd block. The logic is the same as if | |
| 421 // we got a partial write. | |
| 422 bytes_written = 0; | |
| 423 } | |
| 424 | |
| 425 DCHECK_GE(bytes_written, 0); | |
| 426 if (static_cast<size_t>(bytes_written) < bytes_to_write) { | |
| 427 // Partial (or no) write. | |
| 428 write_message_offset_ += static_cast<size_t>(bytes_written); | |
| 429 } else { | |
| 430 // Complete write. | |
| 431 DCHECK_EQ(static_cast<size_t>(bytes_written), bytes_to_write); | |
| 432 write_message_queue_.pop_front(); | |
| 433 delete message; | |
| 434 write_message_offset_ = 0; | |
| 435 } | |
| 436 | |
| 437 return true; | |
| 438 } | |
| 439 | |
| 440 void RawChannelPosix::CancelPendingWritesNoLock() { | |
| 441 write_lock_.AssertAcquired(); | |
| 442 DCHECK(!write_stopped_); | |
| 443 | |
| 444 write_stopped_ = true; | |
| 445 STLDeleteElements(&write_message_queue_); | |
| 446 } | 306 } |
| 447 | 307 |
| 448 } // namespace | 308 } // namespace |
| 449 | 309 |
| 450 // ----------------------------------------------------------------------------- | 310 // ----------------------------------------------------------------------------- |
| 451 | 311 |
| 452 // Static factory method declared in raw_channel.h. | 312 // Static factory method declared in raw_channel.h. |
| 453 // static | 313 // static |
| 454 RawChannel* RawChannel::Create(embedder::ScopedPlatformHandle handle, | 314 RawChannel* RawChannel::Create(embedder::ScopedPlatformHandle handle, |
| 455 Delegate* delegate, | 315 Delegate* delegate, |
| 456 base::MessageLoopForIO* message_loop_for_io) { | 316 base::MessageLoopForIO* message_loop_for_io) { |
| 457 return new RawChannelPosix(handle.Pass(), delegate, message_loop_for_io); | 317 return new RawChannelPosix(handle.Pass(), delegate, message_loop_for_io); |
| 458 } | 318 } |
| 459 | 319 |
| 460 } // namespace system | 320 } // namespace system |
| 461 } // namespace mojo | 321 } // namespace mojo |
| OLD | NEW |