The Tor Browser / file revision

 /* -*- 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