Text: [HELP] SentencePiece is not compatible with DataLoader with the Windows platform

[zhangguanheng66]

We added a test to cover the compatibility between SetencePiece and DataLoader. The test passes in the Linux platform but fails under the Windows platform. We need some experts to help debug.

self = <test.experimental.test_transforms_with_asset.TestTransformsWithAsset testMethod=test_sentencepiece_with_dataloader>

    def test_sentencepiece_with_dataloader(self):
        sp_model_path = download_from_url(PRETRAINED_SP_MODEL['text_bpe_25000'])
        spm_processor = sentencepiece_processor(sp_model_path)
        _path = os.path.join(self.project_root, '.data', 'text_bpe_25000.model')
        os.remove(_path)
        example_strings = ['the pretrained spm model names'] * 64
        ref_results = torch.tensor([[13, 1465, 12824, 304, 24935, 5771, 3776]] * 16, dtype=torch.long)

        def batch_func(data):
            return torch.tensor([spm_processor(text) for text in data], dtype=torch.long)

        dataloader = DataLoader(example_strings, batch_size=16, num_workers=2, collate_fn=batch_func)
>       for item in dataloader:

test\experimental\test_transforms_with_asset.py:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
env\lib\site-packages\torch\utils\data\dataloader.py:359: in __iter__
    return self._get_iterator()
env\lib\site-packages\torch\utils\data\dataloader.py:301: in _get_iterator
    return _MultiProcessingDataLoaderIter(self)
env\lib\site-packages\torch\utils\data\dataloader.py:885: in __init__
    w.start()
env\lib\multiprocessing\process.py:105: in start
    self._popen = self._Popen(self)
env\lib\multiprocessing\context.py:223: in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
env\lib\multiprocessing\context.py:322: in _Popen
    return Popen(process_obj)
env\lib\multiprocessing\popen_spawn_win32.py:65: in __init__
    reduction.dump(process_obj, to_child)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

obj = <Process(Process-1, initial daemon)>, file = <_io.BufferedWriter name=11>
protocol = None

    def dump(obj, file, protocol=None):
        '''Replacement for pickle.dump() using ForkingPickler.'''
>       ForkingPickler(file, protocol).dump(obj)
E       AttributeError: Can't pickle local object 'TestTransformsWithAsset.test_sentencepiece_with_dataloader.<locals>.batch_func'

env\lib\multiprocessing\reduction.py:60: AttributeError

cc @peterjc123 @maxluk @nbcsm @guyang3532 @gunandrose4u @smartcat2010 @mszhanyi

[zhangguanheng66]

cc @peterjc123. This may be another task for the MSFT team.

[peterjc123]

Nested functions are not pickle-able on Windows. Please move it to the global namespace.

[zhangguanheng66]

Thanks. Could you be more specific?

[peterjc123]

Change to sth. like this:

 def batch_func(data):
        return torch.tensor([spm_processor(text) for text in data], dtype=torch.long)

    @unittest.skipIf(platform.system() == "Windows", "Test is known to fail on Windows.")
    def test_sentencepiece_with_dataloader(self):
        sp_model_path = download_from_url(PRETRAINED_SP_MODEL['text_bpe_25000'])
        spm_processor = sentencepiece_processor(sp_model_path)
        _path = os.path.join(self.project_root, '.data', 'text_bpe_25000.model')
        os.remove(_path)
        example_strings = ['the pretrained spm model names'] * 64
        ref_results = torch.tensor([[13, 1465, 12824, 304, 24935, 5771, 3776]] * 16, dtype=torch.long)

        dataloader = DataLoader(example_strings, batch_size=16, num_workers=2, collate_fn=batch_func)
        for item in dataloader:
            self.assertEqual(item, ref_results)

batch_func is a nested function in your PR and it won't work on Windows.

[cpuhrsch]

@zhangguanheng66 - It's an unfortunate reality of working with Python on Windows

[zhangguanheng66]

Change to sth. like this:

 def batch_func(data):
        return torch.tensor([spm_processor(text) for text in data], dtype=torch.long)

    @unittest.skipIf(platform.system() == "Windows", "Test is known to fail on Windows.")
    def test_sentencepiece_with_dataloader(self):
        sp_model_path = download_from_url(PRETRAINED_SP_MODEL['text_bpe_25000'])
        spm_processor = sentencepiece_processor(sp_model_path)
        _path = os.path.join(self.project_root, '.data', 'text_bpe_25000.model')
        os.remove(_path)
        example_strings = ['the pretrained spm model names'] * 64
        ref_results = torch.tensor([[13, 1465, 12824, 304, 24935, 5771, 3776]] * 16, dtype=torch.long)

        dataloader = DataLoader(example_strings, batch_size=16, num_workers=2, collate_fn=batch_func)
        for item in dataloader:
            self.assertEqual(item, ref_results)

batch_func is a nested function in your PR and it won't work on Windows.

Thanks for the help. I see what you means for the nested function.

[zhangguanheng66]

Re-open this issue and help wanted. I see some random Windows CI tests failed in https://github.com/pytorch/text/pull/1068. DataLoader calls out timeout in the error message below.

self = <torch.utils.data.dataloader._MultiProcessingDataLoaderIter object at 0x000001A8B2296898>
timeout = 5.0

    def _try_get_data(self, timeout=_utils.MP_STATUS_CHECK_INTERVAL):
        # Tries to fetch data from `self._data_queue` once for a given timeout.
        # This can also be used as inner loop of fetching without timeout, with
        # the sender status as the loop condition.
        #
        # This raises a `RuntimeError` if any worker died expectedly. This error
        # can come from either the SIGCHLD handler in `_utils/signal_handling.py`
        # (only for non-Windows platforms), or the manual check below on errors
        # and timeouts.
        #
        # Returns a 2-tuple:
        #   (bool: whether successfully get data, any: data if successful else None)
        try:
>           data = self._data_queue.get(timeout=timeout)

env\lib\site-packages\torch\utils\data\dataloader.py:956: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <multiprocessing.queues.Queue object at 0x000001A8B22969B0>, block = True
timeout = 5.0

    def get(self, block=True, timeout=None):
        if block and timeout is None:
            with self._rlock:
                res = self._recv_bytes()
            self._sem.release()
        else:
            if block:
                deadline = time.monotonic() + timeout
            if not self._rlock.acquire(block, timeout):
                raise Empty
            try:
                if block:
                    timeout = deadline - time.monotonic()
                    if not self._poll(timeout):
>                       raise Empty
E                       queue.Empty

env\lib\multiprocessing\queues.py:105: Empty

The above exception was the direct cause of the following exception:

self = <test.experimental.test_transforms_with_asset.TestTransformsWithAsset testMethod=test_sentencepiece_with_dataloader>

    def test_sentencepiece_with_dataloader(self):
        example_strings = ['the pretrained spm model names'] * 64
        ref_results = torch.tensor([[13, 1465, 12824, 304, 24935, 5771, 3776]] * 16, dtype=torch.long)
        dataloader = DataLoader(example_strings, batch_size=16, num_workers=2,
                                collate_fn=batch_func)
>       for item in dataloader:

test\experimental\test_transforms_with_asset.py:187: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
env\lib\site-packages\torch\utils\data\dataloader.py:519: in __next__
    data = self._next_data()
env\lib\site-packages\torch\utils\data\dataloader.py:1152: in _next_data
    idx, data = self._get_data()
env\lib\site-packages\torch\utils\data\dataloader.py:1118: in _get_data
    success, data = self._try_get_data()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <torch.utils.data.dataloader._MultiProcessingDataLoaderIter object at 0x000001A8B2296898>
timeout = 5.0

    def _try_get_data(self, timeout=_utils.MP_STATUS_CHECK_INTERVAL):
        # Tries to fetch data from `self._data_queue` once for a given timeout.
        # This can also be used as inner loop of fetching without timeout, with
        # the sender status as the loop condition.
        #
        # This raises a `RuntimeError` if any worker died expectedly. This error
        # can come from either the SIGCHLD handler in `_utils/signal_handling.py`
        # (only for non-Windows platforms), or the manual check below on errors
        # and timeouts.
        #
        # Returns a 2-tuple:
        #   (bool: whether successfully get data, any: data if successful else None)
        try:
            data = self._data_queue.get(timeout=timeout)
            return (True, data)
        except Exception as e:
            # At timeout and error, we manually check whether any worker has
            # failed. Note that this is the only mechanism for Windows to detect
            # worker failures.
            failed_workers = []
            for worker_id, w in enumerate(self._workers):
                if self._workers_status[worker_id] and not w.is_alive():
                    failed_workers.append(w)
                    self._mark_worker_as_unavailable(worker_id)
            if len(failed_workers) > 0:
                pids_str = ', '.join(str(w.pid) for w in failed_workers)
>               raise RuntimeError('DataLoader worker (pid(s) {}) exited unexpectedly'.format(pids_str)) from e
E               RuntimeError: DataLoader worker (pid(s) 3484) exited unexpectedly

env\lib\site-packages\torch\utils\data\dataloader.py:969: RuntimeError

[peterjc123]

@zhangguanheng66 Does the test pass when num_workers=0?

[zhangguanheng66]

@zhangguanheng66 Does the test pass when num_workers=0?

The tests pass if setting num_workers to 0

[peterjc123]

If the job is urgent, you may make the num_workers=0 on Windows and revert the changes for non-Windows and land the PR. The debugging will take quite a bit time.

[zhangguanheng66]

I will bypass this test for Windows platform.