The event_loop implementation has several issues affecting performance and correctness:

1. Busy-Wait Loop: The nested while True with a non-blocking recv_pyobj creates a busy-wait loop, consuming unnecessary CPU cycles. It's more efficient to use a single blocking recv_pyobj() to wait for messages.
2. Redundant Operations: backup_weights_from_disk() and start_transfer_server() are called for every received request. Since the expert weights on disk are static, these expensive operations should be performed only once when the manager starts.
3. Potential None Access: If backup_weights_from_disk results in self.continuous_buffer being None, the subsequent access self.continuous_buffer.numel() will raise an AttributeError.

I suggest refactoring this method to perform one-time setup and use a blocking receive loop.

    def event_loop(self):
        self.backup_weights_from_disk()
        if self.continuous_buffer is not None:
            self.start_transfer_server()

        while True:
            try:
                recv_req = self.recv_from_expert_location_updater.recv_pyobj()
            except zmq.ZMQError:
                # This can happen if the context is terminated.
                break

            logger.info("recv_req: %s", recv_req)
            # TODO (stage 1): handle recv_req
            print("RECEIVED REQ: ", self.engine_rank)

            buffer_size = 0
            if self.continuous_buffer is not None:
                buffer_size = self.continuous_buffer.numel() * self.continuous_buffer.element_size()

            back_req = BackupDramReq(
                _rank=self.engine_rank,
                _map=self.weight_pointer_map,
                session_id=self.session_id,
                buffer_size=buffer_size,
            )
            self.send_to_expert_location_updater.send_pyobj(back_req)

The current implementation of _receive_loop will sequentially block on receiving from each socket in self.recv_list. This means it will wait for a message from self.recv_list[0], then self.recv_list[1], and so on. This is likely not the intended behavior, as it should process messages from any backup manager as they arrive.

A better approach is to use zmq.Poller to wait for messages on all sockets simultaneously. This avoids both sequential blocking and busy-waiting.

Here's an example of how you could refactor it:

def _receive_loop(self):
    cnt = 0
    poller = zmq.Poller()
    for sock in self.recv_list:
        poller.register(sock, zmq.POLLIN)

    while cnt < self.engine_num:
        try:
            socks = dict(poller.poll())
        except zmq.ZMQError:
            # Context terminated
            break

        for sock in socks:
            if socks[sock] == zmq.POLLIN:
                try:
                    response = sock.recv_pyobj(zmq.NOBLOCK)
                    self.model_runner.dram_map_list[response._rank] = response._map
                    self.model_runner.session_id_list[response._rank] = response.session_id
                    self.model_runner.buffer_size = max(self.model_runner.buffer_size, response.buffer_size)
                    cnt += 1
                    print("RECEIVED: ", response._rank)
                except zmq.ZMQError:
                    continue
    
    if cnt == self.engine_num:
        self.model_runner.if_backup = True
        self.model_runner.start_transfer_client()

Replace print with logger.info for consistent and controllable logging. This allows log levels to be managed centrally and provides more context (like timestamp, module name) in logs.

Please use the logging module instead of print for better log management and consistency.

Please replace print with logger.info for standardized logging. This will make logs more informative and easier to filter.

This loop registers memory for all model parameters with the transfer engine. However, the backup and transfer mechanism seems to be focused on expert weights. If only expert weights are transferred, registering all parameters might be unnecessary and could increase initialization time and memory overhead for the transfer engine. It would be more efficient to only register the memory for the parameters that will actually be updated from the DRAM backup.

This large block of commented-out code for load_weight_from_dram should be removed to improve code clarity if it is no longer in use.

When self.if_backup is true, this function performs weight updates directly via RDMA transfer instead of yielding weights. This behavior is contrary to what its name get_weight_iter suggests, making the control flow confusing. The caller model_load_weights receives an empty iterator and does nothing, while the actual weight update happens inside this function.

Consider refactoring this logic to improve clarity. For example, you could create a separate function like _update_weights_from_dram_backup that is called when self.if_backup is true, and have get_weight_iter consistently behave as an iterator.

the explanation here carries zero information

Yes, we will make it clear to understand

Can we avoid engine crash for this resilient feature?

we will avoid using backup when failing

I feel that we should check elastic_ep_backend here instead

yes！

Does it only work for mooncake ep backend? need to make sure if it is general or we need to check the elastic ep backend == mooncake

it is general, precisely we check elastic ep backend is not None

Looks a little bit hacky. But you can optimize this in the future.