Revert "Stageless: close the finish channel so executor doesn't deadlock (bevyengine#7448)"

This reverts commit ff7d5ff.

Author: maniwani

crates/bevy_ecs/src/schedule_v3/executor/multi_threaded.rs

@@ -1,5 +1,3 @@
-use std::panic::AssertUnwindSafe;
-
 use bevy_tasks::{ComputeTaskPool, Scope, TaskPool, ThreadExecutor};
 use bevy_utils::default;
 use bevy_utils::syncunsafecell::SyncUnsafeCell;
@@ -177,10 +175,11 @@ impl SystemExecutor for MultiThreadedExecutor {
 
                         if self.num_running_systems > 0 {
                             // wait for systems to complete
-                            let index =
-                                self.receiver.recv().await.expect(
-                                    "A system has panicked so the executor cannot continue.",
-                                );
+                            let index = self
+                                .receiver
+                                .recv()
+                                .await
+                                .unwrap_or_else(|error| unreachable!("{}", error));
 
                             self.finish_system_and_signal_dependents(index);
 
@@ -430,22 +429,14 @@ impl MultiThreadedExecutor {
         let task = async move {
             #[cfg(feature = "trace")]
             let system_guard = system_span.enter();
-            let res = std::panic::catch_unwind(AssertUnwindSafe(|| {
-                // SAFETY: access is compatible
-                unsafe { system.run_unsafe((), world) };
-            }));
+            // SAFETY: access is compatible
+            unsafe { system.run_unsafe((), world) };
             #[cfg(feature = "trace")]
             drop(system_guard);
-            if res.is_err() {
-                // close the channel to propagate the error to the
-                // multithreaded executor
-                sender.close();
-            } else {
-                sender
-                    .send(system_index)
-                    .await
-                    .unwrap_or_else(|error| unreachable!("{}", error));
-            }
+            sender
+                .send(system_index)
+                .await
+                .unwrap_or_else(|error| unreachable!("{}", error));
         };
 
         #[cfg(feature = "trace")]
@@ -488,21 +479,13 @@ impl MultiThreadedExecutor {
             let task = async move {
                 #[cfg(feature = "trace")]
                 let system_guard = system_span.enter();
-                let res = std::panic::catch_unwind(AssertUnwindSafe(|| {
-                    apply_system_buffers(&unapplied_systems, systems, world);
-                }));
+                apply_system_buffers(&unapplied_systems, systems, world);
                 #[cfg(feature = "trace")]
                 drop(system_guard);
-                if res.is_err() {
-                    // close the channel to propagate the error to the
-                    // multithreaded executor
-                    sender.close();
-                } else {
-                    sender
-                        .send(system_index)
-                        .await
-                        .unwrap_or_else(|error| unreachable!("{}", error));
-                }
+                sender
+                    .send(system_index)
+                    .await
+                    .unwrap_or_else(|error| unreachable!("{}", error));
             };
 
             #[cfg(feature = "trace")]
@@ -512,21 +495,13 @@ impl MultiThreadedExecutor {
             let task = async move {
                 #[cfg(feature = "trace")]
                 let system_guard = system_span.enter();
-                let res = std::panic::catch_unwind(AssertUnwindSafe(|| {
-                    system.run((), world);
-                }));
+                system.run((), world);
                 #[cfg(feature = "trace")]
                 drop(system_guard);
-                if res.is_err() {
-                    // close the channel to propagate the error to the
-                    // multithreaded executor
-                    sender.close();
-                } else {
-                    sender
-                        .send(system_index)
-                        .await
-                        .unwrap_or_else(|error| unreachable!("{}", error));
-                }
+                sender
+                    .send(system_index)
+                    .await
+                    .unwrap_or_else(|error| unreachable!("{}", error));
             };
 
             #[cfg(feature = "trace")]