class_docstr: Tests for async processing logic in horovod.elastic.
class_docstr_tok: ["Tests","for","async","processing","logic","in","horovod.elastic","."]
docstr: Tests that host add events are sent to the worker notification service and consumed.
docstr_tok: ["Tests","that","host","add","events","are","sent","to","the","worker","notification","service","and","consumed","."]
code: def test_worker_notification_manager(self):
      """Tests that host add events are sent to the worker notification service and consumed."""
      slots = {"host-1": 2}
      discovery = FixedHosts(slots)
  
      rendezvous = RendezvousServer()
      driver = ElasticDriver(
          rendezvous, discovery, min_num_proc=2, max_num_proc=4
      )
      driver.wait_for_available_slots(min_num_proc=2)
      handler = create_rendezvous_handler(driver)
  
      common_intfs = network.get_local_intfs()
      addr = network.get_driver_ip(common_intfs)
      port = rendezvous.start(handler)
      nic = list(common_intfs)[0]
  
      rank_results = {}
  
      class NotificationReceiver:
          def __init__(self):
              self.events = []
  
          def on_hosts_updated(self, timestamp, res):
              self.events.append((timestamp, res))
  
      def add_host():
          slots = {"host-1": 2, "host-2": 2}
          discovery.set(slots)
  
      def remove_host():
          slots = {"host-2": 2}
          discovery.set(slots)
  
      def exec_command(slot_info, events):
          manager = WorkerNotificationManager()
          manager.init(
              rendezvous_addr=addr,
              rendezvous_port=port,
              nic=nic,
              hostname=slot_info.hostname,
              local_rank=slot_info.local_rank,
          )
  
          notification_receiver = NotificationReceiver()
          manager.register_listener(notification_receiver)
  
          driver.record_ready(slot_info.hostname, slot_info.local_rank)
  
          if slot_info.rank == 0:
              add_host()
          driver.wait_for_available_slots(4)
  
          if slot_info.rank == 0:
              remove_host()
  
          # Busy wait for the number of available slots to decrease
          while (
              driver._host_manager.current_hosts.count_available_slots()
              > 2
          ):
              time.sleep(0.01)
  
          rank_results[slot_info.rank] = notification_receiver.events
          return 0, time.time()
  
      driver.start(num_proc=2, create_worker_fn=exec_command)
      res = driver.get_results().worker_results
      driver.stop()
  
      assert len(res) == 2
      for name, (exit_code, timestamp) in res.items():
          assert exit_code == 0, name
  
      assert len(rank_results) == 2
      for rank, events in rank_results.items():
          expected = 2 if rank == 0 else 0
          assert len(events) == expected, rank
          if rank == 0:
              # First update is an add
              assert events[0][1] == HostUpdateResult.added
              # Second update is a removal
              assert events[1][1] == HostUpdateResult.removed
  
      rendezvous.stop()
  
imports: import time
  import unittest
  import warnings
  import mock
  import pytest
  from horovod.runner.util import network
  from horovod.runner.elastic.discovery import FixedHosts, HostManager
  from horovod.runner.elastic.driver import ElasticDriver
  from horovod.runner.elastic.rendezvous import create_rendezvous_handler
  from horovod.runner.elastic.worker import HostUpdateResult, WorkerNotificationManager
  from horovod.runner.http.http_server import RendezvousServer
label_desc: Use image node_id 10 for calling the ElasticDriverTests obj's underlying member method code with example usage: obj.test_worker_notification_manager() and returns: int,  time