Concurrency bug when using same cache among several jobs

## Describe the bug
I used to see this bug with an older version of the datasets. It seems to persist. 

This is my concrete scenario: I launch several evaluation jobs on a cluster in which I share the file system and I share the cache directory used by huggingface libraries. The evaluation jobs read the same *.csv files. If my jobs get all scheduled pretty much at the same time, there are all kinds of weird concurrency errors. Sometime it crashes silently. This time I got lucky that it crashed with a stack trace that I can share and maybe you get to the bottom of this. If you don't have a similar setup available, it may be hard to reproduce as you really need two jobs accessing the same file at the same time to see this type of bug.  

## Steps to reproduce the bug
I'm running a modified version of `run_glue.py` script adapted to my use case. I've seen the same problem when running some glue datasets as well (so it's not specific to loading the datasets from csv files).  

## Expected results
No crash, concurrent access to the (intermediate) files just fine.

## Actual results
Crashes due to races/concurrency bugs. 

## Environment info
<!-- You can run the command `datasets-cli env` and copy-and-paste its output below. -->
- `datasets` version: 2.3.2
- Platform: Linux-4.18.0-348.23.1.el8_5.x86_64-x86_64-with-glibc2.10
- Python version: 3.8.5
- PyArrow version: 8.0.0
- Pandas version: 1.1.0

Stack trace that I just got with the crash (I've obfuscated some names, it should still be quite informative):

```
Running tokenizer on dataset:   0%|          | 0/3 [00:00<?, ?ba/s]
Traceback (most recent call last):
  File "../../src/models//run_*******.py", line 600, in <module>
    main()
  File "../../src/models//run_*******.py", line 444, in main
    raw_datasets = raw_datasets.map(
  File "/*******//envs/tr-crt/lib/python3.8/site-packages/datasets/dataset_dict.py", line 770, in map
    {
  File "/*******//envs/tr-crt/lib/python3.8/site-packages/datasets/dataset_dict.py", line 771, in <dictcomp>
    k: dataset.map(
  File "/*******//envs/tr-crt/lib/python3.8/site-packages/datasets/arrow_dataset.py", line 2376, in map
    return self._map_single(
  File "/*******/envs/tr-crt/lib/python3.8/site-packages/datasets/arrow_dataset.py", line 551, in wrapper
    out: Union["Dataset", "DatasetDict"] = func(self, *args, **kwargs)
  File "/*******//envs/tr-crt/lib/python3.8/site-packages/datasets/arrow_dataset.py", line 518, in wrapper
    out: Union["Dataset", "DatasetDict"] = func(self, *args, **kwargs)
  File "/*******/envs/tr-crt/lib/python3.8/site-packages/datasets/fingerprint.py", line 458, in wrapper
    out = func(self, *args, **kwargs)
  File "/*******//envs/tr-crt/lib/python3.8/site-packages/datasets/arrow_dataset.py", line 2776, in _map_single
    buf_writer, writer, tmp_file = init_buffer_and_writer()
  File "/*******//envs/tr-crt/lib/python3.8/site-packages/datasets/arrow_dataset.py", line 2696, in init_buffer_and_writer
    tmp_file = tempfile.NamedTemporaryFile("wb", dir=os.path.dirname(cache_file_name), delete=False)
  File "/*******//envs/tr-crt/lib/python3.8/tempfile.py", line 541, in NamedTemporaryFile
    (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
  File "/*******//envs/tr-crt/lib/python3.8/tempfile.py", line 250, in _mkstemp_inner
    fd = _os.open(file, flags, 0o600)
FileNotFoundError: [Errno 2] No such file or directory: '/*******/cache-transformers//transformers/csv/default-ef9cd184210742a7/0.0.0/51cce309a08df9c4d82ffd9363bbe090bf173197fc01a71b034e8594995a1a58/tmps8l6j5yc'
```

As I ran 100s of experiments last year for an empirical paper, I ran into this type of bugs several times. I found several bandaid/work-arounds, e.g., run one job first that caches the dataset => eliminate concurrency; OR use unique caches => eliminate concurrency (but increase storage space), etc. and it all works fine. 

I'd like to help you fixing this bug as it's really annoying to always apply the work arounds. Let me know what other info from my side could help you figure out the issue.

Thanks for your help!


Comments:
I can confirm that if I run one job first that processes the dataset, then I can run any jobs in parallel with no problem (no write-concurrency anymore...). 

Hi! That's weird. It seems like the error points to the `mkstemp` function, but the official docs state the following:
```
There are no race conditions in the file’s creation, assuming that the platform properly implements the [os.O_EXCL](https://docs.python.org/3/library/os.html#os.O_EXCL) flag for [os.open()](https://docs.python.org/3/library/os.html#os.open)
```
So this could mean your platform doesn't support that flag.

~~Can you please check if wrapping the temp file creation (the line `tmp_file = tempfile.NamedTemporaryFile("wb", dir=os.path.dirname(cache_file_name), delete=False)` in `_map_single`) with the `multiprocess.Lock` fixes the issue?~~
Perhaps wrapping the temp file creation in `_map_single`  with `filelock` could work:
```python
with FileLock(lock_path):
    tmp_file = tempfile.NamedTemporaryFile("wb", dir=os.path.dirname(cache_file_name), delete=False)
```
Can you please check if that helps?

**Edit**: while writing my comment I took the time the read previous comments. By wrapping `dl_manager.download_and_extract` with a **FileLock** it works like a charm ! Thx @mariosasko 

OS : MacOS 14.7.4 (intel)
Python : 3.12
datasets : 3.5.0

Adding to this, I had a similar problem when 2 process concurrently load different subsets of the same dataset that needs to be extracted. The use case is similar as OP : running a benchmark.

The dataloader needs to download and extract a zip file, with ~20 ~100Mo files.

When 2 processes executes `load_dataset("TESTLOAD.py", name="a", trust_remote_code=True)` at the same time it is fine (there must some lock on `("TESTLOAD.py", "a")`).
But when running `load_dataset("TESTLOAD.py", name="a", trust_remote_code=True)` and `load_dataset("TESTLOAD.py", name="b", trust_remote_code=True)` (cf. `test.py`.

Here is what I managed to understand as a table using the scripts below. Step 3 is attested by the `os.listdir` in `TESTLOAD.py`.

| steps | process a | process b |
|---|---|---|
| 1 | download | wait |
| 2 | end of download | wait |
| 3 | extracting | FAIL to open a not yet extracted file |
| 4 | end of extraction | KO |
| 5 | OK | KO |

<details>

<summary>TESTLOAD.py (dataloader)</summary>

```python
import os
import datasets
# from filelock import FileLock
_URL = "/Users/ygallina/Documents/dr-benchmark/GSC-v1.1_big.zip"

class TESTLOAD(datasets.GeneratorBasedBuilder):

	BUILDER_CONFIGS = [
		datasets.BuilderConfig(name='a'),
		datasets.BuilderConfig(name='b')
	]

	def _info(self):
		features = datasets.Features({
			"id": datasets.Value("string"),
		})
		return datasets.DatasetInfo(features=features)

	def _split_generators(self, dl_manager):
		c2p = {'a': "Medline_GSC_en_fr_man.xml", 'b': "Medline_GSC_en_es_man.xml"}
		# with FileLock("path/to/tmp.lock"):
		data_dir = dl_manager.download_and_extract(_URL)

		print(os.listdir(data_dir))

		data_dir = data_dir + "/" + c2p[self.config.name]
		return [datasets.SplitGenerator(
			name=datasets.Split.TRAIN,
			gen_kwargs={"data_dir": data_dir}
		)]

	def _generate_examples(self, data_dir):
		f = open(data_dir)
		f.close()
		yield 0, {'id': data_dir}
```

</details>


<details>

<summary>test.py (main file)</summary>

Commands to execute the test
```bash
rm -rf "test"
HF_HOME="test" python test.py
```

```python
import os
from datasets import load_dataset

# Forking to make sure access will be concurrent
# Waiting to fork after imports (because it takes a while)
child = os.fork()

if child:
    print('child')
    ds = load_dataset('TESTLOAD.py', 'a', trust_remote_code=True)
    print(f"child, {ds['train'][0]}")
else:
    print('parent')
    ds = load_dataset('TESTLOAD.py', 'b', trust_remote_code=True)
    print(f"parent, {ds['train'][0]}")
```

</details>