ipc_sync_channel.cc | Explore in Territory

// Copyright 2012 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "ipc/ipc_sync_channel.h"

#include <stddef.h>
#include <stdint.h>

#include <memory>
#include <utility>

#include "base/functional/bind.h"
#include "base/location.h"
#include "base/logging.h"
#include "base/memory/ptr_util.h"
#include "base/memory/raw_ptr.h"
#include "base/run_loop.h"
#include "base/synchronization/waitable_event.h"
#include "base/task/sequenced_task_runner.h"
#include "base/task/single_thread_task_runner.h"
#include "base/trace_event/trace_event.h"
#include "build/build_config.h"
#include "ipc/ipc_channel_factory.h"
#include "ipc/ipc_logging.h"
#include "ipc/ipc_message.h"
#include "ipc/ipc_message_macros.h"
#include "ipc/ipc_sync_message.h"
#include "mojo/public/cpp/bindings/sync_event_watcher.h"
#include "third_party/abseil-cpp/absl/base/attributes.h"

#if !BUILDFLAG(IPC_MESSAGE_LOG_ENABLED)
#include "ipc/trace_ipc_message.h"
#endif

WaitableEvent;

namespace IPC {

namespace {

// A generic callback used when watching handles synchronously. Sets |*signal|
// to true.
void OnEventReady(bool* signal) { … }

// Holds a pointer to the per-thread ReceivedSyncMsgQueue object.
ABSL_CONST_INIT thread_local SyncChannel::ReceivedSyncMsgQueue* received_queue = …;

}  // namespace

// When we're blocked in a Send(), we need to process incoming synchronous
// messages right away because it could be blocking our reply (either
// directly from the same object we're calling, or indirectly through one or
// more other channels).  That means that in SyncContext's OnMessageReceived,
// we need to process sync message right away if we're blocked.  However a
// simple check isn't sufficient, because the listener thread can be in the
// process of calling Send.
// To work around this, when SyncChannel filters a sync message, it sets
// an event that the listener thread waits on during its Send() call.  This
// allows us to dispatch incoming sync messages when blocked.  The race
// condition is handled because if Send is in the process of being called, it
// will check the event.  In case the listener thread isn't sending a message,
// we queue a task on the listener thread to dispatch the received messages.
// The messages are stored in this queue object that's shared among all
// SyncChannel objects on the same thread (since one object can receive a
// sync message while another one is blocked).

class SyncChannel::ReceivedSyncMsgQueue :
    public base::RefCountedThreadSafe<ReceivedSyncMsgQueue> { … };

SyncChannel::SyncContext::SyncContext(
    Listener* listener,
    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
    WaitableEvent* shutdown_event)
    : … { … }

void SyncChannel::SyncContext::OnSendDoneEventSignaled(
    base::RunLoop* nested_loop,
    base::WaitableEvent* event) { … }

SyncChannel::SyncContext::~SyncContext() { … }

// Adds information about an outgoing sync message to the context so that
// we know how to deserialize the reply. Returns |true| if the message was added
// to the context or |false| if it was rejected (e.g. due to shutdown.)
bool SyncChannel::SyncContext::Push(SyncMessage* sync_msg) { … }

bool SyncChannel::SyncContext::Pop() { … }

base::WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() { … }

base::WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() { … }

void SyncChannel::SyncContext::DispatchMessages() { … }

bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) { … }

void SyncChannel::SyncContext::Clear() { … }

bool SyncChannel::SyncContext::OnMessageReceived(const Message& msg) { … }

void SyncChannel::SyncContext::OnChannelError() { … }

void SyncChannel::SyncContext::OnChannelOpened() { … }

void SyncChannel::SyncContext::OnChannelClosed() { … }

void SyncChannel::SyncContext::CancelPendingSends() { … }

void SyncChannel::SyncContext::OnShutdownEventSignaled(WaitableEvent* event) { … }

// static
std::unique_ptr<SyncChannel> SyncChannel::Create(
    const IPC::ChannelHandle& channel_handle,
    Channel::Mode mode,
    Listener* listener,
    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
    bool create_pipe_now,
    base::WaitableEvent* shutdown_event) { … }

// static
std::unique_ptr<SyncChannel> SyncChannel::Create(
    Listener* listener,
    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
    WaitableEvent* shutdown_event) { … }

SyncChannel::SyncChannel(
    Listener* listener,
    const scoped_refptr<base::SingleThreadTaskRunner>& ipc_task_runner,
    const scoped_refptr<base::SingleThreadTaskRunner>& listener_task_runner,
    WaitableEvent* shutdown_event)
    : … { … }

void SyncChannel::AddListenerTaskRunner(
    int32_t routing_id,
    scoped_refptr<base::SingleThreadTaskRunner> task_runner) { … }

void SyncChannel::RemoveListenerTaskRunner(int32_t routing_id) { … }

SyncChannel::~SyncChannel() = default;

void SyncChannel::SetRestrictDispatchChannelGroup(int group) { … }

scoped_refptr<SyncMessageFilter> SyncChannel::CreateSyncMessageFilter() { … }

bool SyncChannel::Send(Message* message) { … }

void SyncChannel::WaitForReply(mojo::SyncHandleRegistry* registry,
                               SyncContext* context) { … }

void SyncChannel::OnDispatchEventSignaled(base::WaitableEvent* event) { … }

void SyncChannel::StartWatching() { … }

void SyncChannel::OnChannelInit() { … }

}  // namespace IPC
chromium/ipc/ipc_sync_channel.cc
