Create new eventloop to run coroutines

[ejguan]

Prevent calling asyncio.run within __iter__ function. New event loop would be created rather than reusing the global event loop.
Using policy for the case of custom policy is used as suggested in https://docs.python.org/3.8/library/asyncio-eventloop.html#event-loop

Address the comment

[facebook-github-bot]

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

[ejguan]

I will add a test to iterate over two asyncio DataPipe at the same time to make sure it works.

[dracifer]

thanks

[facebook-github-bot]

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

[NivekT]

LGTM! This should fix the situation where asyncio.run() function cannot be called when another asyncio event loop is running in the same thread, such as in Jupyter notebook.

Non-blocking, it would be nice if we have some tests related to that.

[facebook-github-bot]

@ejguan merged this pull request in f30281b.

[ejguan]

LGTM! This should fix the situation where asyncio.run() function cannot be called when another asyncio event loop is running in the same thread, such as in Jupyter notebook.

Non-blocking, it would be nice if we have some tests related to that.

I just looked deeper on asyncio. I think creating new eventloop is redundant because in theory we should only keep a single eventloop within a process rather than multiple ones or nested one.
The case of Jupyter notebook is special though. If users want to use it, they need to rely on module like nest_asyncio to achieve iterating over this DataPipe. If they want to do multiprocessing, then it's not a problem at all as each sub-process would launch new eventloop excluding jupyter eventloop to execute self.process_batch.

In thoery, users should never make DataLoader2 inside a asyncio.run function as DataLoader2 doesn't provide any async API to users and it gives no benefit at all. This DataPipe should be the place that converts async calls to sync. And, even though I use new loop within AsyncMapper, it's not supported by asyncio.

@dracifer @wenleix @NivekT

[ejguan]

@NivekT also suggests to capture the exception to raise suggestion for users who want to use it in notebook.