Proper way to shut down sub-processes in a dataloader

[bask0]

Hello,

I have a dataloader (with num_workers=0) that creates a number of sub-processes to fill a queue. What is the proper way to signal the dataloader to shut down its processes when the training loop is executed by pytorch lightning?

I can iterate the dataloader manually like this and everything works fine:

class MyIterableDataset():
  def __iter__(self):
    try:
      processes = [mp.Process(...) for _ in range(8)]
      yield ...
    finally:
      process.terminate()
      process.join()

dataloader = Dataloader(MyIterableDataset())

for epoch in range(2):
  for batch in dataloader:
    pass

I simply use the try finally pattern to shut down the sub-processes after each epoch in finally. But when I run a loop with pytorch lightning (via trainer.fit, for example), I get a BrokenPipe error, because the finally block is never executed. Pytorch lightning shuts the loop down from externally, i.e., it never reaches finally. What is the mechanism used by pytorch lightning to stop the iteration (at the end of an epoch or when reaching limit_train_batches etc.)? I tried to figure it out from the source code but did not manage.

Thanks

[bask0]

The issue was not related to pytorch lightning

Turns out the BrokenPipe was caused by me shutting down the queue manager before shutting down the processes