Bug 545342: Cancel dequeue task on channel error. r=cjones

ipc/glue/AsyncChannel.h

@@ -166,7 +166,7 @@ protected:
     void NotifyChannelClosed();
     void NotifyMaybeChannelError();
 
-    void Clear();
+    virtual void Clear();
 
     // Run on the IO thread
 

ipc/glue/RPCChannel.cpp

@@ -102,6 +102,10 @@ RPCChannel::RPCChannel(RPCListener* aListener,
     mCxxStackFrames(0)
 {
     MOZ_COUNT_CTOR(RPCChannel);
+
+    mDequeueOneTask = new RefCountedTask(NewRunnableMethod(
+                                                 this,
+                                                 &RPCChannel::OnMaybeDequeueOne));
 }
 
 RPCChannel::~RPCChannel()
@@ -110,6 +114,14 @@ RPCChannel::~RPCChannel()
     RPC_ASSERT(0 == mCxxStackFrames, "mismatched CxxStackFrame ctor/dtors");
 }
 
+void
+RPCChannel::Clear()
+{
+    mDequeueOneTask->Cancel();
+
+    AsyncChannel::Clear();
+}
+
 #ifdef OS_WIN
 // static
 int RPCChannel::sInnerEventLoopDepth = 0;
@@ -324,7 +336,7 @@ RPCChannel::EnqueuePendingMessages()
     for (size_t i = 0; i < mDeferred.size(); ++i)
         mWorkerLoop->PostTask(
             FROM_HERE,
-            NewRunnableMethod(this, &RPCChannel::OnMaybeDequeueOne));
+            new DequeueTask(mDequeueOneTask));
 
     // XXX performance tuning knob: could process all or k pending
     // messages here, rather than enqueuing for later processing
@@ -332,7 +344,7 @@ RPCChannel::EnqueuePendingMessages()
     for (size_t i = 0; i < mPending.size(); ++i)
         mWorkerLoop->PostTask(
             FROM_HERE,
-            NewRunnableMethod(this, &RPCChannel::OnMaybeDequeueOne));
+            new DequeueTask(mDequeueOneTask));
 }
 
 void
@@ -623,11 +635,10 @@ RPCChannel::OnMessageReceived(const Message& msg)
 
     mPending.push(msg);
 
-    if (0 == StackDepth() && !mBlockedOnParent)
+    if (0 == StackDepth() && !mBlockedOnParent) {
         // the worker thread might be idle, make sure it wakes up
-        mWorkerLoop->PostTask(
-            FROM_HERE,
-            NewRunnableMethod(this, &RPCChannel::OnMaybeDequeueOne));
+        mWorkerLoop->PostTask(FROM_HERE, new DequeueTask(mDequeueOneTask));
+    }
     else if (!AwaitingSyncReply())
         NotifyWorkerThread();
 }
@@ -654,7 +665,6 @@ RPCChannel::OnChannelError()
     AsyncChannel::OnChannelError();
 }
 
-
 } // namespace ipc
 } // namespace mozilla
 

ipc/glue/RPCChannel.h

@@ -44,6 +44,7 @@
 #include <stack>
 
 #include "mozilla/ipc/SyncChannel.h"
+#include "nsAutoPtr.h"
 
 namespace mozilla {
 namespace ipc {
@@ -91,6 +92,9 @@ public:
 
     virtual ~RPCChannel();
 
+    NS_OVERRIDE
+    void Clear();
+
     // Make an RPC to the other side of the channel
     bool Call(Message* msg, Message* reply);
 
@@ -332,6 +336,49 @@ protected:
     // not protected by mMutex.  It is managed exclusively by the
     // helper |class CxxStackFrame|.
     int mCxxStackFrames;
+    
+private:
+
+    //
+    // All dequeuing tasks require a single point of cancellation,
+    // which is handled via a reference-counted task.
+    //
+    class RefCountedTask
+    {
+      public:
+        RefCountedTask(CancelableTask* aTask)
+        : mTask(aTask)
+        , mRefCnt(0) {}
+        ~RefCountedTask() { delete mTask; }
+        void Run() { mTask->Run(); }
+        void Cancel() { mTask->Cancel(); }
+        void AddRef() { ++mRefCnt; }
+        void Release() {
+            if (--mRefCnt == 0)
+                delete this;
+        }
+
+      private:
+        CancelableTask* mTask;
+        nsrefcnt mRefCnt;
+    };
+
+    //
+    // Wrap an existing task which can be cancelled at any time
+    // without the wrapper's knowledge.
+    //
+    class DequeueTask : public Task
+    {
+      public:
+        DequeueTask(RefCountedTask* aTask) : mTask(aTask) {}
+        void Run() { mTask->Run(); }
+        
+      private:
+        nsRefPtr<RefCountedTask> mTask;
+    };
+
+    // A task encapsulating dequeuing one pending task
+    nsRefPtr<RefCountedTask> mDequeueOneTask;
 };