The Tor Browser: comparison ipc/glue/MessageChannel.h

--1:000000000000
+:babd230c25a2
+/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
+* vim: sw=4 ts=4 et :
+*/
+/* This Source Code Form is subject to the terms of the Mozilla Public
+* License, v. 2.0. If a copy of the MPL was not distributed with this
+* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+#ifndef ipc_glue_MessageChannel_h
+#define ipc_glue_MessageChannel_h 1
+#include "base/basictypes.h"
+#include "base/message_loop.h"
+#include "mozilla/Monitor.h"
+#include "mozilla/Vector.h"
+#include "mozilla/WeakPtr.h"
+#include "mozilla/ipc/Transport.h"
+#include "MessageLink.h"
+#include "nsAutoPtr.h"
+#include <deque>
+#include <stack>
+#include <math.h>
+namespace mozilla {
+namespace ipc {
+class MessageChannel;
+class RefCountedMonitor : public Monitor
+{
+public:
+RefCountedMonitor()
+: Monitor("mozilla.ipc.MessageChannel.mMonitor")
+{}
+NS_INLINE_DECL_THREADSAFE_REFCOUNTING(RefCountedMonitor)
+};
+class MessageChannel : HasResultCodes
+{
+friend class ProcessLink;
+friend class ThreadLink;
+friend class AutoEnterRPCTransaction;
+class CxxStackFrame;
+class InterruptFrame;
+typedef mozilla::Monitor Monitor;
+public:
+static const int32_t kNoTimeout;
+typedef IPC::Message Message;
+typedef mozilla::ipc::Transport Transport;
+MessageChannel(MessageListener *aListener);
+~MessageChannel();
+// "Open" from the perspective of the transport layer; the underlying
+// socketpair/pipe should already be created.
+//
+// Returns true if the transport layer was successfully connected,
+// i.e., mChannelState == ChannelConnected.
+bool Open(Transport* aTransport, MessageLoop* aIOLoop=0, Side aSide=UnknownSide);
+// "Open" a connection to another thread in the same process.
+//
+// Returns true if the transport layer was successfully connected,
+// i.e., mChannelState == ChannelConnected.
+//
+// For more details on the process of opening a channel between
+// threads, see the extended comment on this function
+// in MessageChannel.cpp.
+bool Open(MessageChannel *aTargetChan, MessageLoop *aTargetLoop, Side aSide);
+// Close the underlying transport channel.
+void Close();
+// Force the channel to behave as if a channel error occurred. Valid
+// for process links only, not thread links.
+void CloseWithError();
+void SetAbortOnError(bool abort)
+{
+mAbortOnError = true;
+}
+// Misc. behavioral traits consumers can request for this channel
+enum ChannelFlags {
+REQUIRE_DEFAULT                         = 0,
+// Windows: if this channel operates on the UI thread, indicates
+// WindowsMessageLoop code should enable deferred native message
+// handling to prevent deadlocks. Should only be used for protocols
+// that manage child processes which might create native UI, like
+// plugins.
+REQUIRE_DEFERRED_MESSAGE_PROTECTION     = 1 << 0
+};
+void SetChannelFlags(ChannelFlags aFlags) { mFlags = aFlags; }
+ChannelFlags GetChannelFlags() { return mFlags; }
+// Asynchronously send a message to the other side of the channel
+bool Send(Message* aMsg);
+// Asynchronously deliver a message back to this side of the
+// channel
+bool Echo(Message* aMsg);
+// Synchronously send |msg| (i.e., wait for |reply|)
+bool Send(Message* aMsg, Message* aReply);
+// Make an Interrupt call to the other side of the channel
+bool Call(Message* aMsg, Message* aReply);
+bool CanSend() const;
+void SetReplyTimeoutMs(int32_t aTimeoutMs);
+bool IsOnCxxStack() const {
+return !mCxxStackFrames.empty();
+}
+void FlushPendingInterruptQueue();
+// Unsound_IsClosed and Unsound_NumQueuedMessages are safe to call from any
+// thread, but they make no guarantees about whether you'll get an
+// up-to-date value; the values are written on one thread and read without
+// locking, on potentially different threads.  Thus you should only use
+// them when you don't particularly care about getting a recent value (e.g.
+// in a memory report).
+bool Unsound_IsClosed() const {
+return mLink ? mLink->Unsound_IsClosed() : true;
+}
+uint32_t Unsound_NumQueuedMessages() const {
+return mLink ? mLink->Unsound_NumQueuedMessages() : 0;
+}
+static bool IsPumpingMessages() {
+return sIsPumpingMessages;
+}
+static void SetIsPumpingMessages(bool aIsPumping) {
+sIsPumpingMessages = aIsPumping;
+}
+#ifdef OS_WIN
+struct MOZ_STACK_CLASS SyncStackFrame
+{
+SyncStackFrame(MessageChannel* channel, bool interrupt);
+~SyncStackFrame();
+bool mInterrupt;
+bool mSpinNestedEvents;
+bool mListenerNotified;
+MessageChannel* mChannel;
+// The previous stack frame for this channel.
+SyncStackFrame* mPrev;
+// The previous stack frame on any channel.
+SyncStackFrame* mStaticPrev;
+};
+friend struct MessageChannel::SyncStackFrame;
+static bool IsSpinLoopActive() {
+for (SyncStackFrame* frame = sStaticTopFrame; frame; frame = frame->mPrev) {
+if (frame->mSpinNestedEvents)
+return true;
+}
+return false;
+}
+protected:
+// The deepest sync stack frame for this channel.
+SyncStackFrame* mTopFrame;
+bool mIsSyncWaitingOnNonMainThread;
+// The deepest sync stack frame on any channel.
+static SyncStackFrame* sStaticTopFrame;
+public:
+void ProcessNativeEventsInInterruptCall();
+static void NotifyGeckoEventDispatch();
+private:
+void SpinInternalEventLoop();
+#endif
+private:
+void CommonThreadOpenInit(MessageChannel *aTargetChan, Side aSide);
+void OnOpenAsSlave(MessageChannel *aTargetChan, Side aSide);
+void PostErrorNotifyTask();
+void OnNotifyMaybeChannelError();
+void ReportConnectionError(const char* aChannelName) const;
+void ReportMessageRouteError(const char* channelName) const;
+bool MaybeHandleError(Result code, const char* channelName);
+void Clear();
+// Send OnChannelConnected notification to listeners.
+void DispatchOnChannelConnected(int32_t peer_pid);
+// Any protocol that requires blocking until a reply arrives, will send its
+// outgoing message through this function. Currently, two protocols do this:
+//
+//  sync, which can only initiate messages from child to parent.
+//  urgent, which can only initiate messages from parent to child.
+//
+// SendAndWait() expects that the worker thread owns the monitor, and that
+// the message has been prepared to be sent over the link. It returns as
+// soon as a reply has been received, or an error has occurred.
+//
+// Note that while the child is blocked waiting for a sync reply, it can wake
+// up to process urgent calls from the parent.
+bool SendAndWait(Message* aMsg, Message* aReply);
+bool RPCCall(Message* aMsg, Message* aReply);
+bool InterruptCall(Message* aMsg, Message* aReply);
+bool UrgentCall(Message* aMsg, Message* aReply);
+bool InterruptEventOccurred();
+bool ProcessPendingUrgentRequest();
+bool ProcessPendingRPCCall();
+void MaybeUndeferIncall();
+void EnqueuePendingMessages();
+// Executed on the worker thread. Dequeues one pending message.
+bool OnMaybeDequeueOne();
+bool DequeueOne(Message *recvd);
+// Dispatches an incoming message to its appropriate handler.
+void DispatchMessage(const Message &aMsg);
+// DispatchMessage will route to one of these functions depending on the
+// protocol type of the message.
+void DispatchSyncMessage(const Message &aMsg);
+void DispatchUrgentMessage(const Message &aMsg);
+void DispatchAsyncMessage(const Message &aMsg);
+void DispatchRPCMessage(const Message &aMsg);
+void DispatchInterruptMessage(const Message &aMsg, size_t aStackDepth);
+// Return true if the wait ended because a notification was received.
+//
+// Return false if the time elapsed from when we started the process of
+// waiting until afterwards exceeded the currently allotted timeout.
+// That *DOES NOT* mean false => "no event" (== timeout); there are many
+// circumstances that could cause the measured elapsed time to exceed the
+// timeout EVEN WHEN we were notified.
+//
+// So in sum: true is a meaningful return value; false isn't,
+// necessarily.
+bool WaitForSyncNotify();
+bool WaitForInterruptNotify();
+bool WaitResponse(bool aWaitTimedOut);
+bool ShouldContinueFromTimeout();
+// The "remote view of stack depth" can be different than the
+// actual stack depth when there are out-of-turn replies.  When we
+// receive one, our actual Interrupt stack depth doesn't decrease, but
+// the other side (that sent the reply) thinks it has.  So, the
+// "view" returned here is |stackDepth| minus the number of
+// out-of-turn replies.
+//
+// Only called from the worker thread.
+size_t RemoteViewOfStackDepth(size_t stackDepth) const {
+AssertWorkerThread();
+return stackDepth - mOutOfTurnReplies.size();
+}
+int32_t NextSeqno() {
+AssertWorkerThread();
+return (mSide == ChildSide) ? --mNextSeqno : ++mNextSeqno;
+}
+// This helper class manages mCxxStackDepth on behalf of MessageChannel.
+// When the stack depth is incremented from zero to non-zero, it invokes
+// a callback, and similarly for when the depth goes from non-zero to zero.
+void EnteredCxxStack() {
+mListener->OnEnteredCxxStack();
+}
+void ExitedCxxStack();
+void EnteredCall() {
+mListener->OnEnteredCall();
+}
+void ExitedCall() {
+mListener->OnExitedCall();
+}
+MessageListener *Listener() const {
+return mListener.get();
+}
+void DebugAbort(const char* file, int line, const char* cond,
+const char* why,
+bool reply=false) const;
+// This method is only safe to call on the worker thread, or in a
+// debugger with all threads paused.
+void DumpInterruptStack(const char* const pfx="") const;
+private:
+// Called from both threads
+size_t InterruptStackDepth() const {
+mMonitor->AssertCurrentThreadOwns();
+return mInterruptStack.size();
+}
+// Returns true if we're blocking waiting for a reply.
+bool AwaitingSyncReply() const {
+mMonitor->AssertCurrentThreadOwns();
+return mPendingSyncReplies > 0;
+}
+bool AwaitingUrgentReply() const {
+mMonitor->AssertCurrentThreadOwns();
+return mPendingUrgentReplies > 0;
+}
+bool AwaitingRPCReply() const {
+mMonitor->AssertCurrentThreadOwns();
+return mPendingRPCReplies > 0;
+}
+bool AwaitingInterruptReply() const {
+mMonitor->AssertCurrentThreadOwns();
+return !mInterruptStack.empty();
+}
+// Returns true if we're dispatching a sync message's callback.
+bool DispatchingSyncMessage() const {
+return mDispatchingSyncMessage;
+}
+// Returns true if we're dispatching an urgent message's callback.
+bool DispatchingUrgentMessage() const {
+return mDispatchingUrgentMessageCount > 0;
+}
+bool Connected() const;
+private:
+// Executed on the IO thread.
+void NotifyWorkerThread();
+// Return true if |aMsg| is a special message targeted at the IO
+// thread, in which case it shouldn't be delivered to the worker.
+bool MaybeInterceptSpecialIOMessage(const Message& aMsg);
+void OnChannelConnected(int32_t peer_id);
+// Tell the IO thread to close the channel and wait for it to ACK.
+void SynchronouslyClose();
+void OnMessageReceivedFromLink(const Message& aMsg);
+void OnChannelErrorFromLink();
+private:
+// Run on the not current thread.
+void NotifyChannelClosed();
+void NotifyMaybeChannelError();
+private:
+// Can be run on either thread
+void AssertWorkerThread() const
+{
+NS_ABORT_IF_FALSE(mWorkerLoopID == MessageLoop::current()->id(),
+"not on worker thread!");
+}
+// The "link" thread is either the I/O thread (ProcessLink) or the
+// other actor's work thread (ThreadLink).  In either case, it is
+// NOT our worker thread.
+void AssertLinkThread() const
+{
+NS_ABORT_IF_FALSE(mWorkerLoopID != MessageLoop::current()->id(),
+"on worker thread but should not be!");
+}
+private:
+typedef IPC::Message::msgid_t msgid_t;
+typedef std::deque<Message> MessageQueue;
+typedef std::map<size_t, Message> MessageMap;
+// All dequeuing tasks require a single point of cancellation,
+// which is handled via a reference-counted task.
+class RefCountedTask
+{
+public:
+RefCountedTask(CancelableTask* aTask)
+: mTask(aTask)
+{ }
+~RefCountedTask() { delete mTask; }
+void Run() { mTask->Run(); }
+void Cancel() { mTask->Cancel(); }
+NS_INLINE_DECL_THREADSAFE_REFCOUNTING(RefCountedTask)
+private:
+CancelableTask* mTask;
+};
+// Wrap an existing task which can be cancelled at any time
+// without the wrapper's knowledge.
+class DequeueTask : public Task
+{
+public:
+DequeueTask(RefCountedTask* aTask)
+: mTask(aTask)
+{ }
+void Run() { mTask->Run(); }
+private:
+nsRefPtr<RefCountedTask> mTask;
+};
+private:
+mozilla::WeakPtr<MessageListener> mListener;
+ChannelState mChannelState;
+nsRefPtr<RefCountedMonitor> mMonitor;
+Side mSide;
+MessageLink* mLink;
+MessageLoop* mWorkerLoop;           // thread where work is done
+CancelableTask* mChannelErrorTask;  // NotifyMaybeChannelError runnable
+// id() of mWorkerLoop.  This persists even after mWorkerLoop is cleared
+// during channel shutdown.
+int mWorkerLoopID;
+// A task encapsulating dequeuing one pending message.
+nsRefPtr<RefCountedTask> mDequeueOneTask;
+// Timeout periods are broken up in two to prevent system suspension from
+// triggering an abort. This method (called by WaitForEvent with a 'did
+// timeout' flag) decides if we should wait again for half of mTimeoutMs
+// or give up.
+int32_t mTimeoutMs;
+bool mInTimeoutSecondHalf;
+// Worker-thread only; sequence numbers for messages that require
+// synchronous replies.
+int32_t mNextSeqno;
+static bool sIsPumpingMessages;
+class AutoEnterPendingReply {
+public:
+AutoEnterPendingReply(size_t &replyVar)
+: mReplyVar(replyVar)
+{
+mReplyVar++;
+}
+~AutoEnterPendingReply() {
+mReplyVar--;
+}
+private:
+size_t& mReplyVar;
+};
+// Worker-thread only; type we're expecting for the reply to a sync
+// out-message. This will never be greater than 1.
+size_t mPendingSyncReplies;
+// Worker-thread only; Number of urgent and rpc replies we're waiting on.
+// These are mutually exclusive since one channel cannot have outcalls of
+// both kinds.
+size_t mPendingUrgentReplies;
+size_t mPendingRPCReplies;
+// When we send an urgent request from the parent process, we could race
+// with an RPC message that was issued by the child beforehand. In this
+// case, if the parent were to wake up while waiting for the urgent reply,
+// and process the RPC, it could send an additional urgent message. The
+// child would wake up to process the urgent message (as it always will),
+// then send a reply, which could be received by the parent out-of-order
+// with respect to the first urgent reply.
+//
+// To address this problem, urgent or RPC requests are associated with a
+// "transaction". Whenever one side of the channel wishes to start a
+// chain of RPC/urgent messages, it allocates a new transaction ID. Any
+// messages the parent receives, not apart of this transaction, are
+// deferred. When issuing RPC/urgent requests on top of a started
+// transaction, the initiating transaction ID is used.
+//
+// To ensure IDs are unique, we use sequence numbers for transaction IDs,
+// which grow in opposite directions from child to parent.
+// The current transaction ID.
+int32_t mCurrentRPCTransaction;
+class AutoEnterRPCTransaction
+{
+public:
+AutoEnterRPCTransaction(MessageChannel *aChan)
+: mChan(aChan),
+mOldTransaction(mChan->mCurrentRPCTransaction)
+{
+mChan->mMonitor->AssertCurrentThreadOwns();
+if (mChan->mCurrentRPCTransaction == 0)
+mChan->mCurrentRPCTransaction = mChan->NextSeqno();
+}
+AutoEnterRPCTransaction(MessageChannel *aChan, Message *message)
+: mChan(aChan),
+mOldTransaction(mChan->mCurrentRPCTransaction)
+{
+mChan->mMonitor->AssertCurrentThreadOwns();
+if (!message->is_rpc() && !message->is_urgent())
+return;
+MOZ_ASSERT_IF(mChan->mSide == ParentSide,
+!mOldTransaction || mOldTransaction == message->transaction_id());
+mChan->mCurrentRPCTransaction = message->transaction_id();
+}
+~AutoEnterRPCTransaction() {
+mChan->mMonitor->AssertCurrentThreadOwns();
+mChan->mCurrentRPCTransaction = mOldTransaction;
+}
+private:
+MessageChannel *mChan;
+int32_t mOldTransaction;
+};
+// If waiting for the reply to a sync out-message, it will be saved here
+// on the I/O thread and then read and cleared by the worker thread.
+nsAutoPtr<Message> mRecvd;
+// Set while we are dispatching a synchronous message.
+bool mDispatchingSyncMessage;
+// Count of the recursion depth of dispatching urgent messages.
+size_t mDispatchingUrgentMessageCount;
+// Queue of all incoming messages, except for replies to sync and urgent
+// messages, which are delivered directly to mRecvd, and any pending urgent
+// incall, which is stored in mPendingUrgentRequest.
+//
+// If both this side and the other side are functioning correctly, the queue
+// can only be in certain configurations.  Let
+//
+//   |A<| be an async in-message,
+//   |S<| be a sync in-message,
+//   |C<| be an Interrupt in-call,
+//   |R<| be an Interrupt reply.
+//
+// The queue can only match this configuration
+//
+//  A<* (S< | C< | R< (?{mStack.size() == 1} A<* (S< | C<)))
+//
+// The other side can send as many async messages |A<*| as it wants before
+// sending us a blocking message.
+//
+// The first case is |S<|, a sync in-msg.  The other side must be blocked,
+// and thus can't send us any more messages until we process the sync
+// in-msg.
+//
+// The second case is |C<|, an Interrupt in-call; the other side must be blocked.
+// (There's a subtlety here: this in-call might have raced with an
+// out-call, but we detect that with the mechanism below,
+// |mRemoteStackDepth|, and races don't matter to the queue.)
+//
+// Final case, the other side replied to our most recent out-call |R<|.
+// If that was the *only* out-call on our stack, |?{mStack.size() == 1}|,
+// then other side "finished with us," and went back to its own business.
+// That business might have included sending any number of async message
+// |A<*| until sending a blocking message |(S< | C<)|.  If we had more than
+// one Interrupt call on our stack, the other side *better* not have sent us
+// another blocking message, because it's blocked on a reply from us.
+//
+MessageQueue mPending;
+// Note that these two pointers are mutually exclusive. One channel cannot
+// send both urgent requests (parent -> child) and RPC calls (child->parent).
+// Also note that since initiating either requires blocking, they cannot
+// queue up on the other side. One message slot is enough.
+//
+// Normally, all other message types are deferred into into mPending, and
+// only these two types have special treatment (since they wake up blocked
+// requests). However, when an RPC in-call races with an urgent out-call,
+// the RPC message will be put into mPending instead of its slot below.
+nsAutoPtr<Message> mPendingUrgentRequest;
+nsAutoPtr<Message> mPendingRPCCall;
+// Stack of all the out-calls on which this channel is awaiting responses.
+// Each stack refers to a different protocol and the stacks are mutually
+// exclusive: multiple outcalls of the same kind cannot be initiated while
+// another is active.
+std::stack<Message> mInterruptStack;
+// This is what we think the Interrupt stack depth is on the "other side" of this
+// Interrupt channel.  We maintain this variable so that we can detect racy Interrupt
+// calls.  With each Interrupt out-call sent, we send along what *we* think the
+// stack depth of the remote side is *before* it will receive the Interrupt call.
+//
+// After sending the out-call, our stack depth is "incremented" by pushing
+// that pending message onto mPending.
+//
+// Then when processing an in-call |c|, it must be true that
+//
+//   mStack.size() == c.remoteDepth
+//
+// I.e., my depth is actually the same as what the other side thought it
+// was when it sent in-call |c|.  If this fails to hold, we have detected
+// racy Interrupt calls.
+//
+// We then increment mRemoteStackDepth *just before* processing the
+// in-call, since we know the other side is waiting on it, and decrement
+// it *just after* finishing processing that in-call, since our response
+// will pop the top of the other side's |mPending|.
+//
+// One nice aspect of this race detection is that it is symmetric; if one
+// side detects a race, then the other side must also detect the same race.
+size_t mRemoteStackDepthGuess;
+// Approximation of code frames on the C++ stack. It can only be
+// interpreted as the implication:
+//
+//  !mCxxStackFrames.empty() => MessageChannel code on C++ stack
+//
+// This member is only accessed on the worker thread, and so is not
+// protected by mMonitor.  It is managed exclusively by the helper
+// |class CxxStackFrame|.
+mozilla::Vector<InterruptFrame> mCxxStackFrames;
+// Did we process an Interrupt out-call during this stack?  Only meaningful in
+// ExitedCxxStack(), from which this variable is reset.
+bool mSawInterruptOutMsg;
+// Map of replies received "out of turn", because of Interrupt
+// in-calls racing with replies to outstanding in-calls.  See
+// https://bugzilla.mozilla.org/show_bug.cgi?id=521929.
+MessageMap mOutOfTurnReplies;
+// Stack of Interrupt in-calls that were deferred because of race
+// conditions.
+std::stack<Message> mDeferred;
+#ifdef OS_WIN
+HANDLE mEvent;
+#endif
+// Should the channel abort the process from the I/O thread when
+// a channel error occurs?
+bool mAbortOnError;
+// See SetChannelFlags
+ChannelFlags mFlags;
+};
+} // namespace ipc
+} // namespace mozilla
+#endif  // ifndef ipc_glue_MessageChannel_h

The Tor Browser / file comparison

comparison: ipc/glue/MessageChannel.h

ipc/glue/MessageChannel.h