Python (Pytorch) Multiprocessing throwing errors: Connection reset by peer and File Not Found
I'm trying to get something working similarly to keras' "fit_generator" method. Basically, I have a (very) large data file of mini-batches and I want to have my CPU grab mini-batches and populate a queue parallel to my GPU taking mini-batches from the queue and training on them. By having the CPU work in parallel to the GPU (as opposed to having the CPU grab a batch and making the GPU wait for the CPU before it trains on that batch) I should be able to reduce my training time by about half. I have benchmarked how long it takes the CPU to grab a mini-batch, and it's taking a comparable amount of time to how long it takes my GPU to train on one mini-batch, so parallelizing the CPU and GPU should work alright. I haven't found a built-in method in pytorch to do this, if there is one, please let me know. 

So I have tried to use the torch.multiprocessing module to do what I want, but I'm not able to complete the training as I always get some sort of error right before training is completed. The torch.multiprocessing module should be a wrapper with essentially all the same functionalities as the regular multiprocessing module except it allows pytorch tensors to be shared between processes. Basically, I have set up my code to have 2 functions, a loader function, and a trainer function like so:

    def data_gen(que,PATH,epochs,steps_per_epoch,batch_size=32):
        for epoch in range(epochs):
            for j in range(steps_per_epoch):
                with h5py.File(PATH,'r') as f:
                    X = f['X'][j*batch_size:(j+1)*batch_size]
                    Y = f['Y'][j*batch_size:(j+1)*batch_size]

                    X = autograd.Variable(torch.Tensor(X).resize_(batch_size,256,25).cpu())
                    Y = autograd.Variable(torch.Tensor(Y).cpu())

                    que.put((X,Y))
            que.put('stop')
        que.close()
        return

    def train_network(que,net,optimizer,epochs):
        print('Training for %s epochs...' %epochs)
        for epoch in range(epochs):
            while(True):
                data = que.get()
                if(data == 'stop'):
                    break
                net.zero_grad()
                net.hid = net.init_hid()
                inp,labels = data
                inp   = inp.cuda()
                labels = labels.cuda()
                out,hid = net(inp)
                loss = F.binary_cross_entropy(out,labels)
                loss.backward()
                optimizer.step()
            print('Epoch end reached')
        return


And then I run the two processes in parallel like so:

    if __name__ == '__main__':
        tmp.set_start_method('spawn')
        que = tmp.Queue(maxsize=10)
        loader = tmp.Process(target=data_gen, args=(que,PATH,epochs,steps), kwargs={'batch_size':batch_size})
        loader.start()
        trainer = tmp.Process(target=train_network, args=(que,net,optimizer,epochs,steps))
        trainer.start()
        loader.join()
        trainer.join()


I have the que put in a 'stop' value at the end of each epoch so I can break out of the loop in the trainer and go to the next epoch. This "poison pill" method appears to work because the code runs for multiple epochs, and the trainer does in fact print the end of epoch verification message. The code runs, and it does appear to speed up the training process (I've been trying to prototype this code on a small subset of the data so it's sometimes hard to tell how much speed up I'm getting), but at the end of training (and always at the end, no matter how many epochs I specify), I always get an error:

    Process Process-2:
    Traceback (most recent call last):
      File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
        self.run()
      File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
        self._target(*self._args, **self._kwargs)
      File "/media/digitalstorm/Storage/RNN_Prototype/Lazuli_rnnprototype.py", line 307, in train_network
        data = que.get()
      File "/usr/lib/python3.6/multiprocessing/queues.py", line 113, in get
        return _ForkingPickler.loads(res)
      File "/usr/local/lib/python3.6/dist-packages/torch/multiprocessing/reductions.py", line 70, in rebuild_storage_fd
        fd = df.detach()
      File "/usr/lib/python3.6/multiprocessing/resource_sharer.py", line 57, in detach
        with _resource_sharer.get_connection(self._id) as conn:
      File "/usr/lib/python3.6/multiprocessing/resource_sharer.py", line 87, in get_connection
        c = Client(address, authkey=process.current_process().authkey)
      File "/usr/lib/python3.6/multiprocessing/connection.py", line 487, in Client
        c = SocketClient(address)
      File "/usr/lib/python3.6/multiprocessing/connection.py", line 614, in SocketClient
        s.connect(address)
    FileNotFoundError: [Errno 2] No such file or directory


or, if I mess around a bit with various options, I sometimes get an error like this:

    Process Process-2:
    Traceback (most recent call last):
      File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
        self.run()
      File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
        self._target(*self._args, **self._kwargs)
      File "/media/digitalstorm/Storage/RNN_Prototype/Lazuli_rnnprototype.py", line 306, in train_network
        data = que.get()
      File "/usr/lib/python3.6/multiprocessing/queues.py", line 113, in get
        return _ForkingPickler.loads(res)
      File "/usr/local/lib/python3.6/dist-packages/torch/multiprocessing/reductions.py", line 70, in rebuild_storage_fd
        fd = df.detach()
      File "/usr/lib/python3.6/multiprocessing/resource_sharer.py", line 58, in detach
        return reduction.recv_handle(conn)
      File "/usr/lib/python3.6/multiprocessing/reduction.py", line 182, in recv_handle
        return recvfds(s, 1)[0]
      File "/usr/lib/python3.6/multiprocessing/reduction.py", line 153, in recvfds
        msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(bytes_size))
    ConnectionResetError: [Errno 104] Connection reset by peer


I don't know where I'm going wrong. Admittedly, I am a novice at multiprocessing, so it's hard for me to debug what went wrong exactly. Any help would be appreciated, thanks!

Since there hasn't been any movement on this question, I'll just post my own workaround to this problem. Basically the loader process was shutting down the que after it finished processing and queuing up the examples. It wasn't waiting for the trainer process to finish, so when the trainer process was going to get the next minibatch, it couldn't find it. I don't quite know why the loader process was prematurely shutting down the que, the documentation for que.close() says that this should only tell the que no more objects are being sent to the que, but it shouldn't actually shut down the que. Also, deleting que.close() did not solve the issue, so I don't think the error had to do with that command. What solved this issue for me was putting a time.sleep(2) command after the que.close() command. This forces the que to sleep for a couple of seconds after it finishes putting everything in the que, and it allows the program to complete and exit without error.    
