Enable distributed GPU training over Rabit (#7930)

tests/cpp/common/test_quantile.cu

@@ -339,7 +339,6 @@ TEST(GPUQuantile, MultiMerge) {
 TEST(GPUQuantile, AllReduceBasic) {
   // This test is supposed to run by a python test that setups the environment.
   std::string msg {"Skipping AllReduce test"};
-#if defined(__linux__) && defined(XGBOOST_USE_NCCL)
   auto n_gpus = AllVisibleGPUs();
   InitRabitContext(msg, n_gpus);
   auto world = rabit::GetWorldSize();
@@ -420,15 +419,10 @@ TEST(GPUQuantile, AllReduceBasic) {
     }
   });
   rabit::Finalize();
-#else
-  LOG(WARNING) << msg;
-  return;
-#endif  // !defined(__linux__) && defined(XGBOOST_USE_NCCL)
 }
 
 TEST(GPUQuantile, SameOnAllWorkers) {
   std::string msg {"Skipping SameOnAllWorkers test"};
-#if defined(__linux__) && defined(XGBOOST_USE_NCCL)
   auto n_gpus = AllVisibleGPUs();
   InitRabitContext(msg, n_gpus);
   auto world = rabit::GetWorldSize();
@@ -495,10 +489,6 @@ TEST(GPUQuantile, SameOnAllWorkers) {
       offset += size_as_float;
     }
   });
-#else
-  LOG(WARNING) << msg;
-  return;
-#endif  // !defined(__linux__) && defined(XGBOOST_USE_NCCL)
 }
 
 TEST(GPUQuantile, Push) {

tests/cpp/common/test_transform_range.cu

@@ -4,7 +4,6 @@
  */
 #include "test_transform_range.cc"
 
-#if defined(XGBOOST_USE_NCCL)
 namespace xgboost {
 namespace common {
 
@@ -15,7 +14,7 @@ TEST(Transform, MGPU_SpecifiedGpuId) {  // NOLINT
   }
   // Use 1 GPU, Numbering of GPU starts from 1
   auto device = 1;
-  const size_t size {256};
+  auto const size {256};
   std::vector<bst_float> h_in(size);
   std::vector<bst_float> h_out(size);
   std::iota(h_in.begin(), h_in.end(), 0);
@@ -34,4 +33,3 @@ TEST(Transform, MGPU_SpecifiedGpuId) {  // NOLINT
 
 }  // namespace common
 }  // namespace xgboost
-#endif

tests/cpp/metric/test_multiclass_metric.cc

@@ -85,7 +85,7 @@ TEST(Metric, DeclareUnifiedTest(MultiClassLogLoss)) {
   xgboost::CheckDeterministicMetricMultiClass(xgboost::StringView{"mlogloss"}, GPUIDX);
 }
 
-#if defined(XGBOOST_USE_NCCL) && defined(__CUDACC__)
+#if defined(__CUDACC__)
 namespace xgboost {
 namespace common {
 TEST(Metric, MGPU_MultiClassError) {
@@ -109,4 +109,4 @@ TEST(Metric, MGPU_MultiClassError) {
 }
 }  // namespace common
 }  // namespace xgboost
-#endif  // defined(XGBOOST_USE_NCCL)
+#endif  // defined(__CUDACC__)

tests/distributed/runtests-federated.sh

@@ -4,14 +4,14 @@ set -e
 
 rm -f ./*.model* ./agaricus* ./*.pem
 
-world_size=3
+world_size=$(nvidia-smi -L | wc -l)
 
 # Generate server and client certificates.
 openssl req -x509 -newkey rsa:2048 -days 7 -nodes -keyout server-key.pem -out server-cert.pem -subj "/C=US/CN=localhost"
 openssl req -x509 -newkey rsa:2048 -days 7 -nodes -keyout client-key.pem -out client-cert.pem -subj "/C=US/CN=localhost"
 
 # Split train and test files manually to simulate a federated environment.
-split -n l/${world_size} -d ../../demo/data/agaricus.txt.train agaricus.txt.train-
-split -n l/${world_size} -d ../../demo/data/agaricus.txt.test agaricus.txt.test-
+split -n l/"${world_size}" -d ../../demo/data/agaricus.txt.train agaricus.txt.train-
+split -n l/"${world_size}" -d ../../demo/data/agaricus.txt.test agaricus.txt.test-
 
-python test_federated.py ${world_size}
+python test_federated.py "${world_size}"

tests/distributed/test_federated.py

@@ -17,7 +17,7 @@ def run_server(port: int, world_size: int) -> None:
                                            CLIENT_CERT)
 
 
-def run_worker(port: int, world_size: int, rank: int) -> None:
+def run_worker(port: int, world_size: int, rank: int, with_gpu: bool) -> None:
     # Always call this before using distributed module
     rabit_env = [
         f'federated_server_address=localhost:{port}',
@@ -34,6 +34,9 @@ def run_worker(port: int, world_size: int, rank: int) -> None:
 
         # Specify parameters via map, definition are same as c++ version
         param = {'max_depth': 2, 'eta': 1, 'objective': 'binary:logistic'}
+        if with_gpu:
+            param['tree_method'] = 'gpu_hist'
+            param['gpu_id'] = rank
 
         # Specify validations set to watch performance
         watchlist = [(dtest, 'eval'), (dtrain, 'train')]
@@ -49,7 +52,7 @@ def run_worker(port: int, world_size: int, rank: int) -> None:
             xgb.rabit.tracker_print("Finished training\n")
 
 
-def run_test() -> None:
+def run_test(with_gpu: bool = False) -> None:
     port = 9091
     world_size = int(sys.argv[1])
 
@@ -61,7 +64,7 @@ def run_test() -> None:
 
     workers = []
     for rank in range(world_size):
-        worker = multiprocessing.Process(target=run_worker, args=(port, world_size, rank))
+        worker = multiprocessing.Process(target=run_worker, args=(port, world_size, rank, with_gpu))
         workers.append(worker)
         worker.start()
     for worker in workers:
@@ -71,3 +74,4 @@ def run_test() -> None:
 
 if __name__ == '__main__':
     run_test()
+    run_test(with_gpu=True)