Refactor rabit tests (#6096)

* Merge rabit tests into XGBoost.
* Run them On CI.
* Simplification for CMake scripts.

tests/cpp/rabit/allreduce_base_test.cc

@@ -0,0 +1,68 @@
+#define RABIT_CXXTESTDEFS_H
+#if !defined(_WIN32)
+#include <gtest/gtest.h>
+
+#include <string>
+#include <iostream>
+#include "../../../rabit/src/allreduce_base.h"
+
+TEST(AllreduceBase, InitTask)
+{
+  rabit::engine::AllreduceBase base;
+
+  std::string rabit_task_id = "rabit_task_id=1";
+  char cmd[rabit_task_id.size()+1];
+  std::copy(rabit_task_id.begin(), rabit_task_id.end(), cmd);
+  cmd[rabit_task_id.size()] = '\0';
+
+  char* argv[] = {cmd};
+  base.Init(1, argv);
+  EXPECT_EQ(base.task_id, "1");
+}
+
+TEST(AllreduceBase, InitWithCacheOn)
+{
+  rabit::engine::AllreduceBase base;
+
+  std::string rabit_task_id = "rabit_task_id=1";
+  char cmd[rabit_task_id.size()+1];
+  std::copy(rabit_task_id.begin(), rabit_task_id.end(), cmd);
+  cmd[rabit_task_id.size()] = '\0';
+
+  std::string rabit_bootstrap_cache = "rabit_bootstrap_cache=1";
+  char cmd2[rabit_bootstrap_cache.size()+1];
+  std::copy(rabit_bootstrap_cache.begin(), rabit_bootstrap_cache.end(), cmd2);
+  cmd2[rabit_bootstrap_cache.size()] = '\0';
+
+  std::string rabit_debug = "rabit_debug=1";
+  char cmd3[rabit_debug.size()+1];
+  std::copy(rabit_debug.begin(), rabit_debug.end(), cmd3);
+  cmd3[rabit_debug.size()] = '\0';
+
+  char* argv[] = {cmd, cmd2, cmd3};
+  base.Init(3, argv);
+  EXPECT_EQ(base.task_id, "1");
+  EXPECT_TRUE(base.rabit_bootstrap_cache);
+  EXPECT_EQ(base.rabit_debug, 1);
+}
+
+TEST(AllreduceBase, InitWithRingReduce)
+{
+  rabit::engine::AllreduceBase base;
+
+  std::string rabit_task_id = "rabit_task_id=1";
+  char cmd[rabit_task_id.size()+1];
+  std::copy(rabit_task_id.begin(), rabit_task_id.end(), cmd);
+  cmd[rabit_task_id.size()] = '\0';
+
+  std::string rabit_reduce_ring_mincount = "rabit_reduce_ring_mincount=1";
+  char cmd2[rabit_reduce_ring_mincount.size()+1];
+  std::copy(rabit_reduce_ring_mincount.begin(), rabit_reduce_ring_mincount.end(), cmd2);
+  cmd2[rabit_reduce_ring_mincount.size()] = '\0';
+
+  char* argv[] = {cmd, cmd2};
+  base.Init(2, argv);
+  EXPECT_EQ(base.task_id, "1");
+  EXPECT_EQ(base.reduce_ring_mincount, 1ul);
+}
+#endif  // !defined(_WIN32)

tests/cpp/rabit/allreduce_mock_test.cc

@@ -0,0 +1,53 @@
+#define RABIT_CXXTESTDEFS_H
+#if !defined(_WIN32)
+#include <gtest/gtest.h>
+
+#include <string>
+#include <iostream>
+#include "../../../rabit/src/allreduce_mock.h"
+
+TEST(AllreduceMock, MockAllreduce)
+{
+  rabit::engine::AllreduceMock m;
+
+  std::string mock_str = "mock=0,0,0,0";
+  char cmd[mock_str.size()+1];
+  std::copy(mock_str.begin(), mock_str.end(), cmd);
+  cmd[mock_str.size()] = '\0';
+
+  char* argv[] = {cmd};
+  m.Init(1, argv);
+  m.rank = 0;
+  EXPECT_THROW(m.Allreduce(nullptr,0,0,nullptr,nullptr,nullptr), dmlc::Error);
+}
+
+TEST(AllreduceMock, MockBroadcast)
+{
+  rabit::engine::AllreduceMock m;
+  std::string mock_str = "mock=0,1,2,0";
+  char cmd[mock_str.size()+1];
+  std::copy(mock_str.begin(), mock_str.end(), cmd);
+  cmd[mock_str.size()] = '\0';
+  char* argv[] = {cmd};
+  m.Init(1, argv);
+  m.rank = 0;
+  m.version_number=1;
+  m.seq_counter=2;
+  EXPECT_THROW(m.Broadcast(nullptr,0,0), dmlc::Error);
+}
+
+TEST(AllreduceMock, MockGather)
+{
+  rabit::engine::AllreduceMock m;
+  std::string mock_str = "mock=3,13,22,0";
+  char cmd[mock_str.size()+1];
+  std::copy(mock_str.begin(), mock_str.end(), cmd);
+  cmd[mock_str.size()] = '\0';
+  char* argv[] = {cmd};
+  m.Init(1, argv);
+  m.rank = 3;
+  m.version_number=13;
+  m.seq_counter=22;
+  EXPECT_THROW({m.Allgather(nullptr,0,0,0,0);}, dmlc::Error);
+}
+#endif  // !defined(_WIN32)

tests/cpp/rabit/allreduce_robust_test.cc

@@ -0,0 +1,235 @@
+#define RABIT_CXXTESTDEFS_H
+#if !defined(_WIN32)
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <string>
+#include <iostream>
+#include "../../../rabit/src/allreduce_robust.h"
+
+inline void MockErr(const char *fmt, ...) {EXPECT_STRCASEEQ(fmt, "[%d] exit due to time out %d s\n");}
+inline void MockAssert(bool val, const char *fmt, ...) {}
+rabit::engine::AllreduceRobust::ReturnType err_type(rabit::engine::AllreduceRobust::ReturnTypeEnum::kSockError);
+rabit::engine::AllreduceRobust::ReturnType succ_type(rabit::engine::AllreduceRobust::ReturnTypeEnum::kSuccess);
+
+TEST(AllreduceRobust, SyncErrorTimeout)
+{
+  rabit::engine::AllreduceRobust m;
+
+  std::string rabit_timeout = "rabit_timeout=1";
+  char cmd[rabit_timeout.size()+1];
+  std::copy(rabit_timeout.begin(), rabit_timeout.end(), cmd);
+  cmd[rabit_timeout.size()] = '\0';
+
+  std::string rabit_timeout_sec = "rabit_timeout_sec=1";
+  char cmd1[rabit_timeout_sec.size()+1];
+  std::copy(rabit_timeout_sec.begin(), rabit_timeout_sec.end(), cmd1);
+  cmd1[rabit_timeout_sec.size()] = '\0';
+
+  char* argv[] = {cmd,cmd1};
+  m.Init(2, argv);
+  m.rank = 0;
+  m.rabit_bootstrap_cache = true;
+  m.error_ = MockErr;
+  m.assert_ = MockAssert;
+  EXPECT_EQ(m.CheckAndRecover(err_type), false);
+  std::this_thread::sleep_for(std::chrono::milliseconds(1500));
+  EXPECT_EQ(m.rabit_timeout_task_.get(), false);
+}
+
+TEST(AllreduceRobust, SyncErrorReset)
+{
+  rabit::engine::AllreduceRobust m;
+
+  std::string rabit_timeout = "rabit_timeout=1";
+  char cmd[rabit_timeout.size()+1];
+  std::copy(rabit_timeout.begin(), rabit_timeout.end(), cmd);
+  cmd[rabit_timeout.size()] = '\0';
+
+  std::string rabit_timeout_sec = "rabit_timeout_sec=1";
+  char cmd1[rabit_timeout_sec.size()+1];
+  std::copy(rabit_timeout_sec.begin(), rabit_timeout_sec.end(), cmd1);
+  cmd1[rabit_timeout_sec.size()] = '\0';
+
+  std::string rabit_debug = "rabit_debug=1";
+  char cmd2[rabit_debug.size()+1];
+  std::copy(rabit_debug.begin(), rabit_debug.end(), cmd2);
+  cmd2[rabit_debug.size()] = '\0';
+
+  char* argv[] = {cmd, cmd1,cmd2};
+  m.Init(3, argv);
+  m.rank = 0;
+  m.assert_ = MockAssert;
+  EXPECT_EQ(m.CheckAndRecover(err_type), false);
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  EXPECT_EQ(m.CheckAndRecover(succ_type), true);
+  EXPECT_EQ(m.rabit_timeout_task_.get(), true);
+  m.Shutdown();
+}
+
+TEST(AllreduceRobust, SyncSuccessErrorTimeout)
+{
+  rabit::engine::AllreduceRobust m;
+
+  std::string rabit_timeout = "rabit_timeout=1";
+  char cmd[rabit_timeout.size()+1];
+  std::copy(rabit_timeout.begin(), rabit_timeout.end(), cmd);
+  cmd[rabit_timeout.size()] = '\0';
+
+  std::string rabit_timeout_sec = "rabit_timeout_sec=1";
+  char cmd1[rabit_timeout_sec.size()+1];
+  std::copy(rabit_timeout_sec.begin(), rabit_timeout_sec.end(), cmd1);
+  cmd1[rabit_timeout_sec.size()] = '\0';
+
+  std::string rabit_debug = "rabit_debug=1";
+  char cmd2[rabit_debug.size()+1];
+  std::copy(rabit_debug.begin(), rabit_debug.end(), cmd2);
+  cmd2[rabit_debug.size()] = '\0';
+
+  char* argv[] = {cmd, cmd1,cmd2};
+  m.Init(3, argv);
+  m.rank = 0;
+  m.rabit_bootstrap_cache = true;
+  m.assert_ = MockAssert;
+  m.error_ = MockErr;
+  EXPECT_EQ(m.CheckAndRecover(succ_type), true);
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  EXPECT_EQ(m.CheckAndRecover(err_type), false);
+  std::this_thread::sleep_for(std::chrono::milliseconds(1500));
+  EXPECT_EQ(m.rabit_timeout_task_.get(), false);
+}
+
+TEST(AllreduceRobust, SyncSuccessErrorSuccess)
+{
+  rabit::engine::AllreduceRobust m;
+
+  std::string rabit_timeout = "rabit_timeout=1";
+  char cmd[rabit_timeout.size()+1];
+  std::copy(rabit_timeout.begin(), rabit_timeout.end(), cmd);
+  cmd[rabit_timeout.size()] = '\0';
+
+  std::string rabit_timeout_sec = "rabit_timeout_sec=1";
+  char cmd1[rabit_timeout_sec.size()+1];
+  std::copy(rabit_timeout_sec.begin(), rabit_timeout_sec.end(), cmd1);
+  cmd1[rabit_timeout_sec.size()] = '\0';
+
+  std::string rabit_debug = "rabit_debug=1";
+  char cmd2[rabit_debug.size()+1];
+  std::copy(rabit_debug.begin(), rabit_debug.end(), cmd2);
+  cmd2[rabit_debug.size()] = '\0';
+
+  char* argv[] = {cmd, cmd1,cmd2};
+  m.Init(3, argv);
+  m.rank = 0;
+  m.rabit_bootstrap_cache = true;
+  m.assert_ = MockAssert;
+  EXPECT_EQ(m.CheckAndRecover(succ_type), true);
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+  EXPECT_EQ(m.CheckAndRecover(err_type), false);
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  EXPECT_EQ(m.CheckAndRecover(succ_type), true);
+  std::this_thread::sleep_for(std::chrono::milliseconds(1100));
+  EXPECT_EQ(m.rabit_timeout_task_.get(), true);
+  m.Shutdown();
+}
+
+TEST(AllreduceRobust, SyncErrorNoResetTimeout)
+{
+  rabit::engine::AllreduceRobust m;
+
+  std::string rabit_timeout = "rabit_timeout=1";
+  char cmd[rabit_timeout.size()+1];
+  std::copy(rabit_timeout.begin(), rabit_timeout.end(), cmd);
+  cmd[rabit_timeout.size()] = '\0';
+
+  std::string rabit_timeout_sec = "rabit_timeout_sec=1";
+  char cmd1[rabit_timeout_sec.size()+1];
+  std::copy(rabit_timeout_sec.begin(), rabit_timeout_sec.end(), cmd1);
+  cmd1[rabit_timeout_sec.size()] = '\0';
+
+  std::string rabit_debug = "rabit_debug=1";
+  char cmd2[rabit_debug.size()+1];
+  std::copy(rabit_debug.begin(), rabit_debug.end(), cmd2);
+  cmd2[rabit_debug.size()] = '\0';
+
+  char* argv[] = {cmd, cmd1,cmd2};
+  m.Init(3, argv);
+  m.rank = 0;
+  m.rabit_bootstrap_cache = true;
+  m.assert_ = MockAssert;
+  m.error_ = MockErr;
+  auto start = std::chrono::system_clock::now();
+
+  EXPECT_EQ(m.CheckAndRecover(err_type), false);
+  std::this_thread::sleep_for(std::chrono::milliseconds(1100));
+
+  EXPECT_EQ(m.CheckAndRecover(err_type), false);
+
+  m.rabit_timeout_task_.wait();
+  auto end = std::chrono::system_clock::now();
+  std::chrono::duration<double> diff = end-start;
+
+  EXPECT_EQ(m.rabit_timeout_task_.get(), false);
+  // expect second error don't overwrite/reset timeout task
+  EXPECT_LT(diff.count(), 2);
+}
+
+TEST(AllreduceRobust, NoTimeoutShutDown)
+{
+  rabit::engine::AllreduceRobust m;
+
+  std::string rabit_timeout = "rabit_timeout=1";
+  char cmd[rabit_timeout.size()+1];
+  std::copy(rabit_timeout.begin(), rabit_timeout.end(), cmd);
+  cmd[rabit_timeout.size()] = '\0';
+
+  std::string rabit_timeout_sec = "rabit_timeout_sec=1";
+  char cmd1[rabit_timeout_sec.size()+1];
+  std::copy(rabit_timeout_sec.begin(), rabit_timeout_sec.end(), cmd1);
+  cmd1[rabit_timeout_sec.size()] = '\0';
+
+  std::string rabit_debug = "rabit_debug=1";
+  char cmd2[rabit_debug.size()+1];
+  std::copy(rabit_debug.begin(), rabit_debug.end(), cmd2);
+  cmd2[rabit_debug.size()] = '\0';
+
+  char* argv[] = {cmd, cmd1,cmd2};
+  m.Init(3, argv);
+  m.rank = 0;
+
+  EXPECT_EQ(m.CheckAndRecover(succ_type), true);
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  m.Shutdown();
+}
+
+TEST(AllreduceRobust, ShutDownBeforeTimeout)
+{
+  rabit::engine::AllreduceRobust m;
+
+  std::string rabit_timeout = "rabit_timeout=1";
+  char cmd[rabit_timeout.size()+1];
+  std::copy(rabit_timeout.begin(), rabit_timeout.end(), cmd);
+  cmd[rabit_timeout.size()] = '\0';
+
+  std::string rabit_timeout_sec = "rabit_timeout_sec=1";
+  char cmd1[rabit_timeout_sec.size()+1];
+  std::copy(rabit_timeout_sec.begin(), rabit_timeout_sec.end(), cmd1);
+  cmd1[rabit_timeout_sec.size()] = '\0';
+
+  std::string rabit_debug = "rabit_debug=1";
+  char cmd2[rabit_debug.size()+1];
+  std::copy(rabit_debug.begin(), rabit_debug.end(), cmd2);
+  cmd2[rabit_debug.size()] = '\0';
+
+  char* argv[] = {cmd, cmd1,cmd2};
+  m.Init(3, argv);
+  m.rank = 0;
+  rabit::engine::AllreduceRobust::LinkRecord a;
+  m.err_link = &a;
+
+  EXPECT_EQ(m.CheckAndRecover(err_type), false);
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  m.Shutdown();
+}
+#endif  // !defined(_WIN32)

tests/cpp/rabit/test_io.cc

@@ -0,0 +1,18 @@
+/*!
+ *  Copyright (c) 2019 by Contributors
+ */
+#include <gtest/gtest.h>
+#include <rabit/internal/io.h>
+
+#include <vector>
+
+namespace rabit {
+TEST(MemoryFixSizeBuffer, Seek) {
+  size_t constexpr kSize { 64 };
+  std::vector<int32_t> memory( kSize );
+  utils::MemoryFixSizeBuffer buf(memory.data(), memory.size());
+  buf.Seek(utils::MemoryFixSizeBuffer::kSeekEnd);
+  size_t end = buf.Tell();
+  ASSERT_EQ(end, kSize);
+}
+}  // namespace rabit

tests/cpp/rabit/test_utils.cc

@@ -0,0 +1,6 @@
+#include <gtest/gtest.h>
+#include <rabit/internal/utils.h>
+
+TEST(Utils, Assert) {
+  EXPECT_THROW({rabit::utils::Assert(false, "foo");}, dmlc::Error);
+}