code: def test_two_subscribers(ray_start_regular):
      """Tests concurrently subscribing to two channels work."""
  
      address_info = ray_start_regular
      gcs_server_addr = address_info["gcs_address"]
  
      num_messages = 100
  
      errors = []
      error_subscriber = ray._raylet.GcsErrorSubscriber(
          address=gcs_server_addr
      )
      # Make sure subscription is registered before publishing starts.
      error_subscriber.subscribe()
  
      def receive_errors():
          while len(errors) < num_messages:
              _, msg = error_subscriber.poll()
              errors.append(msg)
  
      t1 = threading.Thread(target=receive_errors)
      t1.start()
  
      logs = []
      log_subscriber = ray._raylet.GcsLogSubscriber(
          address=gcs_server_addr
      )
      # Make sure subscription is registered before publishing starts.
      log_subscriber.subscribe()
  
      log_str_pattern = re.compile("^log ([0-9]+)$")
  
      def receive_logs():
          while len(logs) < num_messages:
              log_batch = log_subscriber.poll()
              if log_str_pattern.match(log_batch["lines"][0]):
                  logs.append(log_batch)
  
      t2 = threading.Thread(target=receive_logs)
      t2.start()
  
      publisher = ray._raylet.GcsPublisher(address=gcs_server_addr)
      for i in range(0, num_messages):
          publisher.publish_error(b"msg_id", "", f"error {i}")
          publisher.publish_logs(
              {
                  "ip": "127.0.0.1",
                  "pid": "gcs",
                  "job": "0001",
                  "is_err": False,
                  "lines": [f"log {i}"],
                  "actor_name": "test actor",
                  "task_name": "test task",
              }
          )
  
      t1.join(timeout=10)
      assert len(errors) == num_messages, str(errors)
      assert not t1.is_alive(), str(errors)
  
      t2.join(timeout=10)
      assert len(logs) == num_messages, str(logs)
      assert not t2.is_alive(), str(logs)
  
      for i in range(0, num_messages):
          assert errors[i]["error_message"] == f"error {i}", str(errors)
          assert logs[i]["lines"][0] == f"log {i}", str(logs)
  
file_path: ray/python/ray/tests/test_gcs_pubsub.py
imports: import asyncio
  import sys
  import threading
  import re
  import ray
  from ray._private.gcs_pubsub import GcsAioPublisher, GcsAioErrorSubscriber, GcsAioLogSubscriber, GcsAioResourceUsageSubscriber
  from ray.core.generated.gcs_pb2 import ErrorTableData
  import pytest
label_desc: Use image node_id 3 for calling a global function with example usage: test_two_subscribers(ray_start_regular) without return types