code: def shutdown(cluster_info, grace_secs=0, queues=["input"]):
      """Stops all TensorFlow nodes by feeding ``None`` into the multiprocessing.Queues.
  
      Args:
        :cluster_info: node reservation information for the cluster (e.g. host, executor_id, pid, ports, etc).
        :queues: *INTERNAL_USE*
  
      Returns:
        A nodeRDD.mapPartitions() function
      """
  
      def _shutdown(iter):
          host = util.get_ip_address()
          executor_id = util.read_executor_id()
  
          # reconnect to shared queue
          mgr = _get_manager(cluster_info, host, executor_id)
  
          # send SIGTERM to Tensorboard proc (if running)
          for node in cluster_info:
              if (
                  node["host"] == host
                  and node["executor_id"] == executor_id
              ):
                  tb_pid = node["tb_pid"]
                  if tb_pid != 0:
                      logger.info(
                          "Stopping tensorboard (pid={0})".format(
                              tb_pid
                          )
                      )
                      subprocess.Popen(["kill", str(tb_pid)])
  
          # terminate any listening queues
          logger.info("Stopping all queues")
          for q in queues:
              if q != "error":
                  try:
                      queue = mgr.get_queue(q)
                      logger.info(
                          "Feeding None into {0} queue".format(q)
                      )
                      queue.put(None, block=True)
                  except (AttributeError, KeyError):
                      msg = "Queue '{}' not found on this node, check for exceptions on other nodes.".format(
                          q
                      )
                      raise Exception(msg)
  
          # wait for grace period (after terminating feed queues)
          if grace_secs > 0:
              logger.info(
                  "Waiting for {} second grace period".format(
                      grace_secs
                  )
              )
              time.sleep(grace_secs)
  
          # then check for any late exceptions
          equeue = mgr.get_queue("error")
          if not equeue.empty():
              # note: "peek" this queue, since otherwise Spark might retry this "failed" task, find no errors in queue, and finish the job with SUCCESS
              e_str = equeue.get()
              equeue.put(e_str)
              raise Exception("Exception in worker:\n" + e_str)
  
          logger.info("Setting mgr.state to 'stopped'")
          mgr.set("state", "stopped")
          return [True]
  
      return _shutdown
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import nested_scopes
  from __future__ import print_function
  import json
  import logging
  import multiprocessing
  import os
  import pkg_resources
  import platform
  import socket
  import subprocess
  import sys
  import uuid
  import time
  import traceback
  from packaging import version
  from threading import Thread
  from .None import TFManager
  from .None import TFNode
  from .None import gpu_info
  from .None import marker
  from .None import reservation
  from .None import util
label_desc: Use image node_id 7 for calling a global function with example usage: shutdown(cluster_info, grace_secs, queues) and returns: _shutdown, list