ipc/glue/MessageChannel.cpp

   1 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*-
   2  * vim: sw=2 ts=4 et :
   3  */
   4 /* This Source Code Form is subject to the terms of the Mozilla Public
   5  * License, v. 2.0. If a copy of the MPL was not distributed with this
   6  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   7
   8 #include "mozilla/ipc/MessageChannel.h"
   9
  10 #include <math.h>
  11
  12 #include <utility>
  13
  14 #include "CrashAnnotations.h"
  15 #include "mozilla/Assertions.h"
  16 #include "mozilla/CycleCollectedJSContext.h"
  17 #include "mozilla/DebugOnly.h"
  18 #include "mozilla/Fuzzing.h"
  19 #include "mozilla/IntentionalCrash.h"
  20 #include "mozilla/Logging.h"
  21 #include "mozilla/Monitor.h"
  22 #include "mozilla/Mutex.h"
  23 #include "mozilla/ProfilerMarkers.h"
  24 #include "mozilla/ScopeExit.h"
  25 #include "mozilla/Sprintf.h"
  26 #include "mozilla/StaticMutex.h"
  27 #include "mozilla/Telemetry.h"
  28 #include "mozilla/TimeStamp.h"
  29 #include "mozilla/UniquePtrExtensions.h"
  30 #include "mozilla/dom/ScriptSettings.h"
  31 #include "mozilla/ipc/NodeController.h"
  32 #include "mozilla/ipc/ProcessChild.h"
  33 #include "mozilla/ipc/ProtocolUtils.h"
  34 #include "nsAppRunner.h"
  35 #include "nsContentUtils.h"
  36 #include "nsIDirectTaskDispatcher.h"
  37 #include "nsTHashMap.h"
  38 #include "nsDebug.h"
  39 #include "nsExceptionHandler.h"
  40 #include "nsIMemoryReporter.h"
  41 #include "nsISupportsImpl.h"
  42 #include "nsPrintfCString.h"
  43 #include "nsThreadUtils.h"
  44
  45 #ifdef XP_WIN
  46 #  include "mozilla/gfx/Logging.h"
  47 #endif
  48
  49 #ifdef FUZZING_SNAPSHOT
  50 #  include "mozilla/fuzzing/IPCFuzzController.h"
  51 #endif
  52
  53 // Undo the damage done by mozzconf.h
  54 #undef compress
  55
  56 static mozilla::LazyLogModule sLogModule("ipc");
  57 #define IPC_LOG(...) MOZ_LOG(sLogModule, LogLevel::Debug, (__VA_ARGS__))
  58
  59 /*
  60  * IPC design:
  61  *
  62  * There are two kinds of messages: async and sync. Sync messages are blocking.
  63  *
  64  * Terminology: To dispatch a message Foo is to run the RecvFoo code for
  65  * it. This is also called "handling" the message.
  66  *
  67  * Sync and async messages can sometimes "nest" inside other sync messages
  68  * (i.e., while waiting for the sync reply, we can dispatch the inner
  69  * message). The three possible nesting levels are NOT_NESTED,
  70  * NESTED_INSIDE_SYNC, and NESTED_INSIDE_CPOW.  The intended uses are:
  71  *   NOT_NESTED - most messages.
  72  *   NESTED_INSIDE_SYNC - CPOW-related messages, which are always sync
  73  *                        and can go in either direction.
  74  *   NESTED_INSIDE_CPOW - messages where we don't want to dispatch
  75  *                        incoming CPOWs while waiting for the response.
  76  * These nesting levels are ordered: NOT_NESTED, NESTED_INSIDE_SYNC,
  77  * NESTED_INSIDE_CPOW.  Async messages cannot be NESTED_INSIDE_SYNC but they can
  78  * be NESTED_INSIDE_CPOW.
  79  *
  80  * To avoid jank, the parent process is not allowed to send NOT_NESTED sync
  81  * messages. When a process is waiting for a response to a sync message M0, it
  82  * will dispatch an incoming message M if:
  83  *   1. M has a higher nesting level than M0, or
  84  *   2. if M has the same nesting level as M0 and we're in the child, or
  85  *   3. if M has the same nesting level as M0 and it was sent by the other side
  86  *      while dispatching M0.
  87  * The idea is that messages with higher nesting should take precendence. The
  88  * purpose of rule 2 is to handle a race where both processes send to each other
  89  * simultaneously. In this case, we resolve the race in favor of the parent (so
  90  * the child dispatches first).
  91  *
  92  * Messages satisfy the following properties:
  93  *   A. When waiting for a response to a sync message, we won't dispatch any
  94  *      messages of a lower nesting level.
  95  *   B. Messages of the same nesting level will be dispatched roughly in the
  96  *      order they were sent. The exception is when the parent and child send
  97  *      sync messages to each other simulataneously. In this case, the parent's
  98  *      message is dispatched first. While it is dispatched, the child may send
  99  *      further nested messages, and these messages may be dispatched before the
 100  *      child's original message. We can consider ordering to be preserved here
 101  *      because we pretend that the child's original message wasn't sent until
 102  *      after the parent's message is finished being dispatched.
 103  *
 104  * When waiting for a sync message reply, we dispatch an async message only if
 105  * it is NESTED_INSIDE_CPOW. Normally NESTED_INSIDE_CPOW async
 106  * messages are sent only from the child. However, the parent can send
 107  * NESTED_INSIDE_CPOW async messages when it is creating a bridged protocol.
 108  */
 109
 110 using namespace mozilla;
 111 using namespace mozilla::ipc;
 112
 113 using mozilla::MonitorAutoLock;
 114 using mozilla::MonitorAutoUnlock;
 115 using mozilla::dom::AutoNoJSAPI;
 116
 117 #define IPC_ASSERT(_cond, ...)                                           \
 118   do {                                                                   \
 119     AssertWorkerThread();                                                \
 120     mMonitor->AssertCurrentThreadOwns();                                 \
 121     if (!(_cond)) DebugAbort(__FILE__, __LINE__, #_cond, ##__VA_ARGS__); \
 122   } while (0)
 123
 124 static MessageChannel* gParentProcessBlocker = nullptr;
 125
 126 namespace mozilla {
 127 namespace ipc {
 128
 129 static const uint32_t kMinTelemetryMessageSize = 4096;
 130
 131 // Note: we round the time we spend waiting for a response to the nearest
 132 // millisecond. So a min value of 1 ms actually captures from 500us and above.
 133 // This is used for both the sending and receiving side telemetry for sync IPC,
 134 // (IPC_SYNC_MAIN_LATENCY_MS and IPC_SYNC_RECEIVE_MS).
 135 static const uint32_t kMinTelemetrySyncIPCLatencyMs = 1;
 136
 137 // static
 138 bool MessageChannel::sIsPumpingMessages = false;
 139
 140 class AutoEnterTransaction {
 141  public:
 142   explicit AutoEnterTransaction(MessageChannel* aChan, int32_t aMsgSeqno,
 143                                 int32_t aTransactionID, int aNestedLevel)
 144       MOZ_REQUIRES(*aChan->mMonitor)
 145       : mChan(aChan),
 146         mActive(true),
 147         mOutgoing(true),
 148         mNestedLevel(aNestedLevel),
 149         mSeqno(aMsgSeqno),
 150         mTransaction(aTransactionID),
 151         mNext(mChan->mTransactionStack) {
 152     mChan->mMonitor->AssertCurrentThreadOwns();
 153     mChan->mTransactionStack = this;
 154   }
 155
 156   explicit AutoEnterTransaction(MessageChannel* aChan,
 157                                 const IPC::Message& aMessage)
 158       MOZ_REQUIRES(*aChan->mMonitor)
 159       : mChan(aChan),
 160         mActive(true),
 161         mOutgoing(false),
 162         mNestedLevel(aMessage.nested_level()),
 163         mSeqno(aMessage.seqno()),
 164         mTransaction(aMessage.transaction_id()),
 165         mNext(mChan->mTransactionStack) {
 166     mChan->mMonitor->AssertCurrentThreadOwns();
 167
 168     if (!aMessage.is_sync()) {
 169       mActive = false;
 170       return;
 171     }
 172
 173     mChan->mTransactionStack = this;
 174   }
 175
 176   ~AutoEnterTransaction() {
 177     mChan->mMonitor->AssertCurrentThreadOwns();
 178     if (mActive) {
 179       mChan->mTransactionStack = mNext;
 180     }
 181   }
 182
 183   void Cancel() {
 184     mChan->mMonitor->AssertCurrentThreadOwns();
 185     AutoEnterTransaction* cur = mChan->mTransactionStack;
 186     MOZ_RELEASE_ASSERT(cur == this);
 187     while (cur && cur->mNestedLevel != IPC::Message::NOT_NESTED) {
 188       // Note that, in the following situation, we will cancel multiple
 189       // transactions:
 190       // 1. Parent sends NESTED_INSIDE_SYNC message P1 to child.
 191       // 2. Child sends NESTED_INSIDE_SYNC message C1 to child.
 192       // 3. Child dispatches P1, parent blocks.
 193       // 4. Child cancels.
 194       // In this case, both P1 and C1 are cancelled. The parent will
 195       // remove C1 from its queue when it gets the cancellation message.
 196       MOZ_RELEASE_ASSERT(cur->mActive);
 197       cur->mActive = false;
 198       cur = cur->mNext;
 199     }
 200
 201     mChan->mTransactionStack = cur;
 202
 203     MOZ_RELEASE_ASSERT(IsComplete());
 204   }
 205
 206   bool AwaitingSyncReply() const {
 207     MOZ_RELEASE_ASSERT(mActive);
 208     if (mOutgoing) {
 209       return true;
 210     }
 211     return mNext ? mNext->AwaitingSyncReply() : false;
 212   }
 213
 214   int AwaitingSyncReplyNestedLevel() const {
 215     MOZ_RELEASE_ASSERT(mActive);
 216     if (mOutgoing) {
 217       return mNestedLevel;
 218     }
 219     return mNext ? mNext->AwaitingSyncReplyNestedLevel() : 0;
 220   }
 221
 222   bool DispatchingSyncMessage() const {
 223     MOZ_RELEASE_ASSERT(mActive);
 224     if (!mOutgoing) {
 225       return true;
 226     }
 227     return mNext ? mNext->DispatchingSyncMessage() : false;
 228   }
 229
 230   int DispatchingSyncMessageNestedLevel() const {
 231     MOZ_RELEASE_ASSERT(mActive);
 232     if (!mOutgoing) {
 233       return mNestedLevel;
 234     }
 235     return mNext ? mNext->DispatchingSyncMessageNestedLevel() : 0;
 236   }
 237
 238   int NestedLevel() const {
 239     MOZ_RELEASE_ASSERT(mActive);
 240     return mNestedLevel;
 241   }
 242
 243   int32_t SequenceNumber() const {
 244     MOZ_RELEASE_ASSERT(mActive);
 245     return mSeqno;
 246   }
 247
 248   int32_t TransactionID() const {
 249     MOZ_RELEASE_ASSERT(mActive);
 250     return mTransaction;
 251   }
 252
 253   void ReceivedReply(UniquePtr<IPC::Message> aMessage) {
 254     MOZ_RELEASE_ASSERT(aMessage->seqno() == mSeqno);
 255     MOZ_RELEASE_ASSERT(aMessage->transaction_id() == mTransaction);
 256     MOZ_RELEASE_ASSERT(!mReply);
 257     IPC_LOG("Reply received on worker thread: seqno=%d", mSeqno);
 258     mReply = std::move(aMessage);
 259     MOZ_RELEASE_ASSERT(IsComplete());
 260   }
 261
 262   void HandleReply(UniquePtr<IPC::Message> aMessage) {
 263     mChan->mMonitor->AssertCurrentThreadOwns();
 264     AutoEnterTransaction* cur = mChan->mTransactionStack;
 265     MOZ_RELEASE_ASSERT(cur == this);
 266     while (cur) {
 267       MOZ_RELEASE_ASSERT(cur->mActive);
 268       if (aMessage->seqno() == cur->mSeqno) {
 269         cur->ReceivedReply(std::move(aMessage));
 270         break;
 271       }
 272       cur = cur->mNext;
 273       MOZ_RELEASE_ASSERT(cur);
 274     }
 275   }
 276
 277   bool IsComplete() { return !mActive || mReply; }
 278
 279   bool IsOutgoing() { return mOutgoing; }
 280
 281   bool IsCanceled() { return !mActive; }
 282
 283   bool IsBottom() const { return !mNext; }
 284
 285   bool IsError() {
 286     MOZ_RELEASE_ASSERT(mReply);
 287     return mReply->is_reply_error();
 288   }
 289
 290   UniquePtr<IPC::Message> GetReply() { return std::move(mReply); }
 291
 292  private:
 293   MessageChannel* mChan;
 294
 295   // Active is true if this transaction is on the mChan->mTransactionStack
 296   // stack. Generally we're not on the stack if the transaction was canceled
 297   // or if it was for a message that doesn't require transactions (an async
 298   // message).
 299   bool mActive;
 300
 301   // Is this stack frame for an outgoing message?
 302   bool mOutgoing;
 303
 304   // Properties of the message being sent/received.
 305   int mNestedLevel;
 306   int32_t mSeqno;
 307   int32_t mTransaction;
 308
 309   // Next item in mChan->mTransactionStack.
 310   AutoEnterTransaction* mNext;
 311
 312   // Pointer the a reply received for this message, if one was received.
 313   UniquePtr<IPC::Message> mReply;
 314 };
 315
 316 class PendingResponseReporter final : public nsIMemoryReporter {
 317   ~PendingResponseReporter() = default;
 318
 319  public:
 320   NS_DECL_THREADSAFE_ISUPPORTS
 321
 322   NS_IMETHOD
 323   CollectReports(nsIHandleReportCallback* aHandleReport, nsISupports* aData,
 324                  bool aAnonymize) override {
 325     MOZ_COLLECT_REPORT(
 326         "unresolved-ipc-responses", KIND_OTHER, UNITS_COUNT,
 327         MessageChannel::gUnresolvedResponses,
 328         "Outstanding IPC async message responses that are still not resolved.");
 329     return NS_OK;
 330   }
 331 };
 332
 333 NS_IMPL_ISUPPORTS(PendingResponseReporter, nsIMemoryReporter)
 334
 335 class ChannelCountReporter final : public nsIMemoryReporter {
 336   ~ChannelCountReporter() = default;
 337
 338   struct ChannelCounts {
 339     size_t mNow;
 340     size_t mMax;
 341
 342     ChannelCounts() : mNow(0), mMax(0) {}
 343
 344     void Inc() {
 345       ++mNow;
 346       if (mMax < mNow) {
 347         mMax = mNow;
 348       }
 349     }
 350
 351     void Dec() {
 352       MOZ_ASSERT(mNow > 0);
 353       --mNow;
 354     }
 355   };
 356
 357   using CountTable = nsTHashMap<nsDepCharHashKey, ChannelCounts>;
 358
 359   static StaticMutex sChannelCountMutex;
 360   static CountTable* sChannelCounts MOZ_GUARDED_BY(sChannelCountMutex);
 361
 362  public:
 363   NS_DECL_THREADSAFE_ISUPPORTS
 364
 365   NS_IMETHOD
 366   CollectReports(nsIHandleReportCallback* aHandleReport, nsISupports* aData,
 367                  bool aAnonymize) override {
 368     AutoTArray<std::pair<const char*, ChannelCounts>, 16> counts;
 369     {
 370       StaticMutexAutoLock countLock(sChannelCountMutex);
 371       if (!sChannelCounts) {
 372         return NS_OK;
 373       }
 374       counts.SetCapacity(sChannelCounts->Count());
 375       for (const auto& entry : *sChannelCounts) {
 376         counts.AppendElement(std::pair{entry.GetKey(), entry.GetData()});
 377       }
 378     }
 379
 380     for (const auto& entry : counts) {
 381       nsPrintfCString pathNow("ipc-channels/%s", entry.first);
 382       nsPrintfCString pathMax("ipc-channels-peak/%s", entry.first);
 383       nsPrintfCString descNow(
 384           "Number of IPC channels for"
 385           " top-level actor type %s",
 386           entry.first);
 387       nsPrintfCString descMax(
 388           "Peak number of IPC channels for"
 389           " top-level actor type %s",
 390           entry.first);
 391
 392       aHandleReport->Callback(""_ns, pathNow, KIND_OTHER, UNITS_COUNT,
 393                               entry.second.mNow, descNow, aData);
 394       aHandleReport->Callback(""_ns, pathMax, KIND_OTHER, UNITS_COUNT,
 395                               entry.second.mMax, descMax, aData);
 396     }
 397     return NS_OK;
 398   }
 399
 400   static void Increment(const char* aName) {
 401     StaticMutexAutoLock countLock(sChannelCountMutex);
 402     if (!sChannelCounts) {
 403       sChannelCounts = new CountTable;
 404     }
 405     sChannelCounts->LookupOrInsert(aName).Inc();
 406   }
 407
 408   static void Decrement(const char* aName) {
 409     StaticMutexAutoLock countLock(sChannelCountMutex);
 410     MOZ_ASSERT(sChannelCounts);
 411     sChannelCounts->LookupOrInsert(aName).Dec();
 412   }
 413 };
 414
 415 StaticMutex ChannelCountReporter::sChannelCountMutex;
 416 ChannelCountReporter::CountTable* ChannelCountReporter::sChannelCounts;
 417
 418 NS_IMPL_ISUPPORTS(ChannelCountReporter, nsIMemoryReporter)
 419
 420 // In child processes, the first MessageChannel is created before
 421 // XPCOM is initialized enough to construct the memory reporter
 422 // manager.  This retries every time a MessageChannel is constructed,
 423 // which is good enough in practice.
 424 template <class Reporter>
 425 static void TryRegisterStrongMemoryReporter() {
 426   static Atomic<bool> registered;
 427   if (registered.compareExchange(false, true)) {
 428     RefPtr<Reporter> reporter = new Reporter();
 429     if (NS_FAILED(RegisterStrongMemoryReporter(reporter))) {
 430       registered = false;
 431     }
 432   }
 433 }
 434
 435 Atomic<size_t> MessageChannel::gUnresolvedResponses;
 436
 437 MessageChannel::MessageChannel(const char* aName, IToplevelProtocol* aListener)
 438     : mName(aName), mListener(aListener), mMonitor(new RefCountedMonitor()) {
 439   MOZ_COUNT_CTOR(ipc::MessageChannel);
 440
 441 #ifdef XP_WIN
 442   mEvent = CreateEventW(nullptr, TRUE, FALSE, nullptr);
 443   MOZ_RELEASE_ASSERT(mEvent, "CreateEvent failed! Nothing is going to work!");
 444 #endif
 445
 446   TryRegisterStrongMemoryReporter<PendingResponseReporter>();
 447   TryRegisterStrongMemoryReporter<ChannelCountReporter>();
 448 }
 449
 450 MessageChannel::~MessageChannel() {
 451   MOZ_COUNT_DTOR(ipc::MessageChannel);
 452   MonitorAutoLock lock(*mMonitor);
 453   MOZ_RELEASE_ASSERT(!mOnCxxStack,
 454                      "MessageChannel destroyed while code on CxxStack");
 455 #ifdef XP_WIN
 456   if (mEvent) {
 457     BOOL ok = CloseHandle(mEvent);
 458     mEvent = nullptr;
 459
 460     if (!ok) {
 461       gfxDevCrash(mozilla::gfx::LogReason::MessageChannelCloseFailure)
 462           << "MessageChannel failed to close. GetLastError: " << GetLastError();
 463     }
 464     MOZ_RELEASE_ASSERT(ok);
 465   } else {
 466     gfxDevCrash(mozilla::gfx::LogReason::MessageChannelCloseFailure)
 467         << "MessageChannel destructor ran without an mEvent Handle";
 468   }
 469 #endif
 470
 471   // Make sure that the MessageChannel was closed (and therefore cleared) before
 472   // it was destroyed. We can't properly close the channel at this point, as it
 473   // would be unsafe to invoke our listener's callbacks, and we may be being
 474   // destroyed on a thread other than `mWorkerThread`.
 475   if (!IsClosedLocked()) {
 476     CrashReporter::RecordAnnotationCString(
 477         CrashReporter::Annotation::IPCFatalErrorProtocol, mName);
 478     switch (mChannelState) {
 479       case ChannelConnected:
 480         MOZ_CRASH(
 481             "MessageChannel destroyed without being closed "
 482             "(mChannelState == ChannelConnected).");
 483         break;
 484       case ChannelClosing:
 485         MOZ_CRASH(
 486             "MessageChannel destroyed without being closed "
 487             "(mChannelState == ChannelClosing).");
 488         break;
 489       case ChannelError:
 490         MOZ_CRASH(
 491             "MessageChannel destroyed without being closed "
 492             "(mChannelState == ChannelError).");
 493         break;
 494       default:
 495         MOZ_CRASH("MessageChannel destroyed without being closed.");
 496     }
 497   }
 498
 499   // Double-check other properties for thoroughness.
 500   MOZ_RELEASE_ASSERT(!mLink);
 501   MOZ_RELEASE_ASSERT(mPendingResponses.empty());
 502   MOZ_RELEASE_ASSERT(!mChannelErrorTask);
 503   MOZ_RELEASE_ASSERT(mPending.isEmpty());
 504   MOZ_RELEASE_ASSERT(!mShutdownTask);
 505 }
 506
 507 #ifdef DEBUG
 508 void MessageChannel::AssertMaybeDeferredCountCorrect() {
 509   mMonitor->AssertCurrentThreadOwns();
 510
 511   size_t count = 0;
 512   for (MessageTask* task : mPending) {
 513     task->AssertMonitorHeld(*mMonitor);
 514     if (!IsAlwaysDeferred(*task->Msg())) {
 515       count++;
 516     }
 517   }
 518
 519   MOZ_ASSERT(count == mMaybeDeferredPendingCount);
 520 }
 521 #endif
 522
 523 // This function returns the current transaction ID. Since the notion of a
 524 // "current transaction" can be hard to define when messages race with each
 525 // other and one gets canceled and the other doesn't, we require that this
 526 // function is only called when the current transaction is known to be for a
 527 // NESTED_INSIDE_SYNC message. In that case, we know for sure what the caller is
 528 // looking for.
 529 int32_t MessageChannel::CurrentNestedInsideSyncTransaction() const {
 530   mMonitor->AssertCurrentThreadOwns();
 531   if (!mTransactionStack) {
 532     return 0;
 533   }
 534   MOZ_RELEASE_ASSERT(mTransactionStack->NestedLevel() ==
 535                      IPC::Message::NESTED_INSIDE_SYNC);
 536   return mTransactionStack->TransactionID();
 537 }
 538
 539 bool MessageChannel::TestOnlyIsTransactionComplete() const {
 540   AssertWorkerThread();
 541   MonitorAutoLock lock(*mMonitor);
 542   return !mTransactionStack || mTransactionStack->IsComplete();
 543 }
 544
 545 bool MessageChannel::AwaitingSyncReply() const {
 546   mMonitor->AssertCurrentThreadOwns();
 547   return mTransactionStack ? mTransactionStack->AwaitingSyncReply() : false;
 548 }
 549
 550 int MessageChannel::AwaitingSyncReplyNestedLevel() const {
 551   mMonitor->AssertCurrentThreadOwns();
 552   return mTransactionStack ? mTransactionStack->AwaitingSyncReplyNestedLevel()
 553                            : 0;
 554 }
 555
 556 bool MessageChannel::DispatchingSyncMessage() const {
 557   mMonitor->AssertCurrentThreadOwns();
 558   return mTransactionStack ? mTransactionStack->DispatchingSyncMessage()
 559                            : false;
 560 }
 561
 562 int MessageChannel::DispatchingSyncMessageNestedLevel() const {
 563   mMonitor->AssertCurrentThreadOwns();
 564   return mTransactionStack
 565              ? mTransactionStack->DispatchingSyncMessageNestedLevel()
 566              : 0;
 567 }
 568
 569 static void PrintErrorMessage(Side side, const char* channelName,
 570                               const char* msg) {
 571   printf_stderr("\n###!!! [%s][%s] Error: %s\n\n", StringFromIPCSide(side),
 572                 channelName, msg);
 573 }
 574
 575 bool MessageChannel::Connected() const {
 576   mMonitor->AssertCurrentThreadOwns();
 577   return ChannelConnected == mChannelState;
 578 }
 579
 580 bool MessageChannel::ConnectedOrClosing() const {
 581   mMonitor->AssertCurrentThreadOwns();
 582   return ChannelConnected == mChannelState || ChannelClosing == mChannelState;
 583 }
 584
 585 bool MessageChannel::CanSend() const {
 586   if (!mMonitor) {
 587     return false;
 588   }
 589   MonitorAutoLock lock(*mMonitor);
 590   return Connected();
 591 }
 592
 593 void MessageChannel::Clear() {
 594   AssertWorkerThread();
 595   mMonitor->AssertCurrentThreadOwns();
 596   MOZ_DIAGNOSTIC_ASSERT(IsClosedLocked(), "MessageChannel cleared too early?");
 597   MOZ_ASSERT(ChannelClosed == mChannelState || ChannelError == mChannelState);
 598
 599   // Don't clear mWorkerThread; we use it in AssertWorkerThread().
 600   //
 601   // Also don't clear mListener.  If we clear it, then sending a message
 602   // through this channel after it's Clear()'ed can cause this process to
 603   // crash.
 604
 605   if (mShutdownTask) {
 606     mShutdownTask->Clear();
 607     mWorkerThread->UnregisterShutdownTask(mShutdownTask);
 608   }
 609   mShutdownTask = nullptr;
 610
 611   if (NS_IsMainThread() && gParentProcessBlocker == this) {
 612     gParentProcessBlocker = nullptr;
 613   }
 614
 615   gUnresolvedResponses -= mPendingResponses.size();
 616   {
 617     CallbackMap map = std::move(mPendingResponses);
 618     MonitorAutoUnlock unlock(*mMonitor);
 619     for (auto& pair : map) {
 620       pair.second->Reject(ResponseRejectReason::ChannelClosed);
 621     }
 622   }
 623   mPendingResponses.clear();
 624
 625   SetIsCrossProcess(false);
 626
 627   mLink = nullptr;
 628
 629   if (mChannelErrorTask) {
 630     mChannelErrorTask->Cancel();
 631     mChannelErrorTask = nullptr;
 632   }
 633
 634   if (mFlushLazySendTask) {
 635     mFlushLazySendTask->Cancel();
 636     mFlushLazySendTask = nullptr;
 637   }
 638
 639   // Free up any memory used by pending messages.
 640   mPending.clear();
 641
 642   mMaybeDeferredPendingCount = 0;
 643 }
 644
 645 bool MessageChannel::Open(ScopedPort aPort, Side aSide,
 646                           const nsID& aMessageChannelId,
 647                           nsISerialEventTarget* aEventTarget) {
 648   nsCOMPtr<nsISerialEventTarget> eventTarget =
 649       aEventTarget ? aEventTarget : GetCurrentSerialEventTarget();
 650   MOZ_RELEASE_ASSERT(eventTarget,
 651                      "Must open MessageChannel on a nsISerialEventTarget");
 652   MOZ_RELEASE_ASSERT(eventTarget->IsOnCurrentThread(),
 653                      "Must open MessageChannel from worker thread");
 654
 655   auto shutdownTask = MakeRefPtr<WorkerTargetShutdownTask>(eventTarget, this);
 656   nsresult rv = eventTarget->RegisterShutdownTask(shutdownTask);
 657   MOZ_ASSERT(rv != NS_ERROR_NOT_IMPLEMENTED,
 658              "target for MessageChannel must support shutdown tasks");
 659   if (rv == NS_ERROR_UNEXPECTED) {
 660     // If shutdown tasks have already started running, dispatch our shutdown
 661     // task manually.
 662     NS_WARNING("Opening MessageChannel on EventTarget in shutdown");
 663     rv = eventTarget->Dispatch(shutdownTask->AsRunnable());
 664   }
 665   MOZ_RELEASE_ASSERT(NS_SUCCEEDED(rv),
 666                      "error registering ShutdownTask for MessageChannel");
 667
 668   {
 669     MonitorAutoLock lock(*mMonitor);
 670     MOZ_RELEASE_ASSERT(!mLink, "Open() called > once");
 671     MOZ_RELEASE_ASSERT(ChannelClosed == mChannelState, "Not currently closed");
 672     MOZ_ASSERT(mSide == UnknownSide);
 673
 674     mMessageChannelId = aMessageChannelId;
 675     mWorkerThread = eventTarget;
 676     mShutdownTask = shutdownTask;
 677     mLink = MakeUnique<PortLink>(this, std::move(aPort));
 678     mChannelState = ChannelConnected;
 679     mSide = aSide;
 680   }
 681
 682   // Notify our listener that the underlying IPC channel has been established.
 683   // IProtocol will use this callback to create the ActorLifecycleProxy, and
 684   // perform an `AddRef` call to keep the actor alive until the channel is
 685   // disconnected.
 686   //
 687   // We unlock our monitor before calling `OnIPCChannelOpened` to ensure that
 688   // any calls back into `MessageChannel` do not deadlock. At this point, we may
 689   // be receiving messages on the IO thread, however we cannot process them on
 690   // the worker thread or have notified our listener until after this function
 691   // returns.
 692   mListener->OnIPCChannelOpened();
 693   return true;
 694 }
 695
 696 static Side GetOppSide(Side aSide) {
 697   switch (aSide) {
 698     case ChildSide:
 699       return ParentSide;
 700     case ParentSide:
 701       return ChildSide;
 702     default:
 703       return UnknownSide;
 704   }
 705 }
 706
 707 bool MessageChannel::Open(MessageChannel* aTargetChan,
 708                           nsISerialEventTarget* aEventTarget, Side aSide) {
 709   // Opens a connection to another thread in the same process.
 710
 711   MOZ_ASSERT(aTargetChan, "Need a target channel");
 712
 713   nsID channelId = nsID::GenerateUUID();
 714
 715   std::pair<ScopedPort, ScopedPort> ports =
 716       NodeController::GetSingleton()->CreatePortPair();
 717
 718   // NOTE: This dispatch must be sync as it captures locals by non-owning
 719   // reference, however we can't use `NS_DispatchAndSpinEventLoopUntilComplete`
 720   // as that will spin a nested event loop, and doesn't work with certain types
 721   // of calling event targets.
 722   base::WaitableEvent event(/* manual_reset */ true,
 723                             /* initially_signaled */ false);
 724   MOZ_ALWAYS_SUCCEEDS(aEventTarget->Dispatch(NS_NewCancelableRunnableFunction(
 725       "ipc::MessageChannel::OpenAsOtherThread", [&]() {
 726         aTargetChan->Open(std::move(ports.second), GetOppSide(aSide), channelId,
 727                           aEventTarget);
 728         event.Signal();
 729       })));
 730   bool ok = event.Wait();
 731   MOZ_RELEASE_ASSERT(ok);
 732
 733   // Now that the other side has connected, open the port on our side.
 734   return Open(std::move(ports.first), aSide, channelId);
 735 }
 736
 737 bool MessageChannel::OpenOnSameThread(MessageChannel* aTargetChan,
 738                                       mozilla::ipc::Side aSide) {
 739   auto [porta, portb] = NodeController::GetSingleton()->CreatePortPair();
 740
 741   nsID channelId = nsID::GenerateUUID();
 742
 743   aTargetChan->mIsSameThreadChannel = true;
 744   mIsSameThreadChannel = true;
 745
 746   auto* currentThread = GetCurrentSerialEventTarget();
 747   return aTargetChan->Open(std::move(portb), GetOppSide(aSide), channelId,
 748                            currentThread) &&
 749          Open(std::move(porta), aSide, channelId, currentThread);
 750 }
 751
 752 bool MessageChannel::Send(UniquePtr<Message> aMsg) {
 753   if (aMsg->size() >= kMinTelemetryMessageSize) {
 754     Telemetry::Accumulate(Telemetry::IPC_MESSAGE_SIZE2, aMsg->size());
 755   }
 756
 757   MOZ_RELEASE_ASSERT(!aMsg->is_sync());
 758   MOZ_RELEASE_ASSERT(aMsg->nested_level() != IPC::Message::NESTED_INSIDE_SYNC);
 759
 760   AutoSetValue<bool> setOnCxxStack(mOnCxxStack, true);
 761
 762   AssertWorkerThread();
 763   mMonitor->AssertNotCurrentThreadOwns();
 764   if (MSG_ROUTING_NONE == aMsg->routing_id()) {
 765     ReportMessageRouteError("MessageChannel::Send");
 766     return false;
 767   }
 768
 769   if (aMsg->seqno() == 0) {
 770     aMsg->set_seqno(NextSeqno());
 771   }
 772
 773   MonitorAutoLock lock(*mMonitor);
 774   if (!Connected()) {
 775     ReportConnectionError("Send", aMsg->type());
 776     return false;
 777   }
 778
 779   AddProfilerMarker(*aMsg, MessageDirection::eSending);
 780   SendMessageToLink(std::move(aMsg));
 781   return true;
 782 }
 783
 784 void MessageChannel::SendMessageToLink(UniquePtr<Message> aMsg) {
 785   AssertWorkerThread();
 786   mMonitor->AssertCurrentThreadOwns();
 787
 788   // If the channel is not cross-process, there's no reason to be lazy, so we
 789   // ignore the flag in that case.
 790   if (aMsg->is_lazy_send() && mIsCrossProcess) {
 791     // If this is the first lazy message in the queue and our worker thread
 792     // supports direct task dispatch, dispatch a task to flush messages,
 793     // ensuring we don't leave them pending forever.
 794     if (!mFlushLazySendTask) {
 795       if (nsCOMPtr<nsIDirectTaskDispatcher> dispatcher =
 796               do_QueryInterface(mWorkerThread)) {
 797         mFlushLazySendTask = new FlushLazySendMessagesRunnable(this);
 798         MOZ_ALWAYS_SUCCEEDS(
 799             dispatcher->DispatchDirectTask(do_AddRef(mFlushLazySendTask)));
 800       }
 801     }
 802     if (mFlushLazySendTask) {
 803       mFlushLazySendTask->PushMessage(std::move(aMsg));
 804       return;
 805     }
 806   }
 807
 808   if (mFlushLazySendTask) {
 809     FlushLazySendMessages();
 810   }
 811   mLink->SendMessage(std::move(aMsg));
 812 }
 813
 814 void MessageChannel::FlushLazySendMessages() {
 815   AssertWorkerThread();
 816   mMonitor->AssertCurrentThreadOwns();
 817
 818   // Clean up any SendLazyTask which might be pending.
 819   auto messages = mFlushLazySendTask->TakeMessages();
 820   mFlushLazySendTask = nullptr;
 821
 822   // Send all lazy messages, then clear the queue.
 823   for (auto& msg : messages) {
 824     mLink->SendMessage(std::move(msg));
 825   }
 826 }
 827
 828 UniquePtr<MessageChannel::UntypedCallbackHolder> MessageChannel::PopCallback(
 829     const Message& aMsg, int32_t aActorId) {
 830   auto iter = mPendingResponses.find(aMsg.seqno());
 831   if (iter != mPendingResponses.end() && iter->second->mActorId == aActorId &&
 832       iter->second->mReplyMsgId == aMsg.type()) {
 833     UniquePtr<MessageChannel::UntypedCallbackHolder> ret =
 834         std::move(iter->second);
 835     mPendingResponses.erase(iter);
 836     gUnresolvedResponses--;
 837     return ret;
 838   }
 839   return nullptr;
 840 }
 841
 842 void MessageChannel::RejectPendingResponsesForActor(int32_t aActorId) {
 843   auto itr = mPendingResponses.begin();
 844   while (itr != mPendingResponses.end()) {
 845     if (itr->second.get()->mActorId != aActorId) {
 846       ++itr;
 847       continue;
 848     }
 849     itr->second.get()->Reject(ResponseRejectReason::ActorDestroyed);
 850     // Take special care of advancing the iterator since we are
 851     // removing it while iterating.
 852     itr = mPendingResponses.erase(itr);
 853     gUnresolvedResponses--;
 854   }
 855 }
 856
 857 class BuildIDsMatchMessage : public IPC::Message {
 858  public:
 859   BuildIDsMatchMessage()
 860       : IPC::Message(MSG_ROUTING_NONE, BUILD_IDS_MATCH_MESSAGE_TYPE) {}
 861   void Log(const std::string& aPrefix, FILE* aOutf) const {
 862     fputs("(special `Build IDs match' message)", aOutf);
 863   }
 864 };
 865
 866 // Send the parent a special async message to confirm when the parent and child
 867 // are of the same buildID. Skips sending the message and returns false if the
 868 // buildIDs don't match. This is a minor variation on
 869 // MessageChannel::Send(Message* aMsg).
 870 bool MessageChannel::SendBuildIDsMatchMessage(const char* aParentBuildID) {
 871   MOZ_ASSERT(!XRE_IsParentProcess());
 872
 873   nsCString parentBuildID(aParentBuildID);
 874   nsCString childBuildID(mozilla::PlatformBuildID());
 875
 876   if (parentBuildID != childBuildID) {
 877     // The build IDs didn't match, usually because an update occurred in the
 878     // background.
 879     return false;
 880   }
 881
 882   auto msg = MakeUnique<BuildIDsMatchMessage>();
 883
 884   MOZ_RELEASE_ASSERT(!msg->is_sync());
 885   MOZ_RELEASE_ASSERT(msg->nested_level() != IPC::Message::NESTED_INSIDE_SYNC);
 886
 887   AssertWorkerThread();
 888   mMonitor->AssertNotCurrentThreadOwns();
 889   // Don't check for MSG_ROUTING_NONE.
 890
 891   MonitorAutoLock lock(*mMonitor);
 892   if (!Connected()) {
 893     ReportConnectionError("SendBuildIDsMatchMessage", msg->type());
 894     return false;
 895   }
 896
 897 #if defined(MOZ_DEBUG) && defined(ENABLE_TESTS)
 898   // Technically, the behavior is interesting for any kind of process
 899   // but when exercising tests, we want to crash only a content process and
 900   // avoid making noise with other kind of processes crashing
 901   if (const char* dontSend = PR_GetEnv("MOZ_BUILDID_MATCH_DONTSEND")) {
 902     if (dontSend[0] == '1') {
 903       // Bug 1732999: We are going to crash, so we need to advise leak check
 904       // tooling to avoid intermittent missing leakcheck
 905       NoteIntentionalCrash(XRE_GetProcessTypeString());
 906       if (XRE_IsContentProcess()) {
 907         return false;
 908       }
 909     }
 910   }
 911 #endif
 912
 913   SendMessageToLink(std::move(msg));
 914   return true;
 915 }
 916
 917 class CancelMessage : public IPC::Message {
 918  public:
 919   explicit CancelMessage(int transaction)
 920       : IPC::Message(MSG_ROUTING_NONE, CANCEL_MESSAGE_TYPE) {
 921     set_transaction_id(transaction);
 922   }
 923   static bool Read(const Message* msg) { return true; }
 924   void Log(const std::string& aPrefix, FILE* aOutf) const {
 925     fputs("(special `Cancel' message)", aOutf);
 926   }
 927 };
 928
 929 bool MessageChannel::MaybeInterceptSpecialIOMessage(const Message& aMsg) {
 930   mMonitor->AssertCurrentThreadOwns();
 931
 932   if (MSG_ROUTING_NONE == aMsg.routing_id()) {
 933     if (GOODBYE_MESSAGE_TYPE == aMsg.type()) {
 934       // We've received a GOODBYE message, close the connection and mark
 935       // ourselves as "Closing".
 936       mLink->Close();
 937       mChannelState = ChannelClosing;
 938       if (LoggingEnabledFor(mListener->GetProtocolName(), mSide)) {
 939         printf(
 940             "[%s %u] NOTE: %s%s actor received `Goodbye' message.  Closing "
 941             "channel.\n",
 942             XRE_GeckoProcessTypeToString(XRE_GetProcessType()),
 943             static_cast<uint32_t>(base::GetCurrentProcId()),
 944             mListener->GetProtocolName(), StringFromIPCSide(mSide));
 945       }
 946
 947       // Notify the worker thread that the connection has been closed, as we
 948       // will not receive an `OnChannelErrorFromLink` after calling
 949       // `mLink->Close()`.
 950       if (AwaitingSyncReply()) {
 951         NotifyWorkerThread();
 952       }
 953       PostErrorNotifyTask();
 954       return true;
 955     } else if (CANCEL_MESSAGE_TYPE == aMsg.type()) {
 956       IPC_LOG("Cancel from message");
 957       CancelTransaction(aMsg.transaction_id());
 958       NotifyWorkerThread();
 959       return true;
 960     } else if (BUILD_IDS_MATCH_MESSAGE_TYPE == aMsg.type()) {
 961       IPC_LOG("Build IDs match message");
 962       mBuildIDsConfirmedMatch = true;
 963       return true;
 964     } else if (IMPENDING_SHUTDOWN_MESSAGE_TYPE == aMsg.type()) {
 965       IPC_LOG("Impending Shutdown received");
 966       ProcessChild::NotifiedImpendingShutdown();
 967       return true;
 968     }
 969   }
 970   return false;
 971 }
 972
 973 /* static */
 974 bool MessageChannel::IsAlwaysDeferred(const Message& aMsg) {
 975   // If a message is not NESTED_INSIDE_CPOW and not sync, then we always defer
 976   // it.
 977   return aMsg.nested_level() != IPC::Message::NESTED_INSIDE_CPOW &&
 978          !aMsg.is_sync();
 979 }
 980
 981 bool MessageChannel::ShouldDeferMessage(const Message& aMsg) {
 982   // Never defer messages that have the highest nested level, even async
 983   // ones. This is safe because only the child can send these messages, so
 984   // they can never nest.
 985   if (aMsg.nested_level() == IPC::Message::NESTED_INSIDE_CPOW) {
 986     MOZ_ASSERT(!IsAlwaysDeferred(aMsg));
 987     return false;
 988   }
 989
 990   // Unless they're NESTED_INSIDE_CPOW, we always defer async messages.
 991   // Note that we never send an async NESTED_INSIDE_SYNC message.
 992   if (!aMsg.is_sync()) {
 993     MOZ_RELEASE_ASSERT(aMsg.nested_level() == IPC::Message::NOT_NESTED);
 994     MOZ_ASSERT(IsAlwaysDeferred(aMsg));
 995     return true;
 996   }
 997
 998   MOZ_ASSERT(!IsAlwaysDeferred(aMsg));
 999
1000   int msgNestedLevel = aMsg.nested_level();
1001   int waitingNestedLevel = AwaitingSyncReplyNestedLevel();
1002
1003   // Always defer if the nested level of the incoming message is less than the
1004   // nested level of the message we're awaiting.
1005   if (msgNestedLevel < waitingNestedLevel) return true;
1006
1007   // Never defer if the message has strictly greater nested level.
1008   if (msgNestedLevel > waitingNestedLevel) return false;
1009
1010   // When both sides send sync messages of the same nested level, we resolve the
1011   // race by dispatching in the child and deferring the incoming message in
1012   // the parent. However, the parent still needs to dispatch nested sync
1013   // messages.
1014   //
1015   // Deferring in the parent only sort of breaks message ordering. When the
1016   // child's message comes in, we can pretend the child hasn't quite
1017   // finished sending it yet. Since the message is sync, we know that the
1018   // child hasn't moved on yet.
1019   return mSide == ParentSide &&
1020          aMsg.transaction_id() != CurrentNestedInsideSyncTransaction();
1021 }
1022
1023 void MessageChannel::OnMessageReceivedFromLink(UniquePtr<Message> aMsg) {
1024   mMonitor->AssertCurrentThreadOwns();
1025   MOZ_ASSERT(mChannelState == ChannelConnected);
1026
1027   if (MaybeInterceptSpecialIOMessage(*aMsg)) {
1028     return;
1029   }
1030
1031   mListener->OnChannelReceivedMessage(*aMsg);
1032
1033   // If we're awaiting a sync reply, we know that it needs to be immediately
1034   // handled to unblock us.
1035   if (aMsg->is_sync() && aMsg->is_reply()) {
1036     IPC_LOG("Received reply seqno=%d xid=%d", aMsg->seqno(),
1037             aMsg->transaction_id());
1038
1039     if (aMsg->seqno() == mTimedOutMessageSeqno) {
1040       // Drop the message, but allow future sync messages to be sent.
1041       IPC_LOG("Received reply to timedout message; igoring; xid=%d",
1042               mTimedOutMessageSeqno);
1043       EndTimeout();
1044       return;
1045     }
1046
1047     MOZ_RELEASE_ASSERT(AwaitingSyncReply());
1048     MOZ_RELEASE_ASSERT(!mTimedOutMessageSeqno);
1049
1050     mTransactionStack->HandleReply(std::move(aMsg));
1051     NotifyWorkerThread();
1052     return;
1053   }
1054
1055   // Nested messages cannot be compressed.
1056   MOZ_RELEASE_ASSERT(aMsg->compress_type() == IPC::Message::COMPRESSION_NONE ||
1057                      aMsg->nested_level() == IPC::Message::NOT_NESTED);
1058
1059   if (aMsg->compress_type() == IPC::Message::COMPRESSION_ENABLED &&
1060       !mPending.isEmpty()) {
1061     auto* last = mPending.getLast();
1062     last->AssertMonitorHeld(*mMonitor);
1063     bool compress = last->Msg()->type() == aMsg->type() &&
1064                     last->Msg()->routing_id() == aMsg->routing_id();
1065     if (compress) {
1066       // This message type has compression enabled, and the back of the
1067       // queue was the same message type and routed to the same destination.
1068       // Replace it with the newer message.
1069       MOZ_RELEASE_ASSERT(last->Msg()->compress_type() ==
1070                          IPC::Message::COMPRESSION_ENABLED);
1071       last->Msg() = std::move(aMsg);
1072       return;
1073     }
1074   } else if (aMsg->compress_type() == IPC::Message::COMPRESSION_ALL &&
1075              !mPending.isEmpty()) {
1076     for (MessageTask* p = mPending.getLast(); p; p = p->getPrevious()) {
1077       p->AssertMonitorHeld(*mMonitor);
1078       if (p->Msg()->type() == aMsg->type() &&
1079           p->Msg()->routing_id() == aMsg->routing_id()) {
1080         // This message type has compression enabled, and the queue
1081         // holds a message with the same message type and routed to the
1082         // same destination. Erase it. Note that, since we always
1083         // compress these redundancies, There Can Be Only One.
1084         MOZ_RELEASE_ASSERT(p->Msg()->compress_type() ==
1085                            IPC::Message::COMPRESSION_ALL);
1086         MOZ_RELEASE_ASSERT(IsAlwaysDeferred(*p->Msg()));
1087         p->remove();
1088         break;
1089       }
1090     }
1091   }
1092
1093   bool alwaysDeferred = IsAlwaysDeferred(*aMsg);
1094
1095   bool shouldWakeUp = AwaitingSyncReply() && !ShouldDeferMessage(*aMsg);
1096
1097   IPC_LOG("Receive from link; seqno=%d, xid=%d, shouldWakeUp=%d", aMsg->seqno(),
1098           aMsg->transaction_id(), shouldWakeUp);
1099
1100   // There are two cases we're concerned about, relating to the state of the
1101   // worker thread:
1102   //
1103   // (1) We are waiting on a sync reply - worker thread is blocked on the
1104   //     IPC monitor.
1105   //   - If the message is NESTED_INSIDE_SYNC, we wake up the worker thread to
1106   //     deliver the message depending on ShouldDeferMessage. Otherwise, we
1107   //     leave it in the mPending queue, posting a task to the worker event
1108   //     loop, where it will be processed once the synchronous reply has been
1109   //     received.
1110   //
1111   // (2) We are not waiting on a reply.
1112   //   - We post a task to the worker event loop.
1113   //
1114   // Note that, we may notify the worker thread even though the monitor is not
1115   // blocked. This is okay, since we always check for pending events before
1116   // blocking again.
1117
1118   RefPtr<MessageTask> task = new MessageTask(this, std::move(aMsg));
1119   mPending.insertBack(task);
1120
1121   if (!alwaysDeferred) {
1122     mMaybeDeferredPendingCount++;
1123   }
1124
1125   if (shouldWakeUp) {
1126     NotifyWorkerThread();
1127   }
1128
1129   // Although we usually don't need to post a message task if
1130   // shouldWakeUp is true, it's easier to post anyway than to have to
1131   // guarantee that every Send call processes everything it's supposed to
1132   // before returning.
1133   task->AssertMonitorHeld(*mMonitor);
1134   task->Post();
1135 }
1136
1137 void MessageChannel::PeekMessages(
1138     const std::function<bool(const Message& aMsg)>& aInvoke) {
1139   // FIXME: We shouldn't be holding the lock for aInvoke!
1140   MonitorAutoLock lock(*mMonitor);
1141
1142   for (MessageTask* it : mPending) {
1143     it->AssertMonitorHeld(*mMonitor);
1144     const Message& msg = *it->Msg();
1145     if (!aInvoke(msg)) {
1146       break;
1147     }
1148   }
1149 }
1150
1151 void MessageChannel::ProcessPendingRequests(
1152     ActorLifecycleProxy* aProxy, AutoEnterTransaction& aTransaction) {
1153   mMonitor->AssertCurrentThreadOwns();
1154
1155   AssertMaybeDeferredCountCorrect();
1156   if (mMaybeDeferredPendingCount == 0) {
1157     return;
1158   }
1159
1160   IPC_LOG("ProcessPendingRequests for seqno=%d, xid=%d",
1161           aTransaction.SequenceNumber(), aTransaction.TransactionID());
1162
1163   // Loop until there aren't any more nested messages to process.
1164   for (;;) {
1165     // If we canceled during ProcessPendingRequest, then we need to leave
1166     // immediately because the results of ShouldDeferMessage will be
1167     // operating with weird state (as if no Send is in progress). That could
1168     // cause even NOT_NESTED sync messages to be processed (but not
1169     // NOT_NESTED async messages), which would break message ordering.
1170     if (aTransaction.IsCanceled()) {
1171       return;
1172     }
1173
1174     Vector<UniquePtr<Message>> toProcess;
1175
1176     for (MessageTask* p = mPending.getFirst(); p;) {
1177       p->AssertMonitorHeld(*mMonitor);
1178       UniquePtr<Message>& msg = p->Msg();
1179
1180       MOZ_RELEASE_ASSERT(!aTransaction.IsCanceled(),
1181                          "Calling ShouldDeferMessage when cancelled");
1182       bool defer = ShouldDeferMessage(*msg);
1183
1184       // Only log the interesting messages.
1185       if (msg->is_sync() ||
1186           msg->nested_level() == IPC::Message::NESTED_INSIDE_CPOW) {
1187         IPC_LOG("ShouldDeferMessage(seqno=%d) = %d", msg->seqno(), defer);
1188       }
1189
1190       if (!defer) {
1191         MOZ_ASSERT(!IsAlwaysDeferred(*msg));
1192
1193         if (!toProcess.append(std::move(msg))) MOZ_CRASH();
1194
1195         mMaybeDeferredPendingCount--;
1196
1197         p = p->removeAndGetNext();
1198         continue;
1199       }
1200       p = p->getNext();
1201     }
1202
1203     if (toProcess.empty()) {
1204       break;
1205     }
1206
1207     // Processing these messages could result in more messages, so we
1208     // loop around to check for more afterwards.
1209
1210     for (auto& msg : toProcess) {
1211       ProcessPendingRequest(aProxy, std::move(msg));
1212     }
1213   }
1214
1215   AssertMaybeDeferredCountCorrect();
1216 }
1217
1218 bool MessageChannel::Send(UniquePtr<Message> aMsg, UniquePtr<Message>* aReply) {
1219   mozilla::TimeStamp start = TimeStamp::Now();
1220   if (aMsg->size() >= kMinTelemetryMessageSize) {
1221     Telemetry::Accumulate(Telemetry::IPC_MESSAGE_SIZE2, aMsg->size());
1222   }
1223
1224   // Sanity checks.
1225   AssertWorkerThread();
1226   mMonitor->AssertNotCurrentThreadOwns();
1227   MOZ_RELEASE_ASSERT(!mIsSameThreadChannel,
1228                      "sync send over same-thread channel will deadlock!");
1229
1230   RefPtr<ActorLifecycleProxy> proxy = Listener()->GetLifecycleProxy();
1231
1232 #ifdef XP_WIN
1233   SyncStackFrame frame(this);
1234   NeuteredWindowRegion neuteredRgn(mFlags &
1235                                    REQUIRE_DEFERRED_MESSAGE_PROTECTION);
1236 #endif
1237
1238   AutoSetValue<bool> setOnCxxStack(mOnCxxStack, true);
1239
1240   MonitorAutoLock lock(*mMonitor);
1241
1242   if (mTimedOutMessageSeqno) {
1243     // Don't bother sending another sync message if a previous one timed out
1244     // and we haven't received a reply for it. Once the original timed-out
1245     // message receives a reply, we'll be able to send more sync messages
1246     // again.
1247     IPC_LOG("Send() failed due to previous timeout");
1248     mLastSendError = SyncSendError::PreviousTimeout;
1249     return false;
1250   }
1251
1252   if (DispatchingSyncMessageNestedLevel() == IPC::Message::NOT_NESTED &&
1253       aMsg->nested_level() > IPC::Message::NOT_NESTED) {
1254     // Don't allow sending CPOWs while we're dispatching a sync message.
1255     IPC_LOG("Nested level forbids send");
1256     mLastSendError = SyncSendError::SendingCPOWWhileDispatchingSync;
1257     return false;
1258   }
1259
1260   if (DispatchingSyncMessageNestedLevel() == IPC::Message::NESTED_INSIDE_CPOW ||
1261       DispatchingAsyncMessageNestedLevel() ==
1262           IPC::Message::NESTED_INSIDE_CPOW) {
1263     // Generally only the parent dispatches urgent messages. And the only
1264     // sync messages it can send are NESTED_INSIDE_SYNC. Mainly we want to
1265     // ensure here that we don't return false for non-CPOW messages.
1266     MOZ_RELEASE_ASSERT(aMsg->nested_level() ==
1267                        IPC::Message::NESTED_INSIDE_SYNC);
1268     IPC_LOG("Sending while dispatching urgent message");
1269     mLastSendError = SyncSendError::SendingCPOWWhileDispatchingUrgent;
1270     return false;
1271   }
1272
1273   if (aMsg->nested_level() < DispatchingSyncMessageNestedLevel() ||
1274       aMsg->nested_level() < AwaitingSyncReplyNestedLevel()) {
1275     MOZ_RELEASE_ASSERT(DispatchingSyncMessage() || DispatchingAsyncMessage());
1276     IPC_LOG("Cancel from Send");
1277     auto cancel =
1278         MakeUnique<CancelMessage>(CurrentNestedInsideSyncTransaction());
1279     CancelTransaction(CurrentNestedInsideSyncTransaction());
1280     SendMessageToLink(std::move(cancel));
1281   }
1282
1283   IPC_ASSERT(aMsg->is_sync(), "can only Send() sync messages here");
1284
1285   IPC_ASSERT(aMsg->nested_level() >= DispatchingSyncMessageNestedLevel(),
1286              "can't send sync message of a lesser nested level than what's "
1287              "being dispatched");
1288   IPC_ASSERT(AwaitingSyncReplyNestedLevel() <= aMsg->nested_level(),
1289              "nested sync message sends must be of increasing nested level");
1290   IPC_ASSERT(
1291       DispatchingSyncMessageNestedLevel() != IPC::Message::NESTED_INSIDE_CPOW,
1292       "not allowed to send messages while dispatching urgent messages");
1293
1294   IPC_ASSERT(
1295       DispatchingAsyncMessageNestedLevel() != IPC::Message::NESTED_INSIDE_CPOW,
1296       "not allowed to send messages while dispatching urgent messages");
1297
1298   if (!Connected()) {
1299     ReportConnectionError("SendAndWait", aMsg->type());
1300     mLastSendError = SyncSendError::NotConnectedBeforeSend;
1301     return false;
1302   }
1303
1304   aMsg->set_seqno(NextSeqno());
1305
1306   int32_t seqno = aMsg->seqno();
1307   int nestedLevel = aMsg->nested_level();
1308   msgid_t replyType = aMsg->type() + 1;
1309
1310   AutoEnterTransaction* stackTop = mTransactionStack;
1311
1312   // If the most recent message on the stack is NESTED_INSIDE_SYNC, then our
1313   // message should nest inside that and we use the same transaction
1314   // ID. Otherwise we need a new transaction ID (so we use the seqno of the
1315   // message we're sending).
1316   bool nest =
1317       stackTop && stackTop->NestedLevel() == IPC::Message::NESTED_INSIDE_SYNC;
1318   int32_t transaction = nest ? stackTop->TransactionID() : seqno;
1319   aMsg->set_transaction_id(transaction);
1320
1321   AutoEnterTransaction transact(this, seqno, transaction, nestedLevel);
1322
1323   IPC_LOG("Send seqno=%d, xid=%d", seqno, transaction);
1324
1325   // aMsg will be destroyed soon, let's keep its type.
1326   const char* msgName = aMsg->name();
1327   const msgid_t msgType = aMsg->type();
1328
1329   AddProfilerMarker(*aMsg, MessageDirection::eSending);
1330   SendMessageToLink(std::move(aMsg));
1331
1332   while (true) {
1333     MOZ_RELEASE_ASSERT(!transact.IsCanceled());
1334     ProcessPendingRequests(proxy, transact);
1335     if (transact.IsComplete()) {
1336       break;
1337     }
1338     if (!Connected()) {
1339       ReportConnectionError("Send", msgType);
1340       mLastSendError = SyncSendError::DisconnectedDuringSend;
1341       return false;
1342     }
1343
1344     MOZ_RELEASE_ASSERT(!mTimedOutMessageSeqno);
1345     MOZ_RELEASE_ASSERT(!transact.IsComplete());
1346     MOZ_RELEASE_ASSERT(mTransactionStack == &transact);
1347
1348     bool maybeTimedOut = !WaitForSyncNotify();
1349
1350     if (mListener->NeedArtificialSleep()) {
1351       MonitorAutoUnlock unlock(*mMonitor);
1352       mListener->ArtificialSleep();
1353     }
1354
1355     if (!Connected()) {
1356       ReportConnectionError("SendAndWait", msgType);
1357       mLastSendError = SyncSendError::DisconnectedDuringSend;
1358       return false;
1359     }
1360
1361     if (transact.IsCanceled()) {
1362       break;
1363     }
1364
1365     MOZ_RELEASE_ASSERT(mTransactionStack == &transact);
1366
1367     // We only time out a message if it initiated a new transaction (i.e.,
1368     // if neither side has any other message Sends on the stack).
1369     bool canTimeOut = transact.IsBottom();
1370     if (maybeTimedOut && canTimeOut && !ShouldContinueFromTimeout()) {
1371       // Since ShouldContinueFromTimeout drops the lock, we need to
1372       // re-check all our conditions here. We shouldn't time out if any of
1373       // these things happen because there won't be a reply to the timed
1374       // out message in these cases.
1375       if (transact.IsComplete()) {
1376         break;
1377       }
1378
1379       IPC_LOG("Timing out Send: xid=%d", transaction);
1380
1381       mTimedOutMessageSeqno = seqno;
1382       mTimedOutMessageNestedLevel = nestedLevel;
1383       mLastSendError = SyncSendError::TimedOut;
1384       return false;
1385     }
1386
1387     if (transact.IsCanceled()) {
1388       break;
1389     }
1390   }
1391
1392   if (transact.IsCanceled()) {
1393     IPC_LOG("Other side canceled seqno=%d, xid=%d", seqno, transaction);
1394     mLastSendError = SyncSendError::CancelledAfterSend;
1395     return false;
1396   }
1397
1398   if (transact.IsError()) {
1399     IPC_LOG("Error: seqno=%d, xid=%d", seqno, transaction);
1400     mLastSendError = SyncSendError::ReplyError;
1401     return false;
1402   }
1403
1404   uint32_t latencyMs = round((TimeStamp::Now() - start).ToMilliseconds());
1405   IPC_LOG("Got reply: seqno=%d, xid=%d, msgName=%s, latency=%ums", seqno,
1406           transaction, msgName, latencyMs);
1407
1408   UniquePtr<Message> reply = transact.GetReply();
1409
1410   MOZ_RELEASE_ASSERT(reply);
1411   MOZ_RELEASE_ASSERT(reply->is_reply(), "expected reply");
1412   MOZ_RELEASE_ASSERT(!reply->is_reply_error());
1413   MOZ_RELEASE_ASSERT(reply->seqno() == seqno);
1414   MOZ_RELEASE_ASSERT(reply->type() == replyType, "wrong reply type");
1415   MOZ_RELEASE_ASSERT(reply->is_sync());
1416
1417   AddProfilerMarker(*reply, MessageDirection::eReceiving);
1418
1419   if (reply->size() >= kMinTelemetryMessageSize) {
1420     Telemetry::Accumulate(Telemetry::IPC_REPLY_SIZE,
1421                           nsDependentCString(msgName), reply->size());
1422   }
1423
1424   *aReply = std::move(reply);
1425
1426   // NOTE: Only collect IPC_SYNC_MAIN_LATENCY_MS on the main thread (bug
1427   // 1343729)
1428   if (NS_IsMainThread() && latencyMs >= kMinTelemetrySyncIPCLatencyMs) {
1429     Telemetry::Accumulate(Telemetry::IPC_SYNC_MAIN_LATENCY_MS,
1430                           nsDependentCString(msgName), latencyMs);
1431   }
1432   return true;
1433 }
1434
1435 bool MessageChannel::HasPendingEvents() {
1436   AssertWorkerThread();
1437   mMonitor->AssertCurrentThreadOwns();
1438   return ConnectedOrClosing() && !mPending.isEmpty();
1439 }
1440
1441 bool MessageChannel::ProcessPendingRequest(ActorLifecycleProxy* aProxy,
1442                                            UniquePtr<Message> aUrgent) {
1443   AssertWorkerThread();
1444   mMonitor->AssertCurrentThreadOwns();
1445
1446   IPC_LOG("Process pending: seqno=%d, xid=%d", aUrgent->seqno(),
1447           aUrgent->transaction_id());
1448
1449   // keep the error relevant information
1450   msgid_t msgType = aUrgent->type();
1451
1452   DispatchMessage(aProxy, std::move(aUrgent));
1453   if (!ConnectedOrClosing()) {
1454     ReportConnectionError("ProcessPendingRequest", msgType);
1455     return false;
1456   }
1457
1458   return true;
1459 }
1460
1461 bool MessageChannel::ShouldRunMessage(const Message& aMsg) {
1462   if (!mTimedOutMessageSeqno) {
1463     return true;
1464   }
1465
1466   // If we've timed out a message and we're awaiting the reply to the timed
1467   // out message, we have to be careful what messages we process. Here's what
1468   // can go wrong:
1469   // 1. child sends a NOT_NESTED sync message S
1470   // 2. parent sends a NESTED_INSIDE_SYNC sync message H at the same time
1471   // 3. parent times out H
1472   // 4. child starts processing H and sends a NESTED_INSIDE_SYNC message H'
1473   //    nested within the same transaction
1474   // 5. parent dispatches S and sends reply
1475   // 6. child asserts because it instead expected a reply to H'.
1476   //
1477   // To solve this, we refuse to process S in the parent until we get a reply
1478   // to H. More generally, let the timed out message be M. We don't process a
1479   // message unless the child would need the response to that message in order
1480   // to process M. Those messages are the ones that have a higher nested level
1481   // than M or that are part of the same transaction as M.
1482   if (aMsg.nested_level() < mTimedOutMessageNestedLevel ||
1483       (aMsg.nested_level() == mTimedOutMessageNestedLevel &&
1484        aMsg.transaction_id() != mTimedOutMessageSeqno)) {
1485     return false;
1486   }
1487
1488   return true;
1489 }
1490
1491 void MessageChannel::RunMessage(ActorLifecycleProxy* aProxy,
1492                                 MessageTask& aTask) {
1493   AssertWorkerThread();
1494   mMonitor->AssertCurrentThreadOwns();
1495   aTask.AssertMonitorHeld(*mMonitor);
1496
1497   UniquePtr<Message>& msg = aTask.Msg();
1498
1499   if (!ConnectedOrClosing()) {
1500     ReportConnectionError("RunMessage", msg->type());
1501     return;
1502   }
1503
1504   // Check that we're going to run the first message that's valid to run.
1505 #if 0
1506 #  ifdef DEBUG
1507     for (MessageTask* task : mPending) {
1508         if (task == &aTask) {
1509             break;
1510         }
1511
1512         MOZ_ASSERT(!ShouldRunMessage(*task->Msg()) ||
1513                    aTask.Msg()->priority() != task->Msg()->priority());
1514
1515     }
1516 #  endif
1517 #endif
1518
1519   if (!ShouldRunMessage(*msg)) {
1520     return;
1521   }
1522
1523   MOZ_RELEASE_ASSERT(aTask.isInList());
1524   aTask.remove();
1525
1526   if (!IsAlwaysDeferred(*msg)) {
1527     mMaybeDeferredPendingCount--;
1528   }
1529
1530   DispatchMessage(aProxy, std::move(msg));
1531 }
1532
1533 NS_IMPL_ISUPPORTS_INHERITED(MessageChannel::MessageTask, CancelableRunnable,
1534                             nsIRunnablePriority, nsIRunnableIPCMessageType)
1535
1536 static uint32_t ToRunnablePriority(IPC::Message::PriorityValue aPriority) {
1537   switch (aPriority) {
1538     case IPC::Message::NORMAL_PRIORITY:
1539       return nsIRunnablePriority::PRIORITY_NORMAL;
1540     case IPC::Message::INPUT_PRIORITY:
1541       return nsIRunnablePriority::PRIORITY_INPUT_HIGH;
1542     case IPC::Message::VSYNC_PRIORITY:
1543       return nsIRunnablePriority::PRIORITY_VSYNC;
1544     case IPC::Message::MEDIUMHIGH_PRIORITY:
1545       return nsIRunnablePriority::PRIORITY_MEDIUMHIGH;
1546     case IPC::Message::CONTROL_PRIORITY:
1547       return nsIRunnablePriority::PRIORITY_CONTROL;
1548     default:
1549       MOZ_ASSERT_UNREACHABLE();
1550       return nsIRunnablePriority::PRIORITY_NORMAL;
1551   }
1552 }
1553
1554 MessageChannel::MessageTask::MessageTask(MessageChannel* aChannel,
1555                                          UniquePtr<Message> aMessage)
1556     : CancelableRunnable(aMessage->name()),
1557       mMonitor(aChannel->mMonitor),
1558       mChannel(aChannel),
1559       mMessage(std::move(aMessage)),
1560       mPriority(ToRunnablePriority(mMessage->priority())),
1561       mScheduled(false)
1562 #ifdef FUZZING_SNAPSHOT
1563       ,
1564       mIsFuzzMsg(mMessage->IsFuzzMsg()),
1565       mFuzzStopped(false)
1566 #endif
1567 {
1568   MOZ_DIAGNOSTIC_ASSERT(mMessage, "message may not be null");
1569 #ifdef FUZZING_SNAPSHOT
1570   if (mIsFuzzMsg) {
1571     MOZ_FUZZING_IPC_MT_CTOR();
1572   }
1573 #endif
1574 }
1575
1576 MessageChannel::MessageTask::~MessageTask() {
1577 #ifdef FUZZING_SNAPSHOT
1578   // We track fuzzing messages until their run is complete. To make sure
1579   // that we don't miss messages that are for some reason destroyed without
1580   // being run (e.g. canceled), we catch this condition in the destructor.
1581   if (mIsFuzzMsg && !mFuzzStopped) {
1582     MOZ_FUZZING_IPC_MT_STOP();
1583   } else if (!mIsFuzzMsg && !fuzzing::Nyx::instance().started()) {
1584     MOZ_FUZZING_IPC_PRE_FUZZ_MT_STOP();
1585   }
1586 #endif
1587 }
1588
1589 nsresult MessageChannel::MessageTask::Run() {
1590   mMonitor->AssertNotCurrentThreadOwns();
1591
1592   // Drop the toplevel actor's lifecycle proxy outside of our monitor if we take
1593   // it, as destroying our ActorLifecycleProxy reference can acquire the
1594   // monitor.
1595   RefPtr<ActorLifecycleProxy> proxy;
1596
1597   MonitorAutoLock lock(*mMonitor);
1598
1599   // In case we choose not to run this message, we may need to be able to Post
1600   // it again.
1601   mScheduled = false;
1602
1603   if (!isInList()) {
1604     return NS_OK;
1605   }
1606
1607   Channel()->AssertWorkerThread();
1608   mMonitor->AssertSameMonitor(*Channel()->mMonitor);
1609
1610 #ifdef FUZZING_SNAPSHOT
1611   if (!mIsFuzzMsg) {
1612     if (fuzzing::Nyx::instance().started() && XRE_IsParentProcess() &&
1613         Channel()->IsCrossProcess()) {
1614       // Once we started fuzzing, prevent non-fuzzing tasks from being
1615       // run and potentially blocking worker threads.
1616       //
1617       // TODO: This currently blocks all MessageTasks from running, not
1618       // just those belonging to the target process pair. We currently
1619       // do this for performance reasons, but it should be re-evaluated
1620       // at a later stage when we found a better snapshot point.
1621       return NS_OK;
1622     }
1623     // Record all running tasks prior to fuzzing, so we can wait for
1624     // them to settle before snapshotting.
1625     MOZ_FUZZING_IPC_PRE_FUZZ_MT_RUN();
1626   }
1627 #endif
1628
1629   proxy = Channel()->Listener()->GetLifecycleProxy();
1630   Channel()->RunMessage(proxy, *this);
1631
1632 #ifdef FUZZING_SNAPSHOT
1633   if (mIsFuzzMsg && !mFuzzStopped) {
1634     MOZ_FUZZING_IPC_MT_STOP();
1635     mFuzzStopped = true;
1636   }
1637 #endif
1638   return NS_OK;
1639 }
1640
1641 // Warning: This method removes the receiver from whatever list it might be in.
1642 nsresult MessageChannel::MessageTask::Cancel() {
1643   mMonitor->AssertNotCurrentThreadOwns();
1644
1645   MonitorAutoLock lock(*mMonitor);
1646
1647   if (!isInList()) {
1648     return NS_OK;
1649   }
1650
1651   Channel()->AssertWorkerThread();
1652   mMonitor->AssertSameMonitor(*Channel()->mMonitor);
1653   if (!IsAlwaysDeferred(*Msg())) {
1654     Channel()->mMaybeDeferredPendingCount--;
1655   }
1656
1657   remove();
1658
1659 #ifdef FUZZING_SNAPSHOT
1660   if (mIsFuzzMsg && !mFuzzStopped) {
1661     MOZ_FUZZING_IPC_MT_STOP();
1662     mFuzzStopped = true;
1663   }
1664 #endif
1665
1666   return NS_OK;
1667 }
1668
1669 void MessageChannel::MessageTask::Post() {
1670   mMonitor->AssertCurrentThreadOwns();
1671   mMonitor->AssertSameMonitor(*Channel()->mMonitor);
1672   MOZ_RELEASE_ASSERT(!mScheduled);
1673   MOZ_RELEASE_ASSERT(isInList());
1674
1675   mScheduled = true;
1676
1677   Channel()->mWorkerThread->Dispatch(do_AddRef(this));
1678 }
1679
1680 NS_IMETHODIMP
1681 MessageChannel::MessageTask::GetPriority(uint32_t* aPriority) {
1682   *aPriority = mPriority;
1683   return NS_OK;
1684 }
1685
1686 NS_IMETHODIMP
1687 MessageChannel::MessageTask::GetType(uint32_t* aType) {
1688   mMonitor->AssertNotCurrentThreadOwns();
1689
1690   MonitorAutoLock lock(*mMonitor);
1691   if (!mMessage) {
1692     // If mMessage has been moved already elsewhere, we can't know what the type
1693     // has been.
1694     return NS_ERROR_FAILURE;
1695   }
1696
1697   *aType = mMessage->type();
1698   return NS_OK;
1699 }
1700
1701 void MessageChannel::DispatchMessage(ActorLifecycleProxy* aProxy,
1702                                      UniquePtr<Message> aMsg) {
1703   AssertWorkerThread();
1704   mMonitor->AssertCurrentThreadOwns();
1705
1706   Maybe<AutoNoJSAPI> nojsapi;
1707   if (NS_IsMainThread() && CycleCollectedJSContext::Get()) {
1708     nojsapi.emplace();
1709   }
1710
1711   UniquePtr<Message> reply;
1712
1713 #ifdef FUZZING_SNAPSHOT
1714   if (IsCrossProcess()) {
1715     aMsg = mozilla::fuzzing::IPCFuzzController::instance().replaceIPCMessage(
1716         std::move(aMsg));
1717   }
1718 #endif
1719
1720   IPC_LOG("DispatchMessage: seqno=%d, xid=%d", aMsg->seqno(),
1721           aMsg->transaction_id());
1722   AddProfilerMarker(*aMsg, MessageDirection::eReceiving);
1723
1724   {
1725     AutoEnterTransaction transaction(this, *aMsg);
1726
1727     int id = aMsg->transaction_id();
1728     MOZ_RELEASE_ASSERT(!aMsg->is_sync() || id == transaction.TransactionID());
1729
1730     {
1731       MonitorAutoUnlock unlock(*mMonitor);
1732       AutoSetValue<bool> setOnCxxStack(mOnCxxStack, true);
1733
1734       mListener->ArtificialSleep();
1735
1736       if (aMsg->is_sync()) {
1737         DispatchSyncMessage(aProxy, *aMsg, reply);
1738       } else {
1739         DispatchAsyncMessage(aProxy, *aMsg);
1740       }
1741
1742       mListener->ArtificialSleep();
1743     }
1744
1745     if (reply && transaction.IsCanceled()) {
1746       // The transaction has been canceled. Don't send a reply.
1747       IPC_LOG("Nulling out reply due to cancellation, seqno=%d, xid=%d",
1748               aMsg->seqno(), id);
1749       reply = nullptr;
1750     }
1751   }
1752
1753 #ifdef FUZZING_SNAPSHOT
1754   if (aMsg->IsFuzzMsg()) {
1755     mozilla::fuzzing::IPCFuzzController::instance().syncAfterReplace();
1756   }
1757 #endif
1758
1759   if (reply && ChannelConnected == mChannelState) {
1760     IPC_LOG("Sending reply seqno=%d, xid=%d", aMsg->seqno(),
1761             aMsg->transaction_id());
1762     AddProfilerMarker(*reply, MessageDirection::eSending);
1763
1764     SendMessageToLink(std::move(reply));
1765   }
1766 }
1767
1768 void MessageChannel::DispatchSyncMessage(ActorLifecycleProxy* aProxy,
1769                                          const Message& aMsg,
1770                                          UniquePtr<Message>& aReply) {
1771   AssertWorkerThread();
1772
1773   mozilla::TimeStamp start = TimeStamp::Now();
1774
1775   int nestedLevel = aMsg.nested_level();
1776
1777   MOZ_RELEASE_ASSERT(nestedLevel == IPC::Message::NOT_NESTED ||
1778                      NS_IsMainThread());
1779
1780   MessageChannel* dummy;
1781   MessageChannel*& blockingVar =
1782       mSide == ChildSide && NS_IsMainThread() ? gParentProcessBlocker : dummy;
1783
1784   Result rv;
1785   {
1786     AutoSetValue<MessageChannel*> blocked(blockingVar, this);
1787     rv = aProxy->Get()->OnMessageReceived(aMsg, aReply);
1788   }
1789
1790   uint32_t latencyMs = round((TimeStamp::Now() - start).ToMilliseconds());
1791   if (latencyMs >= kMinTelemetrySyncIPCLatencyMs) {
1792     Telemetry::Accumulate(Telemetry::IPC_SYNC_RECEIVE_MS,
1793                           nsDependentCString(aMsg.name()), latencyMs);
1794   }
1795
1796   if (!MaybeHandleError(rv, aMsg, "DispatchSyncMessage")) {
1797     aReply = Message::ForSyncDispatchError(aMsg.nested_level());
1798   }
1799   aReply->set_seqno(aMsg.seqno());
1800   aReply->set_transaction_id(aMsg.transaction_id());
1801 }
1802
1803 void MessageChannel::DispatchAsyncMessage(ActorLifecycleProxy* aProxy,
1804                                           const Message& aMsg) {
1805   AssertWorkerThread();
1806   MOZ_RELEASE_ASSERT(!aMsg.is_sync());
1807
1808   if (aMsg.routing_id() == MSG_ROUTING_NONE) {
1809     NS_WARNING("unhandled special message!");
1810     MaybeHandleError(MsgNotKnown, aMsg, "DispatchAsyncMessage");
1811     return;
1812   }
1813
1814   Result rv;
1815   {
1816     int nestedLevel = aMsg.nested_level();
1817     AutoSetValue<bool> async(mDispatchingAsyncMessage, true);
1818     AutoSetValue<int> nestedLevelSet(mDispatchingAsyncMessageNestedLevel,
1819                                      nestedLevel);
1820     rv = aProxy->Get()->OnMessageReceived(aMsg);
1821   }
1822   MaybeHandleError(rv, aMsg, "DispatchAsyncMessage");
1823 }
1824
1825 void MessageChannel::EnqueuePendingMessages() {
1826   AssertWorkerThread();
1827   mMonitor->AssertCurrentThreadOwns();
1828
1829   // XXX performance tuning knob: could process all or k pending
1830   // messages here, rather than enqueuing for later processing
1831
1832   RepostAllMessages();
1833 }
1834
1835 bool MessageChannel::WaitResponse(bool aWaitTimedOut) {
1836   AssertWorkerThread();
1837   if (aWaitTimedOut) {
1838     if (mInTimeoutSecondHalf) {
1839       // We've really timed out this time.
1840       return false;
1841     }
1842     // Try a second time.
1843     mInTimeoutSecondHalf = true;
1844   } else {
1845     mInTimeoutSecondHalf = false;
1846   }
1847   return true;
1848 }
1849
1850 #ifndef XP_WIN
1851 bool MessageChannel::WaitForSyncNotify() {
1852   AssertWorkerThread();
1853 #  ifdef DEBUG
1854   // WARNING: We don't release the lock here. We can't because the link
1855   // could signal at this time and we would miss it. Instead we require
1856   // ArtificialTimeout() to be extremely simple.
1857   if (mListener->ArtificialTimeout()) {
1858     return false;
1859   }
1860 #  endif
1861
1862   MOZ_RELEASE_ASSERT(!mIsSameThreadChannel,
1863                      "Wait on same-thread channel will deadlock!");
1864
1865   TimeDuration timeout = (kNoTimeout == mTimeoutMs)
1866                              ? TimeDuration::Forever()
1867                              : TimeDuration::FromMilliseconds(mTimeoutMs);
1868   CVStatus status = mMonitor->Wait(timeout);
1869
1870   // If the timeout didn't expire, we know we received an event. The
1871   // converse is not true.
1872   return WaitResponse(status == CVStatus::Timeout);
1873 }
1874
1875 void MessageChannel::NotifyWorkerThread() { mMonitor->Notify(); }
1876 #endif
1877
1878 bool MessageChannel::ShouldContinueFromTimeout() {
1879   AssertWorkerThread();
1880   mMonitor->AssertCurrentThreadOwns();
1881
1882   bool cont;
1883   {
1884     MonitorAutoUnlock unlock(*mMonitor);
1885     cont = mListener->ShouldContinueFromReplyTimeout();
1886     mListener->ArtificialSleep();
1887   }
1888
1889   static enum {
1890     UNKNOWN,
1891     NOT_DEBUGGING,
1892     DEBUGGING
1893   } sDebuggingChildren = UNKNOWN;
1894
1895   if (sDebuggingChildren == UNKNOWN) {
1896     sDebuggingChildren =
1897         getenv("MOZ_DEBUG_CHILD_PROCESS") || getenv("MOZ_DEBUG_CHILD_PAUSE")
1898             ? DEBUGGING
1899             : NOT_DEBUGGING;
1900   }
1901   if (sDebuggingChildren == DEBUGGING) {
1902     return true;
1903   }
1904
1905   return cont;
1906 }
1907
1908 void MessageChannel::SetReplyTimeoutMs(int32_t aTimeoutMs) {
1909   // Set channel timeout value. Since this is broken up into
1910   // two period, the minimum timeout value is 2ms.
1911   AssertWorkerThread();
1912   mTimeoutMs =
1913       (aTimeoutMs <= 0) ? kNoTimeout : (int32_t)ceil((double)aTimeoutMs / 2.0);
1914 }
1915
1916 void MessageChannel::ReportConnectionError(const char* aFunctionName,
1917                                            const uint32_t aMsgType) const {
1918   AssertWorkerThread();
1919   mMonitor->AssertCurrentThreadOwns();
1920
1921   const char* errorMsg = nullptr;
1922   switch (mChannelState) {
1923     case ChannelClosed:
1924       errorMsg = "Closed channel: cannot send/recv";
1925       break;
1926     case ChannelClosing:
1927       errorMsg = "Channel closing: too late to send, messages will be lost";
1928       break;
1929     case ChannelError:
1930       errorMsg = "Channel error: cannot send/recv";
1931       break;
1932
1933     default:
1934       MOZ_CRASH("unreached");
1935   }
1936
1937   // IPC connection errors are fairly common, especially "Channel closing: too
1938   // late to send/recv, messages will be lost", so shouldn't be being reported
1939   // on release builds, as that's misleading as to their severity.
1940   NS_WARNING(nsPrintfCString("IPC Connection Error: [%s][%s] %s(msgname=%s) %s",
1941                              StringFromIPCSide(mSide), mName, aFunctionName,
1942                              IPC::StringFromIPCMessageType(aMsgType), errorMsg)
1943                  .get());
1944
1945   MonitorAutoUnlock unlock(*mMonitor);
1946   mListener->ProcessingError(MsgDropped, errorMsg);
1947 }
1948
1949 void MessageChannel::ReportMessageRouteError(const char* channelName) const {
1950   PrintErrorMessage(mSide, channelName, "Need a route");
1951   mListener->ProcessingError(MsgRouteError, "MsgRouteError");
1952 }
1953
1954 bool MessageChannel::MaybeHandleError(Result code, const Message& aMsg,
1955                                       const char* channelName) {
1956   if (MsgProcessed == code) return true;
1957
1958 #ifdef FUZZING_SNAPSHOT
1959   mozilla::fuzzing::IPCFuzzController::instance().OnMessageError(code, aMsg);
1960 #endif
1961
1962   const char* errorMsg = nullptr;
1963   switch (code) {
1964     case MsgNotKnown:
1965       errorMsg = "Unknown message: not processed";
1966       break;
1967     case MsgNotAllowed:
1968       errorMsg = "Message not allowed: cannot be sent/recvd in this state";
1969       break;
1970     case MsgPayloadError:
1971       errorMsg = "Payload error: message could not be deserialized";
1972       break;
1973     case MsgProcessingError:
1974       errorMsg =
1975           "Processing error: message was deserialized, but the handler "
1976           "returned false (indicating failure)";
1977       break;
1978     case MsgRouteError:
1979       errorMsg = "Route error: message sent to unknown actor ID";
1980       break;
1981     case MsgValueError:
1982       errorMsg =
1983           "Value error: message was deserialized, but contained an illegal "
1984           "value";
1985       break;
1986
1987     default:
1988       MOZ_CRASH("unknown Result code");
1989       return false;
1990   }
1991
1992   char reason[512];
1993   const char* msgname = aMsg.name();
1994   if (msgname[0] == '?') {
1995     SprintfLiteral(reason, "(msgtype=0x%X) %s", aMsg.type(), errorMsg);
1996   } else {
1997     SprintfLiteral(reason, "%s %s", msgname, errorMsg);
1998   }
1999
2000   PrintErrorMessage(mSide, channelName, reason);
2001
2002   // Error handled in mozilla::ipc::IPCResult.
2003   if (code == MsgProcessingError) {
2004     return false;
2005   }
2006
2007   mListener->ProcessingError(code, reason);
2008
2009   return false;
2010 }
2011
2012 void MessageChannel::OnChannelErrorFromLink() {
2013   mMonitor->AssertCurrentThreadOwns();
2014   MOZ_ASSERT(mChannelState == ChannelConnected);
2015
2016   IPC_LOG("OnChannelErrorFromLink");
2017
2018   if (AwaitingSyncReply()) {
2019     NotifyWorkerThread();
2020   }
2021
2022   if (mAbortOnError) {
2023     // mAbortOnError is set by main actors (e.g., ContentChild) to ensure
2024     // that the process terminates even if normal shutdown is prevented.
2025     // A MOZ_CRASH() here is not helpful because crash reporting relies
2026     // on the parent process which we know is dead or otherwise unusable.
2027     //
2028     // Additionally, the parent process can (and often is) killed on Android
2029     // when apps are backgrounded. We don't need to report a crash for
2030     // normal behavior in that case.
2031     printf_stderr("Exiting due to channel error.\n");
2032     ProcessChild::QuickExit();
2033   }
2034   mChannelState = ChannelError;
2035   mMonitor->Notify();
2036
2037   PostErrorNotifyTask();
2038 }
2039
2040 void MessageChannel::NotifyMaybeChannelError(ReleasableMonitorAutoLock& aLock) {
2041   AssertWorkerThread();
2042   mMonitor->AssertCurrentThreadOwns();
2043   aLock.AssertCurrentThreadOwns();
2044   MOZ_ASSERT(mChannelState != ChannelConnected);
2045
2046   if (ChannelClosing == mChannelState || ChannelClosed == mChannelState) {
2047     // the channel closed, but we received a "Goodbye" message warning us
2048     // about it. no worries
2049     mChannelState = ChannelClosed;
2050     NotifyChannelClosed(aLock);
2051     return;
2052   }
2053
2054   MOZ_ASSERT(ChannelError == mChannelState);
2055
2056   Clear();
2057
2058   // IPDL assumes these notifications do not fire twice, so we do not let
2059   // that happen.
2060   if (mNotifiedChannelDone) {
2061     return;
2062   }
2063   mNotifiedChannelDone = true;
2064
2065   // Let our listener know that the channel errored. This may cause the
2066   // channel to be deleted. Release our caller's `MonitorAutoLock` before
2067   // invoking the listener, as this may call back into MessageChannel, and/or
2068   // cause the channel to be destroyed.
2069   aLock.Unlock();
2070   mListener->OnChannelError();
2071 }
2072
2073 void MessageChannel::OnNotifyMaybeChannelError() {
2074   AssertWorkerThread();
2075   mMonitor->AssertNotCurrentThreadOwns();
2076
2077   // This lock guard may be reset by `NotifyMaybeChannelError` before invoking
2078   // listener callbacks which may destroy this `MessageChannel`.
2079   //
2080   // Acquiring the lock here also allows us to ensure that
2081   // `OnChannelErrorFromLink` has finished running before this task is allowed
2082   // to continue.
2083   ReleasableMonitorAutoLock lock(*mMonitor);
2084
2085   mChannelErrorTask = nullptr;
2086
2087   if (IsOnCxxStack()) {
2088     // This used to post a 10ms delayed task; however not all
2089     // nsISerialEventTarget implementations support delayed dispatch.
2090     // The delay being completely arbitrary, we may not as well have any.
2091     PostErrorNotifyTask();
2092     return;
2093   }
2094
2095   // This call may destroy `this`.
2096   NotifyMaybeChannelError(lock);
2097 }
2098
2099 void MessageChannel::PostErrorNotifyTask() {
2100   mMonitor->AssertCurrentThreadOwns();
2101
2102   if (mChannelErrorTask) {
2103     return;
2104   }
2105
2106   // This must be the last code that runs on this thread!
2107   mChannelErrorTask = NewNonOwningCancelableRunnableMethod(
2108       "ipc::MessageChannel::OnNotifyMaybeChannelError", this,
2109       &MessageChannel::OnNotifyMaybeChannelError);
2110   mWorkerThread->Dispatch(do_AddRef(mChannelErrorTask));
2111 }
2112
2113 // Special async message.
2114 class GoodbyeMessage : public IPC::Message {
2115  public:
2116   GoodbyeMessage() : IPC::Message(MSG_ROUTING_NONE, GOODBYE_MESSAGE_TYPE) {}
2117   static bool Read(const Message* msg) { return true; }
2118   void Log(const std::string& aPrefix, FILE* aOutf) const {
2119     fputs("(special `Goodbye' message)", aOutf);
2120   }
2121 };
2122
2123 void MessageChannel::InduceConnectionError() {
2124   MonitorAutoLock lock(*mMonitor);
2125
2126   // Either connected or closing, immediately convert to an error and notify.
2127   switch (mChannelState) {
2128     case ChannelConnected:
2129       // The channel is still actively connected. Immediately shut down the
2130       // connection with our peer and simulate it invoking
2131       // OnChannelErrorFromLink on us.
2132       //
2133       // This will update the state to ChannelError, preventing new messages
2134       // from being processed, leading to an error being reported asynchronously
2135       // to our listener.
2136       mLink->Close();
2137       OnChannelErrorFromLink();
2138       return;
2139
2140     case ChannelClosing:
2141       // An notify task has already been posted. Update mChannelState to stop
2142       // processing new messages and treat the notification as an error.
2143       mChannelState = ChannelError;
2144       return;
2145
2146     default:
2147       // Either already closed or errored. Nothing to do.
2148       MOZ_ASSERT(mChannelState == ChannelClosed ||
2149                  mChannelState == ChannelError);
2150       return;
2151   }
2152 }
2153
2154 void MessageChannel::NotifyImpendingShutdown() {
2155   UniquePtr<Message> msg =
2156       MakeUnique<Message>(MSG_ROUTING_NONE, IMPENDING_SHUTDOWN_MESSAGE_TYPE);
2157   MonitorAutoLock lock(*mMonitor);
2158   if (Connected()) {
2159     SendMessageToLink(std::move(msg));
2160   }
2161 }
2162
2163 void MessageChannel::Close() {
2164   AssertWorkerThread();
2165   mMonitor->AssertNotCurrentThreadOwns();
2166
2167   // This lock guard may be reset by `Notify{ChannelClosed,MaybeChannelError}`
2168   // before invoking listener callbacks which may destroy this `MessageChannel`.
2169   ReleasableMonitorAutoLock lock(*mMonitor);
2170
2171   switch (mChannelState) {
2172     case ChannelError:
2173       // See bug 538586: if the listener gets deleted while the
2174       // IO thread's NotifyChannelError event is still enqueued
2175       // and subsequently deletes us, then the error event will
2176       // also be deleted and the listener will never be notified
2177       // of the channel error.
2178       NotifyMaybeChannelError(lock);
2179       return;
2180     case ChannelClosed:
2181       // Slightly unexpected but harmless; ignore.  See bug 1554244.
2182       return;
2183
2184     default:
2185       // Notify the other side that we're about to close our socket. If we've
2186       // already received a Goodbye from the other side (and our state is
2187       // ChannelClosing), there's no reason to send one.
2188       if (ChannelConnected == mChannelState) {
2189         SendMessageToLink(MakeUnique<GoodbyeMessage>());
2190       }
2191       mLink->Close();
2192       mChannelState = ChannelClosed;
2193       NotifyChannelClosed(lock);
2194       return;
2195   }
2196 }
2197
2198 void MessageChannel::NotifyChannelClosed(ReleasableMonitorAutoLock& aLock) {
2199   AssertWorkerThread();
2200   mMonitor->AssertCurrentThreadOwns();
2201   aLock.AssertCurrentThreadOwns();
2202
2203   if (ChannelClosed != mChannelState) {
2204     MOZ_CRASH("channel should have been closed!");
2205   }
2206
2207   Clear();
2208
2209   // IPDL assumes these notifications do not fire twice, so we do not let
2210   // that happen.
2211   if (mNotifiedChannelDone) {
2212     return;
2213   }
2214   mNotifiedChannelDone = true;
2215
2216   // Let our listener know that the channel was closed. This may cause the
2217   // channel to be deleted. Release our caller's `MonitorAutoLock` before
2218   // invoking the listener, as this may call back into MessageChannel, and/or
2219   // cause the channel to be destroyed.
2220   aLock.Unlock();
2221   mListener->OnChannelClose();
2222 }
2223
2224 void MessageChannel::DebugAbort(const char* file, int line, const char* cond,
2225                                 const char* why, bool reply) {
2226   AssertWorkerThread();
2227   mMonitor->AssertCurrentThreadOwns();
2228
2229   printf_stderr(
2230       "###!!! [MessageChannel][%s][%s:%d] "
2231       "Assertion (%s) failed.  %s %s\n",
2232       StringFromIPCSide(mSide), file, line, cond, why, reply ? "(reply)" : "");
2233
2234   MessageQueue pending = std::move(mPending);
2235   while (!pending.isEmpty()) {
2236     pending.getFirst()->AssertMonitorHeld(*mMonitor);
2237     printf_stderr("    [ %s%s ]\n",
2238                   pending.getFirst()->Msg()->is_sync() ? "sync" : "async",
2239                   pending.getFirst()->Msg()->is_reply() ? "reply" : "");
2240     pending.popFirst();
2241   }
2242
2243   MOZ_CRASH_UNSAFE(why);
2244 }
2245
2246 void MessageChannel::AddProfilerMarker(const IPC::Message& aMessage,
2247                                        MessageDirection aDirection) {
2248   mMonitor->AssertCurrentThreadOwns();
2249
2250   if (profiler_feature_active(ProfilerFeature::IPCMessages)) {
2251     base::ProcessId pid = mListener->OtherPidMaybeInvalid();
2252     // Only record markers for IPCs with a valid pid.
2253     // And if one of the profiler mutexes is locked on this thread, don't record
2254     // markers, because we don't want to expose profiler IPCs due to the
2255     // profiler itself, and also to avoid possible re-entrancy issues.
2256     if (pid != base::kInvalidProcessId &&
2257         !profiler_is_locked_on_current_thread()) {
2258       // The current timestamp must be given to the `IPCMarker` payload.
2259       [[maybe_unused]] const TimeStamp now = TimeStamp::Now();
2260       bool isThreadBeingProfiled =
2261           profiler_thread_is_being_profiled_for_markers();
2262       PROFILER_MARKER(
2263           "IPC", IPC,
2264           mozilla::MarkerOptions(
2265               mozilla::MarkerTiming::InstantAt(now),
2266               // If the thread is being profiled, add the marker to
2267               // the current thread. If the thread is not being
2268               // profiled, add the marker to the main thread. It
2269               // will appear in the main thread's IPC track. Profiler analysis
2270               // UI correlates all the IPC markers from different threads and
2271               // generates processed markers.
2272               isThreadBeingProfiled ? mozilla::MarkerThreadId::CurrentThread()
2273                                     : mozilla::MarkerThreadId::MainThread()),
2274           IPCMarker, now, now, pid, aMessage.seqno(), aMessage.type(), mSide,
2275           aDirection, MessagePhase::Endpoint, aMessage.is_sync(),
2276           // aOriginThreadId: If the thread is being profiled, do not include a
2277           // thread ID, as it's the same as the markers. Only include this field
2278           // when the marker is being sent from another thread.
2279           isThreadBeingProfiled ? mozilla::MarkerThreadId{}
2280                                 : mozilla::MarkerThreadId::CurrentThread());
2281     }
2282   }
2283 }
2284
2285 void MessageChannel::EndTimeout() {
2286   mMonitor->AssertCurrentThreadOwns();
2287
2288   IPC_LOG("Ending timeout of seqno=%d", mTimedOutMessageSeqno);
2289   mTimedOutMessageSeqno = 0;
2290   mTimedOutMessageNestedLevel = 0;
2291
2292   RepostAllMessages();
2293 }
2294
2295 void MessageChannel::RepostAllMessages() {
2296   mMonitor->AssertCurrentThreadOwns();
2297
2298   bool needRepost = false;
2299   for (MessageTask* task : mPending) {
2300     task->AssertMonitorHeld(*mMonitor);
2301     if (!task->IsScheduled()) {
2302       needRepost = true;
2303       break;
2304     }
2305   }
2306   if (!needRepost) {
2307     // If everything is already scheduled to run, do nothing.
2308     return;
2309   }
2310
2311   // In some cases we may have deferred dispatch of some messages in the
2312   // queue. Now we want to run them again. However, we can't just re-post
2313   // those messages since the messages after them in mPending would then be
2314   // before them in the event queue. So instead we cancel everything and
2315   // re-post all messages in the correct order.
2316   MessageQueue queue = std::move(mPending);
2317   while (RefPtr<MessageTask> task = queue.popFirst()) {
2318     task->AssertMonitorHeld(*mMonitor);
2319     RefPtr<MessageTask> newTask = new MessageTask(this, std::move(task->Msg()));
2320     newTask->AssertMonitorHeld(*mMonitor);
2321     mPending.insertBack(newTask);
2322     newTask->Post();
2323   }
2324
2325   AssertMaybeDeferredCountCorrect();
2326 }
2327
2328 void MessageChannel::CancelTransaction(int transaction) {
2329   mMonitor->AssertCurrentThreadOwns();
2330
2331   // When we cancel a transaction, we need to behave as if there's no longer
2332   // any IPC on the stack. Anything we were dispatching or sending will get
2333   // canceled. Consequently, we have to update the state variables below.
2334   //
2335   // We also need to ensure that when any IPC functions on the stack return,
2336   // they don't reset these values using an RAII class like AutoSetValue. To
2337   // avoid that, these RAII classes check if the variable they set has been
2338   // tampered with (by us). If so, they don't reset the variable to the old
2339   // value.
2340
2341   IPC_LOG("CancelTransaction: xid=%d", transaction);
2342
2343   // An unusual case: We timed out a transaction which the other side then
2344   // cancelled. In this case we just leave the timedout state and try to
2345   // forget this ever happened.
2346   if (transaction == mTimedOutMessageSeqno) {
2347     IPC_LOG("Cancelled timed out message %d", mTimedOutMessageSeqno);
2348     EndTimeout();
2349
2350     // Normally mCurrentTransaction == 0 here. But it can be non-zero if:
2351     // 1. Parent sends NESTED_INSIDE_SYNC message H.
2352     // 2. Parent times out H.
2353     // 3. Child dispatches H and sends nested message H' (same transaction).
2354     // 4. Parent dispatches H' and cancels.
2355     MOZ_RELEASE_ASSERT(!mTransactionStack ||
2356                        mTransactionStack->TransactionID() == transaction);
2357     if (mTransactionStack) {
2358       mTransactionStack->Cancel();
2359     }
2360   } else {
2361     MOZ_RELEASE_ASSERT(mTransactionStack->TransactionID() == transaction);
2362     mTransactionStack->Cancel();
2363   }
2364
2365   bool foundSync = false;
2366   for (MessageTask* p = mPending.getFirst(); p;) {
2367     p->AssertMonitorHeld(*mMonitor);
2368     UniquePtr<Message>& msg = p->Msg();
2369
2370     // If there was a race between the parent and the child, then we may
2371     // have a queued sync message. We want to drop this message from the
2372     // queue since if will get cancelled along with the transaction being
2373     // cancelled. This happens if the message in the queue is
2374     // NESTED_INSIDE_SYNC.
2375     if (msg->is_sync() && msg->nested_level() != IPC::Message::NOT_NESTED) {
2376       MOZ_RELEASE_ASSERT(!foundSync);
2377       MOZ_RELEASE_ASSERT(msg->transaction_id() != transaction);
2378       IPC_LOG("Removing msg from queue seqno=%d xid=%d", msg->seqno(),
2379               msg->transaction_id());
2380       foundSync = true;
2381       if (!IsAlwaysDeferred(*msg)) {
2382         mMaybeDeferredPendingCount--;
2383       }
2384       p = p->removeAndGetNext();
2385       continue;
2386     }
2387
2388     p = p->getNext();
2389   }
2390
2391   AssertMaybeDeferredCountCorrect();
2392 }
2393
2394 void MessageChannel::CancelCurrentTransaction() {
2395   MonitorAutoLock lock(*mMonitor);
2396   if (DispatchingSyncMessageNestedLevel() >= IPC::Message::NESTED_INSIDE_SYNC) {
2397     if (DispatchingSyncMessageNestedLevel() ==
2398             IPC::Message::NESTED_INSIDE_CPOW ||
2399         DispatchingAsyncMessageNestedLevel() ==
2400             IPC::Message::NESTED_INSIDE_CPOW) {
2401       mListener->IntentionalCrash();
2402     }
2403
2404     IPC_LOG("Cancel requested: current xid=%d",
2405             CurrentNestedInsideSyncTransaction());
2406     MOZ_RELEASE_ASSERT(DispatchingSyncMessage());
2407     auto cancel =
2408         MakeUnique<CancelMessage>(CurrentNestedInsideSyncTransaction());
2409     CancelTransaction(CurrentNestedInsideSyncTransaction());
2410     SendMessageToLink(std::move(cancel));
2411   }
2412 }
2413
2414 void CancelCPOWs() {
2415   MOZ_ASSERT(NS_IsMainThread());
2416
2417   if (gParentProcessBlocker) {
2418     mozilla::Telemetry::Accumulate(mozilla::Telemetry::IPC_TRANSACTION_CANCEL,
2419                                    true);
2420     gParentProcessBlocker->CancelCurrentTransaction();
2421   }
2422 }
2423
2424 bool MessageChannel::IsCrossProcess() const {
2425   mMonitor->AssertCurrentThreadOwns();
2426   return mIsCrossProcess;
2427 }
2428
2429 void MessageChannel::SetIsCrossProcess(bool aIsCrossProcess) {
2430   mMonitor->AssertCurrentThreadOwns();
2431   if (aIsCrossProcess == mIsCrossProcess) {
2432     return;
2433   }
2434   mIsCrossProcess = aIsCrossProcess;
2435   if (mIsCrossProcess) {
2436     ChannelCountReporter::Increment(mName);
2437   } else {
2438     ChannelCountReporter::Decrement(mName);
2439   }
2440 }
2441
2442 NS_IMPL_ISUPPORTS(MessageChannel::WorkerTargetShutdownTask,
2443                   nsITargetShutdownTask)
2444
2445 MessageChannel::WorkerTargetShutdownTask::WorkerTargetShutdownTask(
2446     nsISerialEventTarget* aTarget, MessageChannel* aChannel)
2447     : mTarget(aTarget), mChannel(aChannel) {}
2448
2449 void MessageChannel::WorkerTargetShutdownTask::TargetShutdown() {
2450   MOZ_RELEASE_ASSERT(mTarget->IsOnCurrentThread());
2451   IPC_LOG("Closing channel due to event target shutdown");
2452   if (MessageChannel* channel = std::exchange(mChannel, nullptr)) {
2453     channel->Close();
2454   }
2455 }
2456
2457 void MessageChannel::WorkerTargetShutdownTask::Clear() {
2458   MOZ_RELEASE_ASSERT(mTarget->IsOnCurrentThread());
2459   mChannel = nullptr;
2460 }
2461
2462 NS_IMPL_ISUPPORTS_INHERITED0(MessageChannel::FlushLazySendMessagesRunnable,
2463                              CancelableRunnable)
2464
2465 MessageChannel::FlushLazySendMessagesRunnable::FlushLazySendMessagesRunnable(
2466     MessageChannel* aChannel)
2467     : CancelableRunnable("MessageChannel::FlushLazyMessagesRunnable"),
2468       mChannel(aChannel) {}
2469
2470 NS_IMETHODIMP MessageChannel::FlushLazySendMessagesRunnable::Run() {
2471   if (mChannel) {
2472     MonitorAutoLock lock(*mChannel->mMonitor);
2473     MOZ_ASSERT(mChannel->mFlushLazySendTask == this);
2474     mChannel->FlushLazySendMessages();
2475   }
2476   return NS_OK;
2477 }
2478
2479 nsresult MessageChannel::FlushLazySendMessagesRunnable::Cancel() {
2480   mQueue.Clear();
2481   mChannel = nullptr;
2482   return NS_OK;
2483 }
2484
2485 void MessageChannel::FlushLazySendMessagesRunnable::PushMessage(
2486     UniquePtr<Message> aMsg) {
2487   MOZ_ASSERT(mChannel);
2488   mQueue.AppendElement(std::move(aMsg));
2489 }
2490
2491 nsTArray<UniquePtr<IPC::Message>>
2492 MessageChannel::FlushLazySendMessagesRunnable::TakeMessages() {
2493   mChannel = nullptr;
2494   return std::move(mQueue);
2495 }
2496
2497 }  // namespace ipc
2498 }  // namespace mozilla