Mypal/ipc/glue/MessageChannel.h

/* -*- 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/Function.h"
#include "mozilla/DebugOnly.h"
#include "mozilla/Monitor.h"
#include "mozilla/Vector.h"
#if defined(OS_WIN)
#include "mozilla/ipc/Neutering.h"
#endif // defined(OS_WIN)
#include "mozilla/ipc/Transport.h"
#include "MessageLink.h"

#include <deque>
#include <stack>
#include <math.h>

namespace mozilla {
namespace ipc {

class MessageChannel;
class IToplevelProtocol;

class RefCountedMonitor : public Monitor
{
  public:
    RefCountedMonitor()
        : Monitor("mozilla.ipc.MessageChannel.mMonitor")
    {}

    NS_INLINE_DECL_THREADSAFE_REFCOUNTING(RefCountedMonitor)

  private:
    ~RefCountedMonitor() {}
};

enum class SyncSendError {
    SendSuccess,
    PreviousTimeout,
    SendingCPOWWhileDispatchingSync,
    SendingCPOWWhileDispatchingUrgent,
    NotConnectedBeforeSend,
    DisconnectedDuringSend,
    CancelledBeforeSend,
    CancelledAfterSend,
    TimedOut,
    ReplyError,
};

enum ChannelState {
    ChannelClosed,
    ChannelOpening,
    ChannelConnected,
    ChannelTimeout,
    ChannelClosing,
    ChannelError
};

class AutoEnterTransaction;

class MessageChannel : HasResultCodes
{
    friend class ProcessLink;
    friend class ThreadLink;

    class CxxStackFrame;
    class InterruptFrame;

    typedef mozilla::Monitor Monitor;

  public:
    static const int32_t kNoTimeout;

    typedef IPC::Message Message;
    typedef IPC::MessageInfo MessageInfo;
    typedef mozilla::ipc::Transport Transport;

    explicit MessageChannel(IToplevelProtocol *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 CloseWithTimeout();

    void SetAbortOnError(bool abort)
    {
        mAbortOnError = abort;
    }

    // Call aInvoke for each pending message until it returns false.
    // XXX: You must get permission from an IPC peer to use this function
    //      since it requires custom deserialization and re-orders events.
    void PeekMessages(mozilla::function<bool(const Message& aMsg)> aInvoke);

    // 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,
      // Windows: When this flag is specified, any wait that occurs during
      // synchronous IPC will be alertable, thus allowing a11y code in the
      // chrome process to reenter content while content is waiting on a
      // synchronous call.
      REQUIRE_A11Y_REENTRY                    = 1 << 1,
    };
    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);

    // Wait until a message is received
    bool WaitForIncomingMessage();

    bool CanSend() const;

    // If sending a sync message returns an error, this function gives a more
    // descriptive error message.
    SyncSendError LastSendError() const {
        AssertWorkerThread();
        return mLastSendError;
    }

    // Currently only for debugging purposes, doesn't aquire mMonitor.
    ChannelState GetChannelState__TotallyRacy() const {
        return mChannelState;
    }

    void SetReplyTimeoutMs(int32_t aTimeoutMs);

    bool IsOnCxxStack() const {
        return !mCxxStackFrames.empty();
    }

    bool IsInTransaction() const;
    void CancelCurrentTransaction();

    /**
     * This function is used by hang annotation code to determine which IPDL
     * actor is highest in the call stack at the time of the hang. It should
     * be called from the main thread when a sync or intr message is about to
     * be sent.
     */
    int32_t GetTopmostMessageRoutingId() const;

    // 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();
#if defined(ACCESSIBILITY)
    bool WaitForSyncNotifyWithA11yReentry();
#endif // defined(ACCESSIBILITY)
#endif // defined(OS_WIN)

  private:
    void CommonThreadOpenInit(MessageChannel *aTargetChan, Side aSide);
    void OnOpenAsSlave(MessageChannel *aTargetChan, Side aSide);

    void PostErrorNotifyTask();
    void OnNotifyMaybeChannelError();
    void ReportConnectionError(const char* aChannelName, Message* aMsg = nullptr) const;
    void ReportMessageRouteError(const char* channelName) const;
    bool MaybeHandleError(Result code, const Message& aMsg, const char* channelName);

    void Clear();

    // Send OnChannelConnected notification to listeners.
    void DispatchOnChannelConnected();

    bool InterruptEventOccurred();
    bool HasPendingEvents();

    void ProcessPendingRequests(AutoEnterTransaction& aTransaction);
    bool ProcessPendingRequest(Message &&aUrgent);

    void MaybeUndeferIncall();
    void EnqueuePendingMessages();

    // Dispatches an incoming message to its appropriate handler.
    void DispatchMessage(Message &&aMsg);

    // DispatchMessage will route to one of these functions depending on the
    // protocol type of the message.
    void DispatchSyncMessage(const Message &aMsg, Message*& aReply);
    void DispatchUrgentMessage(const Message &aMsg);
    void DispatchAsyncMessage(const Message &aMsg);
    void DispatchRPCMessage(const Message &aMsg);
    void DispatchInterruptMessage(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 aHandleWindowsMessages);
    bool WaitForInterruptNotify();

    bool WaitResponse(bool aWaitTimedOut);

    bool ShouldContinueFromTimeout();

    void EndTimeout();
    void CancelTransaction(int transaction);

    void RepostAllMessages();

    // 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();
    void ExitedCxxStack();

    void EnteredCall();
    void ExitedCall();

    void EnteredSyncSend();
    void ExitedSyncSend();

    IToplevelProtocol *Listener() const {
        return mListener;
    }

    void DebugAbort(const char* file, int line, const char* cond,
                    const char* why,
                    bool reply=false);

    // 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();
    }

    bool AwaitingInterruptReply() const {
        mMonitor->AssertCurrentThreadOwns();
        return !mInterruptStack.empty();
    }
    bool AwaitingIncomingMessage() const {
        mMonitor->AssertCurrentThreadOwns();
        return mIsWaitingForIncoming;
    }

    class MOZ_STACK_CLASS AutoEnterWaitForIncoming
    {
    public:
        explicit AutoEnterWaitForIncoming(MessageChannel& aChannel)
            : mChannel(aChannel)
        {
            aChannel.mMonitor->AssertCurrentThreadOwns();
            aChannel.mIsWaitingForIncoming = true;
        }

        ~AutoEnterWaitForIncoming()
        {
            mChannel.mIsWaitingForIncoming = false;
        }

    private:
        MessageChannel& mChannel;
    };
    friend class AutoEnterWaitForIncoming;

    // Returns true if we're dispatching an async message's callback.
    bool DispatchingAsyncMessage() const {
        AssertWorkerThread();
        return mDispatchingAsyncMessage;
    }

    int DispatchingAsyncMessageNestedLevel() const {
        AssertWorkerThread();
        return mDispatchingAsyncMessageNestedLevel;
    }

    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();

    bool WasTransactionCanceled(int transaction);
    bool ShouldDeferMessage(const Message& aMsg);
    bool ShouldDeferInterruptMessage(const Message& aMsg, size_t aStackDepth);
    void OnMessageReceivedFromLink(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
    {
        MOZ_RELEASE_ASSERT(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
    {
        MOZ_RELEASE_ASSERT(mWorkerLoopID != MessageLoop::current()->id(),
                           "on worker thread but should not be!");
    }

  private:
    class MessageTask :
        public CancelableRunnable,
        public LinkedListElement<RefPtr<MessageTask>>,
        public nsIRunnablePriority
    {
    public:
        explicit MessageTask(MessageChannel* aChannel, Message&& aMessage)
          : mChannel(aChannel), mMessage(Move(aMessage)), mScheduled(false)
        {}

        NS_DECL_ISUPPORTS_INHERITED

        NS_IMETHOD Run() override;
        nsresult Cancel() override;
        NS_IMETHOD GetPriority(uint32_t* aPriority) override;
        void Post();
        void Clear();

        bool IsScheduled() const { return mScheduled; }

        Message& Msg() { return mMessage; }
        const Message& Msg() const { return mMessage; }

    private:
        MessageTask() = delete;
        MessageTask(const MessageTask&) = delete;
        ~MessageTask() {}

        MessageChannel* mChannel;
        Message mMessage;
        bool mScheduled : 1;
    };

    bool ShouldRunMessage(const Message& aMsg);
    void RunMessage(MessageTask& aTask);

    typedef LinkedList<RefPtr<MessageTask>> MessageQueue;
    typedef std::map<size_t, Message> MessageMap;
    typedef IPC::Message::msgid_t msgid_t;

  private:
    // Based on presumption the listener owns and overlives the channel,
    // this is never nullified.
    IToplevelProtocol* mListener;
    ChannelState mChannelState;
    RefPtr<RefCountedMonitor> mMonitor;
    Side mSide;
    MessageLink* mLink;
    MessageLoop* mWorkerLoop;           // thread where work is done
    RefPtr<CancelableRunnable> mChannelErrorTask;  // NotifyMaybeChannelError runnable

    // id() of mWorkerLoop.  This persists even after mWorkerLoop is cleared
    // during channel shutdown.
    int mWorkerLoopID;

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

    // If ::Send returns false, this gives a more descriptive error.
    SyncSendError mLastSendError;

    template<class T>
    class AutoSetValue {
      public:
        explicit AutoSetValue(T &var, const T &newValue)
          : mVar(var), mPrev(var), mNew(newValue)
        {
            mVar = newValue;
        }
        ~AutoSetValue() {
            // The value may have been zeroed if the transaction was
            // canceled. In that case we shouldn't return it to its previous
            // value.
            if (mVar == mNew) {
                mVar = mPrev;
            }
        }
      private:
        T& mVar;
        T mPrev;
        T mNew;
    };

    bool mDispatchingAsyncMessage;
    int mDispatchingAsyncMessageNestedLevel;

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

    friend class AutoEnterTransaction;
    AutoEnterTransaction *mTransactionStack;

    int32_t CurrentNestedInsideSyncTransaction() const;

    bool AwaitingSyncReply() const;
    int AwaitingSyncReplyNestedLevel() const;

    bool DispatchingSyncMessage() const;
    int DispatchingSyncMessageNestedLevel() const;

    // If a sync message times out, we store its sequence number here. Any
    // future sync messages will fail immediately. Once the reply for original
    // sync message is received, we allow sync messages again.
    //
    // When a message times out, nothing is done to inform the other side. The
    // other side will eventually dispatch the message and send a reply. Our
    // side is responsible for replying to all sync messages sent by the other
    // side when it dispatches the timed out message. The response is always an
    // error.
    //
    // A message is only timed out if it initiated a transaction. This avoids
    // hitting a lot of corner cases with message nesting that we don't really
    // care about.
    int32_t mTimedOutMessageSeqno;
    int mTimedOutMessageNestedLevel;

    // Queue of all incoming messages.
    //
    // 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< (?{mInterruptStack.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, |?{mInterruptStack.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;

    // 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<MessageInfo> 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
    //
    //   mInterruptStack.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;

    // Are we waiting on this channel for an incoming message? This is used
    // to implement WaitForIncomingMessage(). Must only be accessed while owning
    // mMonitor.
    bool mIsWaitingForIncoming;

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

    // True if the listener has already been notified of a channel close or
    // error.
    bool mNotifiedChannelDone;

    // See SetChannelFlags
    ChannelFlags mFlags;

    // Task and state used to asynchronously notify channel has been connected
    // safely.  This is necessary to be able to cancel notification if we are
    // closed at the same time.
    RefPtr<CancelableRunnable> mOnChannelConnectedTask;
    bool mPeerPidSet;
    int32_t mPeerPid;
};

void
CancelCPOWs();

} // namespace ipc
} // namespace mozilla

#endif  // ifndef ipc_glue_MessageChannel_h