Back out "Revert D69698406: gloo: async lazy connect"

Differential Revision: D72425729

Pull Request resolved: #426

Author: d4l3k

gloo/test/allgather_test.cc

@@ -167,6 +167,11 @@ TEST_F(AllgatherNewTest, TestTimeout) {
     AllgatherOptions opts(context);
     opts.setInput(input.getPointer(), 1);
     opts.setOutput(output.getPointer(), context->size);
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    allgather(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/allgatherv_test.cc

@@ -95,6 +95,11 @@ TEST_F(AllgathervTest, TestTimeout) {
     std::vector<size_t> counts({1, 1});
     AllgathervOptions opts(context);
     opts.setOutput(output.getPointer(), counts);
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    allgatherv(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/allreduce_test.cc

@@ -389,6 +389,11 @@ TEST_F(AllreduceNewTest, TestTimeout) {
     AllreduceOptions opts(context);
     opts.setOutputs(outputs.getPointers(), 1);
     opts.setReduceFunction(getFunction<uint64_t>());
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    allreduce(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/barrier_test.cc

@@ -127,12 +127,18 @@ INSTANTIATE_TEST_CASE_P(
 TEST_F(BarrierNewTest, TestTimeout) {
   spawn(Transport::TCP, 2, [&](std::shared_ptr<Context> context) {
     BarrierOptions opts(context);
+
+    // Run barrier first so we're measuring the barrier timeout not connection
+    // timeout.
+    barrier(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {
         barrier(opts);
         FAIL() << "Expected exception to be thrown";
       } catch (::gloo::IoException& e) {
+        std::cerr << e.what() << std::endl;
         ASSERT_NE(std::string(e.what()).find("Timed out"), std::string::npos);
       }
     }

gloo/test/base_test.cc

@@ -17,6 +17,7 @@ const char* kDefaultDevice = "localhost";
 // Transports that instantiated algorithms can be tested against.
 const std::vector<Transport> kTransportsForClassAlgorithms = {
     Transport::TCP,
+    Transport::TCP_LAZY,
 #if GLOO_HAVE_TRANSPORT_TCP_TLS
     Transport::TCP_TLS,
 #endif
@@ -27,6 +28,7 @@ const std::vector<Transport> kTransportsForClassAlgorithms = {
 // preferred over the instantiated style.
 const std::vector<Transport> kTransportsForFunctionAlgorithms = {
     Transport::TCP,
+    Transport::TCP_LAZY,
 #if GLOO_HAVE_TRANSPORT_TCP_TLS
     Transport::TCP_TLS,
 #endif
@@ -37,6 +39,8 @@ std::shared_ptr<::gloo::transport::Device> createDevice(Transport transport) {
 #if GLOO_HAVE_TRANSPORT_TCP
   if (transport == Transport::TCP) {
     return ::gloo::transport::tcp::CreateDevice(kDefaultDevice);
+  } else if (transport == Transport::TCP_LAZY) {
+    return ::gloo::transport::tcp::CreateLazyDevice(kDefaultDevice);
   }
 #endif
 #if GLOO_HAVE_TRANSPORT_TCP_TLS

gloo/test/base_test.h

@@ -59,6 +59,7 @@ class Barrier {
 
 enum Transport {
   TCP,
+  TCP_LAZY,
 #if GLOO_HAVE_TRANSPORT_TCP_TLS
   TCP_TLS,
 #endif

gloo/test/broadcast_test.cc

@@ -182,6 +182,11 @@ TEST_F(BroadcastTest, TestTimeout) {
     BroadcastOptions opts(context);
     opts.setOutput(output.getPointer(), 1);
     opts.setRoot(0);
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    broadcast(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/gather_test.cc

@@ -76,6 +76,11 @@ TEST_F(GatherTest, TestTimeout) {
     opts.setInput(input.getPointer(), 1);
     opts.setOutput(output.getPointer(), context->size);
     opts.setRoot(0);
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    gather(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/gatherv_test.cc

@@ -106,6 +106,11 @@ TEST_F(GathervTest, TestTimeout) {
     opts.setRoot(0);
     opts.setInput(input.getPointer(), 1);
     opts.setOutput(output.getPointer(), counts);
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    gatherv(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/multiproc_test.h

@@ -102,11 +102,31 @@ class MultiProcWorker {
     auto device = createDevice(transport);
     context->setTimeout(std::chrono::milliseconds(kMultiProcTimeout));
     context->connectFullMesh(store_, device);
+
+    // Wait for all workers to be ready
+    ringBarrier(context);
+
     device.reset();
     sem_post(semaphore_);
     fn(std::move(context));
   }
 
+  void ringBarrier(std::shared_ptr<::gloo::rendezvous::Context>& context) {
+    int sendScratch = 0;
+    int recvScratch = 0;
+    auto sendBuf =
+        context->createUnboundBuffer(&sendScratch, sizeof(sendScratch));
+    auto recvBuf =
+        context->createUnboundBuffer(&recvScratch, sizeof(recvScratch));
+    const auto leftRank = (context->size + context->rank - 1) % context->size;
+    const auto rightRank = (context->rank + 1) % context->size;
+
+    sendBuf->send(leftRank, 0);
+    recvBuf->recv(rightRank, 0);
+    sendBuf->waitSend();
+    recvBuf->waitRecv();
+  }
+
  protected:
   std::shared_ptr<::gloo::rendezvous::Store> store_;
   sem_t* semaphore_;

gloo/test/reduce_test.cc

@@ -101,6 +101,11 @@ TEST_F(ReduceTest, TestTimeout) {
     opts.setOutput(outputs.getPointer(), 1);
     opts.setRoot(0);
     opts.setReduceFunction(getFunction<uint64_t>());
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    reduce(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/scatter_test.cc

@@ -72,6 +72,11 @@ TEST_F(ScatterTest, TestTimeout) {
     opts.setInputs(input.getPointers(), 1);
     opts.setOutput(output.getPointer(), 1);
     opts.setRoot(0);
+
+    // Run one operation first so we're measuring the operation timeout not
+    // connection timeout.
+    scatter(opts);
+
     opts.setTimeout(std::chrono::milliseconds(10));
     if (context->rank == 0) {
       try {

gloo/test/tcp_test.cc

@@ -8,7 +8,7 @@ namespace transport {
 namespace tcp {
 
 TEST(TcpTest, ConnectTimeout) {
-  auto loop = std::make_shared<Loop>();
+  Loop loop;
 
   std::mutex m;
   std::condition_variable cv;
@@ -25,7 +25,7 @@ TEST(TcpTest, ConnectTimeout) {
     EXPECT_TRUE(e);
     EXPECT_TRUE(dynamic_cast<const TimeoutError*>(&e));
   };
-  connectLoop(*loop, remote, 0, 5, timeout, std::move(fn));
+  connectLoop(loop, remote, 0, 5, timeout, std::move(fn));
 
   std::unique_lock<std::mutex> lock(m);
   cv.wait(lock, [&] { return done; });