Avoid setting event loop policy if within IPython kernel and no runni…

…ng event loop (#3336)

Setting asyncio event loop policy at these two places could cause problems.
1. When policy is set in Jupyter notebook server extension.
This causes the notebook server to hang. This is fixed in #2343.
2. When policy is set in iPython startup config (`~/.ipython/profile_default/startup/whatever.py`) or by setting `get_config().InteractiveShellApp.exec_lines` in `~/.ipython/profile_default/ipython_config.py`.
This causes the kernel to hang. This can be reproduced by running either `jupyter console` or `jupyter notebook`. If running In Jupyter notebook, it will struck at "Kernel starting, please wait...".
Note that manually setting the policy in notebook cell, after the kernel has started, is fine.

In both cases, running `asyncio.get_running_loop()` just before setting the policy will raise `RuntimeError`, meaning there is no running event loop yet.

See #3202.

distributed/utils.py

@@ -1177,25 +1177,33 @@ def reset_logger_locks():
             handler.createLock()
 
 
-# Only bother if asyncio has been loaded by Tornado
-if "asyncio" in sys.modules and tornado.version_info[0] >= 5:
+if tornado.version_info[0] >= 5:
 
-    jupyter_event_loop_initialized = False
+    is_server_extension = False
 
     if "notebook" in sys.modules:
         import traitlets
         from notebook.notebookapp import NotebookApp
 
-        jupyter_event_loop_initialized = traitlets.config.Application.initialized() and isinstance(
+        is_server_extension = traitlets.config.Application.initialized() and isinstance(
             traitlets.config.Application.instance(), NotebookApp
         )
 
-    if not jupyter_event_loop_initialized:
-        import tornado.platform.asyncio
+    if not is_server_extension:
+        is_kernel_and_no_running_loop = False
 
-        asyncio.set_event_loop_policy(
-            tornado.platform.asyncio.AnyThreadEventLoopPolicy()
-        )
+        if is_kernel():
+            try:
+                asyncio.get_running_loop()
+            except RuntimeError:
+                is_kernel_and_no_running_loop = True
+
+        if not is_kernel_and_no_running_loop:
+            import tornado.platform.asyncio
+
+            asyncio.set_event_loop_policy(
+                tornado.platform.asyncio.AnyThreadEventLoopPolicy()
+            )
 
 
 @functools.lru_cache(1000)