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: #include "mozilla/ipc/MessageLink.h"
michael@0: #include "mozilla/ipc/MessageChannel.h"
michael@0: #include "mozilla/ipc/BrowserProcessSubThread.h"
michael@0: #include "mozilla/ipc/ProtocolUtils.h"
michael@0: 
michael@0: #ifdef MOZ_NUWA_PROCESS
michael@0: #include "ipc/Nuwa.h"
michael@0: #include "mozilla/Preferences.h"
michael@0: #endif
michael@0: 
michael@0: #include "nsDebug.h"
michael@0: #include "nsISupportsImpl.h"
michael@0: #include "nsXULAppAPI.h"
michael@0: 
michael@0: using namespace mozilla;
michael@0: using namespace std;
michael@0: 
michael@0: template<>
michael@0: struct RunnableMethodTraits<mozilla::ipc::ProcessLink>
michael@0: {
michael@0:     static void RetainCallee(mozilla::ipc::ProcessLink* obj) { }
michael@0:     static void ReleaseCallee(mozilla::ipc::ProcessLink* obj) { }
michael@0: };
michael@0: 
michael@0: // We rely on invariants about the lifetime of the transport:
michael@0: //
michael@0: //  - outlives this MessageChannel
michael@0: //  - deleted on the IO thread
michael@0: //
michael@0: // These invariants allow us to send messages directly through the
michael@0: // transport without having to worry about orphaned Send() tasks on
michael@0: // the IO thread touching MessageChannel memory after it's been deleted
michael@0: // on the worker thread.  We also don't need to refcount the
michael@0: // Transport, because whatever task triggers its deletion only runs on
michael@0: // the IO thread, and only runs after this MessageChannel is done with
michael@0: // the Transport.
michael@0: template<>
michael@0: struct RunnableMethodTraits<mozilla::ipc::MessageChannel::Transport>
michael@0: {
michael@0:     static void RetainCallee(mozilla::ipc::MessageChannel::Transport* obj) { }
michael@0:     static void ReleaseCallee(mozilla::ipc::MessageChannel::Transport* obj) { }
michael@0: };
michael@0: 
michael@0: namespace mozilla {
michael@0: namespace ipc {
michael@0: 
michael@0: MessageLink::MessageLink(MessageChannel *aChan)
michael@0:   : mChan(aChan)
michael@0: {
michael@0: }
michael@0: 
michael@0: MessageLink::~MessageLink()
michael@0: {
michael@0:     mChan = nullptr;
michael@0: }
michael@0: 
michael@0: ProcessLink::ProcessLink(MessageChannel *aChan)
michael@0:   : MessageLink(aChan),
michael@0:     mExistingListener(nullptr)
michael@0: {
michael@0: }
michael@0: 
michael@0: ProcessLink::~ProcessLink()
michael@0: {
michael@0:     mIOLoop = 0;
michael@0:     if (mTransport) {
michael@0:         mTransport->set_listener(0);
michael@0:         
michael@0:         // we only hold a weak ref to the transport, which is "owned"
michael@0:         // by GeckoChildProcess/GeckoThread
michael@0:         mTransport = 0;
michael@0:     }
michael@0: }
michael@0: 
michael@0: void 
michael@0: ProcessLink::Open(mozilla::ipc::Transport* aTransport, MessageLoop *aIOLoop, Side aSide)
michael@0: {
michael@0:     NS_PRECONDITION(aTransport, "need transport layer");
michael@0: 
michael@0:     // FIXME need to check for valid channel
michael@0: 
michael@0:     mTransport = aTransport;
michael@0: 
michael@0:     // FIXME figure out whether we're in parent or child, grab IO loop
michael@0:     // appropriately
michael@0:     bool needOpen = true;
michael@0:     if(aIOLoop) {
michael@0:         // We're a child or using the new arguments.  Either way, we
michael@0:         // need an open.
michael@0:         needOpen = true;
michael@0:         mChan->mSide = (aSide == UnknownSide) ? ChildSide : aSide;
michael@0:     } else {
michael@0:         NS_PRECONDITION(aSide == UnknownSide, "expected default side arg");
michael@0: 
michael@0:         // parent
michael@0:         mChan->mSide = ParentSide;
michael@0:         needOpen = false;
michael@0:         aIOLoop = XRE_GetIOMessageLoop();
michael@0:     }
michael@0: 
michael@0:     mIOLoop = aIOLoop;
michael@0: 
michael@0:     NS_ASSERTION(mIOLoop, "need an IO loop");
michael@0:     NS_ASSERTION(mChan->mWorkerLoop, "need a worker loop");
michael@0: 
michael@0:     {
michael@0:         MonitorAutoLock lock(*mChan->mMonitor);
michael@0: 
michael@0:         if (needOpen) {
michael@0:             // Transport::Connect() has not been called.  Call it so
michael@0:             // we start polling our pipe and processing outgoing
michael@0:             // messages.
michael@0:             mIOLoop->PostTask(
michael@0:                 FROM_HERE,
michael@0:                 NewRunnableMethod(this, &ProcessLink::OnChannelOpened));
michael@0:         } else {
michael@0:             // Transport::Connect() has already been called.  Take
michael@0:             // over the channel from the previous listener and process
michael@0:             // any queued messages.
michael@0:             mIOLoop->PostTask(
michael@0:                 FROM_HERE,
michael@0:                 NewRunnableMethod(this, &ProcessLink::OnTakeConnectedChannel));
michael@0:         }
michael@0: 
michael@0: #ifdef MOZ_NUWA_PROCESS
michael@0:         if (IsNuwaProcess() &&
michael@0:             Preferences::GetBool("dom.ipc.processPrelaunch.testMode")) {
michael@0:             // The pref value is turned on in a deadlock test against the Nuwa
michael@0:             // process. The sleep here makes it easy to trigger the deadlock
michael@0:             // that an IPC channel is still opening but the worker loop is
michael@0:             // already frozen.
michael@0:             sleep(5);
michael@0:         }
michael@0: #endif
michael@0: 
michael@0:         // Should not wait here if something goes wrong with the channel.
michael@0:         while (!mChan->Connected() && mChan->mChannelState != ChannelError) {
michael@0:             mChan->mMonitor->Wait();
michael@0:         }
michael@0:     }
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::EchoMessage(Message *msg)
michael@0: {
michael@0:     mChan->AssertWorkerThread();
michael@0:     mChan->mMonitor->AssertCurrentThreadOwns();
michael@0: 
michael@0:     mIOLoop->PostTask(
michael@0:         FROM_HERE,
michael@0:         NewRunnableMethod(this, &ProcessLink::OnEchoMessage, msg));
michael@0:     // OnEchoMessage takes ownership of |msg|
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::SendMessage(Message *msg)
michael@0: {
michael@0:     mChan->AssertWorkerThread();
michael@0:     mChan->mMonitor->AssertCurrentThreadOwns();
michael@0: 
michael@0:     mIOLoop->PostTask(
michael@0:         FROM_HERE,
michael@0:         NewRunnableMethod(mTransport, &Transport::Send, msg));
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::SendClose()
michael@0: {
michael@0:     mChan->AssertWorkerThread();
michael@0:     mChan->mMonitor->AssertCurrentThreadOwns();
michael@0: 
michael@0:     mIOLoop->PostTask(
michael@0:         FROM_HERE, NewRunnableMethod(this, &ProcessLink::OnCloseChannel));
michael@0: }
michael@0: 
michael@0: ThreadLink::ThreadLink(MessageChannel *aChan, MessageChannel *aTargetChan)
michael@0:   : MessageLink(aChan),
michael@0:     mTargetChan(aTargetChan)
michael@0: {
michael@0: }
michael@0: 
michael@0: ThreadLink::~ThreadLink()
michael@0: {
michael@0:     // :TODO: MonitorAutoLock lock(*mChan->mMonitor);
michael@0:     // Bug 848949: We need to prevent the other side
michael@0:     // from sending us any more messages to avoid Use-After-Free.
michael@0:     // The setup here is as shown:
michael@0:     //
michael@0:     //          (Us)         (Them)
michael@0:     //       MessageChannel  MessageChannel
michael@0:     //         |  ^     \ /     ^ |
michael@0:     //         |  |      X      | |
michael@0:     //         v  |     / \     | v
michael@0:     //        ThreadLink   ThreadLink
michael@0:     //
michael@0:     // We want to null out the diagonal link from their ThreadLink
michael@0:     // to our MessageChannel.  Note that we must hold the monitor so
michael@0:     // that we do this atomically with respect to them trying to send
michael@0:     // us a message.
michael@0:     if (mTargetChan) {
michael@0:         static_cast<ThreadLink*>(mTargetChan->mLink)->mTargetChan = 0;
michael@0:     }
michael@0:     mTargetChan = 0;
michael@0: }
michael@0: 
michael@0: void
michael@0: ThreadLink::EchoMessage(Message *msg)
michael@0: {
michael@0:     mChan->AssertWorkerThread();
michael@0:     mChan->mMonitor->AssertCurrentThreadOwns();
michael@0: 
michael@0:     mChan->OnMessageReceivedFromLink(*msg);
michael@0:     delete msg;
michael@0: }
michael@0: 
michael@0: void
michael@0: ThreadLink::SendMessage(Message *msg)
michael@0: {
michael@0:     mChan->AssertWorkerThread();
michael@0:     mChan->mMonitor->AssertCurrentThreadOwns();
michael@0: 
michael@0:     if (mTargetChan)
michael@0:         mTargetChan->OnMessageReceivedFromLink(*msg);
michael@0:     delete msg;
michael@0: }
michael@0: 
michael@0: void
michael@0: ThreadLink::SendClose()
michael@0: {
michael@0:     mChan->AssertWorkerThread();
michael@0:     mChan->mMonitor->AssertCurrentThreadOwns();
michael@0: 
michael@0:     mChan->mChannelState = ChannelClosed;
michael@0: 
michael@0:     // In a ProcessLink, we would close our half the channel.  This
michael@0:     // would show up on the other side as an error on the I/O thread.
michael@0:     // The I/O thread would then invoke OnChannelErrorFromLink().
michael@0:     // As usual, we skip that process and just invoke the
michael@0:     // OnChannelErrorFromLink() method directly.
michael@0:     if (mTargetChan)
michael@0:         mTargetChan->OnChannelErrorFromLink();
michael@0: }
michael@0: 
michael@0: bool
michael@0: ThreadLink::Unsound_IsClosed() const
michael@0: {
michael@0:     MonitorAutoLock lock(*mChan->mMonitor);
michael@0:     return mChan->mChannelState == ChannelClosed;
michael@0: }
michael@0: 
michael@0: uint32_t
michael@0: ThreadLink::Unsound_NumQueuedMessages() const
michael@0: {
michael@0:     // ThreadLinks don't have a message queue.
michael@0:     return 0;
michael@0: }
michael@0: 
michael@0: //
michael@0: // The methods below run in the context of the IO thread
michael@0: //
michael@0: 
michael@0: void
michael@0: ProcessLink::OnMessageReceived(const Message& msg)
michael@0: {
michael@0:     AssertIOThread();
michael@0:     NS_ASSERTION(mChan->mChannelState != ChannelError, "Shouldn't get here!");
michael@0:     MonitorAutoLock lock(*mChan->mMonitor);
michael@0:     mChan->OnMessageReceivedFromLink(msg);
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::OnEchoMessage(Message* msg)
michael@0: {
michael@0:     AssertIOThread();
michael@0:     OnMessageReceived(*msg);
michael@0:     delete msg;
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::OnChannelOpened()
michael@0: {
michael@0:     mChan->AssertLinkThread();
michael@0:     {
michael@0:         MonitorAutoLock lock(*mChan->mMonitor);
michael@0: 
michael@0:         mExistingListener = mTransport->set_listener(this);
michael@0: #ifdef DEBUG
michael@0:         if (mExistingListener) {
michael@0:             queue<Message> pending;
michael@0:             mExistingListener->GetQueuedMessages(pending);
michael@0:             MOZ_ASSERT(pending.empty());
michael@0:         }
michael@0: #endif  // DEBUG
michael@0: 
michael@0:         mChan->mChannelState = ChannelOpening;
michael@0:         lock.Notify();
michael@0:     }
michael@0:     /*assert*/mTransport->Connect();
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::OnTakeConnectedChannel()
michael@0: {
michael@0:     AssertIOThread();
michael@0: 
michael@0:     queue<Message> pending;
michael@0:     {
michael@0:         MonitorAutoLock lock(*mChan->mMonitor);
michael@0: 
michael@0:         mChan->mChannelState = ChannelConnected;
michael@0: 
michael@0:         mExistingListener = mTransport->set_listener(this);
michael@0:         if (mExistingListener) {
michael@0:             mExistingListener->GetQueuedMessages(pending);
michael@0:         }
michael@0:         lock.Notify();
michael@0:     }
michael@0: 
michael@0:     // Dispatch whatever messages the previous listener had queued up.
michael@0:     while (!pending.empty()) {
michael@0:         OnMessageReceived(pending.front());
michael@0:         pending.pop();
michael@0:     }
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::OnChannelConnected(int32_t peer_pid)
michael@0: {
michael@0:     AssertIOThread();
michael@0: 
michael@0:     {
michael@0:         MonitorAutoLock lock(*mChan->mMonitor);
michael@0:         mChan->mChannelState = ChannelConnected;
michael@0:         mChan->mMonitor->Notify();
michael@0:     }
michael@0: 
michael@0:     if (mExistingListener)
michael@0:         mExistingListener->OnChannelConnected(peer_pid);
michael@0: 
michael@0:     mChan->OnChannelConnected(peer_pid);
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::OnChannelError()
michael@0: {
michael@0:     AssertIOThread();
michael@0:     MonitorAutoLock lock(*mChan->mMonitor);
michael@0:     mChan->OnChannelErrorFromLink();
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessLink::OnCloseChannel()
michael@0: {
michael@0:     AssertIOThread();
michael@0: 
michael@0:     mTransport->Close();
michael@0: 
michael@0:     MonitorAutoLock lock(*mChan->mMonitor);
michael@0:     mChan->mChannelState = ChannelClosed;
michael@0:     mChan->mMonitor->Notify();
michael@0: }
michael@0: 
michael@0: bool
michael@0: ProcessLink::Unsound_IsClosed() const
michael@0: {
michael@0:     return mTransport->Unsound_IsClosed();
michael@0: }
michael@0: 
michael@0: uint32_t
michael@0: ProcessLink::Unsound_NumQueuedMessages() const
michael@0: {
michael@0:     return mTransport->Unsound_NumQueuedMessages();
michael@0: }
michael@0: 
michael@0: } // namespace ipc
michael@0: } // namespace mozilla