code: def start_cluster_server(ctx, num_gpus=1, rdma=False):
      """Function that wraps the creation of TensorFlow ``tf.train.Server`` for a node in a distributed TensorFlow cluster.
  
      This is intended to be invoked from within the TF ``map_fun``, replacing explicit code to instantiate ``tf.train.ClusterSpec``
      and ``tf.train.Server`` objects.
  
      DEPRECATED for TensorFlow 2.x+
  
      Args:
        :ctx: TFNodeContext containing the metadata specific to this node in the cluster.
        :num_gpu: number of GPUs desired
        :rdma: boolean indicating if RDMA 'iverbs' should be used for cluster communications.
  
      Returns:
        A tuple of (cluster_spec, server)
      """
      import os
      import time
      from . import gpu_info
  
      if version.parse(TF_VERSION) >= version.parse("2.0.0"):
          raise Exception(
              "DEPRECATED: Use higher-level APIs like `tf.keras` or `tf.estimator`"
          )
  
      logging.info(
          "{0}: ======== {1}:{2} ========".format(
              ctx.worker_num, ctx.job_name, ctx.task_index
          )
      )
      cluster_spec = ctx.cluster_spec
      logging.info(
          "{0}: Cluster spec: {1}".format(ctx.worker_num, cluster_spec)
      )
  
      if compat.is_gpu_available() and num_gpus > 0:
          # compute my index relative to other nodes placed on the same host (for GPU allocation)
          my_addr = cluster_spec[ctx.job_name][ctx.task_index]
          my_host = my_addr.split(":")[0]
          flattened = [
              v for sublist in cluster_spec.values() for v in sublist
          ]
          local_peers = [p for p in flattened if p.startswith(my_host)]
          my_index = local_peers.index(my_addr)
  
          # GPU
          gpu_initialized = False
          retries = 3
          while not gpu_initialized and retries > 0:
              try:
                  # override PS jobs to only reserve one GPU
                  if ctx.job_name == "ps":
                      num_gpus = 0
  
                  # Find a free gpu(s) to use
                  gpus_to_use = gpu_info.get_gpus(num_gpus, my_index)
                  gpu_prompt = "GPU" if num_gpus == 1 else "GPUs"
                  logging.info(
                      "{0}: Using {1}: {2}".format(
                          ctx.worker_num, gpu_prompt, gpus_to_use
                      )
                  )
  
                  # Set GPU device to use for TensorFlow
                  os.environ["CUDA_VISIBLE_DEVICES"] = gpus_to_use
  
                  # Import tensorflow after gpu allocation
                  import tensorflow as tf
  
                  # Create a cluster from the parameter server and worker hosts.
                  cluster = tf.train.ClusterSpec(cluster_spec)
  
                  # Create and start a server for the local task.
                  if rdma:
                      server = tf.train.Server(
                          cluster,
                          ctx.job_name,
                          ctx.task_index,
                          protocol="grpc+verbs",
                      )
                  else:
                      server = tf.train.Server(
                          cluster, ctx.job_name, ctx.task_index
                      )
                  gpu_initialized = True
              except Exception as e:
                  print(e)
                  logging.error(
                      "{0}: Failed to allocate GPU, trying again...".format(
                          ctx.worker_num
                      )
                  )
                  retries -= 1
                  time.sleep(10)
          if not gpu_initialized:
              raise Exception("Failed to allocate GPU")
      else:
          # CPU
          import tensorflow as tf
  
          os.environ["CUDA_VISIBLE_DEVICES"] = ""
          logging.info("{0}: Using CPU".format(ctx.worker_num))
  
          # Create a cluster from the parameter server and worker hosts.
          cluster = tf.train.ClusterSpec(cluster_spec)
  
          # Create and start a server for the local task.
          server = tf.train.Server(
              cluster, ctx.job_name, ctx.task_index
          )
  
      return (cluster, server)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import nested_scopes
  from __future__ import print_function
  import getpass
  import logging
  import pkg_resources
  from packaging import version
  from six.moves.queue import Empty
  from .None import compat, marker
label_desc: Use image node_id 2 for calling a global function with example usage: start_cluster_server(ctx, num_gpus, rdma) and returns: cluster,  server