pytorch / pytorch/gloo

Message exchange failure when perform alltoallv (cpus)

Open
#334 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
C++
Stars
1.5k
Forks
368
Avg merge
19h 32m
Merged PRs (30d)
3

Description

When performing an alltoallv message exchange on cpus results in the following error


terminate called after throwing an instance of 'gloo::EnforceNotMet'
what(): [enforce fail at ../third_party/gloo/gloo/transport/tcp/pair.cc:490] op.preamble.length <= op.nbytes. 881392472 vs 881392448

This error is reproducible and a standalone python script is included in this Issue report submission.
This is a very simple script which uses 10 machines/10-processes to reproduce this above mentioned error.
I used 10 machine cluster to reproduce this error repeatedly... however I guess the same may happen on a single machine using 10 processes.

This script performs the following tasks

  1. create the processgroup with 10 ranks
  2. exchanges no of int64's which will be exchanged, this no. is used on the receiving side to allocate buffers.
  3. once the buffers are allocated alltoallv (which is included in the standalone script) is performed to exchange int64's.
  4. The error happens when performing alltoallv using cpus.

Some observations about this error:

  1. This error happens when sending large messages. The same piece of logic works when smaller messages were sent.
  2. This standalone script is created by mimic'ing some of the functionality in the application I am working on at the moment.
  3. The hardcoded no. of int64's is one such instance when this error is deterministically reproducible.

Please use the following standalone script

import numpy as np
import argparse
import torch 
import os
import time
from datetime import timedelta
import torch.distributed as dist
from timeit import default_timer as timer
from datetime import timedelta


def alltoall_cpu(rank, world_size, output_tensor_list, input_tensor_list):
    input_tensor_list = [tensor.to(torch.device('cpu')) for tensor in input_tensor_list]
    for i in range(world_size):
        dist.scatter(output_tensor_list[i], input_tensor_list if i == rank else [], src=i)


def alltoallv_cpu(rank, world_size, output_tensor_list, input_tensor_list):
    senders = []
    for i in range(world_size):
        if i == rank:
            output_tensor_list[i] = input_tensor_list[i].to(torch.device('cpu'))
        else:
            sender = dist.isend(input_tensor_list[i].to(torch.device('cpu')), dst=i, tag=i)
            senders.append(sender)

    for i in range(world_size):
        if i != rank:
            dist.recv(output_tensor_list[i], src=i, tag=i)

    torch.distributed.barrier()

def splitdata_exec(rank, world_size):
    int64_counts = np.array([
        [0,         110105856, 110093280, 110116272, 110097840, 110111128, 110174059, 110087008, 110125040, 110087400],#0 
        [110174059, 0,         110158903, 110160317, 110149564, 110170899, 110166538, 110139263, 110163283, 110154040],#1
        [110251793, 110254110, 0,         110243087, 110249640, 110270594, 110248594, 110249172, 110277587, 110242484],#2
        [110191018, 110171210, 110170046, 0,         110167632, 110165475, 110174676, 110158908, 110171609, 110158631],#3 
        [110197278, 110198689, 110193780, 110198301, 0,         110208663, 110184046, 110194628, 110200308, 110168337],#4 
        [110256343, 110244546, 110248884, 110255858, 110236621, 0,         110247954, 110246921, 110247543, 110243309],#5 
        [110113348, 109915976, 109891208, 109908240, 109916552, 109917544, 0,         109893592, 109930888, 109895912],#6 
        [110024052, 109995591, 110003242, 110013125, 110002038, 110013278, 110003047, 0,         110015547, 109981915],#7 
        [109936439, 109948208, 109937391, 109936696, 109930888, 109941325, 109940259, 109917662, 0,         109917002],#8 
        [110050394, 110029327, 110036926, 110043437, 110021664, 110051453, 110036305, 110039768, 110054324, 0],#9
        ])

    start = timer()

    sizes = int64_counts[rank]
    print('[Rank: ', rank, '] outgoing int64 counts: ', sizes)

    # buffer sizes send/recv
    send_counts = list(torch.Tensor(sizes).type(dtype=torch.int64).chunk(world_size))
    recv_counts = list(torch.zeros([world_size], dtype=torch.int64).chunk(world_size))
    alltoall_cpu(rank, world_size, recv_counts, send_counts)

    #allocate buffers
    recv_nodes = []
    for i in recv_counts:
        recv_nodes.append(torch.zeros(i.tolist(), dtype=torch.int64))

    #form the outgoing message
    send_nodes = []
    for i in range(world_size):
        # sending 
        d = np.ones(shape=(sizes[i]), dtype=np.int64)*rank
        send_nodes.append(torch.from_numpy(d))

    alltoallv_cpu(rank, world_size, recv_nodes, send_nodes)
    end = timer()

    for i in range(world_size): 
        data = recv_nodes[i].numpy()
        assert np.all(data == np.ones(data.shape, dtype=np.int64)*i)

    print('[Rank: ', rank, '] Done with the test...')

def multi_dev_proc_init(params):
    rank = int(os.environ["RANK"])
    dist.init_process_group("gloo", rank=rank, world_size=params.world_size, timeout=timedelta(seconds=5*60))
    splitdata_exec(rank, params.world_size)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Construct graph partitions')
    parser.add_argument('--world-size', help='no. of processes to spawn', default=1, type=int, required=True)
    params = parser.parse_args()
    multi_dev_proc_init(params)

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with the standalone Python reproducer and the assertion at third_party/gloo/gloo/transport/tcp/pair.cc:490, focusing on the large-message alltoallv exchange and the reported preamble.length versus op.nbytes mismatch. Run the ten-process CPU case and trace the TCP pair handling; done means the reproducer completes successfully and the received int64 data assertions pass.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp, python
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.