'Stops TensorFlow nodes'

def shutdown(self, ssc=None):

    logging.info('Stopping    TensorFlow    nodes')
    (ps_list, worker_list) = ([], [])
    for node in self.cluster_info:
        if (node['job_name'] == 'ps'):
            ps_list.append(node)
        else:
            worker_list.append(node)
    if (ssc is not None):
        done = False
        while (not done):
            done = ssc.awaitTerminationOrTimeout(1)
            if ((not done) and self.server.done):
                logging.info('Server    done,    stopping    StreamingContext')
                ssc.stop(stopSparkContext=False, stopGraceFully=True)
            done = (done or self.server.done)
    else:
        if (self.input_mode == InputMode.TENSORFLOW):
            count = 0
            done = False
            while (not done):
                st = self.sc.statusTracker()
                jobs = st.getActiveJobsIds()
                if (len(jobs) > 0):
                    stages = st.getActiveStageIds()
                    for i in stages:
                        si = st.getStageInfo(i)
                        if (si.numActiveTasks == len(ps_list)):
                            count += 1
                            done = (count >= 3)
                            time.sleep(5)
                else:
                    done = True
        workers = len(worker_list)
        workerRDD = self.sc.parallelize(range(workers), workers)
        workerRDD.foreachPartition(TFSparkNode.shutdown(self.cluster_info, self.queues))
    logging.info('Shutting    down    cluster')
    for node in ps_list:
        addr = node['addr']
        authkey = node['authkey']
        m = TFManager.connect(addr, authkey)
        q = m.get_queue('control')
        q.put(None)
        q.join()
    done = False
    while (not done):
        time.sleep(5)
        st = self.sc.statusTracker()
        jobs = st.getActiveJobsIds()
        if (len(jobs) == 0):
            break
