Added file_lock in IoPathSaverIterDataPipe

[msaroufim]

Untested, made this PR to make communication easier

Partially fixes #144

The main change is to lock a file when multiple processes are writing to the same file

The fix was upstreamed from an internal fix @hudeven made

with iopath.file_lock(filepath):
    if not os.path.exists(filepath):
        with self.pathmgr.open(filepath, self.mode) as f:
            f.write(data)

And the test was extended from a gist @ejguan sent me

It does seem to be working as intended right now, @parmeet please let me know if this works for you

Logs with fix

Ran 17 tests in 3.982s

OK (skipped=1)

Logs without fix

Traceback (most recent call last):
  File "/home/ubuntu/data/test/test_local_io.py", line 638, in test_io_path_saver_iterdatapipe
    self.assertEqual(expected_paths, res_file_paths)
AssertionError: Lists differ: ['/tmp/tmpbgt32c3h/1.txt', '/tmp/tmpbgt32c3h/2.txt', '/tmp/tmpbgt32c3h/3.txt'] != ['/tmp/tmpbgt32c3h/3.txt']

First differing element 0:
'/tmp/tmpbgt32c3h/1.txt'
'/tmp/tmpbgt32c3h/3.txt'

First list contains 2 additional elements.
First extra element 1:
'/tmp/tmpbgt32c3h/2.txt'

- ['/tmp/tmpbgt32c3h/1.txt', '/tmp/tmpbgt32c3h/2.txt', '/tmp/tmpbgt32c3h/3.txt']
+ ['/tmp/tmpbgt32c3h/3.txt']

[msaroufim]

As far as the test goes it seems like the IoPathSaver is tested here only https://github.com/pytorch/data/blob/main/test/test_local_io.py#L635

So will extend this test to multiple processes and see how CI works on Windows and Linux

@parmeet do you have some existing test/script I can leverage? Just wanna make sure your usecase is solved

[parmeet]

do you have some existing test/script I can leverage? Just wanna make sure your usecase is solved

Unfortunately, no. we observed this issue when running DDP while training doc classification workflow using TorchRecipes internally. cc: @hudeven

[VitalyFedyunin]

It does not fixes #144 as on_disk_cache using

data/torchdata/datapipes/iter/util/saver.py

Line 18 in 2623265

class SaverIterDataPipe(IterDataPipe[str]):

to save files.

Also you probably need to cross lock writing with

data/torchdata/datapipes/iter/util/cacheholder.py

Line 185 in c1d89fe

def _cache_check_fn(data, filepath_fn, hash_dict, hash_type, extra_check_fn):

@ejguan please correct me if I'm wrong.

[VitalyFedyunin]

Side comment: I'm ok to have portalocker as hard dependency.

[ejguan]

It does not fixes #144 as on_disk_cache using

data/torchdata/datapipes/iter/util/saver.py

Line 18 in 2623265

class SaverIterDataPipe(IterDataPipe[str]):

to save files.
Also you probably need to cross lock writing with

data/torchdata/datapipes/iter/util/cacheholder.py

Line 185 in c1d89fe

def _cache_check_fn(data, filepath_fn, hash_dict, hash_type, extra_check_fn):

@ejguan please correct me if I'm wrong.

Completely correct. We are using this iopath file to test portalocker. We can add another PR to fix Saver and CacheHolder

[ejguan]

LGTM with one comment on test.

[facebook-github-bot]

@msaroufim has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

[ejguan]

Could you please fix the followings before landing? Thank you!

[facebook-github-bot]

@msaroufim has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

[ejguan]

Could you please fix lint as well in this file? Really appreciate.

[msaroufim]

Note to self - there's now a genuine failure since I refactored self.fn to be a global function in the test class, as far as this saver goes, I need to make sure this gets fixed

    def __iter__(self) -> Iterator[str]:
        for filepath, data in self.source_datapipe:
            if self.fn is not None:
>               filepath = self.fn(filepath)
E               TypeError: filepath_fn() missing 1 required positional argument: 'name'
E               This exception is thrown by __iter__ of SaverIterDataPipe()

[facebook-github-bot]

@msaroufim has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

[facebook-github-bot]

@msaroufim has imported this pull request. If you are a Meta employee, you can view this diff on Phabricator.

[ejguan]

I think you might need to rebase onto main branch again. There are lots of changes that are not supposed to be in this PR.