Prints log messages from workers on all of the nodes.

    Args:
        redis_client: A client to the primary Redis shard.
        threads_stopped (threading.Event): A threading event used to signal to
            the thread that it should exit.

def print_logs(redis_client, threads_stopped):
    """Prints log messages from workers on all of the nodes.

    Args:
        redis_client: A client to the primary Redis shard.
        threads_stopped (threading.Event): A threading event used to signal to
            the thread that it should exit.
    """
    pubsub_client = redis_client.pubsub(ignore_subscribe_messages=True)
    pubsub_client.subscribe(ray.gcs_utils.LOG_FILE_CHANNEL)
    localhost = services.get_node_ip_address()
    try:
        # Keep track of the number of consecutive log messages that have been
        # received with no break in between. If this number grows continually,
        # then the worker is probably not able to process the log messages as
        # rapidly as they are coming in.
        num_consecutive_messages_received = 0
        while True:
            # Exit if we received a signal that we should stop.
            if threads_stopped.is_set():
                return

            msg = pubsub_client.get_message()
            if msg is None:
                num_consecutive_messages_received = 0
                threads_stopped.wait(timeout=0.01)
                continue
            num_consecutive_messages_received += 1

            data = json.loads(ray.utils.decode(msg["data"]))
            if data["ip"] == localhost:
                for line in data["lines"]:
                    print("{}{}(pid={}){} {}".format(
                        colorama.Style.DIM, colorama.Fore.CYAN, data["pid"],
                        colorama.Style.RESET_ALL, line))
            else:
                for line in data["lines"]:
                    print("{}{}(pid={}, ip={}){} {}".format(
                        colorama.Style.DIM, colorama.Fore.CYAN, data["pid"],
                        data["ip"], colorama.Style.RESET_ALL, line))

            if (num_consecutive_messages_received % 100 == 0
                    and num_consecutive_messages_received > 0):
                logger.warning(
                    "The driver may not be able to keep up with the "
                    "stdout/stderr of the workers. To avoid forwarding logs "
                    "to the driver, use 'ray.init(log_to_driver=False)'.")
    finally:
        # Close the pubsub client to avoid leaking file descriptors.
        pubsub_client.close()