Fixing tsan races (#1307)

test/integration/fake_upstream.cc

@@ -281,6 +281,7 @@ FakeHttpConnectionPtr FakeUpstream::waitForHttpConnection(Event::Dispatcher& cli
   ASSERT(!new_connections_.empty());
   FakeHttpConnectionPtr connection(
       new FakeHttpConnection(std::move(new_connections_.front()), stats_store_, http_type_));
+  connection->initialize();
   new_connections_.pop_front();
   connection->readDisable(false);
   return connection;
@@ -295,6 +296,7 @@ FakeRawConnectionPtr FakeUpstream::waitForRawConnection() {
 
   ASSERT(!new_connections_.empty());
   FakeRawConnectionPtr connection(new FakeRawConnection(std::move(new_connections_.front())));
+  connection->initialize();
   new_connections_.pop_front();
   connection->readDisable(false);
   return connection;

test/integration/fake_upstream.h

@@ -85,11 +85,15 @@ class QueuedConnectionWrapper : public Network::ConnectionCallbacks {
       : connection_(connection), parented_(false) {
     connection_.addConnectionCallbacks(*this);
   }
-  void set_parented() { parented_ = true; }
+  void set_parented() {
+    std::unique_lock<std::mutex> lock(lock_);
+    parented_ = true;
+  }
   Network::Connection& connection() const { return connection_; }
 
   // Network::ConnectionCallbacks
   void onEvent(uint32_t events) override {
+    std::unique_lock<std::mutex> lock(lock_);
     RELEASE_ASSERT(parented_ || (!(events & Network::ConnectionEvent::RemoteClose) &&
                                  !(events & Network::ConnectionEvent::LocalClose)));
   }
@@ -99,6 +103,7 @@ class QueuedConnectionWrapper : public Network::ConnectionCallbacks {
 private:
   Network::Connection& connection_;
   bool parented_;
+  std::mutex lock_;
 };
 
 typedef std::unique_ptr<QueuedConnectionWrapper> QueuedConnectionWrapperPtr;
@@ -108,6 +113,7 @@ typedef std::unique_ptr<QueuedConnectionWrapper> QueuedConnectionWrapperPtr;
  */
 class FakeConnectionBase : public Network::ConnectionCallbacks {
 public:
+  ~FakeConnectionBase() { ASSERT(initialized_); }
   void close();
   void readDisable(bool disable);
   // By default waitForDisconnect assumes the next event is a disconnect and
@@ -120,18 +126,22 @@ class FakeConnectionBase : public Network::ConnectionCallbacks {
   void onAboveWriteBufferHighWatermark() override {}
   void onBelowWriteBufferLowWatermark() override {}
 
-protected:
-  FakeConnectionBase(QueuedConnectionWrapperPtr connection_wrapper)
-      : connection_(connection_wrapper->connection()),
-        connection_wrapper_(std::move(connection_wrapper)) {
+  void initialize() {
+    initialized_ = true;
     connection_wrapper_->set_parented();
     connection_.dispatcher().post([this]() -> void { connection_.addConnectionCallbacks(*this); });
   }
 
+protected:
+  FakeConnectionBase(QueuedConnectionWrapperPtr connection_wrapper)
+      : connection_(connection_wrapper->connection()),
+        connection_wrapper_(std::move(connection_wrapper)) {}
+
   Network::Connection& connection_;
   std::mutex lock_;
   std::condition_variable connection_event_;
   bool disconnected_{};
+  bool initialized_{false};
 
 private:
   // We hold on to this as connection callbacks live for the entire life of the