Stops the distributed TensorFlow cluster.

    For InputMode.SPARK, this will be executed AFTER the `TFCluster.train()` or `TFCluster.inference()` method completes.
    For InputMode.TENSORFLOW, this will be executed IMMEDIATELY after `TFCluster.run()` and will wait until the TF worker nodes complete.

    Args:
      :ssc: *For Streaming applications only*. Spark StreamingContext
      :grace_secs: Grace period to wait after all executors have completed their tasks before terminating the Spark application, e.g. to allow the chief worker to perform any final/cleanup duties like exporting or evaluating the model.  Default is 0.
      :timeout: Time in seconds to wait for TF cluster to complete before terminating the Spark application.  This can be useful if the TF code hangs for any reason.  Default is 3 days.  Use -1 to disable timeout.

def shutdown(self, ssc=None, grace_secs=0, timeout=259200):
    """Stops the distributed TensorFlow cluster.

    For InputMode.SPARK, this will be executed AFTER the `TFCluster.train()` or `TFCluster.inference()` method completes.
    For InputMode.TENSORFLOW, this will be executed IMMEDIATELY after `TFCluster.run()` and will wait until the TF worker nodes complete.

    Args:
      :ssc: *For Streaming applications only*. Spark StreamingContext
      :grace_secs: Grace period to wait after all executors have completed their tasks before terminating the Spark application, e.g. to allow the chief worker to perform any final/cleanup duties like exporting or evaluating the model.  Default is 0.
      :timeout: Time in seconds to wait for TF cluster to complete before terminating the Spark application.  This can be useful if the TF code hangs for any reason.  Default is 3 days.  Use -1 to disable timeout.
    """
    logging.info("Stopping TensorFlow nodes")

    # identify ps/workers
    ps_list, worker_list, eval_list = [], [], []
    for node in self.cluster_info:
      (ps_list if node['job_name'] == 'ps' else eval_list if node['job_name'] == 'evaluator' else worker_list).append(node)

    # setup execution timeout
    if timeout > 0:
      def timeout_handler(signum, frame):
        logging.error("TensorFlow execution timed out, exiting Spark application with error status")
        self.sc.cancelAllJobs()
        self.sc.stop()
        sys.exit(1)

      signal.signal(signal.SIGALRM, timeout_handler)
      signal.alarm(timeout)

    # wait for Spark Streaming termination or TF app completion for InputMode.TENSORFLOW
    if ssc is not None:
      # Spark Streaming
      while not ssc.awaitTerminationOrTimeout(1):
        if self.server.done:
          logging.info("Server done, stopping StreamingContext")
          ssc.stop(stopSparkContext=False, stopGraceFully=True)
          break
    elif self.input_mode == InputMode.TENSORFLOW:
      # in TENSORFLOW mode, there is no "data feeding" job, only a "start" job, so we must wait for the TensorFlow workers
      # to complete all tasks, while accounting for any PS tasks which run indefinitely.
      count = 0
      while count < 3:
        st = self.sc.statusTracker()
        jobs = st.getActiveJobsIds()
        if len(jobs) == 0:
          break
        stages = st.getActiveStageIds()
        for i in stages:
          si = st.getStageInfo(i)
          if si.numActiveTasks == len(ps_list) + len(eval_list):
            # if we only have PS tasks left, check that we see this condition a couple times
            count += 1
        time.sleep(5)

    # shutdown queues and managers for "worker" executors.
    # note: in SPARK mode, this job will immediately queue up behind the "data feeding" job.
    # in TENSORFLOW mode, this will only run after all workers have finished.
    workers = len(worker_list)
    workerRDD = self.sc.parallelize(range(workers), workers)
    workerRDD.foreachPartition(TFSparkNode.shutdown(self.cluster_info, self.queues))
    time.sleep(grace_secs)

    # exit Spark application w/ err status if TF job had any errors
    if 'error' in tf_status:
      logging.error("Exiting Spark application with error status.")
      self.sc.cancelAllJobs()
      self.sc.stop()
      sys.exit(1)

    logging.info("Shutting down cluster")
    # shutdown queues and managers for "PS" executors.
    # note: we have to connect/shutdown from the spark driver, because these executors are "busy" and won't accept any other tasks.
    for node in ps_list + eval_list:
      addr = node['addr']
      authkey = node['authkey']
      m = TFManager.connect(addr, authkey)
      q = m.get_queue('control')
      q.put(None)
      q.join()

    # wait for all jobs to finish
    while True:
      time.sleep(5)
      st = self.sc.statusTracker()
      jobs = st.getActiveJobsIds()
      if len(jobs) == 0:
        break