docstr: 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.
docstr_tok: ["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","."]
code: 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.
      """
      logger.info("Waiting for TensorFlow nodes to complete...")
  
      # 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):
              logger.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:
                  logger.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, grace_secs, self.queues
          )
      )
  
      # exit Spark application w/ err status if TF job had any errors
      if "error" in tf_status:
          logger.error("Exiting Spark application with error status.")
          self.sc.cancelAllJobs()
          self.sc.stop()
          sys.exit(1)
  
      logger.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
  
      # stop reservation server
      self.server.stop()
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import nested_scopes
  from __future__ import print_function
  import logging
  import os
  import random
  import signal
  import sys
  import threading
  import time
  from pyspark.streaming import DStream
  from .None import reservation
  from .None import TFManager
  from .None import TFSparkNode
label_desc: Use image node_id 3 for calling the TFCluster obj's underlying member method code with example usage: obj.shutdown(ssc, grace_secs, timeout) without return types