Support async reward functions and parallelize call to reward functions.

[pramodith]

What does this PR do?

1. Checks if a reward function is not a nn.Module
2. If not checks if the reward function is synchronous or async.
3. Execution of synchronous reward functions are parallelized by running each on the next available thread.
4. Execution on non-async reward functions are parallelized by asyncio io's current event loop.
5. We await until all the created tasks are complete.
6. To avoid conflicts with any existing event loop that the code is running under we check if we're in an active event-loop if so we create a new event loop and run all the reward functions under the new event loop.

Fixes #4130

Before submitting

• This PR fixes a typo or improves the docs (you can dismiss the other checks if that's the case).
• Did you read the contributor guideline,
  Pull Request section?
• Was this discussed/approved via a GitHub issue? Please add a link
  to it if that's the case.
• Did you make sure to update the documentation with your changes?
• Did you write any new necessary tests?

Who can review?

Anyone in the community is free to review the PR once the tests have passed. Feel free to tag
members/contributors who may be interested in your PR.

[HuggingFaceDocBuilderDev]

The docs for this PR live here. All of your documentation changes will be reflected on that endpoint. The docs are available until 30 days after the last update.

[pramodith]

On revision I think it might be more prudent to use ThreadPoolExecutor for synchronous reward functions since these are going to be CPU bound/compute heavy, running them in an async fashion might not lead to true parallelism. Happy to hear your thoughts on this.

[qgallouedec]

On revision I think it might be more prudent to use ThreadPoolExecutor for synchronous reward functions since these are going to be CPU bound/compute heavy, running them in an async fashion might not lead to true parallelism. Happy to hear your thoughts on this.

I think I agree with this. I don't have a big background with asyncio though

[pramodith]

Done! Runs each synchronous reward function in its own thread now.

[qgallouedec]

I think the implementation looks good.

However, I'm still wondering if we should constraint the reward function to be synchronous. If execution is parallelized, we should achieve the same performance, no?

[pramodith]

I think the implementation looks good.

However, I'm still wondering if we should constraint the reward function to be synchronous. If execution is parallelized, we should achieve the same performance, no?

There aren't any constraints the reward function can sync or async. Sync functions use ThreadPoolExecutor, async functions use async.gather.

Perhaps I'm misunderstanding you?

if asyncio.iscoroutinefunction(reward_func):
          async_funcs_info.append((i, reward_func, reward_func_name))
else:
    sync_funcs_info.append((i, reward_func, reward_func_name))

The test case added also tests for this by including one sync and one async function.