Method remove_node

python/ray/cluster_utils.py:288–325

Kills all processes associated with worker node. Args: node: Worker node of which all associated processes will be removed.

(self, node, allow_graceful=True)

        return node

    def remove_node(self, node, allow_graceful=True):
        """Kills all processes associated with worker node.

        Args:
            node: Worker node of which all associated processes
                will be removed.
        """
        global_node = ray._private.worker.global_worker.node
        if global_node is not None:
            if node._raylet_socket_name == global_node._raylet_socket_name:
                ray.shutdown()
                raise ValueError(
                    "Removing a node that is connected to this Ray client "
                    "is not allowed because it will break the driver. "
                    "You can use the get_other_node utility to avoid removing "
                    "a node that the Ray client is connected."
                )

        node.destroy_external_storage()
        if self.head_node == node:
            # We have to wait to prevent the raylet becomes a zombie which will prevent
            # worker from exiting
            self.head_node.kill_all_processes(
                check_alive=False, allow_graceful=allow_graceful, wait=True
            )
            self.head_node = None
            # TODO(rliaw): Do we need to kill all worker processes?
        else:
            # We have to wait to prevent the raylet becomes a zombie which will prevent
            # worker from exiting
            node.kill_all_processes(
                check_alive=False, allow_graceful=allow_graceful, wait=True
            )
            self.worker_nodes.remove(node)

        assert (
            not node.any_processes_alive()
        ), "There are zombie processes left over after killing."

    def _wait_for_node(self, node, timeout: float = 30):
        """Wait until this node has appeared in the client table.