Google Git
Sign in
chromium / chromium / src / lkgr / . / mojo / core / channel_fuchsia.cc
blob: db4a322530c7be22a447a633a3d7d7719e116cf0 [file] [log] [blame]
Avi Drissmand387f0922022-09-14 20:51:31[diff] [blame]1// Copyright 2017 The Chromium Authors
Wez9736ba782017-08-10 16:35:36[diff] [blame]2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
Arthur Sonzognid4ee5452024-07-18 21:51:54[diff] [blame]5#ifdef UNSAFE_BUFFERS_BUILD
6// TODO(crbug.com/351564777): Remove this and convert code to safer constructs.
7#pragma allow_unsafe_buffers
8#endif
9
Ken Rockotdba46db2018-07-04 18:41:04[diff] [blame]10#include "mojo/core/channel.h"
Wez9736ba782017-08-10 16:35:36[diff] [blame]11
Wezab912ed2019-02-19 22:05:17[diff] [blame]12#include <lib/fdio/fd.h>
Wez5c3c6f152018-06-09 18:24:02[diff] [blame]13#include <lib/fdio/limits.h>
Wez157707d62018-07-10 22:48:47[diff] [blame]14#include <lib/zx/channel.h>
15#include <lib/zx/handle.h>
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]16#include <zircon/processargs.h>
17#include <zircon/status.h>
18#include <zircon/syscalls.h>
Avi Drissmanebac4302022-01-10 21:56:58[diff] [blame]19
Wez9736ba782017-08-10 16:35:36[diff] [blame]20#include <algorithm>
Peter Boström84d239f2021-05-13 23:02:44[diff] [blame]21#include <memory>
Avi Drissmanebac4302022-01-10 21:56:58[diff] [blame]22#include <tuple>
Wez9736ba782017-08-10 16:35:36[diff] [blame]23
Brett Wilson55ff1475e2017-09-26 00:28:48[diff] [blame]24#include "base/containers/circular_deque.h"
Wezda440922017-08-22 22:54:31[diff] [blame]25#include "base/files/scoped_file.h"
Wez157707d62018-07-10 22:48:47[diff] [blame]26#include "base/fuchsia/fuchsia_logging.h"
Avi Drissmand70f89a2023-01-11 23:52:55[diff] [blame]27#include "base/functional/bind.h"
Wez9736ba782017-08-10 16:35:36[diff] [blame]28#include "base/location.h"
Wez9736ba782017-08-10 16:35:36[diff] [blame]29#include "base/memory/ref_counted.h"
Gabriel Charette19d2ae62018-04-10 14:10:58[diff] [blame]30#include "base/message_loop/message_pump_for_io.h"
Wez9736ba782017-08-10 16:35:36[diff] [blame]31#include "base/synchronization/lock.h"
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]32#include "base/task/current_thread.h"
Sean Mahere672a662023-01-09 21:42:28[diff] [blame]33#include "base/task/single_thread_task_runner.h"
Patrick Monette643cdf62021-10-15 19:13:42[diff] [blame]34#include "base/task/task_runner.h"
Ken Rockotdba46db2018-07-04 18:41:04[diff] [blame]35#include "mojo/core/platform_handle_in_transit.h"
Wez9736ba782017-08-10 16:35:36[diff] [blame]36
37namespace mojo {
Ken Rockotdba46db2018-07-04 18:41:04[diff] [blame]38namespace core {
Wez9736ba782017-08-10 16:35:36[diff] [blame]39
40namespace {
41
42const size_t kMaxBatchReadCapacity = 256 * 1024;
43
Wezab912ed2019-02-19 22:05:17[diff] [blame]44bool UnwrapFdioHandle(PlatformHandleInTransit handle,
45 PlatformHandleInTransit* out_handle,
46 Channel::Message::HandleInfoEntry* info) {
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]47 DCHECK(handle.handle().is_valid());
Wezda440922017-08-22 22:54:31[diff] [blame]48
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]49 if (!handle.handle().is_valid_fd()) {
Wezab912ed2019-02-19 22:05:17[diff] [blame]50 info->is_file_descriptor = false;
51 *out_handle = std::move(handle);
Wezda440922017-08-22 22:54:31[diff] [blame]52 return true;
53 }
Wezcf4155ae2017-08-16 21:01:14[diff] [blame]54
Wez6961d3b2021-10-08 12:22:43[diff] [blame]55 // Try to transfer the FD if possible, otherwise take a clone of it.
56 // This allows non-dup()d FDs to be efficiently unwrapped, while dup()d FDs
57 // have a new handle attached to the same underlying resource created.
Wezab912ed2019-02-19 22:05:17[diff] [blame]58 zx::handle result;
Wez6961d3b2021-10-08 12:22:43[diff] [blame]59 zx_status_t status = fdio_fd_transfer_or_clone(
60 handle.TakeHandle().ReleaseFD(), result.reset_and_get_address());
Wezab912ed2019-02-19 22:05:17[diff] [blame]61 if (status != ZX_OK) {
Wez6961d3b2021-10-08 12:22:43[diff] [blame]62 ZX_DLOG(ERROR, status) << "fdio_fd_transfer_or_clone";
Wez160e4352017-10-19 16:53:11[diff] [blame]63 return false;
64 }
Weze905d432017-10-19 03:54:44[diff] [blame]65
Wezab912ed2019-02-19 22:05:17[diff] [blame]66 info->is_file_descriptor = true;
67 *out_handle = PlatformHandleInTransit(PlatformHandle(std::move(result)));
Wezda440922017-08-22 22:54:31[diff] [blame]68 return true;
69}
70
Wezab912ed2019-02-19 22:05:17[diff] [blame]71PlatformHandle WrapFdioHandle(zx::handle handle,
72 Channel::Message::HandleInfoEntry info) {
73 if (!info.is_file_descriptor)
74 return PlatformHandle(std::move(handle));
Wezda440922017-08-22 22:54:31[diff] [blame]75
Wezab912ed2019-02-19 22:05:17[diff] [blame]76 base::ScopedFD out_fd;
77 zx_status_t status =
78 fdio_fd_create(handle.release(), base::ScopedFD::Receiver(out_fd).get());
79 if (status != ZX_OK) {
80 ZX_DLOG(ERROR, status) << "fdio_fd_create";
81 return PlatformHandle();
Wezda440922017-08-22 22:54:31[diff] [blame]82 }
Wezab912ed2019-02-19 22:05:17[diff] [blame]83 return PlatformHandle(std::move(out_fd));
Wezcf4155ae2017-08-16 21:01:14[diff] [blame]84}
85
Wez9736ba782017-08-10 16:35:36[diff] [blame]86// A view over a Channel::Message object. The write queue uses these since
87// large messages may need to be sent in chunks.
88class MessageView {
89 public:
90 // Owns |message|. |offset| indexes the first unsent byte in the message.
91 MessageView(Channel::MessagePtr message, size_t offset)
92 : message_(std::move(message)),
93 offset_(offset),
Robert Sesek5cd63eeb2019-08-02 21:18:07[diff] [blame]94 handles_(message_->TakeHandles()) {
Wez9736ba782017-08-10 16:35:36[diff] [blame]95 DCHECK_GT(message_->data_num_bytes(), offset_);
96 }
97
Anand K Mistry3357f672020-06-18 00:10:00[diff] [blame]98 MessageView(MessageView&& other) = default;
Wez9736ba782017-08-10 16:35:36[diff] [blame]99
Anand K Mistry3357f672020-06-18 00:10:00[diff] [blame]100 MessageView& operator=(MessageView&& other) = default;
Wez9736ba782017-08-10 16:35:36[diff] [blame]101
Peter Boströma8176282021-09-23 22:33:56[diff] [blame]102 MessageView(const MessageView&) = delete;
103 MessageView& operator=(const MessageView&) = delete;
104
Anand K Mistry48d93df2020-06-18 00:04:22[diff] [blame]105 ~MessageView() = default;
Wez9736ba782017-08-10 16:35:36[diff] [blame]106
107 const void* data() const {
108 return static_cast<const char*>(message_->data()) + offset_;
109 }
110
111 size_t data_num_bytes() const { return message_->data_num_bytes() - offset_; }
112
113 size_t data_offset() const { return offset_; }
114 void advance_data_offset(size_t num_bytes) {
115 DCHECK_GT(message_->data_num_bytes(), offset_ + num_bytes);
116 offset_ += num_bytes;
117 }
118
Ken Rockot36d44a572022-08-15 23:19:57[diff] [blame]119 std::vector<PlatformHandleInTransit> TakeHandles(bool unwrap_fds) {
120 if (handles_.empty() || !unwrap_fds)
121 return std::move(handles_);
Wezda440922017-08-22 22:54:31[diff] [blame]122
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]123 // We can only pass Fuchsia handles via IPC, so unwrap any FDIO file-
Wezab912ed2019-02-19 22:05:17[diff] [blame]124 // descriptors in |handles_| into the underlying handles, with metadata in
125 // the extra header to note which belong to FDIO.
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]126 auto* handles_info = reinterpret_cast<Channel::Message::HandleInfoEntry*>(
127 message_->mutable_extra_header());
128 memset(handles_info, 0, message_->extra_header_size());
Wezda440922017-08-22 22:54:31[diff] [blame]129
Wezab912ed2019-02-19 22:05:17[diff] [blame]130 // Since file descriptors unwrap to a single handle, we can unwrap in-place
131 // in the |handles_| vector.
132 for (size_t i = 0; i < handles_.size(); i++) {
133 if (!UnwrapFdioHandle(std::move(handles_[i]), &handles_[i],
134 &handles_info[i])) {
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]135 return std::vector<PlatformHandleInTransit>();
Wezab912ed2019-02-19 22:05:17[diff] [blame]136 }
Wezcf4155ae2017-08-16 21:01:14[diff] [blame]137 }
138 return std::move(handles_);
Wez9736ba782017-08-10 16:35:36[diff] [blame]139 }
140
141 private:
142 Channel::MessagePtr message_;
143 size_t offset_;
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]144 std::vector<PlatformHandleInTransit> handles_;
Wez9736ba782017-08-10 16:35:36[diff] [blame]145};
146
147class ChannelFuchsia : public Channel,
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]148 public base::CurrentThread::DestructionObserver,
Gabriel Charette19d2ae62018-04-10 14:10:58[diff] [blame]149 public base::MessagePumpForIO::ZxHandleWatcher {
Wez9736ba782017-08-10 16:35:36[diff] [blame]150 public:
151 ChannelFuchsia(Delegate* delegate,
152 ConnectionParams connection_params,
Ken Rockotfada58122018-12-11 16:49:54[diff] [blame]153 HandlePolicy handle_policy,
Gabriel Charettee926fc12019-12-16 19:00:02[diff] [blame]154 scoped_refptr<base::SingleThreadTaskRunner> io_task_runner)
Ken Rockotfada58122018-12-11 16:49:54[diff] [blame]155 : Channel(delegate, handle_policy),
Wez9736ba782017-08-10 16:35:36[diff] [blame]156 self_(this),
Wez157707d62018-07-10 22:48:47[diff] [blame]157 handle_(
158 connection_params.TakeEndpoint().TakePlatformHandle().TakeHandle()),
Wez9736ba782017-08-10 16:35:36[diff] [blame]159 io_task_runner_(io_task_runner) {
160 CHECK(handle_.is_valid());
161 }
162
Peter Boströmfeef05a2021-10-05 21:35:08[diff] [blame]163 ChannelFuchsia(const ChannelFuchsia&) = delete;
164 ChannelFuchsia& operator=(const ChannelFuchsia&) = delete;
165
Wez9736ba782017-08-10 16:35:36[diff] [blame]166 void Start() override {
167 if (io_task_runner_->RunsTasksInCurrentSequence()) {
168 StartOnIOThread();
169 } else {
170 io_task_runner_->PostTask(
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]171 FROM_HERE, base::BindOnce(&ChannelFuchsia::StartOnIOThread, this));
Wez9736ba782017-08-10 16:35:36[diff] [blame]172 }
173 }
174
175 void ShutDownImpl() override {
176 // Always shut down asynchronously when called through the public interface.
177 io_task_runner_->PostTask(
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]178 FROM_HERE, base::BindOnce(&ChannelFuchsia::ShutDownOnIOThread, this));
Wez9736ba782017-08-10 16:35:36[diff] [blame]179 }
180
181 void Write(MessagePtr message) override {
182 bool write_error = false;
183 {
184 base::AutoLock lock(write_lock_);
185 if (reject_writes_)
186 return;
187 if (!WriteNoLock(MessageView(std::move(message), 0)))
188 reject_writes_ = write_error = true;
189 }
190 if (write_error) {
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]191 // Do not synchronously invoke OnWriteError(). Write() may have been
192 // called by the delegate and we don't want to re-enter it.
Wez9736ba782017-08-10 16:35:36[diff] [blame]193 io_task_runner_->PostTask(
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]194 FROM_HERE, base::BindOnce(&ChannelFuchsia::OnWriteError, this,
195 Error::kDisconnected));
Wez9736ba782017-08-10 16:35:36[diff] [blame]196 }
197 }
198
199 void LeakHandle() override {
200 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
201 leak_handle_ = true;
202 }
203
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]204 bool GetReadPlatformHandles(const void* payload,
205 size_t payload_size,
206 size_t num_handles,
207 const void* extra_header,
208 size_t extra_header_size,
209 std::vector<PlatformHandle>* handles,
210 bool* deferred) override {
Wez9736ba782017-08-10 16:35:36[diff] [blame]211 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
212 if (num_handles > std::numeric_limits<uint16_t>::max())
213 return false;
Wezcf4155ae2017-08-16 21:01:14[diff] [blame]214
215 // Locate the handle info and verify there is enough of it.
216 if (!extra_header)
217 return false;
218 const auto* handles_info =
219 reinterpret_cast<const Channel::Message::HandleInfoEntry*>(
220 extra_header);
221 size_t handles_info_size = sizeof(handles_info[0]) * num_handles;
222 if (handles_info_size > extra_header_size)
223 return false;
224
Wezda440922017-08-22 22:54:31[diff] [blame]225 // If there are too few handles then we're not ready yet, so return true
226 // indicating things are OK, but leave |handles| empty.
Wezab912ed2019-02-19 22:05:17[diff] [blame]227 if (incoming_handles_.size() < num_handles)
Wez9736ba782017-08-10 16:35:36[diff] [blame]228 return true;
Wez9736ba782017-08-10 16:35:36[diff] [blame]229
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]230 handles->reserve(num_handles);
Wez9736ba782017-08-10 16:35:36[diff] [blame]231 for (size_t i = 0; i < num_handles; ++i) {
Wezab912ed2019-02-19 22:05:17[diff] [blame]232 handles->emplace_back(WrapFdioHandle(std::move(incoming_handles_.front()),
233 handles_info[i]));
234 DCHECK(handles->back().is_valid());
235 incoming_handles_.pop_front();
Wez9736ba782017-08-10 16:35:36[diff] [blame]236 }
Wez9736ba782017-08-10 16:35:36[diff] [blame]237 return true;
238 }
239
Ken Rockot36d44a572022-08-15 23:19:57[diff] [blame]240 bool GetReadPlatformHandlesForIpcz(
241 size_t num_handles,
242 std::vector<PlatformHandle>& handles) override {
243 if (incoming_handles_.size() < num_handles) {
244 return true;
245 }
246
247 DCHECK(handles.empty());
248 handles.reserve(num_handles);
249 for (size_t i = 0; i < num_handles; ++i) {
250 handles.emplace_back(std::move(incoming_handles_.front()));
251 incoming_handles_.pop_front();
252 }
253 return true;
254 }
255
Wez9736ba782017-08-10 16:35:36[diff] [blame]256 private:
Ken Rockotdba46db2018-07-04 18:41:04[diff] [blame]257 ~ChannelFuchsia() override { DCHECK(!read_watch_); }
Wez9736ba782017-08-10 16:35:36[diff] [blame]258
259 void StartOnIOThread() {
260 DCHECK(!read_watch_);
261
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]262 base::CurrentThread::Get()->AddDestructionObserver(this);
Wez9736ba782017-08-10 16:35:36[diff] [blame]263
Peter Boström84d239f2021-05-13 23:02:44[diff] [blame]264 read_watch_ =
265 std::make_unique<base::MessagePumpForIO::ZxHandleWatchController>(
266 FROM_HERE);
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]267 base::CurrentIOThread::Get()->WatchZxHandle(
Wez157707d62018-07-10 22:48:47[diff] [blame]268 handle_.get(), true /* persistent */,
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]269 ZX_CHANNEL_READABLE | ZX_CHANNEL_PEER_CLOSED, read_watch_.get(), this);
Wez9736ba782017-08-10 16:35:36[diff] [blame]270 }
271
272 void ShutDownOnIOThread() {
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]273 base::CurrentThread::Get()->RemoveDestructionObserver(this);
Wez9736ba782017-08-10 16:35:36[diff] [blame]274
275 read_watch_.reset();
276 if (leak_handle_)
Avi Drissmanebac4302022-01-10 21:56:58[diff] [blame]277 std::ignore = handle_.release();
Wez9736ba782017-08-10 16:35:36[diff] [blame]278 handle_.reset();
279
280 // May destroy the |this| if it was the last reference.
281 self_ = nullptr;
282 }
283
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]284 // base::CurrentThread::DestructionObserver:
Wez9736ba782017-08-10 16:35:36[diff] [blame]285 void WillDestroyCurrentMessageLoop() override {
286 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
287 if (self_)
288 ShutDownOnIOThread();
289 }
290
Gabriel Charette19d2ae62018-04-10 14:10:58[diff] [blame]291 // base::MessagePumpForIO::ZxHandleWatcher:
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]292 void OnZxHandleSignalled(zx_handle_t handle, zx_signals_t signals) override {
Wez9736ba782017-08-10 16:35:36[diff] [blame]293 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
Wez157707d62018-07-10 22:48:47[diff] [blame]294 CHECK_EQ(handle, handle_.get());
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]295 DCHECK((ZX_CHANNEL_READABLE | ZX_CHANNEL_PEER_CLOSED) & signals);
Wez9736ba782017-08-10 16:35:36[diff] [blame]296
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]297 // We always try to read message(s), even if ZX_CHANNEL_PEER_CLOSED, since
Wez9736ba782017-08-10 16:35:36[diff] [blame]298 // the peer may have closed while messages were still unread, in the pipe.
299
300 bool validation_error = false;
301 bool read_error = false;
302 size_t next_read_size = 0;
303 size_t buffer_capacity = 0;
304 size_t total_bytes_read = 0;
305 do {
306 buffer_capacity = next_read_size;
307 char* buffer = GetReadBuffer(&buffer_capacity);
308 DCHECK_GT(buffer_capacity, 0u);
309
310 uint32_t bytes_read = 0;
311 uint32_t handles_read = 0;
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]312 zx_handle_t handles[ZX_CHANNEL_MAX_MSG_HANDLES] = {};
Wez9736ba782017-08-10 16:35:36[diff] [blame]313
Wez157707d62018-07-10 22:48:47[diff] [blame]314 zx_status_t read_result =
Daniel Cheng37aad832022-02-27 01:05:30[diff] [blame]315 handle_.read(0, buffer, handles, buffer_capacity, std::size(handles),
Wez64f1c0c2019-04-17 00:38:25[diff] [blame]316 &bytes_read, &handles_read);
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]317 if (read_result == ZX_OK) {
Wez9736ba782017-08-10 16:35:36[diff] [blame]318 for (size_t i = 0; i < handles_read; ++i) {
Wez157707d62018-07-10 22:48:47[diff] [blame]319 incoming_handles_.emplace_back(handles[i]);
Wez9736ba782017-08-10 16:35:36[diff] [blame]320 }
321 total_bytes_read += bytes_read;
322 if (!OnReadComplete(bytes_read, &next_read_size)) {
323 read_error = true;
324 validation_error = true;
325 break;
326 }
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]327 } else if (read_result == ZX_ERR_BUFFER_TOO_SMALL) {
Daniel Cheng37aad832022-02-27 01:05:30[diff] [blame]328 DCHECK_LE(handles_read, std::size(handles));
Wez9736ba782017-08-10 16:35:36[diff] [blame]329 next_read_size = bytes_read;
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]330 } else if (read_result == ZX_ERR_SHOULD_WAIT) {
Wez9736ba782017-08-10 16:35:36[diff] [blame]331 break;
332 } else {
Wez157707d62018-07-10 22:48:47[diff] [blame]333 ZX_DLOG_IF(ERROR, read_result != ZX_ERR_PEER_CLOSED, read_result)
334 << "zx_channel_read";
Wez9736ba782017-08-10 16:35:36[diff] [blame]335 read_error = true;
336 break;
337 }
338 } while (total_bytes_read < kMaxBatchReadCapacity && next_read_size > 0);
339 if (read_error) {
340 // Stop receiving read notifications.
341 read_watch_.reset();
342 if (validation_error)
343 OnError(Error::kReceivedMalformedData);
344 else
345 OnError(Error::kDisconnected);
346 }
347 }
348
349 // Attempts to write a message directly to the channel. If the full message
350 // cannot be written, it's queued and a wait is initiated to write the message
351 // ASAP on the I/O thread.
352 bool WriteNoLock(MessageView message_view) {
353 uint32_t write_bytes = 0;
354 do {
355 message_view.advance_data_offset(write_bytes);
356
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]357 std::vector<PlatformHandleInTransit> outgoing_handles =
Ken Rockot36d44a572022-08-15 23:19:57[diff] [blame]358 message_view.TakeHandles(/*unwrap_fds=*/!is_for_ipcz());
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]359 zx_handle_t handles[ZX_CHANNEL_MAX_MSG_HANDLES] = {};
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]360 size_t handles_count = outgoing_handles.size();
Wez9736ba782017-08-10 16:35:36[diff] [blame]361
Daniel Cheng37aad832022-02-27 01:05:30[diff] [blame]362 DCHECK_LE(handles_count, std::size(handles));
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]363 for (size_t i = 0; i < handles_count; ++i) {
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]364 DCHECK(outgoing_handles[i].handle().is_valid());
365 handles[i] = outgoing_handles[i].handle().GetHandle().get();
Wez9736ba782017-08-10 16:35:36[diff] [blame]366 }
367
368 write_bytes = std::min(message_view.data_num_bytes(),
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]369 static_cast<size_t>(ZX_CHANNEL_MAX_MSG_BYTES));
Wez157707d62018-07-10 22:48:47[diff] [blame]370 zx_status_t result = handle_.write(0, message_view.data(), write_bytes,
371 handles, handles_count);
Wez676b5fae2018-06-20 22:34:08[diff] [blame]372 // zx_channel_write() consumes |handles| whether or not it succeeds, so
373 // release() our copies now, to avoid them being double-closed.
374 for (auto& outgoing_handle : outgoing_handles)
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]375 outgoing_handle.CompleteTransit();
Wez676b5fae2018-06-20 22:34:08[diff] [blame]376
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]377 if (result != ZX_OK) {
Alison Gale923a33e2024-04-22 23:34:28[diff] [blame]378 // TODO(crbug.com/42050611): Handle ZX_ERR_SHOULD_WAIT flow-control
Fabrice de Gans-Riberi69f65b42020-09-30 23:37:14[diff] [blame]379 // errors, once the platform starts generating them.
Wez157707d62018-07-10 22:48:47[diff] [blame]380 ZX_DLOG_IF(ERROR, result != ZX_ERR_PEER_CLOSED, result)
381 << "WriteNoLock(zx_channel_write)";
Wez9736ba782017-08-10 16:35:36[diff] [blame]382 return false;
383 }
384
Wez9736ba782017-08-10 16:35:36[diff] [blame]385 } while (write_bytes < message_view.data_num_bytes());
386
387 return true;
388 }
389
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]390 void OnWriteError(Error error) {
391 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
392 DCHECK(reject_writes_);
393
394 if (error == Error::kDisconnected) {
395 // If we can't write because the pipe is disconnected then continue
396 // reading to fetch any in-flight messages, relying on end-of-stream to
397 // signal the actual disconnection.
398 if (read_watch_) {
Alison Gale923a33e2024-04-22 23:34:28[diff] [blame]399 // TODO(crbug.com/42050611): When we add flow-control for writes, we
400 // also need to reset the write-watcher here.
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]401 return;
402 }
403 }
404
405 OnError(error);
406 }
407
Wez9736ba782017-08-10 16:35:36[diff] [blame]408 // Keeps the Channel alive at least until explicit shutdown on the IO thread.
409 scoped_refptr<Channel> self_;
410
Wez157707d62018-07-10 22:48:47[diff] [blame]411 zx::channel handle_;
Gabriel Charettee926fc12019-12-16 19:00:02[diff] [blame]412 scoped_refptr<base::SingleThreadTaskRunner> io_task_runner_;
Wez9736ba782017-08-10 16:35:36[diff] [blame]413
414 // These members are only used on the IO thread.
Gabriel Charette19d2ae62018-04-10 14:10:58[diff] [blame]415 std::unique_ptr<base::MessagePumpForIO::ZxHandleWatchController> read_watch_;
Wez157707d62018-07-10 22:48:47[diff] [blame]416 base::circular_deque<zx::handle> incoming_handles_;
Wez9736ba782017-08-10 16:35:36[diff] [blame]417 bool leak_handle_ = false;
418
419 base::Lock write_lock_;
420 bool reject_writes_ = false;
Wez9736ba782017-08-10 16:35:36[diff] [blame]421};
422
423} // namespace
424
425// static
426scoped_refptr<Channel> Channel::Create(
427 Delegate* delegate,
428 ConnectionParams connection_params,
Ken Rockotfada58122018-12-11 16:49:54[diff] [blame]429 HandlePolicy handle_policy,
Gabriel Charettee926fc12019-12-16 19:00:02[diff] [blame]430 scoped_refptr<base::SingleThreadTaskRunner> io_task_runner) {
Wez9736ba782017-08-10 16:35:36[diff] [blame]431 return new ChannelFuchsia(delegate, std::move(connection_params),
Ken Rockotfada58122018-12-11 16:49:54[diff] [blame]432 handle_policy, std::move(io_task_runner));
Wez9736ba782017-08-10 16:35:36[diff] [blame]433}
434
Ken Rockotdba46db2018-07-04 18:41:04[diff] [blame]435} // namespace core
Wez9736ba782017-08-10 16:35:36[diff] [blame]436} // namespace mojo