Looks like this is an issue with the SHM transport. I bumped up the UV transport priority to over-shadow SHM, and this problem disappeared.

  

      github.com
  

  
    pytorch/pytorch/blob/ad1041576aeb7cd9f8065acb26b596ade8e6ecaa/torch/csrc/distributed/rpc/tensorpipe_agent.h#L46 1



      
      
          namespace distributed {
          namespace rpc {
          
          
// These priorities instruct TensorPipe on which transport/channel to pick
          // during handshake. Higher priorities will take precedence over lower ones.
          // The transport with lowest priority will be the one used to bootstrap pipes.
          
          
constexpr int64_t kShmTransportPriority = 200;
          constexpr int64_t kIbvTransportPriority = 100;
          // The UV transport just uses TCP and should work everywhere, thus keep it last.
          constexpr int64_t kUvTransportPriority = 0;
          
          
constexpr int64_t kCmaChannelPriority = 1200;
          constexpr int64_t kMultiplexedUvChannelPriority = 1100;
          // The basic channel reuses a transport as a channel, and is thus our fallback.
          constexpr int64_t kBasicChannelPriority = 1000;
          
          
// CPU channel have higher priority than CUDA channels, since the latter might
          // handle CPU-to-CPU transfers, but will always be less efficient than their
          // CPU-only counterparts.
          constexpr int64_t kCudaIpcChannelPriority = 300;
      
    
  


  

  
    
    
  

  


Hey @akshay-raj-dhamija to unblock you, could you please try adding a _transports=["uv"] kill-switch to rpc_backend_options to disable SHM transport?
Below is the code:
import os
import time
import torch
#torch.multiprocessing.set_sharing_strategy('file_system')
import torch.multiprocessing as mp
import torch.distributed.rpc as rpc

no_of_saver_processes = 15
#world_size = torch.cuda.device_count()
world_size = 4

def cpu_process_initialization(rank):
    os.environ['MASTER_ADDR'] = 'localhost'
    os.environ['MASTER_PORT'] = '9867'
    os.environ["CUDA_VISIBLE_DEVICES"] = f"{rank % torch.cuda.device_count()}"
    rpc.init_rpc(f"{rank}",
                 rank=rank,
                 world_size = world_size + no_of_saver_processes,
                 backend=rpc.BackendType.TENSORPIPE,
                 rpc_backend_options=rpc.TensorPipeRpcBackendOptions(
                     rpc_timeout=6000,
                     init_method='env://',
                     _transports=["uv"],
                 )
    )
    print(f"Started CPU process {rank}")
    print(f"Process {rank}: avaialable device {torch.cuda.current_device()}")
    # Do something rather than sleeping example disk or cpu bound operations
    time.sleep(30)
    rpc.shutdown()
    return

def cuda_process_initialization(rank):
    #torch.cuda.set_device(rank)
    os.environ["CUDA_VISIBLE_DEVICES"] = f"{rank}"
    os.environ['MASTER_ADDR'] = 'localhost'
    os.environ['MASTER_PORT'] = '9867'
    rpc.init_rpc(f"{rank}",
                 rank=rank,
                 world_size=world_size + no_of_saver_processes,
                 backend=rpc.BackendType.TENSORPIPE,
                 rpc_backend_options=rpc.TensorPipeRpcBackendOptions(
                     #num_send_recv_threads=args.world_size*3,
                     rpc_timeout=6000,
                     init_method='env://',
                     _transports=["uv"],
                 )
    )
    print(f"Started CUDA process on gpu {rank}")
    # Do some cuda operations
    print(f"Process {rank}: avaialable device {torch.cuda.current_device()}")
    time.sleep(30)
    rpc.shutdown()
    return

if __name__ == "__main__":
    mp.set_start_method('spawn', force=True)
    trainer_processes = mp.spawn(cuda_process_initialization,
                                 nprocs=world_size,
                                 join=False)
    cpu_processes = []
    for rank in range(world_size,world_size+no_of_saver_processes):
        p = mp.Process(target=cpu_process_initialization,
                       args=(rank,))
        p.start()
        cpu_processes.append(p)
    for p in cpu_processes: p.join()
    trainer_processes.join()