зеркало из https://github.com/mozilla/gecko-dev.git
Bug 1253123 - Remove ipc_sync_message (r=jld)
This commit is contained in:
Bill McCloskey
Родитель
3888bea25e
Коммит
2bdeb1e82b
|
|
@ -36,8 +36,6 @@ UNIFIED_SOURCES += [
|
|||
'src/chrome/common/ipc_channel.cc',
|
||||
'src/chrome/common/ipc_channel_proxy.cc',
|
||||
'src/chrome/common/ipc_message.cc',
|
||||
'src/chrome/common/ipc_sync_channel.cc',
|
||||
'src/chrome/common/ipc_sync_message.cc',
|
||||
'src/chrome/common/message_router.cc',
|
||||
'src/chrome/common/notification_service.cc',
|
||||
]
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@
|
|||
#define CHROME_COMMON_CHILD_THREAD_H_
|
||||
|
||||
#include "base/thread.h"
|
||||
#include "chrome/common/ipc_sync_channel.h"
|
||||
#include "chrome/common/message_router.h"
|
||||
#include "mozilla/UniquePtr.h"
|
||||
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@
|
|||
#if defined(OS_POSIX)
|
||||
#include "chrome/common/file_descriptor_set_posix.h"
|
||||
#endif
|
||||
#include "chrome/common/ipc_sync_message.h"
|
||||
#include "chrome/common/ipc_message.h"
|
||||
#include "chrome/common/transport_dib.h"
|
||||
|
||||
namespace IPC {
|
||||
|
|
|
|||
|
|
@ -1,452 +0,0 @@
|
|||
// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file.
|
||||
|
||||
#include "chrome/common/ipc_sync_channel.h"
|
||||
|
||||
#include "base/logging.h"
|
||||
#include "base/thread_local.h"
|
||||
#include "base/message_loop.h"
|
||||
#include "base/waitable_event.h"
|
||||
#include "base/waitable_event_watcher.h"
|
||||
#include "chrome/common/ipc_sync_message.h"
|
||||
#include "nsISupportsImpl.h"
|
||||
|
||||
using base::TimeDelta;
|
||||
using base::TimeTicks;
|
||||
using base::WaitableEvent;
|
||||
|
||||
namespace IPC {
|
||||
// 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:
|
||||
NS_INLINE_DECL_THREADSAFE_REFCOUNTING(SyncChannel::ReceivedSyncMsgQueue)
|
||||
|
||||
static base::ThreadLocalPointer<ReceivedSyncMsgQueue>& get_tls_ptr() {
|
||||
static base::ThreadLocalPointer<ReceivedSyncMsgQueue> tls_ptr;
|
||||
return tls_ptr;
|
||||
}
|
||||
|
||||
// Returns the ReceivedSyncMsgQueue instance for this thread, creating one
|
||||
// if necessary. Call RemoveContext on the same thread when done.
|
||||
static ReceivedSyncMsgQueue* AddContext() {
|
||||
// We want one ReceivedSyncMsgQueue per listener thread (i.e. since multiple
|
||||
// SyncChannel objects can block the same thread).
|
||||
ReceivedSyncMsgQueue* rv = get_tls_ptr().Get();
|
||||
if (!rv) {
|
||||
rv = new ReceivedSyncMsgQueue();
|
||||
get_tls_ptr().Set(rv);
|
||||
}
|
||||
rv->listener_count_++;
|
||||
return rv;
|
||||
}
|
||||
|
||||
// Called on IPC thread when a synchronous message or reply arrives.
|
||||
void QueueMessage(const Message& msg, SyncChannel::SyncContext* context) {
|
||||
bool was_task_pending;
|
||||
{
|
||||
AutoLock auto_lock(message_lock_);
|
||||
|
||||
was_task_pending = task_pending_;
|
||||
task_pending_ = true;
|
||||
|
||||
// We set the event in case the listener thread is blocked (or is about
|
||||
// to). In case it's not, the PostTask dispatches the messages.
|
||||
message_queue_.push_back(QueuedMessage(new Message(msg), context));
|
||||
}
|
||||
|
||||
dispatch_event_.Signal();
|
||||
if (!was_task_pending) {
|
||||
listener_message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
|
||||
this, &ReceivedSyncMsgQueue::DispatchMessagesTask));
|
||||
}
|
||||
}
|
||||
|
||||
void QueueReply(const Message &msg, SyncChannel::SyncContext* context) {
|
||||
received_replies_.push_back(QueuedMessage(new Message(msg), context));
|
||||
}
|
||||
|
||||
// Called on the listener's thread to process any queues synchronous
|
||||
// messages.
|
||||
void DispatchMessagesTask() {
|
||||
{
|
||||
AutoLock auto_lock(message_lock_);
|
||||
task_pending_ = false;
|
||||
}
|
||||
DispatchMessages();
|
||||
}
|
||||
|
||||
void DispatchMessages() {
|
||||
while (true) {
|
||||
Message* message;
|
||||
RefPtr<SyncChannel::SyncContext> context;
|
||||
{
|
||||
AutoLock auto_lock(message_lock_);
|
||||
if (message_queue_.empty())
|
||||
break;
|
||||
|
||||
message = message_queue_.front().message;
|
||||
context = message_queue_.front().context;
|
||||
message_queue_.pop_front();
|
||||
}
|
||||
|
||||
context->OnDispatchMessage(*message);
|
||||
delete message;
|
||||
}
|
||||
}
|
||||
|
||||
// SyncChannel calls this in its destructor.
|
||||
void RemoveContext(SyncContext* context) {
|
||||
AutoLock auto_lock(message_lock_);
|
||||
|
||||
SyncMessageQueue::iterator iter = message_queue_.begin();
|
||||
while (iter != message_queue_.end()) {
|
||||
if (iter->context == context) {
|
||||
delete iter->message;
|
||||
iter = message_queue_.erase(iter);
|
||||
} else {
|
||||
iter++;
|
||||
}
|
||||
}
|
||||
|
||||
if (--listener_count_ == 0) {
|
||||
DCHECK(get_tls_ptr().Get());
|
||||
get_tls_ptr().Set(NULL);
|
||||
}
|
||||
}
|
||||
|
||||
WaitableEvent* dispatch_event() { return &dispatch_event_; }
|
||||
MessageLoop* listener_message_loop() { return listener_message_loop_; }
|
||||
|
||||
// Called on the ipc thread to check if we can unblock any current Send()
|
||||
// calls based on a queued reply.
|
||||
void DispatchReplies() {
|
||||
for (size_t i = 0; i < received_replies_.size(); ++i) {
|
||||
Message* message = received_replies_[i].message;
|
||||
if (received_replies_[i].context->TryToUnblockListener(message)) {
|
||||
delete message;
|
||||
received_replies_.erase(received_replies_.begin() + i);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected:
|
||||
~ReceivedSyncMsgQueue() {}
|
||||
|
||||
private:
|
||||
// See the comment in SyncChannel::SyncChannel for why this event is created
|
||||
// as manual reset.
|
||||
ReceivedSyncMsgQueue() :
|
||||
dispatch_event_(true, false),
|
||||
listener_message_loop_(MessageLoop::current()),
|
||||
task_pending_(false),
|
||||
listener_count_(0) {
|
||||
}
|
||||
|
||||
// Holds information about a queued synchronous message or reply.
|
||||
struct QueuedMessage {
|
||||
QueuedMessage(Message* m, SyncContext* c) : message(m), context(c) { }
|
||||
Message* message;
|
||||
RefPtr<SyncChannel::SyncContext> context;
|
||||
};
|
||||
|
||||
typedef std::deque<QueuedMessage> SyncMessageQueue;
|
||||
SyncMessageQueue message_queue_;
|
||||
|
||||
std::vector<QueuedMessage> received_replies_;
|
||||
|
||||
// Set when we got a synchronous message that we must respond to as the
|
||||
// sender needs its reply before it can reply to our original synchronous
|
||||
// message.
|
||||
WaitableEvent dispatch_event_;
|
||||
MessageLoop* listener_message_loop_;
|
||||
Lock message_lock_;
|
||||
bool task_pending_;
|
||||
int listener_count_;
|
||||
};
|
||||
|
||||
SyncChannel::SyncContext::SyncContext(
|
||||
Channel::Listener* listener,
|
||||
MessageFilter* filter,
|
||||
MessageLoop* ipc_thread,
|
||||
WaitableEvent* shutdown_event)
|
||||
: ChannelProxy::Context(listener, filter, ipc_thread),
|
||||
received_sync_msgs_(ReceivedSyncMsgQueue::AddContext()),
|
||||
shutdown_event_(shutdown_event) {
|
||||
}
|
||||
|
||||
SyncChannel::SyncContext::~SyncContext() {
|
||||
while (!deserializers_.empty())
|
||||
Pop();
|
||||
}
|
||||
|
||||
// Adds information about an outgoing sync message to the context so that
|
||||
// we know how to deserialize the reply. Returns a handle that's set when
|
||||
// the reply has arrived.
|
||||
void SyncChannel::SyncContext::Push(SyncMessage* sync_msg) {
|
||||
// The event is created as manual reset because in between Signal and
|
||||
// OnObjectSignalled, another Send can happen which would stop the watcher
|
||||
// from being called. The event would get watched later, when the nested
|
||||
// Send completes, so the event will need to remain set.
|
||||
PendingSyncMsg pending(SyncMessage::GetMessageId(*sync_msg),
|
||||
sync_msg->GetReplyDeserializer(),
|
||||
new WaitableEvent(true, false));
|
||||
AutoLock auto_lock(deserializers_lock_);
|
||||
deserializers_.push_back(pending);
|
||||
}
|
||||
|
||||
bool SyncChannel::SyncContext::Pop() {
|
||||
bool result;
|
||||
{
|
||||
AutoLock auto_lock(deserializers_lock_);
|
||||
PendingSyncMsg msg = deserializers_.back();
|
||||
delete msg.deserializer;
|
||||
delete msg.done_event;
|
||||
msg.done_event = NULL;
|
||||
deserializers_.pop_back();
|
||||
result = msg.send_result;
|
||||
}
|
||||
|
||||
// We got a reply to a synchronous Send() call that's blocking the listener
|
||||
// thread. However, further down the call stack there could be another
|
||||
// blocking Send() call, whose reply we received after we made this last
|
||||
// Send() call. So check if we have any queued replies available that
|
||||
// can now unblock the listener thread.
|
||||
ipc_message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
|
||||
received_sync_msgs_.get(), &ReceivedSyncMsgQueue::DispatchReplies));
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
WaitableEvent* SyncChannel::SyncContext::GetSendDoneEvent() {
|
||||
AutoLock auto_lock(deserializers_lock_);
|
||||
return deserializers_.back().done_event;
|
||||
}
|
||||
|
||||
WaitableEvent* SyncChannel::SyncContext::GetDispatchEvent() {
|
||||
return received_sync_msgs_->dispatch_event();
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::DispatchMessages() {
|
||||
received_sync_msgs_->DispatchMessages();
|
||||
}
|
||||
|
||||
bool SyncChannel::SyncContext::TryToUnblockListener(const Message* msg) {
|
||||
AutoLock auto_lock(deserializers_lock_);
|
||||
if (deserializers_.empty() ||
|
||||
!SyncMessage::IsMessageReplyTo(*msg, deserializers_.back().id)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!msg->is_reply_error()) {
|
||||
deserializers_.back().send_result = deserializers_.back().deserializer->
|
||||
SerializeOutputParameters(*msg);
|
||||
}
|
||||
deserializers_.back().done_event->Signal();
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::Clear() {
|
||||
CancelPendingSends();
|
||||
received_sync_msgs_->RemoveContext(this);
|
||||
|
||||
Context::Clear();
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::OnMessageReceived(const Message& msg) {
|
||||
// Give the filters a chance at processing this message.
|
||||
if (TryFilters(msg))
|
||||
return;
|
||||
|
||||
if (TryToUnblockListener(&msg))
|
||||
return;
|
||||
|
||||
if (msg.should_unblock()) {
|
||||
received_sync_msgs_->QueueMessage(msg, this);
|
||||
return;
|
||||
}
|
||||
|
||||
if (msg.is_reply()) {
|
||||
received_sync_msgs_->QueueReply(msg, this);
|
||||
return;
|
||||
}
|
||||
|
||||
return Context::OnMessageReceivedNoFilter(msg);
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::OnChannelError() {
|
||||
CancelPendingSends();
|
||||
shutdown_watcher_.StopWatching();
|
||||
Context::OnChannelError();
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::OnChannelOpened() {
|
||||
shutdown_watcher_.StartWatching(shutdown_event_, this);
|
||||
Context::OnChannelOpened();
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::OnChannelClosed() {
|
||||
shutdown_watcher_.StopWatching();
|
||||
Context::OnChannelClosed();
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::OnSendTimeout(int message_id) {
|
||||
AutoLock auto_lock(deserializers_lock_);
|
||||
PendingSyncMessageQueue::iterator iter;
|
||||
for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++) {
|
||||
if (iter->id == message_id) {
|
||||
iter->done_event->Signal();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::CancelPendingSends() {
|
||||
AutoLock auto_lock(deserializers_lock_);
|
||||
PendingSyncMessageQueue::iterator iter;
|
||||
for (iter = deserializers_.begin(); iter != deserializers_.end(); iter++)
|
||||
iter->done_event->Signal();
|
||||
}
|
||||
|
||||
void SyncChannel::SyncContext::OnWaitableEventSignaled(WaitableEvent* event) {
|
||||
DCHECK(event == shutdown_event_);
|
||||
// Process shut down before we can get a reply to a synchronous message.
|
||||
// Cancel pending Send calls, which will end up setting the send done event.
|
||||
CancelPendingSends();
|
||||
}
|
||||
|
||||
|
||||
SyncChannel::SyncChannel(
|
||||
const std::wstring& channel_id, Channel::Mode mode,
|
||||
Channel::Listener* listener, MessageFilter* filter,
|
||||
MessageLoop* ipc_message_loop, bool create_pipe_now,
|
||||
WaitableEvent* shutdown_event)
|
||||
: ChannelProxy(
|
||||
channel_id, mode, ipc_message_loop,
|
||||
new SyncContext(listener, filter, ipc_message_loop, shutdown_event),
|
||||
create_pipe_now),
|
||||
sync_messages_with_no_timeout_allowed_(true) {
|
||||
// Ideally we only want to watch this object when running a nested message
|
||||
// loop. However, we don't know when it exits if there's another nested
|
||||
// message loop running under it or not, so we wouldn't know whether to
|
||||
// stop or keep watching. So we always watch it, and create the event as
|
||||
// manual reset since the object watcher might otherwise reset the event
|
||||
// when we're doing a WaitMany.
|
||||
dispatch_watcher_.StartWatching(sync_context()->GetDispatchEvent(), this);
|
||||
}
|
||||
|
||||
SyncChannel::~SyncChannel() {
|
||||
}
|
||||
|
||||
bool SyncChannel::Send(Message* message) {
|
||||
return SendWithTimeout(message, base::kNoTimeout);
|
||||
}
|
||||
|
||||
bool SyncChannel::SendWithTimeout(Message* message, int timeout_ms) {
|
||||
if (!message->is_sync()) {
|
||||
ChannelProxy::Send(message);
|
||||
return true;
|
||||
}
|
||||
|
||||
// *this* might get deleted in WaitForReply.
|
||||
RefPtr<SyncContext> context(sync_context());
|
||||
if (context->shutdown_event()->IsSignaled()) {
|
||||
delete message;
|
||||
return false;
|
||||
}
|
||||
|
||||
DCHECK(sync_messages_with_no_timeout_allowed_ ||
|
||||
timeout_ms != base::kNoTimeout);
|
||||
SyncMessage* sync_msg = static_cast<SyncMessage*>(message);
|
||||
context->Push(sync_msg);
|
||||
int message_id = SyncMessage::GetMessageId(*sync_msg);
|
||||
WaitableEvent* pump_messages_event = sync_msg->pump_messages_event();
|
||||
|
||||
ChannelProxy::Send(message);
|
||||
|
||||
if (timeout_ms != base::kNoTimeout) {
|
||||
// We use the sync message id so that when a message times out, we don't
|
||||
// confuse it with another send that is either above/below this Send in
|
||||
// the call stack.
|
||||
context->ipc_message_loop()->PostDelayedTask(FROM_HERE,
|
||||
NewRunnableMethod(context.get(),
|
||||
&SyncContext::OnSendTimeout, message_id), timeout_ms);
|
||||
}
|
||||
|
||||
// Wait for reply, or for any other incoming synchronous messages.
|
||||
WaitForReply(pump_messages_event);
|
||||
|
||||
return context->Pop();
|
||||
}
|
||||
|
||||
void SyncChannel::WaitForReply(WaitableEvent* pump_messages_event) {
|
||||
while (true) {
|
||||
WaitableEvent* objects[] = {
|
||||
sync_context()->GetDispatchEvent(),
|
||||
sync_context()->GetSendDoneEvent(),
|
||||
pump_messages_event
|
||||
};
|
||||
|
||||
unsigned count = pump_messages_event ? 3: 2;
|
||||
unsigned result = WaitableEvent::WaitMany(objects, count);
|
||||
if (result == 0 /* dispatch event */) {
|
||||
// We're waiting for a reply, but we received a blocking synchronous
|
||||
// call. We must process it or otherwise a deadlock might occur.
|
||||
sync_context()->GetDispatchEvent()->Reset();
|
||||
sync_context()->DispatchMessages();
|
||||
continue;
|
||||
}
|
||||
|
||||
if (result == 2 /* pump_messages_event */)
|
||||
WaitForReplyWithNestedMessageLoop(); // Start a nested message loop.
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
void SyncChannel::WaitForReplyWithNestedMessageLoop() {
|
||||
WaitableEvent* old_done_event = send_done_watcher_.GetWatchedEvent();
|
||||
send_done_watcher_.StopWatching();
|
||||
send_done_watcher_.StartWatching(sync_context()->GetSendDoneEvent(), this);
|
||||
bool old_state = MessageLoop::current()->NestableTasksAllowed();
|
||||
MessageLoop::current()->SetNestableTasksAllowed(true);
|
||||
MessageLoop::current()->Run();
|
||||
MessageLoop::current()->SetNestableTasksAllowed(old_state);
|
||||
if (old_done_event)
|
||||
send_done_watcher_.StartWatching(old_done_event, this);
|
||||
}
|
||||
|
||||
void SyncChannel::OnWaitableEventSignaled(WaitableEvent* event) {
|
||||
WaitableEvent* dispatch_event = sync_context()->GetDispatchEvent();
|
||||
if (event == dispatch_event) {
|
||||
// The call to DispatchMessages might delete this object, so reregister
|
||||
// the object watcher first.
|
||||
dispatch_event->Reset();
|
||||
dispatch_watcher_.StartWatching(dispatch_event, this);
|
||||
sync_context()->DispatchMessages();
|
||||
} else {
|
||||
// We got the reply, timed out or the process shutdown.
|
||||
DCHECK(event == sync_context()->GetSendDoneEvent());
|
||||
MessageLoop::current()->Quit();
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace IPC
|
||||
|
|
@ -1,160 +0,0 @@
|
|||
// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file.
|
||||
|
||||
#ifndef CHROME_COMMON_IPC_SYNC_SENDER_H__
|
||||
#define CHROME_COMMON_IPC_SYNC_SENDER_H__
|
||||
|
||||
#include <string>
|
||||
#include <deque>
|
||||
#include "base/basictypes.h"
|
||||
#include "base/lock.h"
|
||||
#include "base/scoped_handle.h"
|
||||
#include "base/waitable_event.h"
|
||||
#include "base/waitable_event_watcher.h"
|
||||
#include "chrome/common/ipc_channel_proxy.h"
|
||||
|
||||
#include "nsAutoPtr.h"
|
||||
|
||||
namespace IPC {
|
||||
|
||||
class SyncMessage;
|
||||
class MessageReplyDeserializer;
|
||||
|
||||
// This is similar to IPC::ChannelProxy, with the added feature of supporting
|
||||
// sending synchronous messages.
|
||||
// Note that care must be taken that the lifetime of the ipc_thread argument
|
||||
// is more than this object. If the message loop goes away while this object
|
||||
// is running and it's used to send a message, then it will use the invalid
|
||||
// message loop pointer to proxy it to the ipc thread.
|
||||
class SyncChannel : public ChannelProxy,
|
||||
public base::WaitableEventWatcher::Delegate {
|
||||
public:
|
||||
SyncChannel(const std::wstring& channel_id, Channel::Mode mode,
|
||||
Channel::Listener* listener, MessageFilter* filter,
|
||||
MessageLoop* ipc_message_loop, bool create_pipe_now,
|
||||
base::WaitableEvent* shutdown_event);
|
||||
~SyncChannel();
|
||||
|
||||
virtual bool Send(Message* message);
|
||||
virtual bool SendWithTimeout(Message* message, int timeout_ms);
|
||||
|
||||
// Whether we allow sending messages with no time-out.
|
||||
void set_sync_messages_with_no_timeout_allowed(bool value) {
|
||||
sync_messages_with_no_timeout_allowed_ = value;
|
||||
}
|
||||
|
||||
protected:
|
||||
class ReceivedSyncMsgQueue;
|
||||
friend class ReceivedSyncMsgQueue;
|
||||
|
||||
// SyncContext holds the per object data for SyncChannel, so that SyncChannel
|
||||
// can be deleted while it's being used in a different thread. See
|
||||
// ChannelProxy::Context for more information.
|
||||
class SyncContext : public Context,
|
||||
public base::WaitableEventWatcher::Delegate {
|
||||
public:
|
||||
SyncContext(Channel::Listener* listener,
|
||||
MessageFilter* filter,
|
||||
MessageLoop* ipc_thread,
|
||||
base::WaitableEvent* shutdown_event);
|
||||
|
||||
~SyncContext();
|
||||
|
||||
// Adds information about an outgoing sync message to the context so that
|
||||
// we know how to deserialize the reply.
|
||||
void Push(IPC::SyncMessage* sync_msg);
|
||||
|
||||
// Cleanly remove the top deserializer (and throw it away). Returns the
|
||||
// result of the Send call for that message.
|
||||
bool Pop();
|
||||
|
||||
// Returns an event that's set when the send is complete, timed out or the
|
||||
// process shut down.
|
||||
base::WaitableEvent* GetSendDoneEvent();
|
||||
|
||||
// Returns an event that's set when an incoming message that's not the reply
|
||||
// needs to get dispatched (by calling SyncContext::DispatchMessages).
|
||||
base::WaitableEvent* GetDispatchEvent();
|
||||
|
||||
void DispatchMessages();
|
||||
|
||||
// Checks if the given message is blocking the listener thread because of a
|
||||
// synchronous send. If it is, the thread is unblocked and true is
|
||||
// returned. Otherwise the function returns false.
|
||||
bool TryToUnblockListener(const Message* msg);
|
||||
|
||||
// Called on the IPC thread when a sync send that runs a nested message loop
|
||||
// times out.
|
||||
void OnSendTimeout(int message_id);
|
||||
|
||||
base::WaitableEvent* shutdown_event() { return shutdown_event_; }
|
||||
|
||||
private:
|
||||
// IPC::ChannelProxy methods that we override.
|
||||
|
||||
// Called on the listener thread.
|
||||
virtual void Clear();
|
||||
|
||||
// Called on the IPC thread.
|
||||
virtual void OnMessageReceived(const Message& msg);
|
||||
virtual void OnChannelError();
|
||||
virtual void OnChannelOpened();
|
||||
virtual void OnChannelClosed();
|
||||
|
||||
// Cancels all pending Send calls.
|
||||
void CancelPendingSends();
|
||||
|
||||
// WaitableEventWatcher::Delegate implementation.
|
||||
virtual void OnWaitableEventSignaled(base::WaitableEvent* arg);
|
||||
|
||||
// When sending a synchronous message, this structure contains an object
|
||||
// that knows how to deserialize the response.
|
||||
struct PendingSyncMsg {
|
||||
PendingSyncMsg(int id, IPC::MessageReplyDeserializer* d,
|
||||
base::WaitableEvent* e) :
|
||||
id(id), deserializer(d), done_event(e), send_result(false) { }
|
||||
int id;
|
||||
IPC::MessageReplyDeserializer* deserializer;
|
||||
base::WaitableEvent* done_event;
|
||||
bool send_result;
|
||||
};
|
||||
|
||||
typedef std::deque<PendingSyncMsg> PendingSyncMessageQueue;
|
||||
PendingSyncMessageQueue deserializers_;
|
||||
Lock deserializers_lock_;
|
||||
|
||||
RefPtr<ReceivedSyncMsgQueue> received_sync_msgs_;
|
||||
|
||||
base::WaitableEvent* shutdown_event_;
|
||||
base::WaitableEventWatcher shutdown_watcher_;
|
||||
};
|
||||
|
||||
private:
|
||||
// WaitableEventWatcher::Delegate implementation.
|
||||
virtual void OnWaitableEventSignaled(base::WaitableEvent* arg);
|
||||
|
||||
SyncContext* sync_context() {
|
||||
return reinterpret_cast<SyncContext*>(context());
|
||||
}
|
||||
|
||||
// Both these functions wait for a reply, timeout or process shutdown. The
|
||||
// latter one also runs a nested message loop in the meantime.
|
||||
void WaitForReply(base::WaitableEvent* pump_messages_event);
|
||||
|
||||
// Runs a nested message loop until a reply arrives, times out, or the process
|
||||
// shuts down.
|
||||
void WaitForReplyWithNestedMessageLoop();
|
||||
|
||||
bool sync_messages_with_no_timeout_allowed_;
|
||||
|
||||
// Used to signal events between the IPC and listener threads.
|
||||
base::WaitableEventWatcher send_done_watcher_;
|
||||
base::WaitableEventWatcher dispatch_watcher_;
|
||||
|
||||
DISALLOW_EVIL_CONSTRUCTORS(SyncChannel);
|
||||
};
|
||||
|
||||
} // namespace IPC
|
||||
|
||||
#endif // CHROME_COMMON_IPC_SYNC_SENDER_H__
|
||||
|
|
@ -1,125 +0,0 @@
|
|||
// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file.
|
||||
|
||||
#include "build/build_config.h"
|
||||
|
||||
#if defined(OS_WIN)
|
||||
#include <windows.h>
|
||||
#endif
|
||||
#include <stack>
|
||||
|
||||
#include "base/logging.h"
|
||||
#include "base/waitable_event.h"
|
||||
#include "chrome/common/ipc_sync_message.h"
|
||||
|
||||
namespace IPC {
|
||||
|
||||
uint32_t SyncMessage::next_id_ = 0;
|
||||
#define kSyncMessageHeaderSize 4
|
||||
|
||||
SyncMessage::SyncMessage(
|
||||
int32_t routing_id,
|
||||
uint16_t type,
|
||||
PriorityValue priority,
|
||||
MessageReplyDeserializer* deserializer)
|
||||
: Message(routing_id, type, priority),
|
||||
deserializer_(deserializer),
|
||||
pump_messages_event_(NULL)
|
||||
{
|
||||
set_sync();
|
||||
set_unblock(true);
|
||||
|
||||
// Add synchronous message data before the message payload.
|
||||
SyncHeader header;
|
||||
header.message_id = ++next_id_;
|
||||
WriteSyncHeader(this, header);
|
||||
}
|
||||
|
||||
MessageReplyDeserializer* SyncMessage::GetReplyDeserializer() {
|
||||
MessageReplyDeserializer* rv = deserializer_;
|
||||
DCHECK(rv);
|
||||
deserializer_ = NULL;
|
||||
return rv;
|
||||
}
|
||||
|
||||
void SyncMessage::EnableMessagePumping() {
|
||||
static base::WaitableEvent* dummy_event = new base::WaitableEvent(true, true);
|
||||
DCHECK(!pump_messages_event_);
|
||||
set_pump_messages_event(dummy_event);
|
||||
}
|
||||
|
||||
bool SyncMessage::IsMessageReplyTo(const Message& msg, int request_id) {
|
||||
if (!msg.is_reply())
|
||||
return false;
|
||||
|
||||
return GetMessageId(msg) == request_id;
|
||||
}
|
||||
|
||||
void* SyncMessage::GetDataIterator(const Message* msg) {
|
||||
void* iter = const_cast<char*>(msg->payload());
|
||||
UpdateIter(&iter, kSyncMessageHeaderSize);
|
||||
return iter;
|
||||
}
|
||||
|
||||
int SyncMessage::GetMessageId(const Message& msg) {
|
||||
if (!msg.is_sync() && !msg.is_reply())
|
||||
return 0;
|
||||
|
||||
SyncHeader header;
|
||||
if (!ReadSyncHeader(msg, &header))
|
||||
return 0;
|
||||
|
||||
return header.message_id;
|
||||
}
|
||||
|
||||
Message* SyncMessage::GenerateReply(const Message* msg) {
|
||||
DCHECK(msg->is_sync());
|
||||
|
||||
Message* reply = new Message(msg->routing_id(), IPC_REPLY_ID,
|
||||
msg->priority());
|
||||
reply->set_reply();
|
||||
|
||||
SyncHeader header;
|
||||
|
||||
// use the same message id, but this time reply bit is set
|
||||
header.message_id = GetMessageId(*msg);
|
||||
WriteSyncHeader(reply, header);
|
||||
|
||||
return reply;
|
||||
}
|
||||
|
||||
bool SyncMessage::ReadSyncHeader(const Message& msg, SyncHeader* header) {
|
||||
DCHECK(msg.is_sync() || msg.is_reply());
|
||||
|
||||
void* iter = NULL;
|
||||
bool result = msg.ReadInt(&iter, &header->message_id);
|
||||
if (!result) {
|
||||
NOTREACHED();
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool SyncMessage::WriteSyncHeader(Message* msg, const SyncHeader& header) {
|
||||
DCHECK(msg->is_sync() || msg->is_reply());
|
||||
DCHECK(msg->payload_size() == 0);
|
||||
bool result = msg->WriteInt(header.message_id);
|
||||
if (!result) {
|
||||
NOTREACHED();
|
||||
return false;
|
||||
}
|
||||
|
||||
// Note: if you add anything here, you need to update kSyncMessageHeaderSize.
|
||||
DCHECK(kSyncMessageHeaderSize == msg->payload_size());
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
bool MessageReplyDeserializer::SerializeOutputParameters(const Message& msg) {
|
||||
return SerializeOutputParameters(msg, SyncMessage::GetDataIterator(&msg));
|
||||
}
|
||||
|
||||
} // namespace IPC
|
||||
|
|
@ -1,97 +0,0 @@
|
|||
// Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style license that can be
|
||||
// found in the LICENSE file.
|
||||
|
||||
#ifndef CHROME_COMMON_IPC_SYNC_MESSAGE_H__
|
||||
#define CHROME_COMMON_IPC_SYNC_MESSAGE_H__
|
||||
|
||||
#if defined(OS_WIN)
|
||||
#include <windows.h>
|
||||
#endif
|
||||
#include <string>
|
||||
#include "base/basictypes.h"
|
||||
#include "chrome/common/ipc_message.h"
|
||||
|
||||
namespace base {
|
||||
class WaitableEvent;
|
||||
}
|
||||
|
||||
namespace IPC {
|
||||
|
||||
class MessageReplyDeserializer;
|
||||
|
||||
class SyncMessage : public Message {
|
||||
public:
|
||||
SyncMessage(int32_t routing_id, uint16_t type, PriorityValue priority,
|
||||
MessageReplyDeserializer* deserializer);
|
||||
|
||||
// Call this to get a deserializer for the output parameters.
|
||||
// Note that this can only be called once, and the caller is responsible
|
||||
// for deleting the deserializer when they're done.
|
||||
MessageReplyDeserializer* GetReplyDeserializer();
|
||||
|
||||
// If this message can cause the receiver to block while waiting for user
|
||||
// input (i.e. by calling MessageBox), then the caller needs to pump window
|
||||
// messages and dispatch asynchronous messages while waiting for the reply.
|
||||
// If this event is passed in, then window messages will start being pumped
|
||||
// when it's set. Note that this behavior will continue even if the event is
|
||||
// later reset. The event must be valid until after the Send call returns.
|
||||
void set_pump_messages_event(base::WaitableEvent* event) {
|
||||
pump_messages_event_ = event;
|
||||
if (event) {
|
||||
header()->flags |= PUMPING_MSGS_BIT;
|
||||
} else {
|
||||
header()->flags &= ~PUMPING_MSGS_BIT;
|
||||
}
|
||||
}
|
||||
|
||||
// Call this if you always want to pump messages. You can call this method
|
||||
// or set_pump_messages_event but not both.
|
||||
void EnableMessagePumping();
|
||||
|
||||
base::WaitableEvent* pump_messages_event() const {
|
||||
return pump_messages_event_;
|
||||
}
|
||||
|
||||
// Returns true if the message is a reply to the given request id.
|
||||
static bool IsMessageReplyTo(const Message& msg, int request_id);
|
||||
|
||||
// Given a reply message, returns an iterator to the beginning of the data
|
||||
// (i.e. skips over the synchronous specific data).
|
||||
static void* GetDataIterator(const Message* msg);
|
||||
|
||||
// Given a synchronous message (or its reply), returns its id.
|
||||
static int GetMessageId(const Message& msg);
|
||||
|
||||
// Generates a reply message to the given message.
|
||||
static Message* GenerateReply(const Message* msg);
|
||||
|
||||
private:
|
||||
struct SyncHeader {
|
||||
// unique ID (unique per sender)
|
||||
int message_id;
|
||||
};
|
||||
|
||||
static bool ReadSyncHeader(const Message& msg, SyncHeader* header);
|
||||
static bool WriteSyncHeader(Message* msg, const SyncHeader& header);
|
||||
|
||||
MessageReplyDeserializer* deserializer_;
|
||||
base::WaitableEvent* pump_messages_event_;
|
||||
|
||||
static uint32_t next_id_; // for generation of unique ids
|
||||
};
|
||||
|
||||
// Used to deserialize parameters from a reply to a synchronous message
|
||||
class MessageReplyDeserializer {
|
||||
public:
|
||||
bool SerializeOutputParameters(const Message& msg);
|
||||
virtual ~MessageReplyDeserializer() {}
|
||||
private:
|
||||
// Derived classes need to implement this, using the given iterator (which
|
||||
// is skipped past the header for synchronous messages).
|
||||
virtual bool SerializeOutputParameters(const Message& msg, void* iter) = 0;
|
||||
};
|
||||
|
||||
} // namespace IPC
|
||||
|
||||
#endif // CHROME_COMMON_IPC_SYNC_MESSAGE_H__
|
||||
Загрузка…
Ссылка в новой задаче