Bug 925623 - Support delivery of WebSocket events off-main-thread in WebSocketChannel r=jduell
authorSteve Workman <sworkman@mozilla.com>
Thu, 27 Mar 2014 13:58:19 -0700
changeset 175693 272ead5346d3b82bf455620a22eb3d8c99928f9c
parent 175692 f0e0c591f7d0a2a02e15a373b44279b14df131f8
child 175694 24a1126b26b7247888f59c6d1efcd1ff7ed772cf
push id41595
push usersworkman@mozilla.com
push dateThu, 27 Mar 2014 20:59:29 +0000
treeherdermozilla-inbound@272ead5346d3 [default view] [failures only]
perfherder[talos] [build metrics] [platform microbench] (compared to previous push)
reviewersjduell
bugs925623
milestone31.0a1
first release with
nightly linux32
nightly linux64
nightly mac
nightly win32
nightly win64
last release without
nightly linux32
nightly linux64
nightly mac
nightly win32
nightly win64
Bug 925623 - Support delivery of WebSocket events off-main-thread in WebSocketChannel r=jduell
netwerk/ipc/ChannelEventQueue.cpp
netwerk/ipc/ChannelEventQueue.h
netwerk/protocol/websocket/BaseWebSocketChannel.cpp
netwerk/protocol/websocket/BaseWebSocketChannel.h
netwerk/protocol/websocket/WebSocketChannel.cpp
netwerk/protocol/websocket/WebSocketChannel.h
netwerk/protocol/websocket/WebSocketChannelChild.cpp
netwerk/protocol/websocket/WebSocketChannelChild.h
--- a/netwerk/ipc/ChannelEventQueue.cpp
+++ b/netwerk/ipc/ChannelEventQueue.cpp
@@ -49,14 +49,32 @@ ChannelEventQueue::Resume()
   MOZ_ASSERT(mSuspendCount > 0);
   if (mSuspendCount <= 0) {
     return;
   }
 
   if (!--mSuspendCount) {
     nsRefPtr<nsRunnableMethod<ChannelEventQueue> > event =
       NS_NewRunnableMethod(this, &ChannelEventQueue::CompleteResume);
-    NS_DispatchToCurrentThread(event);
+    if (mTargetThread) {
+      mTargetThread->Dispatch(event, NS_DISPATCH_NORMAL);
+    } else {
+      MOZ_RELEASE_ASSERT(NS_IsMainThread());
+      NS_DispatchToCurrentThread(event);
+    }
   }
 }
 
+nsresult
+ChannelEventQueue::RetargetDeliveryTo(nsIEventTarget* aTargetThread)
+{
+  MOZ_RELEASE_ASSERT(NS_IsMainThread());
+  MOZ_RELEASE_ASSERT(!mTargetThread);
+  MOZ_RELEASE_ASSERT(aTargetThread);
+
+  mTargetThread = do_QueryInterface(aTargetThread);
+  MOZ_RELEASE_ASSERT(mTargetThread);
+
+  return NS_OK;
+}
+
 }
 }
--- a/netwerk/ipc/ChannelEventQueue.h
+++ b/netwerk/ipc/ChannelEventQueue.h
@@ -7,16 +7,18 @@
 
 #ifndef mozilla_net_ChannelEventQueue_h
 #define mozilla_net_ChannelEventQueue_h
 
 #include <nsTArray.h>
 #include <nsAutoPtr.h>
 
 class nsISupports;
+class nsIEventTarget;
+class nsIThread;
 
 namespace mozilla {
 namespace net {
 
 class ChannelEvent
 {
  public:
   ChannelEvent() { MOZ_COUNT_CTOR(ChannelEvent); }
@@ -66,31 +68,37 @@ class ChannelEventQueue
   // Suspend/resume event queue.  ShouldEnqueue() will return true and no events
   // will be run/flushed until resume is called.  These should be called when
   // the channel owning the event queue is suspended/resumed.
   inline void Suspend();
   // Resume flushes the queue asynchronously, i.e. items in queue will be
   // dispatched in a new event on the current thread.
   void Resume();
 
+  // Retargets delivery of events to the target thread specified.
+  nsresult RetargetDeliveryTo(nsIEventTarget* aTargetThread);
+
  private:
   inline void MaybeFlushQueue();
   void FlushQueue();
   inline void CompleteResume();
 
   nsTArray<nsAutoPtr<ChannelEvent> > mEventQueue;
 
   uint32_t mSuspendCount;
   bool     mSuspended;
   bool mForced;
   bool mFlushing;
 
   // Keep ptr to avoid refcount cycle: only grab ref during flushing.
   nsISupports *mOwner;
 
+  // Target thread for delivery of events.
+  nsCOMPtr<nsIThread> mTargetThread;
+
   friend class AutoEventEnqueuer;
 };
 
 inline bool
 ChannelEventQueue::ShouldEnqueue()
 {
   bool answer =  mForced || mSuspended || mFlushing;
 
--- a/netwerk/protocol/websocket/BaseWebSocketChannel.cpp
+++ b/netwerk/protocol/websocket/BaseWebSocketChannel.cpp
@@ -1,16 +1,17 @@
 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
 /* vim: set sw=2 ts=8 et tw=80 : */
 /* This Source Code Form is subject to the terms of the Mozilla Public
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 #include "WebSocketLog.h"
 #include "BaseWebSocketChannel.h"
+#include "MainThreadUtils.h"
 #include "nsILoadGroup.h"
 #include "nsIInterfaceRequestor.h"
 #include "nsAutoPtr.h"
 #include "nsStandardURL.h"
 
 #if defined(PR_LOGGING)
 PRLogModuleInfo *webSocketLog = nullptr;
 #endif
@@ -237,10 +238,27 @@ BaseWebSocketChannel::AllowPort(int32_t 
 {
   LOG(("BaseWebSocketChannel::AllowPort() %p\n", this));
 
   // do not override any blacklisted ports
   *_retval = false;
   return NS_OK;
 }
 
+//-----------------------------------------------------------------------------
+// BaseWebSocketChannel::nsIThreadRetargetableRequest
+//-----------------------------------------------------------------------------
+
+NS_IMETHODIMP
+BaseWebSocketChannel::RetargetDeliveryTo(nsIEventTarget* aTargetThread)
+{
+  MOZ_ASSERT(NS_IsMainThread());
+  MOZ_ASSERT(aTargetThread);
+  MOZ_ASSERT(!mTargetThread, "Delivery target should be set once, before AsyncOpen");
+  MOZ_ASSERT(!mWasOpened, "Should not be called after AsyncOpen!");
+
+  mTargetThread = do_QueryInterface(aTargetThread);
+  MOZ_ASSERT(mTargetThread);
+  return NS_OK;
+}
+
 } // namespace net
 } // namespace mozilla
--- a/netwerk/protocol/websocket/BaseWebSocketChannel.h
+++ b/netwerk/protocol/websocket/BaseWebSocketChannel.h
@@ -5,32 +5,36 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
 #ifndef mozilla_net_BaseWebSocketChannel_h
 #define mozilla_net_BaseWebSocketChannel_h
 
 #include "nsIWebSocketChannel.h"
 #include "nsIWebSocketListener.h"
 #include "nsIProtocolHandler.h"
+#include "nsIThread.h"
+#include "nsIThreadRetargetableRequest.h"
 #include "nsCOMPtr.h"
 #include "nsString.h"
 
 namespace mozilla {
 namespace net {
 
 const static int32_t kDefaultWSPort     = 80;
 const static int32_t kDefaultWSSPort    = 443;
 
 class BaseWebSocketChannel : public nsIWebSocketChannel,
-                             public nsIProtocolHandler
+                             public nsIProtocolHandler,
+                             public nsIThreadRetargetableRequest
 {
  public:
   BaseWebSocketChannel();
 
   NS_DECL_NSIPROTOCOLHANDLER
+  NS_DECL_NSITHREADRETARGETABLEREQUEST
 
   NS_IMETHOD QueryInterface(const nsIID & uuid, void **result) = 0;
   NS_IMETHOD_(MozExternalRefCountType ) AddRef(void) = 0;
   NS_IMETHOD_(MozExternalRefCountType ) Release(void) = 0;
 
   // Partial implementation of nsIWebSocketChannel
   //
   NS_IMETHOD GetOriginalURI(nsIURI **aOriginalURI);
@@ -49,16 +53,17 @@ class BaseWebSocketChannel : public nsIW
 
  protected:
   nsCOMPtr<nsIURI>                mOriginalURI;
   nsCOMPtr<nsIURI>                mURI;
   nsCOMPtr<nsIWebSocketListener>  mListener;
   nsCOMPtr<nsISupports>           mContext;
   nsCOMPtr<nsIInterfaceRequestor> mCallbacks;
   nsCOMPtr<nsILoadGroup>          mLoadGroup;
+  nsCOMPtr<nsIThread>             mTargetThread;
 
   nsCString                       mProtocol;
   nsCString                       mOrigin;
 
   nsCString                       mNegotiatedExtensions;
 
   uint32_t                        mEncrypted                 : 1;
   uint32_t                        mWasOpened                 : 1;
--- a/netwerk/protocol/websocket/WebSocketChannel.cpp
+++ b/netwerk/protocol/websocket/WebSocketChannel.cpp
@@ -29,27 +29,29 @@
 #include "nsIAsyncVerifyRedirectCallback.h"
 #include "nsIDashboardEventNotifier.h"
 #include "nsIEventTarget.h"
 #include "nsIHttpChannel.h"
 #include "nsILoadGroup.h"
 #include "nsIProtocolHandler.h"
 #include "nsIRandomGenerator.h"
 #include "nsISocketTransport.h"
+#include "nsThreadUtils.h"
 
 #include "nsAutoPtr.h"
 #include "nsNetCID.h"
 #include "nsServiceManagerUtils.h"
 #include "nsCRT.h"
 #include "nsThreadUtils.h"
 #include "nsError.h"
 #include "nsStringStream.h"
 #include "nsAlgorithm.h"
 #include "nsProxyRelease.h"
 #include "nsNetUtil.h"
+#include "mozilla/StaticMutex.h"
 #include "mozilla/Telemetry.h"
 #include "mozilla/TimeStamp.h"
 
 #include "plbase64.h"
 #include "prmem.h"
 #include "prnetdb.h"
 #include "zlib.h"
 #include <algorithm>
@@ -64,29 +66,30 @@
 
 extern PRThread *gSocketThread;
 
 using namespace mozilla;
 
 namespace mozilla {
 namespace net {
 
-NS_IMPL_ISUPPORTS12(WebSocketChannel,
+NS_IMPL_ISUPPORTS13(WebSocketChannel,
                     nsIWebSocketChannel,
                     nsIHttpUpgradeListener,
                     nsIRequestObserver,
                     nsIStreamListener,
                     nsIProtocolHandler,
                     nsIInputStreamCallback,
                     nsIOutputStreamCallback,
                     nsITimerCallback,
                     nsIDNSListener,
                     nsIProtocolProxyCallback,
                     nsIInterfaceRequestor,
-                    nsIChannelEventSink)
+                    nsIChannelEventSink,
+                    nsIThreadRetargetableRequest)
 
 // We implement RFC 6455, which uses Sec-WebSocket-Version: 13 on the wire.
 #define SEC_WEBSOCKET_VERSION "13"
 
 /*
  * About SSL unsigned certificates
  *
  * wss will not work to a host using an unsigned certificate unless there
@@ -312,157 +315,198 @@ private:
 // 1) Ensures that only one websocket at a time is CONNECTING to a given IP
 //    address (or hostname, if using proxy), per RFC 6455 Section 4.1.
 // 2) Delays reconnects to IP/host after connection failure, per Section 7.2.3
 //-----------------------------------------------------------------------------
 
 class nsWSAdmissionManager
 {
 public:
-  nsWSAdmissionManager() : mSessionCount(0)
-  {
-    MOZ_COUNT_CTOR(nsWSAdmissionManager);
-  }
-
-  class nsOpenConn
+  static void Init()
   {
-  public:
-    nsOpenConn(nsCString &addr, WebSocketChannel *channel)
-      : mAddress(addr), mChannel(channel) { MOZ_COUNT_CTOR(nsOpenConn); }
-    ~nsOpenConn() { MOZ_COUNT_DTOR(nsOpenConn); }
-
-    nsCString mAddress;
-    WebSocketChannel *mChannel;
-  };
-
-  ~nsWSAdmissionManager()
+    StaticMutexAutoLock lock(sLock);
+    if (!sManager) {
+      sManager = new nsWSAdmissionManager();
+    }
+  }
+
+  static void Shutdown()
   {
-    MOZ_COUNT_DTOR(nsWSAdmissionManager);
-    for (uint32_t i = 0; i < mQueue.Length(); i++)
-      delete mQueue[i];
+    StaticMutexAutoLock lock(sLock);
+    delete sManager;
+    sManager = nullptr;
   }
 
   // Determine if we will open connection immediately (returns true), or
   // delay/queue the connection (returns false)
-  void ConditionallyConnect(WebSocketChannel *ws)
+  static void ConditionallyConnect(WebSocketChannel *ws)
   {
     NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
     NS_ABORT_IF_FALSE(ws->mConnecting == NOT_CONNECTING, "opening state");
 
+    StaticMutexAutoLock lock(sLock);
+    if (!sManager) {
+      return;
+    }
+
     // If there is already another WS channel connecting to this IP address,
     // defer BeginOpen and mark as waiting in queue.
-    bool found = (IndexOf(ws->mAddress) >= 0);
+    bool found = (sManager->IndexOf(ws->mAddress) >= 0);
 
     // Always add ourselves to queue, even if we'll connect immediately
     nsOpenConn *newdata = new nsOpenConn(ws->mAddress, ws);
-    mQueue.AppendElement(newdata);
+    sManager->mQueue.AppendElement(newdata);
 
     if (found) {
       ws->mConnecting = CONNECTING_QUEUED;
     } else {
-      mFailures.DelayOrBegin(ws);
+      sManager->mFailures.DelayOrBegin(ws);
     }
   }
 
-  void OnConnected(WebSocketChannel *aChannel)
+  static void OnConnected(WebSocketChannel *aChannel)
   {
     NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
     NS_ABORT_IF_FALSE(aChannel->mConnecting == CONNECTING_IN_PROGRESS,
                       "Channel completed connect, but not connecting?");
 
+    StaticMutexAutoLock lock(sLock);
+    if (!sManager) {
+      return;
+    }
+
     aChannel->mConnecting = NOT_CONNECTING;
 
     // Remove from queue
-    RemoveFromQueue(aChannel);
+    sManager->RemoveFromQueue(aChannel);
 
     // Connection succeeded, so stop keeping track of any previous failures
-    mFailures.Remove(aChannel->mAddress, aChannel->mPort);
+    sManager->mFailures.Remove(aChannel->mAddress, aChannel->mPort);
 
     // Check for queued connections to same host.
     // Note: still need to check for failures, since next websocket with same
     // host may have different port
-    ConnectNext(aChannel->mAddress);
+    sManager->ConnectNext(aChannel->mAddress);
   }
 
   // Called every time a websocket channel ends its session (including going away
   // w/o ever successfully creating a connection)
-  void OnStopSession(WebSocketChannel *aChannel, nsresult aReason)
+  static void OnStopSession(WebSocketChannel *aChannel, nsresult aReason)
   {
     NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
 
+    StaticMutexAutoLock lock(sLock);
+    if (!sManager) {
+      return;
+    }
+
     if (NS_FAILED(aReason)) {
       // Have we seen this failure before?
-      FailDelay *knownFailure = mFailures.Lookup(aChannel->mAddress,
-                                                 aChannel->mPort);
+      FailDelay *knownFailure = sManager->mFailures.Lookup(aChannel->mAddress,
+                                                           aChannel->mPort);
       if (knownFailure) {
         if (aReason == NS_ERROR_NOT_CONNECTED) {
           // Don't count close() before connection as a network error
           LOG(("Websocket close() before connection to %s, %d completed"
                " [this=%p]", aChannel->mAddress.get(), (int)aChannel->mPort,
                aChannel));
         } else {
           // repeated failure to connect: increase delay for next connection
           knownFailure->FailedAgain();
         }
       } else {
         // new connection failure: record it.
         LOG(("WebSocket: connection to %s, %d failed: [this=%p]",
               aChannel->mAddress.get(), (int)aChannel->mPort, aChannel));
-        mFailures.Add(aChannel->mAddress, aChannel->mPort);
+        sManager->mFailures.Add(aChannel->mAddress, aChannel->mPort);
       }
     }
 
     if (aChannel->mConnecting) {
       // Only way a connecting channel may get here w/o failing is if it was
       // closed with GOING_AWAY (1001) because of navigation, tab close, etc.
       NS_ABORT_IF_FALSE(NS_FAILED(aReason) ||
                         aChannel->mScriptCloseCode == CLOSE_GOING_AWAY,
                         "websocket closed while connecting w/o failing?");
 
-      RemoveFromQueue(aChannel);
+      sManager->RemoveFromQueue(aChannel);
 
       bool wasNotQueued = (aChannel->mConnecting != CONNECTING_QUEUED);
       aChannel->mConnecting = NOT_CONNECTING;
       if (wasNotQueued) {
-        ConnectNext(aChannel->mAddress);
+        sManager->ConnectNext(aChannel->mAddress);
       }
     }
   }
 
+  static void IncrementSessionCount()
+  {
+    StaticMutexAutoLock lock(sLock);
+    if (!sManager) {
+      return;
+    }
+    sManager->mSessionCount++;
+  }
+
+  static void DecrementSessionCount()
+  {
+    StaticMutexAutoLock lock(sLock);
+    if (!sManager) {
+      return;
+    }
+    sManager->mSessionCount--;
+  }
+
+  static void GetSessionCount(int32_t &aSessionCount)
+  {
+    StaticMutexAutoLock lock(sLock);
+    if (!sManager) {
+      return;
+    }
+    aSessionCount = sManager->mSessionCount;
+  }
+
+private:
+  nsWSAdmissionManager() : mSessionCount(0)
+  {
+    MOZ_COUNT_CTOR(nsWSAdmissionManager);
+  }
+
+  ~nsWSAdmissionManager()
+  {
+    MOZ_COUNT_DTOR(nsWSAdmissionManager);
+    for (uint32_t i = 0; i < mQueue.Length(); i++)
+      delete mQueue[i];
+  }
+
+  class nsOpenConn
+  {
+  public:
+    nsOpenConn(nsCString &addr, WebSocketChannel *channel)
+      : mAddress(addr), mChannel(channel) { MOZ_COUNT_CTOR(nsOpenConn); }
+    ~nsOpenConn() { MOZ_COUNT_DTOR(nsOpenConn); }
+
+    nsCString mAddress;
+    WebSocketChannel *mChannel;
+  };
+
   void ConnectNext(nsCString &hostName)
   {
     int32_t index = IndexOf(hostName);
     if (index >= 0) {
       WebSocketChannel *chan = mQueue[index]->mChannel;
 
       NS_ABORT_IF_FALSE(chan->mConnecting == CONNECTING_QUEUED,
                         "transaction not queued but in queue");
       LOG(("WebSocket: ConnectNext: found channel [this=%p] in queue", chan));
 
       mFailures.DelayOrBegin(chan);
     }
   }
 
-  void IncrementSessionCount()
-  {
-    mSessionCount++;
-  }
-
-  void DecrementSessionCount()
-  {
-    mSessionCount--;
-  }
-
-  int32_t SessionCount()
-  {
-    return mSessionCount;
-  }
-
-private:
-
   void RemoveFromQueue(WebSocketChannel *aChannel)
   {
     int32_t index = IndexOf(aChannel);
     NS_ABORT_IF_FALSE(index >= 0, "connection to remove not in queue");
     if (index >= 0) {
       nsOpenConn *olddata = mQueue[index];
       mQueue.RemoveElementAt(index);
       delete olddata;
@@ -494,19 +538,23 @@ private:
   // CONNECTING_IN_PROGRESS or CONNECTING_DELAYED.  Later ones with the same
   // hostname must be CONNECTING_QUEUED.
   //
   // We could hash hostnames instead of using a single big vector here, but the
   // dataset is expected to be small.
   nsTArray<nsOpenConn *> mQueue;
 
   FailDelayManager       mFailures;
+
+  static nsWSAdmissionManager *sManager;
+  static StaticMutex           sLock;
 };
 
-static nsWSAdmissionManager *sWebSocketAdmissions = nullptr;
+nsWSAdmissionManager *nsWSAdmissionManager::sManager;
+StaticMutex           nsWSAdmissionManager::sLock;
 
 //-----------------------------------------------------------------------------
 // CallOnMessageAvailable
 //-----------------------------------------------------------------------------
 
 class CallOnMessageAvailable MOZ_FINAL : public nsIRunnable
 {
 public:
@@ -516,16 +564,18 @@ public:
                          nsCString        &aData,
                          int32_t           aLen)
     : mChannel(aChannel),
       mData(aData),
       mLen(aLen) {}
 
   NS_IMETHOD Run()
   {
+    MOZ_ASSERT(NS_GetCurrentThread() == mChannel->mTargetThread);
+
     if (mLen < 0)
       mChannel->mListener->OnMessageAvailable(mChannel->mContext, mData);
     else
       mChannel->mListener->OnBinaryMessageAvailable(mChannel->mContext, mData);
     return NS_OK;
   }
 
 private:
@@ -548,19 +598,19 @@ public:
 
   CallOnStop(WebSocketChannel *aChannel,
              nsresult          aReason)
     : mChannel(aChannel),
       mReason(aReason) {}
 
   NS_IMETHOD Run()
   {
-    NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
-
-    sWebSocketAdmissions->OnStopSession(mChannel, mReason);
+    MOZ_ASSERT(NS_GetCurrentThread() == mChannel->mTargetThread);
+
+    nsWSAdmissionManager::OnStopSession(mChannel, mReason);
 
     if (mChannel->mListener) {
       mChannel->mListener->OnStop(mChannel->mContext, mReason);
       mChannel->mListener = nullptr;
       mChannel->mContext = nullptr;
     }
     return NS_OK;
   }
@@ -586,16 +636,18 @@ public:
                     uint16_t          aCode,
                     nsCString        &aReason)
     : mChannel(aChannel),
       mCode(aCode),
       mReason(aReason) {}
 
   NS_IMETHOD Run()
   {
+    MOZ_ASSERT(NS_GetCurrentThread() == mChannel->mTargetThread);
+
     mChannel->mListener->OnServerClose(mChannel->mContext, mCode, mReason);
     return NS_OK;
   }
 
 private:
   ~CallOnServerClose() {}
 
   nsRefPtr<WebSocketChannel>        mChannel;
@@ -615,16 +667,18 @@ public:
 
   CallAcknowledge(WebSocketChannel *aChannel,
                   uint32_t          aSize)
     : mChannel(aChannel),
       mSize(aSize) {}
 
   NS_IMETHOD Run()
   {
+    MOZ_ASSERT(NS_GetCurrentThread() == mChannel->mTargetThread);
+
     LOG(("WebSocketChannel::CallAcknowledge: Size %u\n", mSize));
     mChannel->mListener->OnAcknowledge(mChannel->mContext, mSize);
     return NS_OK;
   }
 
 private:
   ~CallAcknowledge() {}
 
@@ -976,18 +1030,17 @@ WebSocketChannel::WebSocketChannel() :
   mCountRecv(0),
   mCountSent(0),
   mAppId(NECKO_NO_APP_ID)
 {
   NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
 
   LOG(("WebSocketChannel::WebSocketChannel() %p\n", this));
 
-  if (!sWebSocketAdmissions)
-    sWebSocketAdmissions = new nsWSAdmissionManager();
+  nsWSAdmissionManager::Init();
 
   mFramePtr = mBuffer = static_cast<uint8_t *>(moz_xmalloc(mBufferSize));
 
   nsresult rv;
   mConnectionLogService = do_GetService("@mozilla.org/network/dashboard;1",&rv);
   if (NS_FAILED(rv))
     LOG(("Failed to initiate dashboard service."));
 
@@ -1048,18 +1101,17 @@ WebSocketChannel::~WebSocketChannel()
     mLoadGroup.forget(&forgettableGroup);
     NS_ProxyRelease(mainThread, forgettableGroup, false);
   }
 }
 
 void
 WebSocketChannel::Shutdown()
 {
-  delete sWebSocketAdmissions;
-  sWebSocketAdmissions = nullptr;
+  nsWSAdmissionManager::Shutdown();
 }
 
 void
 WebSocketChannel::BeginOpen()
 {
   LOG(("WebSocketChannel::BeginOpen() %p\n", this));
   NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
 
@@ -1354,25 +1406,21 @@ WebSocketChannel::ProcessInput(uint8_t *
           return NS_ERROR_OUT_OF_MEMORY;
 
         // Section 8.1 says to fail connection if invalid utf-8 in text message
         if (!IsUTF8(utf8Data, false)) {
           LOG(("WebSocketChannel:: text frame invalid utf-8\n"));
           return NS_ERROR_CANNOT_CONVERT_DATA;
         }
 
-        NS_DispatchToMainThread(new CallOnMessageAvailable(this, utf8Data, -1));
-        nsresult rv;
+        mTargetThread->Dispatch(new CallOnMessageAvailable(this, utf8Data, -1),
+                                NS_DISPATCH_NORMAL);
         if (mConnectionLogService && !mPrivateBrowsing) {
-          nsAutoCString host;
-          rv = mURI->GetHostPort(host);
-          if (NS_SUCCEEDED(rv)) {
-            mConnectionLogService->NewMsgReceived(host, mSerial, count);
-            LOG(("Added new msg received for %s",host.get()));
-          }
+          mConnectionLogService->NewMsgReceived(mHost, mSerial, count);
+          LOG(("Added new msg received for %s", mHost.get()));
         }
       }
     } else if (opcode & kControlFrameMask) {
       // control frames
       if (payloadLength > 125) {
         LOG(("WebSocketChannel:: bad control frame code %d length %d\n",
              opcode, payloadLength));
         return NS_ERROR_ILLEGAL_VALUE;
@@ -1406,18 +1454,19 @@ WebSocketChannel::ProcessInput(uint8_t *
           }
         }
 
         if (mCloseTimer) {
           mCloseTimer->Cancel();
           mCloseTimer = nullptr;
         }
         if (mListener) {
-          NS_DispatchToMainThread(new CallOnServerClose(this, mServerCloseCode,
-                                                        mServerCloseReason));
+          mTargetThread->Dispatch(new CallOnServerClose(this, mServerCloseCode,
+                                                        mServerCloseReason),
+                                  NS_DISPATCH_NORMAL);
         }
 
         if (mClientClosed)
           ReleaseSession();
       } else if (opcode == kPing) {
         LOG(("WebSocketChannel:: ping received\n"));
         GeneratePong(payload, payloadLength);
       } else if (opcode == kPong) {
@@ -1442,27 +1491,23 @@ WebSocketChannel::ProcessInput(uint8_t *
         if (mBuffered)
           mBuffered -= framingLength + payloadLength;
         payloadLength = 0;
       }
     } else if (opcode == kBinary) {
       LOG(("WebSocketChannel:: binary frame received\n"));
       if (mListener) {
         nsCString binaryData((const char *)payload, payloadLength);
-        NS_DispatchToMainThread(new CallOnMessageAvailable(this, binaryData,
-                                                           payloadLength));
+        mTargetThread->Dispatch(new CallOnMessageAvailable(this, binaryData,
+                                                           payloadLength),
+                                NS_DISPATCH_NORMAL);
         // To add the header to 'Networking Dashboard' log
-        nsresult rv;
         if (mConnectionLogService && !mPrivateBrowsing) {
-          nsAutoCString host;
-          rv = mURI->GetHostPort(host);
-          if (NS_SUCCEEDED(rv)) {
-            mConnectionLogService->NewMsgReceived(host, mSerial, count);
-            LOG(("Added new received msg for %s",host.get()));
-          }
+          mConnectionLogService->NewMsgReceived(mHost, mSerial, count);
+          LOG(("Added new received msg for %s", mHost.get()));
         }
       }
     } else if (opcode != kContinuation) {
       /* unknown opcode */
       LOG(("WebSocketChannel:: unknown op code %d\n", opcode));
       return NS_ERROR_ILLEGAL_VALUE;
     }
 
@@ -1865,22 +1910,18 @@ WebSocketChannel::CleanupConnection()
 
   if (mTransport) {
     mTransport->SetSecurityCallbacks(nullptr);
     mTransport->SetEventSink(nullptr, nullptr);
     mTransport->Close(NS_BASE_STREAM_CLOSED);
     mTransport = nullptr;
   }
 
-  nsresult rv;
   if (mConnectionLogService && !mPrivateBrowsing) {
-    nsAutoCString host;
-    rv = mURI->GetHostPort(host);
-    if (NS_SUCCEEDED(rv))
-      mConnectionLogService->RemoveHost(host, mSerial);
+    mConnectionLogService->RemoveHost(mHost, mSerial);
   }
 
   DecrementSessionCount();
 }
 
 void
 WebSocketChannel::StopSession(nsresult reason)
 {
@@ -1935,18 +1976,20 @@ WebSocketChannel::StopSession(nsresult r
       total += count;
       rv = mSocketIn->Read(buffer, 512, &count);
       if (rv != NS_BASE_STREAM_WOULD_BLOCK &&
         (NS_FAILED(rv) || count == 0))
         mTCPClosed = true;
     } while (NS_SUCCEEDED(rv) && count > 0 && total < 32000);
   }
 
-  if (!mTCPClosed && mTransport && sWebSocketAdmissions &&
-      sWebSocketAdmissions->SessionCount() < kLingeringCloseThreshold) {
+  int32_t sessionCount = kLingeringCloseThreshold;
+  nsWSAdmissionManager::GetSessionCount(sessionCount);
+
+  if (!mTCPClosed && mTransport && sessionCount < kLingeringCloseThreshold) {
 
     // 7.1.1 says that the client SHOULD wait for the server to close the TCP
     // connection. This is so we can reuse port numbers before 2 MSL expires,
     // which is not really as much of a concern for us as the amount of state
     // that might be accrued by keeping this channel object around waiting for
     // the server. We handle the SHOULD by waiting a short time in the common
     // case, but not waiting in the case of high concurrency.
     //
@@ -1975,17 +2018,18 @@ WebSocketChannel::StopSession(nsresult r
   mInflateReader = nullptr;
   mInflateStream = nullptr;
 
   delete mCompressor;
   mCompressor = nullptr;
 
   if (!mCalledOnStop) {
     mCalledOnStop = 1;
-    NS_DispatchToMainThread(new CallOnStop(this, reason));
+    mTargetThread->Dispatch(new CallOnStop(this, reason),
+                            NS_DISPATCH_NORMAL);
   }
 
   return;
 }
 
 void
 WebSocketChannel::AbortSession(nsresult reason)
 {
@@ -2034,30 +2078,30 @@ WebSocketChannel::ReleaseSession()
     return;
   StopSession(NS_OK);
 }
 
 void
 WebSocketChannel::IncrementSessionCount()
 {
   if (!mIncrementedSessionCount) {
-    sWebSocketAdmissions->IncrementSessionCount();
+    nsWSAdmissionManager::IncrementSessionCount();
     mIncrementedSessionCount = 1;
   }
 }
 
 void
 WebSocketChannel::DecrementSessionCount()
 {
   // Make sure we decrement session count only once, and only if we incremented it.
   // This code is thread-safe: sWebSocketAdmissions->DecrementSessionCount is
   // atomic, and mIncrementedSessionCount/mDecrementedSessionCount are set at
   // times when they'll never be a race condition for checking/setting them.
   if (mIncrementedSessionCount && !mDecrementedSessionCount) {
-    sWebSocketAdmissions->DecrementSessionCount();
+    nsWSAdmissionManager::DecrementSessionCount();
     mDecrementedSessionCount = 1;
   }
 }
 
 nsresult
 WebSocketChannel::HandleExtensions()
 {
   LOG(("WebSocketChannel::HandleExtensions() %p\n", this));
@@ -2261,17 +2305,17 @@ WebSocketChannel::StartWebsocketData()
 {
   LOG(("WebSocketChannel::StartWebsocketData() %p", this));
   NS_ABORT_IF_FALSE(!mDataStarted, "StartWebsocketData twice");
   mDataStarted = 1;
 
   // We're now done CONNECTING, which means we can now open another,
   // perhaps parallel, connection to the same host if one
   // is pending
-  sWebSocketAdmissions->OnConnected(this);
+  nsWSAdmissionManager::OnConnected(this);
 
   LOG(("WebSocketChannel::StartWebsocketData Notifying Listener %p\n",
        mListener.get()));
 
   if (mListener)
     mListener->OnStart(mContext);
 
   // Start keepalive ping timer, if we're using keepalive.
@@ -2349,17 +2393,17 @@ WebSocketChannel::OnLookupComplete(nsICa
     mURI->GetHost(mAddress);
   } else {
     nsresult rv = aRecord->GetNextAddrAsString(mAddress);
     if (NS_FAILED(rv))
       LOG(("WebSocketChannel::OnLookupComplete: Failed GetNextAddr\n"));
   }
 
   LOG(("WebSocket OnLookupComplete: Proceeding to ConditionallyConnect\n"));
-  sWebSocketAdmissions->ConditionallyConnect(this);
+  nsWSAdmissionManager::ConditionallyConnect(this);
 
   return NS_OK;
 }
 
 // nsIProtocolProxyCallback
 NS_IMETHODIMP
 WebSocketChannel::OnProxyAvailable(nsICancelable *aRequest, nsIURI *aURI,
                                    nsIProxyInfo *pi, nsresult status)
@@ -2506,17 +2550,17 @@ WebSocketChannel::AsyncOnChannelRedirect
   // Redirected-to URI may need to be delayed by 1-connecting-per-host and
   // delay-after-fail algorithms.  So hold off calling OnRedirectVerifyCallback
   // until BeginOpen, when we know it's OK to proceed with new channel.
   mRedirectCallback = callback;
 
   // Mark old channel as successfully connected so we'll clear any FailDelay
   // associated with the old URI.  Note: no need to also call OnStopSession:
   // it's a no-op for successful, already-connected channels.
-  sWebSocketAdmissions->OnConnected(this);
+  nsWSAdmissionManager::OnConnected(this);
 
   // ApplyForAdmission as if we were starting from fresh...
   mAddress.Truncate();
   mOpenedHttpChannel = 0;
   rv = ApplyForAdmission();
   if (NS_FAILED(rv)) {
     LOG(("WebSocketChannel: Redirect failed due to DNS failure\n"));
     mRedirectCallback = nullptr;
@@ -2624,16 +2668,21 @@ WebSocketChannel::AsyncOpen(nsIURI *aURI
     return NS_ERROR_UNEXPECTED;
   }
 
   if (mListener || mWasOpened)
     return NS_ERROR_ALREADY_OPENED;
 
   nsresult rv;
 
+  // Ensure target thread is set.
+  if (!mTargetThread) {
+    mTargetThread = do_GetMainThread();
+  }
+
   mSocketThread = do_GetService(NS_SOCKETTRANSPORTSERVICE_CONTRACTID, &rv);
   if (NS_FAILED(rv)) {
     NS_WARNING("unable to continue without socket transport service");
     return rv;
   }
 
   mRandomGenerator =
     do_GetService("@mozilla.org/security/random-generator;1", &rv);
@@ -2683,34 +2732,36 @@ WebSocketChannel::AsyncOpen(nsIURI *aURI
     }
     rv = prefService->GetIntPref
       ("network.websocket.max-connections", &intpref);
     if (NS_SUCCEEDED(rv)) {
       mMaxConcurrentConnections = clamped(intpref, 1, 0xffff);
     }
   }
 
-  if (sWebSocketAdmissions)
+  int32_t sessionCount = -1;
+  nsWSAdmissionManager::GetSessionCount(sessionCount);
+  if (sessionCount >= 0) {
     LOG(("WebSocketChannel::AsyncOpen %p sessionCount=%d max=%d\n", this,
-         sWebSocketAdmissions->SessionCount(), mMaxConcurrentConnections));
-
-  if (sWebSocketAdmissions &&
-      sWebSocketAdmissions->SessionCount() >= mMaxConcurrentConnections)
-  {
+         sessionCount, mMaxConcurrentConnections));
+  }
+
+  if (sessionCount >= mMaxConcurrentConnections) {
     LOG(("WebSocketChannel: max concurrency %d exceeded (%d)",
          mMaxConcurrentConnections,
-         sWebSocketAdmissions->SessionCount()));
+         sessionCount));
 
     // WebSocket connections are expected to be long lived, so return
     // an error here instead of queueing
     return NS_ERROR_SOCKET_CREATE_FAILED;
   }
 
   mOriginalURI = aURI;
   mURI = mOriginalURI;
+  mURI->GetHostPort(mHost);
   mOrigin = aOrigin;
 
   nsCOMPtr<nsIURI> localURI;
   nsCOMPtr<nsIChannel> localChannel;
 
   mURI->Clone(getter_AddRefs(localURI));
   if (mEncrypted)
     rv = localURI->SetScheme(NS_LITERAL_CSTRING("https"));
@@ -2751,21 +2802,18 @@ WebSocketChannel::AsyncOpen(nsIURI *aURI
 
   rv = SetupRequest();
   if (NS_FAILED(rv))
     return rv;
 
   mPrivateBrowsing = NS_UsePrivateBrowsing(localChannel);
 
   if (mConnectionLogService && !mPrivateBrowsing) {
-    nsAutoCString host;
-    rv = mURI->GetHostPort(host);
-    if (NS_SUCCEEDED(rv)) {
-      mConnectionLogService->AddHost(host, mSerial, BaseWebSocketChannel::mEncrypted);
-    }
+    mConnectionLogService->AddHost(mHost, mSerial,
+                                   BaseWebSocketChannel::mEncrypted);
   }
 
   rv = ApplyForAdmission();
   if (NS_FAILED(rv))
     return rv;
 
   // Only set these if the open was successful:
   //
@@ -2839,17 +2887,17 @@ WebSocketChannel::SendBinaryStream(nsIIn
 
   return SendMsgCommon(nullptr, true, aLength, aStream);
 }
 
 nsresult
 WebSocketChannel::SendMsgCommon(const nsACString *aMsg, bool aIsBinary,
                                 uint32_t aLength, nsIInputStream *aStream)
 {
-  NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
+  NS_ABORT_IF_FALSE(NS_GetCurrentThread() == mTargetThread, "not target thread");
 
   if (mRequestedClose) {
     LOG(("WebSocketChannel:: Error: send when closed\n"));
     return NS_ERROR_UNEXPECTED;
   }
 
   if (mStopped) {
     LOG(("WebSocketChannel:: Error: send when stopped\n"));
@@ -2857,24 +2905,19 @@ WebSocketChannel::SendMsgCommon(const ns
   }
 
   NS_ABORT_IF_FALSE(mMaxMessageSize >= 0, "max message size negative");
   if (aLength > static_cast<uint32_t>(mMaxMessageSize)) {
     LOG(("WebSocketChannel:: Error: message too big\n"));
     return NS_ERROR_FILE_TOO_BIG;
   }
 
-  nsresult rv;
   if (mConnectionLogService && !mPrivateBrowsing) {
-    nsAutoCString host;
-    rv = mURI->GetHostPort(host);
-    if (NS_SUCCEEDED(rv)) {
-      mConnectionLogService->NewMsgSent(host, mSerial, aLength);
-      LOG(("Added new msg sent for %s",host.get()));
-    }
+    mConnectionLogService->NewMsgSent(mHost, mSerial, aLength);
+    LOG(("Added new msg sent for %s", mHost.get()));
   }
 
   return mSocketThread->Dispatch(
     aStream ? new OutboundEnqueuer(this, new OutboundMessage(aStream, aLength))
             : new OutboundEnqueuer(this,
                      new OutboundMessage(aIsBinary ? kMsgTypeBinaryString
                                                    : kMsgTypeString,
                                          new nsCString(*aMsg))),
@@ -3191,17 +3234,17 @@ WebSocketChannel::OnOutputStreamReady(ns
       rv = mSocketOut->Write(sndBuf, toSend, &amtSent);
       LOG(("WebSocketChannel::OnOutputStreamReady: write %u rv %x\n",
            amtSent, rv));
 
       // accumulate sent bytes
       CountSentBytes(amtSent);
 
       if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
-        mSocketOut->AsyncWait(this, 0, 0, nullptr);
+        mSocketOut->AsyncWait(this, 0, 0, mSocketThread);
         return NS_OK;
       }
 
       if (NS_FAILED(rv)) {
         AbortSession(rv);
         return NS_OK;
       }
     }
@@ -3212,18 +3255,19 @@ WebSocketChannel::OnOutputStreamReady(ns
         mHdrOutToSend = 0;
       } else {
         mHdrOut += amtSent;
         mHdrOutToSend -= amtSent;
       }
     } else {
       if (amtSent == toSend) {
         if (!mStopped) {
-          NS_DispatchToMainThread(new CallAcknowledge(this,
-                                                      mCurrentOut->Length()));
+          mTargetThread->Dispatch(new CallAcknowledge(this,
+                                                      mCurrentOut->Length()),
+                                  NS_DISPATCH_NORMAL);
         }
         DeleteCurrentOutGoingMessage();
         PrimeNewOutgoingMessage();
       } else {
         mCurrentOutSent += amtSent;
       }
     }
   }
--- a/netwerk/protocol/websocket/WebSocketChannel.h
+++ b/netwerk/protocol/websocket/WebSocketChannel.h
@@ -175,16 +175,19 @@ private:
 
   nsCString                       mHashedSecret;
 
   // Used as key for connection managment: Initially set to hostname from URI,
   // then to IP address (unless we're leaving DNS resolution to a proxy server)
   nsCString                       mAddress;
   int32_t                         mPort;          // WS server port
 
+  // Used for off main thread access to the URI string.
+  nsCString                       mHost;
+
   nsCOMPtr<nsISocketTransport>    mTransport;
   nsCOMPtr<nsIAsyncInputStream>   mSocketIn;
   nsCOMPtr<nsIAsyncOutputStream>  mSocketOut;
 
   nsCOMPtr<nsITimer>              mCloseTimer;
   uint32_t                        mCloseTimeout;  /* milliseconds */
 
   nsCOMPtr<nsITimer>              mOpenTimer;
--- a/netwerk/protocol/websocket/WebSocketChannelChild.cpp
+++ b/netwerk/protocol/websocket/WebSocketChannelChild.cpp
@@ -42,21 +42,24 @@ NS_IMETHODIMP_(MozExternalRefCountType) 
   }
   return mRefCnt;
 }
 
 NS_INTERFACE_MAP_BEGIN(WebSocketChannelChild)
   NS_INTERFACE_MAP_ENTRY(nsIWebSocketChannel)
   NS_INTERFACE_MAP_ENTRY(nsIProtocolHandler)
   NS_INTERFACE_MAP_ENTRY_AMBIGUOUS(nsISupports, nsIWebSocketChannel)
+  NS_INTERFACE_MAP_ENTRY(nsIThreadRetargetableRequest)
 NS_INTERFACE_MAP_END
 
 WebSocketChannelChild::WebSocketChannelChild(bool aSecure)
  : mIPCOpen(false)
 {
+  NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
+
   LOG(("WebSocketChannelChild::WebSocketChannelChild() %p\n", this));
   BaseWebSocketChannel::mEncrypted = aSecure;
   mEventQ = new ChannelEventQueue(static_cast<nsIWebSocketChannel*>(this));
 }
 
 WebSocketChannelChild::~WebSocketChannelChild()
 {
   LOG(("WebSocketChannelChild::~WebSocketChannelChild() %p\n", this));
@@ -73,16 +76,44 @@ WebSocketChannelChild::AddIPDLReference(
 void
 WebSocketChannelChild::ReleaseIPDLReference()
 {
   NS_ABORT_IF_FALSE(mIPCOpen, "Attempt to release nonexistent IPDL reference");
   mIPCOpen = false;
   Release();
 }
 
+class WrappedChannelEvent : public nsRunnable
+{
+public:
+  WrappedChannelEvent(ChannelEvent *aChannelEvent)
+    : mChannelEvent(aChannelEvent)
+  {
+    MOZ_RELEASE_ASSERT(aChannelEvent);
+  }
+  NS_IMETHOD Run()
+  {
+    mChannelEvent->Run();
+    return NS_OK;
+  }
+private:
+  nsAutoPtr<ChannelEvent> mChannelEvent;
+};
+
+void
+WebSocketChannelChild::DispatchToTargetThread(ChannelEvent *aChannelEvent)
+{
+  MOZ_RELEASE_ASSERT(NS_IsMainThread());
+  MOZ_RELEASE_ASSERT(mTargetThread);
+  MOZ_RELEASE_ASSERT(aChannelEvent);
+
+  mTargetThread->Dispatch(new WrappedChannelEvent(aChannelEvent),
+                          NS_DISPATCH_NORMAL);
+}
+
 class StartEvent : public ChannelEvent
 {
  public:
   StartEvent(WebSocketChannelChild* aChild,
              const nsCString& aProtocol,
              const nsCString& aExtensions)
   : mChild(aChild)
   , mProtocol(aProtocol)
@@ -100,16 +131,18 @@ class StartEvent : public ChannelEvent
 };
 
 bool
 WebSocketChannelChild::RecvOnStart(const nsCString& aProtocol,
                                    const nsCString& aExtensions)
 {
   if (mEventQ->ShouldEnqueue()) {
     mEventQ->Enqueue(new StartEvent(this, aProtocol, aExtensions));
+  } else if (mTargetThread) {
+    DispatchToTargetThread(new StartEvent(this, aProtocol, aExtensions));
   } else {
     OnStart(aProtocol, aExtensions);
   }
   return true;
 }
 
 void
 WebSocketChannelChild::OnStart(const nsCString& aProtocol,
@@ -143,16 +176,18 @@ class StopEvent : public ChannelEvent
   nsresult mStatusCode;
 };
 
 bool
 WebSocketChannelChild::RecvOnStop(const nsresult& aStatusCode)
 {
   if (mEventQ->ShouldEnqueue()) {
     mEventQ->Enqueue(new StopEvent(this, aStatusCode));
+  } else if (mTargetThread) {
+    DispatchToTargetThread(new StopEvent(this, aStatusCode));
   } else {
     OnStop(aStatusCode);
   }
   return true;
 }
 
 void
 WebSocketChannelChild::OnStop(const nsresult& aStatusCode)
@@ -189,17 +224,19 @@ class MessageEvent : public ChannelEvent
   bool mBinary;
 };
 
 bool
 WebSocketChannelChild::RecvOnMessageAvailable(const nsCString& aMsg)
 {
   if (mEventQ->ShouldEnqueue()) {
     mEventQ->Enqueue(new MessageEvent(this, aMsg, false));
-  } else {
+  } else if (mTargetThread) {
+    DispatchToTargetThread(new MessageEvent(this, aMsg, false));
+   } else {
     OnMessageAvailable(aMsg);
   }
   return true;
 }
 
 void
 WebSocketChannelChild::OnMessageAvailable(const nsCString& aMsg)
 {
@@ -210,16 +247,18 @@ WebSocketChannelChild::OnMessageAvailabl
   }
 }
 
 bool
 WebSocketChannelChild::RecvOnBinaryMessageAvailable(const nsCString& aMsg)
 {
   if (mEventQ->ShouldEnqueue()) {
     mEventQ->Enqueue(new MessageEvent(this, aMsg, true));
+  } else if (mTargetThread) {
+    DispatchToTargetThread(new MessageEvent(this, aMsg, true));
   } else {
     OnBinaryMessageAvailable(aMsg);
   }
   return true;
 }
 
 void
 WebSocketChannelChild::OnBinaryMessageAvailable(const nsCString& aMsg)
@@ -249,16 +288,18 @@ class AcknowledgeEvent : public ChannelE
   uint32_t mSize;
 };
 
 bool
 WebSocketChannelChild::RecvOnAcknowledge(const uint32_t& aSize)
 {
   if (mEventQ->ShouldEnqueue()) {
     mEventQ->Enqueue(new AcknowledgeEvent(this, aSize));
+  } else if (mTargetThread) {
+    DispatchToTargetThread(new AcknowledgeEvent(this, aSize));
   } else {
     OnAcknowledge(aSize);
   }
   return true;
 }
 
 void
 WebSocketChannelChild::OnAcknowledge(const uint32_t& aSize)
@@ -292,16 +333,18 @@ class ServerCloseEvent : public ChannelE
 };
 
 bool
 WebSocketChannelChild::RecvOnServerClose(const uint16_t& aCode,
                                          const nsCString& aReason)
 {
   if (mEventQ->ShouldEnqueue()) {
     mEventQ->Enqueue(new ServerCloseEvent(this, aCode, aReason));
+  } else if (mTargetThread) {
+    DispatchToTargetThread(new ServerCloseEvent(this, aCode, aReason));
   } else {
     OnServerClose(aCode, aReason);
   }
   return true;
 }
 
 void
 WebSocketChannelChild::OnServerClose(const uint16_t& aCode,
@@ -317,16 +360,17 @@ WebSocketChannelChild::OnServerClose(con
 NS_IMETHODIMP
 WebSocketChannelChild::AsyncOpen(nsIURI *aURI,
                                  const nsACString &aOrigin,
                                  nsIWebSocketListener *aListener,
                                  nsISupports *aContext)
 {
   LOG(("WebSocketChannelChild::AsyncOpen() %p\n", this));
 
+  NS_ABORT_IF_FALSE(NS_IsMainThread(), "not main thread");
   NS_ABORT_IF_FALSE(aURI && aListener && !mListener, 
                     "Invalid state for WebSocketChannelChild::AsyncOpen");
 
   mozilla::dom::TabChild* tabChild = nullptr;
   nsCOMPtr<nsITabChild> iTabChild;
   NS_QueryNotificationCallbacks(mCallbacks, mLoadGroup,
                                 NS_GET_IID(nsITabChild),
                                 getter_AddRefs(iTabChild));
@@ -355,64 +399,182 @@ WebSocketChannelChild::AsyncOpen(nsIURI 
   mListener = aListener;
   mContext = aContext;
   mOrigin = aOrigin;
   mWasOpened = 1;
 
   return NS_OK;
 }
 
+class CloseEvent : public nsRunnable
+{
+public:
+  CloseEvent(WebSocketChannelChild *aChild,
+             uint16_t aCode,
+             const nsACString& aReason)
+    : mChild(aChild)
+    , mCode(aCode)
+    , mReason(aReason)
+  {
+    MOZ_RELEASE_ASSERT(!NS_IsMainThread());
+    MOZ_ASSERT(aChild);
+  }
+  NS_IMETHOD Run()
+  {
+    MOZ_RELEASE_ASSERT(NS_IsMainThread());
+    mChild->Close(mCode, mReason);
+    return NS_OK;
+  }
+private:
+  nsRefPtr<WebSocketChannelChild> mChild;
+  uint16_t                        mCode;
+  nsCString                       mReason;
+};
+
 NS_IMETHODIMP
 WebSocketChannelChild::Close(uint16_t code, const nsACString & reason)
 {
+  if (!NS_IsMainThread()) {
+    MOZ_RELEASE_ASSERT(NS_GetCurrentThread() == mTargetThread);
+    return NS_DispatchToMainThread(new CloseEvent(this, code, reason));
+  }
   LOG(("WebSocketChannelChild::Close() %p\n", this));
 
   if (!mIPCOpen || !SendClose(code, nsCString(reason)))
     return NS_ERROR_UNEXPECTED;
   return NS_OK;
 }
 
+class MsgEvent : public nsRunnable
+{
+public:
+  MsgEvent(WebSocketChannelChild *aChild,
+           const nsACString &aMsg,
+           bool aBinaryMsg)
+    : mChild(aChild)
+    , mMsg(aMsg)
+    , mBinaryMsg(aBinaryMsg)
+  {
+    MOZ_RELEASE_ASSERT(!NS_IsMainThread());
+    MOZ_ASSERT(aChild);
+  }
+  NS_IMETHOD Run()
+  {
+    MOZ_RELEASE_ASSERT(NS_IsMainThread());
+    if (mBinaryMsg) {
+      mChild->SendBinaryMsg(mMsg);
+    } else {
+      mChild->SendMsg(mMsg);
+    }
+    return NS_OK;
+  }
+private:
+  nsRefPtr<WebSocketChannelChild> mChild;
+  nsCString                       mMsg;
+  bool                            mBinaryMsg;
+};
+
 NS_IMETHODIMP
 WebSocketChannelChild::SendMsg(const nsACString &aMsg)
 {
+  if (!NS_IsMainThread()) {
+    MOZ_RELEASE_ASSERT(NS_GetCurrentThread() == mTargetThread);
+    return NS_DispatchToMainThread(new MsgEvent(this, aMsg, false));
+  }
   LOG(("WebSocketChannelChild::SendMsg() %p\n", this));
 
   if (!mIPCOpen || !SendSendMsg(nsCString(aMsg)))
     return NS_ERROR_UNEXPECTED;
   return NS_OK;
 }
 
 NS_IMETHODIMP
 WebSocketChannelChild::SendBinaryMsg(const nsACString &aMsg)
 {
+  if (!NS_IsMainThread()) {
+    MOZ_RELEASE_ASSERT(NS_GetCurrentThread() == mTargetThread);
+    return NS_DispatchToMainThread(new MsgEvent(this, aMsg, true));
+  }
   LOG(("WebSocketChannelChild::SendBinaryMsg() %p\n", this));
 
   if (!mIPCOpen || !SendSendBinaryMsg(nsCString(aMsg)))
     return NS_ERROR_UNEXPECTED;
   return NS_OK;
 }
 
+class BinaryStreamEvent : public nsRunnable
+{
+public:
+  BinaryStreamEvent(WebSocketChannelChild *aChild,
+                    OptionalInputStreamParams *aStream,
+                    uint32_t aLength)
+    : mChild(aChild)
+    , mStream(aStream)
+    , mLength(aLength)
+  {
+    MOZ_RELEASE_ASSERT(!NS_IsMainThread());
+    MOZ_ASSERT(aChild);
+  }
+  NS_IMETHOD Run()
+  {
+    MOZ_ASSERT(NS_IsMainThread());
+    mChild->SendBinaryStream(mStream, mLength);
+    return NS_OK;
+  }
+private:
+  nsRefPtr<WebSocketChannelChild>      mChild;
+  nsAutoPtr<OptionalInputStreamParams> mStream;
+  uint32_t                             mLength;
+};
+
 NS_IMETHODIMP
 WebSocketChannelChild::SendBinaryStream(nsIInputStream *aStream,
                                         uint32_t aLength)
 {
+  OptionalInputStreamParams *stream = new OptionalInputStreamParams();
+  nsTArray<mozilla::ipc::FileDescriptor> fds;
+  SerializeInputStream(aStream, *stream, fds);
+
+  MOZ_ASSERT(fds.IsEmpty());
+
+  if (!NS_IsMainThread()) {
+    MOZ_RELEASE_ASSERT(NS_GetCurrentThread() == mTargetThread);
+    return NS_DispatchToMainThread(new BinaryStreamEvent(this, stream, aLength),
+                                   NS_DISPATCH_NORMAL);
+  }
+  return SendBinaryStream(stream, aLength);
+}
+
+nsresult
+WebSocketChannelChild::SendBinaryStream(OptionalInputStreamParams *aStream,
+                                        uint32_t aLength)
+{
   LOG(("WebSocketChannelChild::SendBinaryStream() %p\n", this));
 
-  OptionalInputStreamParams stream;
-  nsTArray<mozilla::ipc::FileDescriptor> fds;
-  SerializeInputStream(aStream, stream, fds);
+  nsAutoPtr<OptionalInputStreamParams> stream(aStream);
 
-  MOZ_ASSERT(fds.IsEmpty());
-
-  if (!mIPCOpen || !SendSendBinaryStream(stream, aLength))
+  if (!mIPCOpen || !SendSendBinaryStream(*stream, aLength))
     return NS_ERROR_UNEXPECTED;
   return NS_OK;
 }
 
 NS_IMETHODIMP
 WebSocketChannelChild::GetSecurityInfo(nsISupports **aSecurityInfo)
 {
   LOG(("WebSocketChannelChild::GetSecurityInfo() %p\n", this));
   return NS_ERROR_NOT_AVAILABLE;
 }
 
+//-----------------------------------------------------------------------------
+// WebSocketChannelChild::nsIThreadRetargetableRequest
+//-----------------------------------------------------------------------------
+
+NS_IMETHODIMP
+WebSocketChannelChild::RetargetDeliveryTo(nsIEventTarget* aTargetThread)
+{
+  nsresult rv = BaseWebSocketChannel::RetargetDeliveryTo(aTargetThread);
+  MOZ_RELEASE_ASSERT(NS_SUCCEEDED(rv));
+
+  return mEventQ->RetargetDeliveryTo(aTargetThread);
+}
+
 } // namespace net
 } // namespace mozilla
--- a/netwerk/protocol/websocket/WebSocketChannelChild.h
+++ b/netwerk/protocol/websocket/WebSocketChannelChild.h
@@ -9,35 +9,38 @@
 
 #include "mozilla/net/PWebSocketChild.h"
 #include "mozilla/net/BaseWebSocketChannel.h"
 #include "nsString.h"
 
 namespace mozilla {
 namespace net {
 
+class ChannelEvent;
 class ChannelEventQueue;
 
 class WebSocketChannelChild : public BaseWebSocketChannel,
                               public PWebSocketChild
 {
  public:
   WebSocketChannelChild(bool aSecure);
   ~WebSocketChannelChild();
 
   NS_DECL_ISUPPORTS
+  NS_DECL_NSITHREADRETARGETABLEREQUEST
 
   // nsIWebSocketChannel methods BaseWebSocketChannel didn't implement for us
   //
   NS_IMETHOD AsyncOpen(nsIURI *aURI, const nsACString &aOrigin,
                        nsIWebSocketListener *aListener, nsISupports *aContext);
   NS_IMETHOD Close(uint16_t code, const nsACString & reason);
   NS_IMETHOD SendMsg(const nsACString &aMsg);
   NS_IMETHOD SendBinaryMsg(const nsACString &aMsg);
   NS_IMETHOD SendBinaryStream(nsIInputStream *aStream, uint32_t aLength);
+  nsresult SendBinaryStream(OptionalInputStreamParams *aStream, uint32_t aLength);
   NS_IMETHOD GetSecurityInfo(nsISupports **aSecurityInfo);
 
   void AddIPDLReference();
   void ReleaseIPDLReference();
 
  private:
   bool RecvOnStart(const nsCString& aProtocol, const nsCString& aExtensions) MOZ_OVERRIDE;
   bool RecvOnStop(const nsresult& aStatusCode) MOZ_OVERRIDE;
@@ -49,16 +52,18 @@ class WebSocketChannelChild : public Bas
   void OnStart(const nsCString& aProtocol, const nsCString& aExtensions);
   void OnStop(const nsresult& aStatusCode);
   void OnMessageAvailable(const nsCString& aMsg);
   void OnBinaryMessageAvailable(const nsCString& aMsg);
   void OnAcknowledge(const uint32_t& aSize);
   void OnServerClose(const uint16_t& aCode, const nsCString& aReason);
   void AsyncOpenFailed();  
 
+  void DispatchToTargetThread(ChannelEvent *aChannelEvent);
+
   nsRefPtr<ChannelEventQueue> mEventQ;
   bool mIPCOpen;
 
   friend class StartEvent;
   friend class StopEvent;
   friend class MessageEvent;
   friend class AcknowledgeEvent;
   friend class ServerCloseEvent;