Function send_requests_to_machine

python/dgl/distributed/rpc.py:919–962

Send requests to the remote machines. This operation isn't block. It returns immediately once it sends all requests. Parameters ---------- target_and_requests : list[(int, Request)] A list of requests and the machine they should be sent to. timeout : int, optional

(target_and_requests)

def send_requests_to_machine(target_and_requests):
    """Send requests to the remote machines.

    This operation isn't block. It returns immediately once it sends all requests.

    Parameters
    ----------
    target_and_requests : list[(int, Request)]
        A list of requests and the machine they should be sent to.
    timeout : int, optional
        The timeout value in milliseconds. If zero, wait indefinitely.

    Returns
    -------
    msgseq2pos : dict
        map the message sequence number to its position in the input list.
    """
    msgseq2pos = {}
    for pos, (target, request) in enumerate(target_and_requests):
        # send request
        service_id = request.service_id
        msg_seq = incr_msg_seq()
        client_id = get_rank()

        server_id = random.randint(
            target * get_num_server_per_machine(),
            (target + 1) * get_num_server_per_machine() - 1,
        )
        data, tensors = serialize_to_payload(request)
        msg = RPCMessage(
            service_id,
            msg_seq,
            client_id,
            server_id,
            data,
            tensors,
            get_group_id(),
        )
        send_rpc_message(msg, server_id)
        # check if has response
        res_cls = get_service_property(service_id)[1]
        if res_cls is not None:
            msgseq2pos[msg_seq] = pos
    return msgseq2pos


def recv_responses(msgseq2pos, timeout=0):