code: def run_rabit_ops(client, n_workers):
      from xgboost.dask import (
          CommunicatorContext,
          _get_dask_config,
          _get_rabit_args,
      )
  
      workers = tm.get_client_workers(client)
      rabit_args = client.sync(
          _get_rabit_args, len(workers), _get_dask_config(), client
      )
      assert not collective.is_distributed()
      n_workers_from_dask = len(workers)
      assert n_workers == n_workers_from_dask
  
      def local_test(worker_id):
          with CommunicatorContext(**rabit_args):
              a = 1
              assert collective.is_distributed()
              a = np.array([a])
              reduced = collective.allreduce(a, collective.Op.SUM)
              assert reduced[0] == n_workers
  
              worker_id = np.array([worker_id])
              reduced = collective.allreduce(
                  worker_id, collective.Op.MAX
              )
              assert reduced == n_workers - 1
  
              return 1
  
      futures = client.map(
          local_test, range(len(workers)), workers=workers
      )
      results = client.gather(futures)
      assert sum(results) == n_workers
  
file_path: xgboost/tests/python/test_tracker.py
imports: import re
  import sys
  import numpy
  import pytest
  import xgboost
  from xgboost import RabitTracker, collective
  from xgboost import testing
label_desc: Use image node_id 3 for calling a global function with example usage: run_rabit_ops(client, n_workers) and returns: int