TimeoutError in map
CONTRIBUTOR
From my current understanding, this timeout is only used when we need to get the results.

One of:
1. All tasks are done
2. One worker died

Your function should work fine and it's definitely a bug if it doesn't.
2024-04-08T14:47:16Z
https://github.com/huggingface/datasets/issues/6787#issuecomment-2042955176
https://api.github.com/repos/huggingface/datasets/issues/6787
IC_kwDODunzps55xQWo
https://api.github.com/repos/huggingface/datasets/issues/comments/2042955176/reactions
2024-04-08T14:47:16Z
https://api.github.com/repos/huggingface/datasets/issues/comments/2042955176
https://avatars.githubusercontent.com/u/8976546?v=4
https://api.github.com/users/Dref360/events{/privacy}
https://api.github.com/users/Dref360/followers
https://api.github.com/users/Dref360/following{/other_user}
https://api.github.com/users/Dref360/gists{/gist_id}
https://github.com/Dref360
Dref360
MDQ6VXNlcjg5NzY1NDY=
https://api.github.com/users/Dref360/orgs
https://api.github.com/users/Dref360/received_events
https://api.github.com/users/Dref360/repos
https://api.github.com/users/Dref360/starred{/owner}{/repo}
https://api.github.com/users/Dref360/subscriptions
User
https://api.github.com/users/Dref360
public
COLLABORATOR
When one of the `map`'s worker processes crashes, the linked code re-raises an error from the crash and returns it to the caller.

If your question is how to limit the time of long-running tasks/worker processes, such functionality doesn't exist in `datasets` (yet), which means you need to implement it yourself.

E.g., you can implement it using the built-in `signal` module like this:
```python
import time
import signal
from contextlib import contextmanager

from datasets import Dataset


@contextmanager
def max_exec_time(t):
  def raise_timeout_handler(signum, frame):
    raise TimeoutError
  
  orig_handler = signal.getsignal(signal.SIGALRM)
  signal.signal(signal.SIGALRM, raise_timeout_handler)
  try:
    signal.alarm(t)
    yield
  finally:
    signal.alarm(0)
    signal.signal(signal.SIGALRM, orig_handler)


def worker(example, rank):
    try:
      with max_exec_time(20):  # 20 sec execution limit
        if rank % 2 == 0:
          time.sleep(50)  # simulate a long-running task
        example["a"] = 100
    except TimeoutError:
      example["a"] = None  # Or return empty batches here in the "batched" mode
    return example

data = Dataset.from_list([{"a": 1}, {"a": 2}])
data = data.map(worker, num_proc=2, with_rank=True)
print(data[0])
```
2024-04-11T18:41:16Z
https://github.com/huggingface/datasets/issues/6787#issuecomment-2050298541
https://api.github.com/repos/huggingface/datasets/issues/6787
IC_kwDODunzps56NRKt
https://api.github.com/repos/huggingface/datasets/issues/comments/2050298541/reactions
2024-04-11T18:41:16Z
https://api.github.com/repos/huggingface/datasets/issues/comments/2050298541
https://avatars.githubusercontent.com/u/47462742?v=4
https://api.github.com/users/mariosasko/events{/privacy}
https://api.github.com/users/mariosasko/followers
https://api.github.com/users/mariosasko/following{/other_user}
https://api.github.com/users/mariosasko/gists{/gist_id}
https://github.com/mariosasko
mariosasko
MDQ6VXNlcjQ3NDYyNzQy
https://api.github.com/users/mariosasko/orgs
https://api.github.com/users/mariosasko/received_events
https://api.github.com/users/mariosasko/repos
https://api.github.com/users/mariosasko/starred{/owner}{/repo}
https://api.github.com/users/mariosasko/subscriptions
User
https://api.github.com/users/mariosasko
public
CONTRIBUTOR
> From my current understanding, this timeout is only used when we need to get the results.
> 
> One of:
> 
> 1. All tasks are done
> 2. One worker died
> 
> Your function should work fine and it's definitely a bug if it doesn't.

thanks for responding! can you reproduce the stuck with the above example code?
2024-04-12T15:10:27Z
https://github.com/huggingface/datasets/issues/6787#issuecomment-2051951707
https://api.github.com/repos/huggingface/datasets/issues/6787
IC_kwDODunzps56Tkxb
https://api.github.com/repos/huggingface/datasets/issues/comments/2051951707/reactions
2024-04-12T15:10:27Z
https://api.github.com/repos/huggingface/datasets/issues/comments/2051951707
https://avatars.githubusercontent.com/u/48146603?v=4
https://api.github.com/users/Jiaxin-Wen/events{/privacy}
https://api.github.com/users/Jiaxin-Wen/followers
https://api.github.com/users/Jiaxin-Wen/following{/other_user}
https://api.github.com/users/Jiaxin-Wen/gists{/gist_id}
https://github.com/Jiaxin-Wen
Jiaxin-Wen
MDQ6VXNlcjQ4MTQ2NjAz
https://api.github.com/users/Jiaxin-Wen/orgs
https://api.github.com/users/Jiaxin-Wen/received_events
https://api.github.com/users/Jiaxin-Wen/repos
https://api.github.com/users/Jiaxin-Wen/starred{/owner}{/repo}
https://api.github.com/users/Jiaxin-Wen/subscriptions
User
https://api.github.com/users/Jiaxin-Wen
public
CONTRIBUTOR
> When one of the `map`'s worker processes crashes, the linked code re-raises an error from the crash and returns it to the caller.
> 
> If your question is how to limit the time of long-running tasks/worker processes, such functionality doesn't exist in `datasets` (yet), which means you need to implement it yourself.
> 
> E.g., you can implement it using the built-in `signal` module like this:
> 
> ```python
> import time
> import signal
> from contextlib import contextmanager
> 
> from datasets import Dataset
> 
> 
> @contextmanager
> def max_exec_time(t):
>   def raise_timeout_handler(signum, frame):
>     raise TimeoutError
>   
>   orig_handler = signal.getsignal(signal.SIGALRM)
>   signal.signal(signal.SIGALRM, raise_timeout_handler)
>   try:
>     signal.alarm(t)
>     yield
>   finally:
>     signal.alarm(0)
>     signal.signal(signal.SIGALRM, orig_handler)
> 
> 
> def worker(example, rank):
>     try:
>       with max_exec_time(20):  # 20 sec execution limit
>         if rank % 2 == 0:
>           time.sleep(50)  # simulate a long-running task
>         example["a"] = 100
>     except TimeoutError:
>       example["a"] = None  # Or return empty batches here in the "batched" mode
>     return example
> 
> data = Dataset.from_list([{"a": 1}, {"a": 2}])
> data = data.map(worker, num_proc=2, with_rank=True)
> print(data[0])
> ```

thanks for responding! However, I don't think we should use `signal` in the context of multiprocessing since sometimes it will crash one process and raise the following error
https://github.com/huggingface/datasets/blob/c3ddb1ef00334a6f973679a51e783905fbc9ef0b/src/datasets/utils/py_utils.py#L664
2024-04-12T15:12:40Z
https://github.com/huggingface/datasets/issues/6787#issuecomment-2051955393
https://api.github.com/repos/huggingface/datasets/issues/6787
IC_kwDODunzps56TlrB
https://api.github.com/repos/huggingface/datasets/issues/comments/2051955393/reactions
2024-04-12T15:12:40Z
https://api.github.com/repos/huggingface/datasets/issues/comments/2051955393
https://avatars.githubusercontent.com/u/48146603?v=4
https://api.github.com/users/Jiaxin-Wen/events{/privacy}
https://api.github.com/users/Jiaxin-Wen/followers
https://api.github.com/users/Jiaxin-Wen/following{/other_user}
https://api.github.com/users/Jiaxin-Wen/gists{/gist_id}
https://github.com/Jiaxin-Wen
Jiaxin-Wen
MDQ6VXNlcjQ4MTQ2NjAz
https://api.github.com/users/Jiaxin-Wen/orgs
https://api.github.com/users/Jiaxin-Wen/received_events
https://api.github.com/users/Jiaxin-Wen/repos
https://api.github.com/users/Jiaxin-Wen/starred{/owner}{/repo}
https://api.github.com/users/Jiaxin-Wen/subscriptions
User
https://api.github.com/users/Jiaxin-Wen
public
COLLABORATOR
> thanks for responding! However, I don't think we should use signal in the context of multiprocessing since sometimes it will crash one process and raise the following error

The above code has `try/except` to catch the error from the handler. Or do you get an error other than `TimeoutError`?
2024-04-12T17:08:03Z
https://github.com/huggingface/datasets/issues/6787#issuecomment-2052149289
https://api.github.com/repos/huggingface/datasets/issues/6787
IC_kwDODunzps56UVAp
https://api.github.com/repos/huggingface/datasets/issues/comments/2052149289/reactions
2024-04-12T17:08:03Z
https://api.github.com/repos/huggingface/datasets/issues/comments/2052149289
https://avatars.githubusercontent.com/u/47462742?v=4
https://api.github.com/users/mariosasko/events{/privacy}
https://api.github.com/users/mariosasko/followers
https://api.github.com/users/mariosasko/following{/other_user}
https://api.github.com/users/mariosasko/gists{/gist_id}
https://github.com/mariosasko
mariosasko
MDQ6VXNlcjQ3NDYyNzQy
https://api.github.com/users/mariosasko/orgs
https://api.github.com/users/mariosasko/received_events
https://api.github.com/users/mariosasko/repos
https://api.github.com/users/mariosasko/starred{/owner}{/repo}
https://api.github.com/users/mariosasko/subscriptions
User
https://api.github.com/users/mariosasko
public
CONTRIBUTOR
> > thanks for responding! However, I don't think we should use signal in the context of multiprocessing since sometimes it will crash one process and raise the following error
> 
> The above code has `try/except` to catch the error from the handler. Or do you get an error other than `TimeoutError`?

yup, it will raise the RuntimeError: https://github.com/huggingface/datasets/blob/c3ddb1ef00334a6f973679a51e783905fbc9ef0b/src/datasets/utils/py_utils.py#L667C19-L670C22

```
 raise RuntimeError(
                        "One of the subprocesses has abruptly died during map operation."
                        "To debug the error, disable multiprocessing."
                    )
```
2024-04-13T06:34:59Z
https://github.com/huggingface/datasets/issues/6787#issuecomment-2053525936
https://api.github.com/repos/huggingface/datasets/issues/6787
IC_kwDODunzps56ZlGw
https://api.github.com/repos/huggingface/datasets/issues/comments/2053525936/reactions
2024-04-13T06:34:59Z
https://api.github.com/repos/huggingface/datasets/issues/comments/2053525936
https://avatars.githubusercontent.com/u/48146603?v=4
https://api.github.com/users/Jiaxin-Wen/events{/privacy}
https://api.github.com/users/Jiaxin-Wen/followers
https://api.github.com/users/Jiaxin-Wen/following{/other_user}
https://api.github.com/users/Jiaxin-Wen/gists{/gist_id}
https://github.com/Jiaxin-Wen
Jiaxin-Wen
MDQ6VXNlcjQ4MTQ2NjAz
https://api.github.com/users/Jiaxin-Wen/orgs
https://api.github.com/users/Jiaxin-Wen/received_events
https://api.github.com/users/Jiaxin-Wen/repos
https://api.github.com/users/Jiaxin-Wen/starred{/owner}{/repo}
https://api.github.com/users/Jiaxin-Wen/subscriptions
User
https://api.github.com/users/Jiaxin-Wen
public
NONE
What @mariosasko proposed it's very useful for debugging. Thank you!
2024-08-13T15:53:49Z
https://github.com/huggingface/datasets/issues/6787#issuecomment-2286589041
https://api.github.com/repos/huggingface/datasets/issues/6787
IC_kwDODunzps6ISpRx
https://api.github.com/repos/huggingface/datasets/issues/comments/2286589041/reactions
2024-08-14T02:09:57Z
https://api.github.com/repos/huggingface/datasets/issues/comments/2286589041
https://avatars.githubusercontent.com/u/37418346?v=4
https://api.github.com/users/williamberrios/events{/privacy}
https://api.github.com/users/williamberrios/followers
https://api.github.com/users/williamberrios/following{/other_user}
https://api.github.com/users/williamberrios/gists{/gist_id}
https://github.com/williamberrios
williamberrios
MDQ6VXNlcjM3NDE4MzQ2
https://api.github.com/users/williamberrios/orgs
https://api.github.com/users/williamberrios/received_events
https://api.github.com/users/williamberrios/repos
https://api.github.com/users/williamberrios/starred{/owner}{/repo}
https://api.github.com/users/williamberrios/subscriptions
User
https://api.github.com/users/williamberrios
public
### Describe the bug

```python
from datasets import Dataset

def worker(example):
    while True:
        continue
    example['a'] = 100
    return example

data = Dataset.from_list([{"a": 1}, {"a": 2}])
data = data.map(worker)
print(data[0])
```

I'm implementing a worker function whose runtime will depend on specific examples (e.g., while most examples take 0.01s in worker, several examples may take 50s).

Therefore, I would like to know how the current implementation will handle those subprocesses that require a long (e.g., >= 5min) or even infinite time.

I notice that the current implementation set a timeout of 0.05 second
https://github.com/huggingface/datasets/blob/c3ddb1ef00334a6f973679a51e783905fbc9ef0b/src/datasets/utils/py_utils.py#L674

However, this example code still gets stuck.

### Steps to reproduce the bug

run the example above

### Expected behavior

I want to set a default worker to handle these timeout cases, instead of getting stuck

### Environment info

main branch version