Bug 1894419: Various gtest TSan fixes. r=mjf

Differential Revision: https://phabricator.services.mozilla.com/D210979

dom/media/webrtc/transport/test/gtest_ringbuffer_dumper.h

@@ -47,12 +47,14 @@ class RingbufferDumper : public ::testing::EmptyTestEventListener {
   }
 
   virtual void OnTestStart(const ::testing::TestInfo& testInfo) override {
+    running_ = true;
     mozilla::SyncRunnable::DispatchToThread(
         test_utils_->sts_target(),
         WrapRunnable(this, &RingbufferDumper::ClearRingBuffer_s));
   }
 
   virtual void OnTestEnd(const ::testing::TestInfo& testInfo) override {
+    running_ = false;
     mozilla::SyncRunnable::DispatchToThread(
         test_utils_->sts_target(),
         WrapRunnable(this, &RingbufferDumper::DestroyRingBuffer_s));
@@ -62,6 +64,13 @@ class RingbufferDumper : public ::testing::EmptyTestEventListener {
   virtual void OnTestPartResult(
       const ::testing::TestPartResult& testResult) override {
     if (testResult.failed()) {
+      if (!running_) {
+        // Why does gtest sometimes call this without ever calling OnTestStart?
+        running_ = true;
+        mozilla::SyncRunnable::DispatchToThread(
+            test_utils_->sts_target(),
+            WrapRunnable(this, &RingbufferDumper::ClearRingBuffer_s));
+      }
       // Dump (and empty) the RLogConnector
       mozilla::SyncRunnable::DispatchToThread(
           test_utils_->sts_target(),
@@ -71,6 +80,7 @@ class RingbufferDumper : public ::testing::EmptyTestEventListener {
 
  private:
   MtransportTestUtils* test_utils_;
+  bool running_ = false;
 };
 
 }  // namespace test

dom/media/webrtc/transport/test/ice_unittest.cpp

@@ -19,6 +19,7 @@
 #include "logging.h"
 #include "ssl.h"
 
+#include "mozilla/DataMutex.h"
 #include "mozilla/Preferences.h"
 #include "nsThreadUtils.h"
 #include "nsXPCOM.h"
@@ -128,39 +129,31 @@ class StunTest : public MtransportTest {
       ASSERT_TRUE(!stun_server_address_.empty());
     }
 
-    // Make sure NrIceCtx is in a testable state.
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&NrIceCtx::internal_DeinitializeGlobal));
+    test_utils_->SyncDispatchToSTS(WrapRunnable(this, &StunTest::SetUp_s));
+  }
 
-    // NB: NrIceCtx::internal_DeinitializeGlobal destroys the RLogConnector
-    // singleton.
+  void SetUp_s() {
+    // Make sure NrIceCtx is in a testable state.
+    NrIceCtx::internal_DeinitializeGlobal();
     RLogConnector::CreateInstance();
 
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunServer::GetInstance, AF_INET));
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunServer::GetInstance, AF_INET6));
+    TestStunServer::GetInstance(AF_INET);
+    TestStunServer::GetInstance(AF_INET6);
 
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunTcpServer::GetInstance, AF_INET));
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunTcpServer::GetInstance, AF_INET6));
+    TestStunTcpServer::GetInstance(AF_INET);
+    TestStunTcpServer::GetInstance(AF_INET6);
   }
 
   void TearDown() override {
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&NrIceCtx::internal_DeinitializeGlobal));
-
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunServer::ShutdownInstance));
-
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunTcpServer::ShutdownInstance));
-
-    RLogConnector::DestroyInstance();
-
+    test_utils_->SyncDispatchToSTS(WrapRunnable(this, &StunTest::TearDown_s));
     MtransportTest::TearDown();
   }
+
+  void TearDown_s() {
+    NrIceCtx::internal_DeinitializeGlobal();
+    TestStunServer::ShutdownInstance();
+    TestStunTcpServer::ShutdownInstance();
+  }
 };
 
 enum TrickleMode { TRICKLE_NONE, TRICKLE_SIMULATE, TRICKLE_REAL };
@@ -357,6 +350,8 @@ class IceTestPeer : public sigslot::has_slots<> {
         offerer_(offerer),
         stream_counter_(0),
         shutting_down_(false),
+        mConnectionStates("IceTestPeer::mConnectionStates"),
+        mGatheringStates("IceTestPeer::mGatheringStates"),
         ready_ct_(0),
         ice_reached_checking_(false),
         received_(0),
@@ -425,8 +420,14 @@ class IceTestPeer : public sigslot::has_slots<> {
     stream->SignalPacketReceived.connect(this, &IceTestPeer::PacketReceived);
     stream->SignalGatheringStateChange.connect(
         this, &IceTestPeer::GatheringStateChange);
-    mConnectionStates[id] = NrIceCtx::ICE_CTX_INIT;
-    mGatheringStates[id] = NrIceMediaStream::ICE_STREAM_GATHER_INIT;
+    {
+      auto lock = mConnectionStates.Lock();
+      lock.ref()[id] = NrIceCtx::ICE_CTX_INIT;
+    }
+    {
+      auto lock = mGatheringStates.Lock();
+      lock.ref()[id] = NrIceMediaStream::ICE_STREAM_GATHER_INIT;
+    }
   }
 
   void AddStream(int components) {
@@ -437,8 +438,14 @@ class IceTestPeer : public sigslot::has_slots<> {
   void RemoveStream_s(size_t index) {
     const std::string id = MakeTransportId(index);
     ice_ctx_->DestroyStream(id);
-    mConnectionStates.erase(id);
-    mGatheringStates.erase(id);
+    {
+      auto lock = mConnectionStates.Lock();
+      lock->erase(id);
+    }
+    {
+      auto lock = mGatheringStates.Lock();
+      lock->erase(id);
+    }
   }
 
   void RemoveStream(size_t index) {
@@ -537,30 +544,59 @@ class IceTestPeer : public sigslot::has_slots<> {
     ice_ctx_->SetStunAddrs(addrs);
   }
 
-  void UseNat() { nat_->enabled_ = true; }
+  void UseNat() {
+    test_utils_->SyncDispatchToSTS(WrapRunnable(this, &IceTestPeer::UseNat_s));
+  }
+
+  void UseNat_s() { nat_->enabled_ = true; }
 
   void SetTimerDivider(int div) { ice_ctx_->internal_SetTimerAccelarator(div); }
 
   void SetStunResponseDelay(uint32_t delay) {
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &IceTestPeer::SetStunResponseDelay_s, delay));
+  }
+
+  void SetStunResponseDelay_s(uint32_t delay) {
     nat_->delay_stun_resp_ms_ = delay;
   }
 
   void SetFilteringType(TestNat::NatBehavior type) {
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &IceTestPeer::SetFilteringType_s, type));
+  }
+
+  void SetFilteringType_s(TestNat::NatBehavior type) {
     MOZ_ASSERT(!nat_->has_port_mappings());
     nat_->filtering_type_ = type;
   }
 
   void SetMappingType(TestNat::NatBehavior type) {
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &IceTestPeer::SetMappingType_s, type));
+  }
+
+  void SetMappingType_s(TestNat::NatBehavior type) {
     MOZ_ASSERT(!nat_->has_port_mappings());
     nat_->mapping_type_ = type;
   }
 
   void SetBlockUdp(bool block) {
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &IceTestPeer::SetBlockUdp_s, block));
+  }
+
+  void SetBlockUdp_s(bool block) {
     MOZ_ASSERT(!nat_->has_port_mappings());
     nat_->block_udp_ = block;
   }
 
-  void SetBlockStun(bool block) { nat_->block_stun_ = block; }
+  void SetBlockStun(bool block) {
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &IceTestPeer::SetBlockStun_s, block));
+  }
+
+  void SetBlockStun_s(bool block) { nat_->block_stun_ = block; }
 
   // Get various pieces of state
   std::vector<std::string> GetGlobalAttributes() {
@@ -655,7 +691,8 @@ class IceTestPeer : public sigslot::has_slots<> {
   }
 
   bool gathering_complete() {
-    for (const auto& [id, state] : mGatheringStates) {
+    auto lock = mGatheringStates.Lock();
+    for (const auto& [id, state] : lock.ref()) {
       Unused << id;
       if (state != NrIceMediaStream::ICE_STREAM_GATHER_COMPLETE) {
         return false;
@@ -679,7 +716,8 @@ class IceTestPeer : public sigslot::has_slots<> {
     return result;
   }
   bool ice_connected() {
-    for (const auto& [id, state] : mConnectionStates) {
+    auto lock = mConnectionStates.Lock();
+    for (const auto& [id, state] : lock.ref()) {
       if (state != NrIceCtx::ICE_CTX_CONNECTED) {
         return false;
       }
@@ -687,7 +725,8 @@ class IceTestPeer : public sigslot::has_slots<> {
     return true;
   }
   bool ice_failed() {
-    for (const auto& [id, state] : mConnectionStates) {
+    auto lock = mConnectionStates.Lock();
+    for (const auto& [id, state] : lock.ref()) {
       if (state == NrIceCtx::ICE_CTX_FAILED) {
         return true;
       }
@@ -698,7 +737,8 @@ class IceTestPeer : public sigslot::has_slots<> {
     if (ice_failed() || ice_connected()) {
       return false;
     }
-    for (const auto& [id, state] : mConnectionStates) {
+    auto lock = mConnectionStates.Lock();
+    for (const auto& [id, state] : lock.ref()) {
       if (state == NrIceCtx::ICE_CTX_CHECKING) {
         return true;
       }
@@ -717,9 +757,14 @@ class IceTestPeer : public sigslot::has_slots<> {
   void RestartIce_s() {
     for (auto& stream : ice_ctx_->GetStreams()) {
       SetIceCredentials_s(*stream);
-      mConnectionStates[stream->GetId()] = NrIceCtx::ICE_CTX_INIT;
-      mGatheringStates[stream->GetId()] =
-          NrIceMediaStream::ICE_STREAM_GATHER_INIT;
+      {
+        auto lock = mConnectionStates.Lock();
+        lock.ref()[stream->GetId()] = NrIceCtx::ICE_CTX_INIT;
+      }
+      {
+        auto lock = mGatheringStates.Lock();
+        lock.ref()[stream->GetId()] = NrIceMediaStream::ICE_STREAM_GATHER_INIT;
+      }
     }
     // take care of some local bookkeeping
     ready_ct_ = 0;
@@ -985,7 +1030,11 @@ class IceTestPeer : public sigslot::has_slots<> {
     if (shutting_down_) {
       return;
     }
-    mGatheringStates[aTransportId] = state;
+
+    {
+      auto lock = mGatheringStates.Lock();
+      lock.ref()[aTransportId] = state;
+    }
 
     if (!gathering_complete()) {
       return;
@@ -1177,7 +1226,10 @@ class IceTestPeer : public sigslot::has_slots<> {
 
   void ConnectionStateChange(NrIceMediaStream* stream,
                              NrIceCtx::ConnectionState state) {
-    mConnectionStates[stream->GetId()] = state;
+    {
+      auto lock = mConnectionStates.Lock();
+      lock.ref()[stream->GetId()] = state;
+    }
     if (ice_checking()) {
       ice_reached_checking_ = true;
     }
@@ -1381,12 +1433,13 @@ class IceTestPeer : public sigslot::has_slots<> {
   std::map<std::string, std::pair<std::string, std::string>> mOldIceCredentials;
   size_t stream_counter_;
   bool shutting_down_;
-  std::map<std::string, NrIceCtx::ConnectionState> mConnectionStates;
-  std::map<std::string, NrIceMediaStream::GatheringState> mGatheringStates;
-  int ready_ct_;
+  DataMutex<std::map<std::string, NrIceCtx::ConnectionState>> mConnectionStates;
+  DataMutex<std::map<std::string, NrIceMediaStream::GatheringState>>
+      mGatheringStates;
+  std::atomic<int> ready_ct_;
   bool ice_reached_checking_;
-  size_t received_;
-  size_t sent_;
+  std::atomic<size_t> received_;
+  std::atomic<size_t> sent_;
   struct timeval consent_timestamp_;
   NrIceResolverFake fake_resolver_;
   RefPtr<NrIceResolver> dns_resolver_;
@@ -1950,10 +2003,8 @@ class WebRtcIcePacketFilterTest : public StunTest {
   void SetUp() {
     StunTest::SetUp();
 
-    NrIceCtx::InitializeGlobals(NrIceCtx::GlobalConfig());
-
-    // Set up enough of the ICE ctx to allow the packet filter to work
-    ice_ctx_ = NrIceCtx::Create("test");
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &WebRtcIcePacketFilterTest::SetUp_s));
 
     nsCOMPtr<nsISocketFilterHandler> udp_handler =
         do_GetService(NS_STUN_UDP_SOCKET_FILTER_HANDLER_CONTRACTID);
@@ -1966,6 +2017,12 @@ class WebRtcIcePacketFilterTest : public StunTest {
     tcp_handler->NewFilter(getter_AddRefs(tcp_filter_));
   }
 
+  void SetUp_s() {
+    NrIceCtx::InitializeGlobals(NrIceCtx::GlobalConfig());
+    // Set up enough of the ICE ctx to allow the packet filter to work
+    ice_ctx_ = NrIceCtx::Create("test");
+  }
+
   void TearDown() {
     test_utils_->SyncDispatchToSTS(
         WrapRunnable(this, &WebRtcIcePacketFilterTest::TearDown_s));

dom/media/webrtc/transport/test/multi_tcp_socket_unittest.cpp

@@ -37,13 +37,15 @@ class MultiTcpSocketTest : public MtransportTest {
   void SetUp() {
     MtransportTest::SetUp();
 
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &MultiTcpSocketTest::SetUp_s));
+  }
+
+  void SetUp_s() {
     NrIceCtx::InitializeGlobals(NrIceCtx::GlobalConfig());
     ice_ctx_ = NrIceCtx::Create("stun");
-
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunTcpServer::GetInstance, AF_INET));
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableNM(&TestStunTcpServer::GetInstance, AF_INET6));
+    TestStunTcpServer::GetInstance(AF_INET);
+    TestStunTcpServer::GetInstance(AF_INET6);
   }
 
   void TearDown() {

dom/media/webrtc/transport/test/nrappkit_unittest.cpp

@@ -95,7 +95,7 @@ class TimerTest : public MtransportTest {
 
  protected:
   void* handle_;
-  bool fired_;
+  std::atomic<bool> fired_;
 };
 }  // namespace
 

dom/media/webrtc/transport/test/proxy_tunnel_socket_unittest.cpp

@@ -33,6 +33,8 @@ class NrTcpSocketTest : public MtransportTest {
         mConnected(false) {}
 
   void SetUp() override {
+    MtransportTest::SetUp();
+
     mSProxy = new NrTcpSocket(nullptr);
     int r = nr_socket_create_int((void*)mSProxy.get(), mSProxy->vtbl(),
                                  &nr_socket_);
@@ -42,7 +44,10 @@ class NrTcpSocketTest : public MtransportTest {
     mSProxy->AssignChannel_DoNotUse(new WebrtcTCPSocketWrapper(nullptr));
   }
 
-  void TearDown() override { mSProxy->close(); }
+  void TearDown() override {
+    mSProxy->close();
+    MtransportTest::TearDown();
+  }
 
   static void readable_cb(NR_SOCKET s, int how, void* cb_arg) {
     NrTcpSocketTest* test = (NrTcpSocketTest*)cb_arg;

dom/media/webrtc/transport/test/sctp_unittest.cpp

@@ -120,8 +120,7 @@ class TransportTestPeer : public sigslot::has_slots<> {
     usrsctp_deregister_address(static_cast<void*>(this));
 
     test_utils_->SyncDispatchToSTS(
-        WrapRunnable(this, &TransportTestPeer::Disconnect_s));
-
+        WrapRunnable(this, &TransportTestPeer::DeleteFlow_s));
     std::cerr << "~TransportTestPeer() completed" << std::endl;
   }
 
@@ -153,14 +152,21 @@ class TransportTestPeer : public sigslot::has_slots<> {
     ASSERT_GE(0, r);
   }
 
-  void Disconnect_s() {
-    disconnect_all();
+  void DeleteFlow_s() {
     if (flow_) {
       flow_ = nullptr;
     }
   }
 
-  void Disconnect() { loopback_->Disconnect(); }
+  void Disconnect_s() {
+    loopback_->Disconnect();
+    disconnect_all();
+  }
+
+  void Disconnect() {
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &TransportTestPeer::Disconnect_s));
+  }
 
   void StartTransfer(size_t to_send) {
     periodic_ = new SendPeriodic(this, to_send);
@@ -193,7 +199,7 @@ class TransportTestPeer : public sigslot::has_slots<> {
   bool connected() const { return connected_; }
 
   static TransportResult SendPacket_s(UniquePtr<MediaPacket> packet,
-                                      const RefPtr<TransportFlow>& flow,
+                                      RefPtr<TransportFlow> flow,
                                       TransportLayer* layer) {
     return layer->SendPacket(*packet);
   }
@@ -278,9 +284,9 @@ class TransportTestPeer : public sigslot::has_slots<> {
 
  private:
   std::string name_;
-  bool connected_;
-  size_t sent_;
-  size_t received_;
+  std::atomic<bool> connected_;
+  std::atomic<size_t> sent_;
+  std::atomic<size_t> received_;
   // Owns the TransportLayerLoopback, but basically does nothing else.
   RefPtr<TransportFlow> flow_;
   TransportLayerLoopback* loopback_;

dom/media/webrtc/transport/test/sockettransportservice_unittest.cpp

@@ -52,10 +52,10 @@ class SocketTransportServiceTest : public MtransportTest {
  private:
   nsCOMPtr<nsISocketTransportService> stservice_;
   nsCOMPtr<nsIEventTarget> target_;
-  size_t received_;
+  std::atomic<size_t> received_;
   PRFileDesc* readpipe_;
   PRFileDesc* writepipe_;
-  bool registered_;
+  std::atomic<bool> registered_;
 };
 
 // Received an event.

dom/media/webrtc/transport/test/test_nr_socket_ice_unittest.cpp

@@ -301,9 +301,12 @@ class TestNrSocketIceUnitTest : public ::testing::Test {
     test_utils_ = new MtransportTestUtils();
     test_utils2_ = new MtransportTestUtils();
 
-    NrIceCtx::InitializeGlobals(NrIceCtx::GlobalConfig());
+    test_utils_->SyncDispatchToSTS(
+        WrapRunnable(this, &TestNrSocketIceUnitTest::SetUp_s));
   }
 
+  void SetUp_s() { NrIceCtx::InitializeGlobals(NrIceCtx::GlobalConfig()); }
+
   void TearDown() override {
     delete test_utils_;
     delete test_utils2_;

dom/media/webrtc/transport/test/transport_unittests.cpp

@@ -651,31 +651,30 @@ class TransportTestPeer : public sigslot::has_slots<> {
   // Gathering complete, so send our candidates and start
   // connecting on the other peer.
   void GatheringComplete() {
-    nsresult res;
-
     // Don't send to the other side
     if (!peer_) {
       gathering_complete_ = true;
       return;
     }
 
-    // First send attributes
     test_utils_->SyncDispatchToSTS(
-        WrapRunnableRet(&res, peer_->ice_ctx_, &NrIceCtx::ParseGlobalAttributes,
-                        ice_ctx_->GetGlobalAttributes()));
+        WrapRunnable(this, &TransportTestPeer::GatheringComplete_s));
+  }
+
+  void GatheringComplete_s() {
+    // First send attributes
+    nsresult res =
+        peer_->ice_ctx_->ParseGlobalAttributes(ice_ctx_->GetGlobalAttributes());
     ASSERT_TRUE(NS_SUCCEEDED(res));
 
     for (size_t i = 0; i < streams_.size(); ++i) {
-      test_utils_->SyncDispatchToSTS(WrapRunnableRet(
-          &res, peer_->streams_[i], &NrIceMediaStream::ConnectToPeer, "ufrag",
-          "pass", streams_[i]->GetAttributes()));
-
+      res = peer_->streams_[i]->ConnectToPeer("ufrag", "pass",
+                                              streams_[i]->GetAttributes());
       ASSERT_TRUE(NS_SUCCEEDED(res));
     }
 
     // Start checks on the other peer.
-    test_utils_->SyncDispatchToSTS(
-        WrapRunnableRet(&res, peer_->ice_ctx_, &NrIceCtx::StartChecks));
+    res = peer_->ice_ctx_->StartChecks();
     ASSERT_TRUE(NS_SUCCEEDED(res));
   }
 
@@ -778,8 +777,8 @@ class TransportTestPeer : public sigslot::has_slots<> {
   std::string name_;
   bool offerer_;
   nsCOMPtr<nsIEventTarget> target_;
-  size_t received_packets_;
-  size_t received_bytes_;
+  std::atomic<size_t> received_packets_;
+  std::atomic<size_t> received_bytes_;
   RefPtr<TransportFlow> flow_;
   TransportLayerLoopback* loopback_;
   TransportLayerLogging* logging_;

dom/media/webrtc/transport/test/turn_unittest.cpp

@@ -135,6 +135,7 @@ class TurnClient : public MtransportTest {
 
   void TearDown() {
     test_utils_->SyncDispatchToSTS(WrapRunnable(this, &TurnClient::TearDown_s));
+    MtransportTest::TearDown();
   }
 
   void Allocate_s() {

dom/media/webrtc/transport/test/webrtcproxychannel_unittest.cpp

@@ -282,7 +282,9 @@ class WebrtcTCPSocketTestInputStream : public nsIAsyncInputStream,
   NS_DECL_NSIINPUTSTREAM
 
   WebrtcTCPSocketTestInputStream()
-      : mMaxReadSize(1024 * 1024), mAllowCallbacks(false) {}
+      : mMaxReadSize(1024 * 1024),
+        mMutex("WebrtcTCPSocketTestInputStream::mMutex"),
+        mAllowCallbacks(false) {}
 
   void DoCallback();
   void CallCallback(const nsCOMPtr<nsIInputStreamCallback>& aCallback);
@@ -294,6 +296,7 @@ class WebrtcTCPSocketTestInputStream : public nsIAsyncInputStream,
   virtual ~WebrtcTCPSocketTestInputStream() = default;
 
  private:
+  mutable Mutex mMutex;
   nsCOMPtr<nsIInputStreamCallback> mCallback;
   nsCOMPtr<nsIEventTarget> mCallbackTarget;
 
@@ -308,8 +311,11 @@ nsresult WebrtcTCPSocketTestInputStream::AsyncWait(
     uint32_t aRequestedCount, nsIEventTarget* aEventTarget) {
   MOZ_ASSERT(!aEventTarget, "no event target should be set");
 
-  mCallback = aCallback;
-  mCallbackTarget = NS_GetCurrentThread();
+  {
+    MutexAutoLock lock(mMutex);
+    mCallback = aCallback;
+    mCallbackTarget = NS_GetCurrentThread();
+  }
 
   if (mAllowCallbacks && DataLength() > 0) {
     DoCallback();
@@ -362,6 +368,7 @@ void WebrtcTCPSocketTestInputStream::CallCallback(
 }
 
 void WebrtcTCPSocketTestInputStream::DoCallback() {
+  MutexAutoLock lock(mMutex);
   if (mCallback) {
     mCallbackTarget->Dispatch(
         NewRunnableMethod<const nsCOMPtr<nsIInputStreamCallback>&>(
@@ -380,7 +387,9 @@ class WebrtcTCPSocketTestOutputStream : public nsIAsyncOutputStream,
   NS_DECL_NSIASYNCOUTPUTSTREAM
   NS_DECL_NSIOUTPUTSTREAM
 
-  WebrtcTCPSocketTestOutputStream() : mMaxWriteSize(1024 * 1024) {}
+  WebrtcTCPSocketTestOutputStream()
+      : mMaxWriteSize(1024 * 1024),
+        mMutex("WebrtcTCPSocketTestOutputStream::mMutex") {}
 
   void DoCallback();
   void CallCallback(const nsCOMPtr<nsIOutputStreamCallback>& aCallback);
@@ -393,6 +402,7 @@ class WebrtcTCPSocketTestOutputStream : public nsIAsyncOutputStream,
   virtual ~WebrtcTCPSocketTestOutputStream() = default;
 
  private:
+  mutable Mutex mMutex;
   nsCOMPtr<nsIOutputStreamCallback> mCallback;
   nsCOMPtr<nsIEventTarget> mCallbackTarget;
 };
@@ -405,8 +415,11 @@ nsresult WebrtcTCPSocketTestOutputStream::AsyncWait(
     uint32_t aRequestedCount, nsIEventTarget* aEventTarget) {
   MOZ_ASSERT(!aEventTarget, "no event target should be set");
 
-  mCallback = aCallback;
-  mCallbackTarget = NS_GetCurrentThread();
+  {
+    MutexAutoLock lock(mMutex);
+    mCallback = aCallback;
+    mCallbackTarget = NS_GetCurrentThread();
+  }
 
   return NS_OK;
 }
@@ -459,6 +472,7 @@ void WebrtcTCPSocketTestOutputStream::CallCallback(
 }
 
 void WebrtcTCPSocketTestOutputStream::DoCallback() {
+  MutexAutoLock lock(mMutex);
   if (mCallback) {
     mCallbackTarget->Dispatch(
         NewRunnableMethod<const nsCOMPtr<nsIOutputStreamCallback>&>(
@@ -520,6 +534,13 @@ class WebrtcTCPSocketTest : public MtransportTest {
 
   void SetUp() override;
   void TearDown() override;
+  size_t CountUnwrittenBytes() {
+    size_t result;
+    test_utils_->SyncDispatchToSTS(WrapRunnableRet(
+        &result, this, &WebrtcTCPSocketTest::CountUnwrittenBytes_s));
+    return result;
+  }
+  size_t CountUnwrittenBytes_s() { return mChannel->CountUnwrittenBytes(); }
 
   void DoTransportAvailable();
 
@@ -534,8 +555,8 @@ class WebrtcTCPSocketTest : public MtransportTest {
   RefPtr<FakeWebrtcTCPSocket> mChannel;
   RefPtr<WebrtcTCPSocketTestCallback> mCallback;
 
-  bool mOnCloseCalled;
-  bool mOnConnectedCalled;
+  std::atomic<bool> mOnCloseCalled;
+  std::atomic<bool> mOnConnectedCalled;
 
   size_t ReadDataLength();
   template <typename T>
@@ -566,6 +587,8 @@ class WebrtcTCPSocketTestCallback : public WebrtcTCPSocketCallback {
 };
 
 void WebrtcTCPSocketTest::SetUp() {
+  MtransportTest::SetUp();
+
   nsresult rv;
   // WebrtcTCPSocket's threading model is the same as mtransport
   // all socket operations are done on the socket thread
@@ -580,7 +603,7 @@ void WebrtcTCPSocketTest::SetUp() {
   mChannel = new FakeWebrtcTCPSocket(mCallback.get());
 }
 
-void WebrtcTCPSocketTest::TearDown() {}
+void WebrtcTCPSocketTest::TearDown() { MtransportTest::TearDown(); }
 
 // WebrtcTCPSocketCallback
 void WebrtcTCPSocketTest::OnRead(nsTArray<uint8_t>&& aReadData) {
@@ -674,7 +697,7 @@ TEST_F(WebrtcTCPSocketTest, Write) {
   data.AppendElements(kReadData, kReadDataLength);
   mChannel->Write(std::move(data));
 
-  ASSERT_TRUE_WAIT(mChannel->CountUnwrittenBytes() == kReadDataLength,
+  ASSERT_TRUE_WAIT(CountUnwrittenBytes() == kReadDataLength,
                    kDefaultTestTimeout);
 
   mOutputStream->DoCallback();
@@ -703,7 +726,7 @@ TEST_F(WebrtcTCPSocketTest, WriteFail) {
   array.AppendElements(kReadData, kReadDataLength);
   mChannel->Write(std::move(array));
 
-  ASSERT_TRUE_WAIT(mChannel->CountUnwrittenBytes() == kReadDataLength,
+  ASSERT_TRUE_WAIT(CountUnwrittenBytes() == kReadDataLength,
                    kDefaultTestTimeout);
 
   mOutputStream->Fail();
@@ -742,8 +765,7 @@ TEST_F(WebrtcTCPSocketTest, WriteLarge) {
     mChannel->Write(std::move(array));
   }
 
-  ASSERT_TRUE_WAIT(mChannel->CountUnwrittenBytes() == data.length(),
-                   kDefaultTestTimeout);
+  ASSERT_TRUE_WAIT(CountUnwrittenBytes() == data.length(), kDefaultTestTimeout);
 
   // make sure writing loops more than once per write request
   mOutputStream->mMaxWriteSize = 1024;