[Core] Show_webui segfault fix.
diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx
index 930e45ecb0903..7173860ec53d3 100644
--- a/python/ray/_raylet.pyx
+++ b/python/ray/_raylet.pyx
@@ -634,6 +634,12 @@ cdef shared_ptr[CBuffer] string_to_buffer(c_string& c_str):
                 <uint8_t*>(c_str.data()), c_str.size(), True))
 
 
+cdef void terminate_asyncio_thread() nogil:
+    with gil:
+        core_worker = ray.worker.global_worker.core_worker
+        core_worker.destroy_event_loop_if_exists()
+
+
 cdef class CoreWorker:
 
     def __cinit__(self, is_driver, store_socket, raylet_socket,
@@ -667,6 +673,7 @@ cdef class CoreWorker:
         options.is_local_mode = local_mode
         options.num_workers = 1
         options.kill_main = kill_main_task
+        options.terminate_asyncio_thread = terminate_asyncio_thread
 
         CCoreWorkerProcess.Initialize(options)
 
@@ -1171,7 +1178,8 @@ cdef class CoreWorker:
 
         if self.async_thread is None:
             self.async_thread = threading.Thread(
-                target=lambda: self.async_event_loop.run_forever()
+                target=lambda: self.async_event_loop.run_forever(),
+                name="AsyncIO Thread"
             )
             # Making the thread a daemon causes it to exit
             # when the main thread exits.
@@ -1192,8 +1200,12 @@ cdef class CoreWorker:
                 .YieldCurrentFiber(event))
         return future.result()
 
-    def destory_event_loop_if_exists(self):
+    def destroy_event_loop_if_exists(self):
         if self.async_event_loop is not None:
+            # We should stop the monitor first because otherwise,
+            # loop.stop() will continue forever as monitor
+            # main loop will not be terminated.
+            self.async_event_loop.monitor_state.kill()
             self.async_event_loop.stop()
         if self.async_thread is not None:
             self.async_thread.join()
diff --git a/python/ray/async_compat.py b/python/ray/async_compat.py
index df103f5df3961..4bfe7bffadcda 100644
--- a/python/ray/async_compat.py
+++ b/python/ray/async_compat.py
@@ -107,7 +107,8 @@ def __init__(self, loop):
         self.names_lock = threading.Lock()
 
         self.sleep_time = 1.0
-        asyncio.ensure_future(self.monitor(), loop=loop)
+        self.monitor_loop_future = asyncio.ensure_future(
+            self.monitor(), loop=loop)
 
     async def monitor(self):
         while True:
@@ -131,3 +132,11 @@ def get_all_task_names(self):
         with self.names_lock:
             names = list(self.names.values())
         return names
+
+    def kill(self):
+        """Kill the monitor's loop
+
+        This should be called in order to clean an event loop
+        that this monitor is running.
+        """
+        self.monitor_loop_future.cancel()
diff --git a/python/ray/includes/libcoreworker.pxd b/python/ray/includes/libcoreworker.pxd
index 6524139e097b7..a2698ee51093b 100644
--- a/python/ray/includes/libcoreworker.pxd
+++ b/python/ray/includes/libcoreworker.pxd
@@ -217,6 +217,7 @@ cdef extern from "ray/core_worker/core_worker.h" nogil:
         int num_workers
         (c_bool() nogil) kill_main
         CCoreWorkerOptions()
+        (void() nogil) terminate_asyncio_thread
 
     cdef cppclass CCoreWorkerProcess "ray::CoreWorkerProcess":
         @staticmethod
diff --git a/python/ray/ray_perf.py b/python/ray/ray_perf.py
index eae62e396fd34..55c3c100c4e1d 100644
--- a/python/ray/ray_perf.py
+++ b/python/ray/ray_perf.py
@@ -92,6 +92,7 @@ def timeit(name, fn, multiplier=1):
 def main():
     print("Tip: set TESTS_TO_RUN='pattern' to run a subset of benchmarks")
     ray.init()
+
     value = ray.put(0)
     arr = np.zeros(100 * 1024 * 1024, dtype=np.int64)
 
diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc
index a3e823a11e359..7739ee867ec9a 100644
--- a/src/ray/core_worker/core_worker.cc
+++ b/src/ray/core_worker/core_worker.cc
@@ -590,6 +590,13 @@ void CoreWorker::WaitForShutdown() {
   }
   if (options_.worker_type == WorkerType::WORKER) {
     RAY_CHECK(task_execution_service_.stopped());
+    // Asyncio coroutines could still run after CoreWorker is removed because it is
+    // running in a different thread. This can cause segfault because coroutines try to
+    // access CoreWorker methods that are already garbage collected. We should complete
+    // all coroutines before shutting down in order to prevent this.
+    if (worker_context_.CurrentActorIsAsync()) {
+      options_.terminate_asyncio_thread();
+    }
   }
 }
 
diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h
index 27c374ccf4169..5b6ea601b948d 100644
--- a/src/ray/core_worker/core_worker.h
+++ b/src/ray/core_worker/core_worker.h
@@ -114,6 +114,8 @@ struct CoreWorkerOptions {
   bool is_local_mode;
   /// The number of workers to be started in the current process.
   int num_workers;
+  /// The function to destroy asyncio event and loops.
+  std::function<void()> terminate_asyncio_thread;
 };
 
 /// Lifecycle management of one or more `CoreWorker` instances in a process.

<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

## Why are these changes needed?

This will resolve https://github.com/ray-project/ray/issues/8309

The problem was that CoreWorker removes its cpp instance before default_worker.py's shutdown is called (where threads are joined). That means that async thread's coroutines are still running and using CoreWorker that was already GC'ed. This PR will make sure those coroutines are cleaned properly before CoreWorker is GC'ed.

This rarely happens in sync cases as well (https://github.com/ray-project/ray/issues/8291) due to import threads (look at Simon's comment). It is so much more rare than async cases because there is a mechanism to "stop thread" before they are joined on shutdown. I didn't handle this in this PR because of two reasons. 

- It hasn't been issues for a long time (although I am sure it was there from before). This is very rare (it is detected only once out of 8 long running tests for 3 days whereas async cases occur almost every microbenchmark script run) and fixing this won't probably have big impact compared to efforts required.
- Fixing this will require uncertain amount of refactoring because we need to change the cleanup mechanism. Moving thread join logic to CoreWorker didn't work. It will break Java code as well because we need to run Python code inside Cpp code. So, I decided to deprioritize this for now. I will keep the other issue open (https://github.com/ray-project/ray/issues/8291) so I can revisit later.
This has been manually tested. I ran the reproducible script 10 times, and verified the error didn't occur. It didn't properly handle this issue yet https://github.com/ray-project/ray/issues/8291 because it was a bit tricky to fix. 

## Related issue number

https://github.com/ray-project/ray/issues/8309

## Checks

- [x] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/latest/.
- [x] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failure rates at https://ray-travis-tracker.herokuapp.com/.
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested (please justify below)
