async def METHOD_NAME(self):
    # When arming the worker, the following must be true:
    #  - No more progress in UCX (see doc of ucp_worker_arm())
    #  - All asyncio tasks that isn't waiting on UCX must be executed
    #    so that the asyncio's next state is epoll wait.
    #    See <https://github.com/rapidsai/ucx-py/issues/413>
    while True:
        worker = self.weakref_worker()
        if worker is None or not worker.initialized:
            return
        worker.progress()
        # Cancel inflight messages that couldn't be completed. This may
        # happen if the user called ep.recv() but the remote worker
        # errored before sending the message.
        if worker.cancel_inflight_messages() > 0:
            worker.progress()
        del worker
        # This IO task returns when all non-IO tasks are finished.
        # Notice, we do NOT hold a reference to `worker` while waiting.
        await self.event_loop.sock_recv(self.rsock, 1)
        worker = self.weakref_worker()
        if worker is None or not worker.initialized:
            return
        if worker.arm():
            # At this point we know that asyncio's next state is
            # epoll wait.
            break