File exists: '/000000_epoch_shape' when using the ddp strategy from pytorch lightning

Environment

• OS: Databricks runtime 15.3ML with mosaicml streaming 0.8.1.
• Hardware (GPU, or instance type): g4dn.12xlarge

To reproduce

Steps to reproduce the behavior:

def get_dataloader_with_mosaic(path, batch_size, shuffle=False):
  # Utility function to clean up stale shared memory during distributed training
  clean_stale_shared_memory()

  # Creating the `StreamingDataset` object and the `StreamingDataLoader` object.
  dataset = StreamingDataset(local=path, shuffle=shuffle, batch_size=batch_size)
  return StreamingDataLoader(dataset, batch_size=batch_size, num_workers=31, drop_last=True, persistent_workers=True), dataset

eval_dataloader, eval_dataset = get_dataloader_with_mosaic(f"{data_storage_location}/mds_{experiment_name}_val", batch_size=256, shuffle=False)
train_dataloader, train_dataset = get_dataloader_with_mosaic(f"{data_storage_location}/mds_{experiment_name}_train", batch_size=32, shuffle=True)

trainer = pl.Trainer(
    accelerator='gpu', 
    devices=4, 
    strategy='ddp_notebook',
    max_epochs=10, 
    num_sanity_val_steps=0
)

trainer.fit(pretrainer, train_dataloader, val_dataloaders=eval_dataloader)

Expected behavior

I'd expect training to begin.

Additional context

-- Process 2 terminated with the following error:
Traceback (most recent call last):
  File "/databricks/python/lib/python3.11/site-packages/torch/multiprocessing/spawn.py", line 75, in _wrap
    fn(i, *args)
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/strategies/launchers/multiprocessing.py", line 173, in _wrapping_function
    results = function(*args, **kwargs)
              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/trainer/trainer.py", line 574, in _fit_impl
    self._run(model, ckpt_path=ckpt_path)
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/trainer/trainer.py", line 981, in _run
    results = self._run_stage()
              ^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/trainer/trainer.py", line 1025, in _run_stage
    self.fit_loop.run()
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/loops/fit_loop.py", line 205, in run
    self.advance()
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/loops/fit_loop.py", line 363, in advance
    self.epoch_loop.run(self._data_fetcher)
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/loops/training_epoch_loop.py", line 140, in run
    self.advance(data_fetcher)
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/loops/training_epoch_loop.py", line 212, in advance
    batch, _, __ = next(data_fetcher)
                   ^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/loops/fetchers.py", line 133, in __next__
    batch = super().__next__()
            ^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/loops/fetchers.py", line 60, in __next__
    batch = next(self.iterator)
            ^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/utilities/combined_loader.py", line 341, in __next__
    out = next(self._iterator)
          ^^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/pytorch_lightning/utilities/combined_loader.py", line 78, in __next__
    out[i] = next(self.iterators[i])
             ^^^^^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/streaming/base/dataloader.py", line 58, in __iter__
    for batch in super().__iter__():
  File "/databricks/python/lib/python3.11/site-packages/torch/utils/data/dataloader.py", line 631, in __next__
    data = self._next_data()
           ^^^^^^^^^^^^^^^^^
  File "/databricks/python/lib/python3.11/site-packages/torch/utils/data/dataloader.py", line 1346, in _next_data
    return self._process_data(data)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/databricks/python/lib/python3.11/site-packages/torch/utils/data/dataloader.py", line 1372, in _process_data
    data.reraise()
  File "/databricks/python/lib/python3.11/site-packages/torch/_utils.py", line 705, in reraise
    raise exception
FileExistsError: Caught FileExistsError in DataLoader worker process 0.
Original Traceback (most recent call last):
  File "/databricks/python/lib/python3.11/site-packages/torch/utils/data/_utils/worker.py", line 308, in _worker_loop
    data = fetcher.fetch(index)  # type: ignore[possibly-undefined]
           ^^^^^^^^^^^^^^^^^^^^
  File "/databricks/python/lib/python3.11/site-packages/torch/utils/data/_utils/fetch.py", line 32, in fetch
    data.append(next(self.dataset_iter))
                ^^^^^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/streaming/base/dataset.py", line 1501, in __iter__
    sample_ids = self._get_work(epoch, sample_in_epoch)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/streaming/base/dataset.py", line 1038, in _get_work
    shape_shm, data_shm = self._share_work(epoch_sample_ids)
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/streaming/base/dataset.py", line 953, in _share_work
    shape_shm = SharedMemory(name=name, create=True, size=size, auto_cleanup=False)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-dc84ac28-3e23-4bab-908e-384148539e68/lib/python3.11/site-packages/streaming/base/shared/memory.py", line 41, in __init__
    shm = BuiltinSharedMemory(name, create, size)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/multiprocessing/shared_memory.py", line 104, in __init__
    self._fd = _posixshmem.shm_open(
               ^^^^^^^^^^^^^^^^^^^^^
FileExistsError: [Errno 17] File exists: '/000000_epoch_shape'

Start with streaming/base/dataset.py, especially _get_work and _share_work, and then inspect streaming/base/shared/memory.py where SharedMemory is created. Reproduce the Databricks/PyTorch Lightning DDP setup with the two StreamingDataLoader instances and persistent workers. Done means training begins without the FileExistsError for /000000_epoch_shape.