code: def test_host_standalone(serve_instance):
      host = ray.remote(LongPollHost).remote()
  
      # Write two values
      ray.get(host.notify_changed.remote("key_1", 999))
      ray.get(host.notify_changed.remote("key_2", 999))
      object_ref = host.listen_for_change.remote(
          {"key_1": -1, "key_2": -1}
      )
  
      # We should be able to get the result immediately
      result: Dict[str, UpdatedObject] = ray.get(object_ref)
      assert set(result.keys()) == {"key_1", "key_2"}
      assert {v.object_snapshot for v in result.values()} == {999}
  
      # Now try to pull it again, nothing should happen
      # because we have the updated snapshot_id
      new_snapshot_ids = {k: v.snapshot_id for k, v in result.items()}
      object_ref = host.listen_for_change.remote(new_snapshot_ids)
      _, not_done = ray.wait([object_ref], timeout=0.2)
      assert len(not_done) == 1
  
      # Now update the value, we should immediately get updated value
      ray.get(host.notify_changed.remote("key_2", 999))
      result = ray.get(object_ref)
      assert len(result) == 1
      assert "key_2" in result
  
file_path: ray/python/ray/serve/tests/test_long_poll.py
imports: import asyncio
  import os
  import sys
  import time
  from typing import Dict
  import pytest
  import ray
  from ray._private.test_utils import async_wait_for_condition
  from ray._private.utils import get_or_create_event_loop
  from ray.serve._private.common import EndpointInfo, EndpointTag, RunningReplicaInfo
  from ray.serve._private.long_poll import LongPollClient, LongPollHost, LongPollNamespace, LongPollState, UpdatedObject
  from ray.serve.generated.serve_pb2 import ActorNameList, EndpointSet, LongPollRequest, LongPollResult
label_desc: Use image node_id 2 for calling a global function with example usage: test_host_standalone(serve_instance) without return types