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