Google Git
Sign in
chromium / chromium / src / lkgr / . / mojo / core / channel_fuchsia.cc
blob: a2016a233db781f7afc6ae63fc33323f714395f0 [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,
Egor Paskoa2a00112025-09-03 17:45:22[diff] [blame]209 std::vector<PlatformHandle>* handles) override {
Wez9736ba782017-08-10 16:35:36[diff] [blame]210 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
211 if (num_handles > std::numeric_limits<uint16_t>::max())
212 return false;
Wezcf4155ae2017-08-16 21:01:14[diff] [blame]213
214 // Locate the handle info and verify there is enough of it.
215 if (!extra_header)
216 return false;
217 const auto* handles_info =
218 reinterpret_cast<const Channel::Message::HandleInfoEntry*>(
219 extra_header);
220 size_t handles_info_size = sizeof(handles_info[0]) * num_handles;
221 if (handles_info_size > extra_header_size)
222 return false;
223
Wezda440922017-08-22 22:54:31[diff] [blame]224 // If there are too few handles then we're not ready yet, so return true
225 // indicating things are OK, but leave |handles| empty.
Wezab912ed2019-02-19 22:05:17[diff] [blame]226 if (incoming_handles_.size() < num_handles)
Wez9736ba782017-08-10 16:35:36[diff] [blame]227 return true;
Wez9736ba782017-08-10 16:35:36[diff] [blame]228
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]229 handles->reserve(num_handles);
Wez9736ba782017-08-10 16:35:36[diff] [blame]230 for (size_t i = 0; i < num_handles; ++i) {
Wezab912ed2019-02-19 22:05:17[diff] [blame]231 handles->emplace_back(WrapFdioHandle(std::move(incoming_handles_.front()),
232 handles_info[i]));
233 DCHECK(handles->back().is_valid());
234 incoming_handles_.pop_front();
Wez9736ba782017-08-10 16:35:36[diff] [blame]235 }
Wez9736ba782017-08-10 16:35:36[diff] [blame]236 return true;
237 }
238
Ken Rockot36d44a572022-08-15 23:19:57[diff] [blame]239 bool GetReadPlatformHandlesForIpcz(
240 size_t num_handles,
241 std::vector<PlatformHandle>& handles) override {
242 if (incoming_handles_.size() < num_handles) {
243 return true;
244 }
245
246 DCHECK(handles.empty());
247 handles.reserve(num_handles);
248 for (size_t i = 0; i < num_handles; ++i) {
249 handles.emplace_back(std::move(incoming_handles_.front()));
250 incoming_handles_.pop_front();
251 }
252 return true;
253 }
254
Wez9736ba782017-08-10 16:35:36[diff] [blame]255 private:
Ken Rockotdba46db2018-07-04 18:41:04[diff] [blame]256 ~ChannelFuchsia() override { DCHECK(!read_watch_); }
Wez9736ba782017-08-10 16:35:36[diff] [blame]257
258 void StartOnIOThread() {
259 DCHECK(!read_watch_);
260
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]261 base::CurrentThread::Get()->AddDestructionObserver(this);
Wez9736ba782017-08-10 16:35:36[diff] [blame]262
Peter Boström84d239f2021-05-13 23:02:44[diff] [blame]263 read_watch_ =
264 std::make_unique<base::MessagePumpForIO::ZxHandleWatchController>(
265 FROM_HERE);
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]266 base::CurrentIOThread::Get()->WatchZxHandle(
Wez157707d62018-07-10 22:48:47[diff] [blame]267 handle_.get(), true /* persistent */,
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]268 ZX_CHANNEL_READABLE | ZX_CHANNEL_PEER_CLOSED, read_watch_.get(), this);
Wez9736ba782017-08-10 16:35:36[diff] [blame]269 }
270
271 void ShutDownOnIOThread() {
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]272 base::CurrentThread::Get()->RemoveDestructionObserver(this);
Wez9736ba782017-08-10 16:35:36[diff] [blame]273
274 read_watch_.reset();
275 if (leak_handle_)
Avi Drissmanebac4302022-01-10 21:56:58[diff] [blame]276 std::ignore = handle_.release();
Wez9736ba782017-08-10 16:35:36[diff] [blame]277 handle_.reset();
278
279 // May destroy the |this| if it was the last reference.
280 self_ = nullptr;
281 }
282
Carlos Caballerob25fe8472020-07-17 10:27:17[diff] [blame]283 // base::CurrentThread::DestructionObserver:
Wez9736ba782017-08-10 16:35:36[diff] [blame]284 void WillDestroyCurrentMessageLoop() override {
285 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
286 if (self_)
287 ShutDownOnIOThread();
288 }
289
Gabriel Charette19d2ae62018-04-10 14:10:58[diff] [blame]290 // base::MessagePumpForIO::ZxHandleWatcher:
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]291 void OnZxHandleSignalled(zx_handle_t handle, zx_signals_t signals) override {
Wez9736ba782017-08-10 16:35:36[diff] [blame]292 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
Wez157707d62018-07-10 22:48:47[diff] [blame]293 CHECK_EQ(handle, handle_.get());
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]294 DCHECK((ZX_CHANNEL_READABLE | ZX_CHANNEL_PEER_CLOSED) & signals);
Wez9736ba782017-08-10 16:35:36[diff] [blame]295
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]296 // We always try to read message(s), even if ZX_CHANNEL_PEER_CLOSED, since
Wez9736ba782017-08-10 16:35:36[diff] [blame]297 // the peer may have closed while messages were still unread, in the pipe.
298
299 bool validation_error = false;
300 bool read_error = false;
301 size_t next_read_size = 0;
302 size_t buffer_capacity = 0;
303 size_t total_bytes_read = 0;
304 do {
305 buffer_capacity = next_read_size;
306 char* buffer = GetReadBuffer(&buffer_capacity);
307 DCHECK_GT(buffer_capacity, 0u);
308
309 uint32_t bytes_read = 0;
310 uint32_t handles_read = 0;
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]311 zx_handle_t handles[ZX_CHANNEL_MAX_MSG_HANDLES] = {};
Wez9736ba782017-08-10 16:35:36[diff] [blame]312
Wez157707d62018-07-10 22:48:47[diff] [blame]313 zx_status_t read_result =
Daniel Cheng37aad832022-02-27 01:05:30[diff] [blame]314 handle_.read(0, buffer, handles, buffer_capacity, std::size(handles),
Wez64f1c0c2019-04-17 00:38:25[diff] [blame]315 &bytes_read, &handles_read);
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]316 if (read_result == ZX_OK) {
Wez9736ba782017-08-10 16:35:36[diff] [blame]317 for (size_t i = 0; i < handles_read; ++i) {
Wez157707d62018-07-10 22:48:47[diff] [blame]318 incoming_handles_.emplace_back(handles[i]);
Wez9736ba782017-08-10 16:35:36[diff] [blame]319 }
320 total_bytes_read += bytes_read;
321 if (!OnReadComplete(bytes_read, &next_read_size)) {
322 read_error = true;
323 validation_error = true;
324 break;
325 }
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]326 } else if (read_result == ZX_ERR_BUFFER_TOO_SMALL) {
Daniel Cheng37aad832022-02-27 01:05:30[diff] [blame]327 DCHECK_LE(handles_read, std::size(handles));
Wez9736ba782017-08-10 16:35:36[diff] [blame]328 next_read_size = bytes_read;
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]329 } else if (read_result == ZX_ERR_SHOULD_WAIT) {
Wez9736ba782017-08-10 16:35:36[diff] [blame]330 break;
331 } else {
Wez157707d62018-07-10 22:48:47[diff] [blame]332 ZX_DLOG_IF(ERROR, read_result != ZX_ERR_PEER_CLOSED, read_result)
333 << "zx_channel_read";
Wez9736ba782017-08-10 16:35:36[diff] [blame]334 read_error = true;
335 break;
336 }
337 } while (total_bytes_read < kMaxBatchReadCapacity && next_read_size > 0);
338 if (read_error) {
339 // Stop receiving read notifications.
340 read_watch_.reset();
341 if (validation_error)
342 OnError(Error::kReceivedMalformedData);
343 else
344 OnError(Error::kDisconnected);
345 }
346 }
347
348 // Attempts to write a message directly to the channel. If the full message
349 // cannot be written, it's queued and a wait is initiated to write the message
350 // ASAP on the I/O thread.
351 bool WriteNoLock(MessageView message_view) {
352 uint32_t write_bytes = 0;
353 do {
354 message_view.advance_data_offset(write_bytes);
355
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]356 std::vector<PlatformHandleInTransit> outgoing_handles =
Ken Rockot36d44a572022-08-15 23:19:57[diff] [blame]357 message_view.TakeHandles(/*unwrap_fds=*/!is_for_ipcz());
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]358 zx_handle_t handles[ZX_CHANNEL_MAX_MSG_HANDLES] = {};
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]359 size_t handles_count = outgoing_handles.size();
Wez9736ba782017-08-10 16:35:36[diff] [blame]360
Daniel Cheng37aad832022-02-27 01:05:30[diff] [blame]361 DCHECK_LE(handles_count, std::size(handles));
Wez6d83a0b2017-11-28 01:17:15[diff] [blame]362 for (size_t i = 0; i < handles_count; ++i) {
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]363 DCHECK(outgoing_handles[i].handle().is_valid());
364 handles[i] = outgoing_handles[i].handle().GetHandle().get();
Wez9736ba782017-08-10 16:35:36[diff] [blame]365 }
366
367 write_bytes = std::min(message_view.data_num_bytes(),
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]368 static_cast<size_t>(ZX_CHANNEL_MAX_MSG_BYTES));
Wez157707d62018-07-10 22:48:47[diff] [blame]369 zx_status_t result = handle_.write(0, message_view.data(), write_bytes,
370 handles, handles_count);
Wez676b5fae2018-06-20 22:34:08[diff] [blame]371 // zx_channel_write() consumes |handles| whether or not it succeeds, so
372 // release() our copies now, to avoid them being double-closed.
373 for (auto& outgoing_handle : outgoing_handles)
Ken Rockot043152da62018-06-29 03:22:16[diff] [blame]374 outgoing_handle.CompleteTransit();
Wez676b5fae2018-06-20 22:34:08[diff] [blame]375
Scott Grahamfe0e9f462017-09-18 21:25:04[diff] [blame]376 if (result != ZX_OK) {
Alison Gale923a33e2024-04-22 23:34:28[diff] [blame]377 // TODO(crbug.com/42050611): Handle ZX_ERR_SHOULD_WAIT flow-control
Fabrice de Gans-Riberi69f65b42020-09-30 23:37:14[diff] [blame]378 // errors, once the platform starts generating them.
Wez157707d62018-07-10 22:48:47[diff] [blame]379 ZX_DLOG_IF(ERROR, result != ZX_ERR_PEER_CLOSED, result)
380 << "WriteNoLock(zx_channel_write)";
Wez9736ba782017-08-10 16:35:36[diff] [blame]381 return false;
382 }
383
Wez9736ba782017-08-10 16:35:36[diff] [blame]384 } while (write_bytes < message_view.data_num_bytes());
385
386 return true;
387 }
388
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]389 void OnWriteError(Error error) {
390 DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
391 DCHECK(reject_writes_);
392
393 if (error == Error::kDisconnected) {
394 // If we can't write because the pipe is disconnected then continue
395 // reading to fetch any in-flight messages, relying on end-of-stream to
396 // signal the actual disconnection.
397 if (read_watch_) {
Alison Gale923a33e2024-04-22 23:34:28[diff] [blame]398 // TODO(crbug.com/42050611): When we add flow-control for writes, we
399 // also need to reset the write-watcher here.
Wez3e64a8a2018-03-13 05:36:06[diff] [blame]400 return;
401 }
402 }
403
404 OnError(error);
405 }
406
Wez9736ba782017-08-10 16:35:36[diff] [blame]407 // Keeps the Channel alive at least until explicit shutdown on the IO thread.
408 scoped_refptr<Channel> self_;
409
Wez157707d62018-07-10 22:48:47[diff] [blame]410 zx::channel handle_;
Gabriel Charettee926fc12019-12-16 19:00:02[diff] [blame]411 scoped_refptr<base::SingleThreadTaskRunner> io_task_runner_;
Wez9736ba782017-08-10 16:35:36[diff] [blame]412
413 // These members are only used on the IO thread.
Gabriel Charette19d2ae62018-04-10 14:10:58[diff] [blame]414 std::unique_ptr<base::MessagePumpForIO::ZxHandleWatchController> read_watch_;
Wez157707d62018-07-10 22:48:47[diff] [blame]415 base::circular_deque<zx::handle> incoming_handles_;
Wez9736ba782017-08-10 16:35:36[diff] [blame]416 bool leak_handle_ = false;
417
418 base::Lock write_lock_;
419 bool reject_writes_ = false;
Wez9736ba782017-08-10 16:35:36[diff] [blame]420};
421
422} // namespace
423
424// static
425scoped_refptr<Channel> Channel::Create(
426 Delegate* delegate,
427 ConnectionParams connection_params,
Ken Rockotfada58122018-12-11 16:49:54[diff] [blame]428 HandlePolicy handle_policy,
Gabriel Charettee926fc12019-12-16 19:00:02[diff] [blame]429 scoped_refptr<base::SingleThreadTaskRunner> io_task_runner) {
Wez9736ba782017-08-10 16:35:36[diff] [blame]430 return new ChannelFuchsia(delegate, std::move(connection_params),
Ken Rockotfada58122018-12-11 16:49:54[diff] [blame]431 handle_policy, std::move(io_task_runner));
Wez9736ba782017-08-10 16:35:36[diff] [blame]432}
433
Ken Rockotdba46db2018-07-04 18:41:04[diff] [blame]434} // namespace core
Wez9736ba782017-08-10 16:35:36[diff] [blame]435} // namespace mojo