diff --git a/crates/bevy_core/src/lib.rs b/crates/bevy_core/src/lib.rs
index 75fd87adb5ebb..9c97e6bdf8b29 100644
--- a/crates/bevy_core/src/lib.rs
+++ b/crates/bevy_core/src/lib.rs
@@ -46,6 +46,24 @@ impl Plugin for CorePlugin {
             .unwrap_or_else(DefaultTaskPoolOptions::default)
             .create_default_pools(app.world_mut());
 
+        #[cfg(not(target_arch = "wasm32"))]
+        app.add_system(
+            task_pool_options::handle_task_pool_panicking_threads_system::<bevy_tasks::IoTaskPool>
+                .system(),
+        )
+        .add_system(
+            task_pool_options::handle_task_pool_panicking_threads_system::<
+                bevy_tasks::ComputeTaskPool,
+            >
+                .system(),
+        )
+        .add_system(
+            task_pool_options::handle_task_pool_panicking_threads_system::<
+                bevy_tasks::AsyncComputeTaskPool,
+            >
+                .system(),
+        );
+
         app.init_resource::<Time>()
             .init_resource::<EntityLabels>()
             .init_resource::<FixedTimesteps>()
diff --git a/crates/bevy_core/src/task_pool_options.rs b/crates/bevy_core/src/task_pool_options.rs
index 19c9dad5bfe2f..67d99e88c1e67 100644
--- a/crates/bevy_core/src/task_pool_options.rs
+++ b/crates/bevy_core/src/task_pool_options.rs
@@ -1,10 +1,24 @@
+#[cfg(not(target_arch = "wasm32"))]
+use bevy_ecs::system::Res;
 use bevy_ecs::world::World;
-use bevy_tasks::{AsyncComputeTaskPool, ComputeTaskPool, IoTaskPool, TaskPoolBuilder};
+use bevy_tasks::{
+    AsyncComputeTaskPool, ComputeTaskPool, DerefTaskPool, IoTaskPool, TaskPoolBuilder,
+    TaskPoolThreadPanicPolicy,
+};
 use bevy_utils::tracing::trace;
 
+/// The set of policies describing how the according task pool behaves
+#[derive(Clone, Debug)]
+pub struct TaskPoolPolicies {
+    /// Used to determine number of threads to allocate
+    pub assignment_policy: TaskPoolThreadAssignmentPolicy,
+    /// Used to determine the panic policy of the task pool
+    pub panic_policy: TaskPoolThreadPanicPolicy,
+}
+
 /// Defines a simple way to determine how many threads to use given the number of remaining cores
 /// and number of total cores
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct TaskPoolThreadAssignmentPolicy {
     /// Force using at least this many threads
     pub min_threads: usize,
@@ -42,13 +56,12 @@ pub struct DefaultTaskPoolOptions {
     /// If the number of physical cores is grater than max_total_threads, force using
     /// max_total_threads
     pub max_total_threads: usize,
-
-    /// Used to determine number of IO threads to allocate
-    pub io: TaskPoolThreadAssignmentPolicy,
-    /// Used to determine number of async compute threads to allocate
-    pub async_compute: TaskPoolThreadAssignmentPolicy,
-    /// Used to determine number of compute threads to allocate
-    pub compute: TaskPoolThreadAssignmentPolicy,
+    /// Used to configure the IOTaskPool's inner policies
+    pub io: TaskPoolPolicies,
+    /// Used to configure the AsyncTaskPool's inner policies
+    pub async_compute: TaskPoolPolicies,
+    /// Used to configure the ComputeTaskPool's inner policies
+    pub compute: TaskPoolPolicies,
 }
 
 impl Default for DefaultTaskPoolOptions {
@@ -59,24 +72,33 @@ impl Default for DefaultTaskPoolOptions {
             max_total_threads: std::usize::MAX,
 
             // Use 25% of cores for IO, at least 1, no more than 4
-            io: TaskPoolThreadAssignmentPolicy {
-                min_threads: 1,
-                max_threads: 4,
-                percent: 0.25,
+            io: TaskPoolPolicies {
+                assignment_policy: TaskPoolThreadAssignmentPolicy {
+                    min_threads: 1,
+                    max_threads: 4,
+                    percent: 0.25,
+                },
+                panic_policy: TaskPoolThreadPanicPolicy::Restart,
             },
 
             // Use 25% of cores for async compute, at least 1, no more than 4
-            async_compute: TaskPoolThreadAssignmentPolicy {
-                min_threads: 1,
-                max_threads: 4,
-                percent: 0.25,
+            async_compute: TaskPoolPolicies {
+                assignment_policy: TaskPoolThreadAssignmentPolicy {
+                    min_threads: 1,
+                    max_threads: 4,
+                    percent: 0.25,
+                },
+                panic_policy: TaskPoolThreadPanicPolicy::Restart,
             },
 
             // Use all remaining cores for compute (at least 1)
-            compute: TaskPoolThreadAssignmentPolicy {
-                min_threads: 1,
-                max_threads: std::usize::MAX,
-                percent: 1.0, // This 1.0 here means "whatever is left over"
+            compute: TaskPoolPolicies {
+                assignment_policy: TaskPoolThreadAssignmentPolicy {
+                    min_threads: 1,
+                    max_threads: std::usize::MAX,
+                    percent: 1.0, // This 1.0 here means "whatever is left over"
+                },
+                panic_policy: TaskPoolThreadPanicPolicy::Restart,
             },
         }
     }
@@ -104,6 +126,7 @@ impl DefaultTaskPoolOptions {
             // Determine the number of IO threads we will use
             let io_threads = self
                 .io
+                .assignment_policy
                 .get_number_of_threads(remaining_threads, total_threads);
 
             trace!("IO Threads: {}", io_threads);
@@ -113,6 +136,7 @@ impl DefaultTaskPoolOptions {
                 TaskPoolBuilder::default()
                     .num_threads(io_threads)
                     .thread_name("IO Task Pool".to_string())
+                    .panic_policy(self.io.panic_policy)
                     .build(),
             ));
         }
@@ -121,6 +145,7 @@ impl DefaultTaskPoolOptions {
             // Determine the number of async compute threads we will use
             let async_compute_threads = self
                 .async_compute
+                .assignment_policy
                 .get_number_of_threads(remaining_threads, total_threads);
 
             trace!("Async Compute Threads: {}", async_compute_threads);
@@ -130,6 +155,7 @@ impl DefaultTaskPoolOptions {
                 TaskPoolBuilder::default()
                     .num_threads(async_compute_threads)
                     .thread_name("Async Compute Task Pool".to_string())
+                    .panic_policy(self.async_compute.panic_policy)
                     .build(),
             ));
         }
@@ -139,6 +165,7 @@ impl DefaultTaskPoolOptions {
             // This is intentionally last so that an end user can specify 1.0 as the percent
             let compute_threads = self
                 .compute
+                .assignment_policy
                 .get_number_of_threads(remaining_threads, total_threads);
 
             trace!("Compute Threads: {}", compute_threads);
@@ -146,8 +173,18 @@ impl DefaultTaskPoolOptions {
                 TaskPoolBuilder::default()
                     .num_threads(compute_threads)
                     .thread_name("Compute Task Pool".to_string())
+                    .panic_policy(self.compute.panic_policy)
                     .build(),
             ));
         }
     }
 }
+
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) fn handle_task_pool_panicking_threads_system<
+    T: DerefTaskPool + Send + Sync + 'static,
+>(
+    task_pool: Res<T>,
+) {
+    bevy_tasks::handle_task_pool_panicking_threads(&**task_pool);
+}
diff --git a/crates/bevy_tasks/Cargo.toml b/crates/bevy_tasks/Cargo.toml
index f2c1bf41f396b..a36d81becfd3c 100644
--- a/crates/bevy_tasks/Cargo.toml
+++ b/crates/bevy_tasks/Cargo.toml
@@ -16,6 +16,8 @@ keywords = ["bevy"]
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+bevy_utils = { path = "../bevy_utils", version = "0.5.0" }
+parking_lot = "0.11.0"
 futures-lite = "1.4.0"
 event-listener = "2.4.0"
 async-executor = "1.3.0"
diff --git a/crates/bevy_tasks/src/lib.rs b/crates/bevy_tasks/src/lib.rs
index 08d7dfc42b988..074f13e718b27 100644
--- a/crates/bevy_tasks/src/lib.rs
+++ b/crates/bevy_tasks/src/lib.rs
@@ -4,10 +4,13 @@ pub use slice::{ParallelSlice, ParallelSliceMut};
 mod task;
 pub use task::Task;
 
+mod task_pool_common;
+pub use task_pool_common::TaskPoolThreadPanicPolicy;
+
 #[cfg(not(target_arch = "wasm32"))]
 mod task_pool;
 #[cfg(not(target_arch = "wasm32"))]
-pub use task_pool::{Scope, TaskPool, TaskPoolBuilder};
+pub use task_pool::{handle_task_pool_panicking_threads, Scope, TaskPool, TaskPoolBuilder};
 
 #[cfg(target_arch = "wasm32")]
 mod single_threaded_task_pool;
@@ -15,7 +18,7 @@ mod single_threaded_task_pool;
 pub use single_threaded_task_pool::{Scope, TaskPool, TaskPoolBuilder};
 
 mod usages;
-pub use usages::{AsyncComputeTaskPool, ComputeTaskPool, IoTaskPool};
+pub use usages::{AsyncComputeTaskPool, ComputeTaskPool, DerefTaskPool, IoTaskPool};
 
 mod countdown_event;
 pub use countdown_event::CountdownEvent;
diff --git a/crates/bevy_tasks/src/single_threaded_task_pool.rs b/crates/bevy_tasks/src/single_threaded_task_pool.rs
index 8f248e1005c38..f76cd87d909de 100644
--- a/crates/bevy_tasks/src/single_threaded_task_pool.rs
+++ b/crates/bevy_tasks/src/single_threaded_task_pool.rs
@@ -1,3 +1,4 @@
+use crate::TaskPoolThreadPanicPolicy;
 use std::{
     future::Future,
     mem,
@@ -26,6 +27,10 @@ impl TaskPoolBuilder {
         self
     }
 
+    pub fn panic_policy(self, _policy: TaskPoolThreadPanicPolicy) -> Self {
+        self
+    }
+
     pub fn build(self) -> TaskPool {
         TaskPool::new_internal()
     }
diff --git a/crates/bevy_tasks/src/task_pool.rs b/crates/bevy_tasks/src/task_pool.rs
index 9331799120583..9b83a6a4dbe87 100644
--- a/crates/bevy_tasks/src/task_pool.rs
+++ b/crates/bevy_tasks/src/task_pool.rs
@@ -1,15 +1,18 @@
+use crate::{Task, TaskPoolThreadPanicPolicy};
+use bevy_utils::tracing::{error, warn};
+use futures_lite::{future, pin};
+use parking_lot::RwLock;
 use std::{
     future::Future,
     mem,
     pin::Pin,
-    sync::Arc,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
     thread::{self, JoinHandle},
 };
 
-use futures_lite::{future, pin};
-
-use crate::Task;
-
 /// Used to create a TaskPool
 #[derive(Debug, Default, Clone)]
 pub struct TaskPoolBuilder {
@@ -21,6 +24,8 @@ pub struct TaskPoolBuilder {
     /// Allows customizing the name of the threads - helpful for debugging. If set, threads will
     /// be named <thread_name> (<thread_index>), i.e. "MyThreadPool (2)"
     thread_name: Option<String>,
+    /// Allows customizing the policy for when a [`TaskPool`]'s thread(s) panic.
+    panic_policy: Option<TaskPoolThreadPanicPolicy>,
 }
 
 impl TaskPoolBuilder {
@@ -49,19 +54,53 @@ impl TaskPoolBuilder {
         self
     }
 
+    pub fn panic_policy(mut self, policy: TaskPoolThreadPanicPolicy) -> Self {
+        self.panic_policy = Some(policy);
+        self
+    }
+
     /// Creates a new ThreadPoolBuilder based on the current options.
     pub fn build(self) -> TaskPool {
         TaskPool::new_internal(
             self.num_threads,
             self.stack_size,
             self.thread_name.as_deref(),
+            self.panic_policy,
         )
     }
 }
 
+#[derive(Debug)]
+struct ThreadState {
+    handle: JoinHandle<()>,
+    panicking: Arc<AtomicBool>,
+}
+
+impl ThreadState {
+    pub fn panicking(&self) -> bool {
+        self.panicking.load(Ordering::Acquire)
+    }
+
+    pub fn thread(&self) -> &std::thread::Thread {
+        self.handle.thread()
+    }
+}
+
+struct PanicState {
+    panicking: Arc<AtomicBool>,
+}
+
+impl Drop for PanicState {
+    fn drop(&mut self) {
+        if thread::panicking() {
+            self.panicking.store(true, Ordering::Release);
+        }
+    }
+}
+
 #[derive(Debug)]
 struct TaskPoolInner {
-    threads: Vec<JoinHandle<()>>,
+    threads: Vec<ThreadState>,
     shutdown_tx: async_channel::Sender<()>,
 }
 
@@ -70,8 +109,8 @@ impl Drop for TaskPoolInner {
         self.shutdown_tx.close();
 
         let panicking = thread::panicking();
-        for join_handle in self.threads.drain(..) {
-            let res = join_handle.join();
+        for state in self.threads.drain(..) {
+            let res = state.handle.join();
             if !panicking {
                 res.expect("Task thread panicked while executing.");
             }
@@ -91,7 +130,14 @@ pub struct TaskPool {
     executor: Arc<async_executor::Executor<'static>>,
 
     /// Inner state of the pool
-    inner: Arc<TaskPoolInner>,
+    inner: Arc<RwLock<TaskPoolInner>>,
+
+    /// Panic policy of the inner thread pool.
+    panic_policy: TaskPoolThreadPanicPolicy,
+
+    /// Receiving channel used when an inner thread panics and
+    /// another one is spawned in its place.
+    shutdown_rx: async_channel::Receiver<()>,
 }
 
 impl TaskPool {
@@ -104,10 +150,97 @@ impl TaskPool {
         TaskPoolBuilder::new().build()
     }
 
+    fn any_panicking_threads(&self) -> bool {
+        self.inner
+            .read()
+            .threads
+            .iter()
+            .any(|state| state.panicking())
+    }
+
+    pub(crate) fn handle_panicking_threads(&self) {
+        if self.any_panicking_threads() {
+            match self.panic_policy {
+                TaskPoolThreadPanicPolicy::Propagate => {
+                    for state in self.inner.write().threads.drain(..) {
+                        let thread = state.thread().clone();
+                        if let Err(err) = state.handle.join() {
+                            error!("TaskPool's inner thread '{:?}' panicked!", thread);
+                            std::panic::resume_unwind(err);
+                        }
+                    }
+                }
+                TaskPoolThreadPanicPolicy::Restart => {
+                    for (idx, state) in self
+                        .inner
+                        .write()
+                        .threads
+                        .iter_mut()
+                        .enumerate()
+                        .filter(|(_, state)| state.panicking())
+                    {
+                        let thread_name = match state.thread().name() {
+                            Some(name) => name.to_owned(),
+                            None => format!("TaskPool ({})", idx),
+                        };
+
+                        let old_state = std::mem::replace(
+                            state,
+                            Self::spawn_thread_internal(
+                                None,
+                                thread_name,
+                                self.executor.clone(),
+                                self.shutdown_rx.clone(),
+                            ),
+                        );
+
+                        warn!(
+                            "TaskPool's inner thread '{:?}' panicked!",
+                            old_state.thread()
+                        );
+
+                        // join the panicked thread handle
+                        old_state.handle.join().unwrap_err();
+                    }
+                }
+            }
+        }
+    }
+
+    fn spawn_thread_internal(
+        stack_size: Option<usize>,
+        thread_name: String,
+        executor: Arc<async_executor::Executor<'static>>,
+        shutdown_rx: async_channel::Receiver<()>,
+    ) -> ThreadState {
+        let mut thread_builder = thread::Builder::new().name(thread_name);
+
+        if let Some(stack_size) = stack_size {
+            thread_builder = thread_builder.stack_size(stack_size);
+        }
+
+        let panicking = Arc::new(AtomicBool::new(false));
+        let panicking_clone = panicking.clone();
+
+        let handle = thread_builder
+            .spawn(move || {
+                let _panic_state = PanicState {
+                    panicking: panicking_clone,
+                };
+                let shutdown_future = executor.run(shutdown_rx.recv());
+                // Use unwrap_err because we expect a Closed error
+                future::block_on(shutdown_future).unwrap_err();
+            })
+            .expect("Failed to spawn thread.");
+
+        ThreadState { handle, panicking }
+    }
+
     fn new_internal(
         num_threads: Option<usize>,
         stack_size: Option<usize>,
         thread_name: Option<&str>,
+        panic_policy: Option<TaskPoolThreadPanicPolicy>,
     ) -> Self {
         let (shutdown_tx, shutdown_rx) = async_channel::unbounded::<()>();
 
@@ -117,43 +250,37 @@ impl TaskPool {
 
         let threads = (0..num_threads)
             .map(|i| {
-                let ex = Arc::clone(&executor);
-                let shutdown_rx = shutdown_rx.clone();
-
                 let thread_name = if let Some(thread_name) = thread_name {
                     format!("{} ({})", thread_name, i)
                 } else {
                     format!("TaskPool ({})", i)
                 };
 
-                let mut thread_builder = thread::Builder::new().name(thread_name);
-
-                if let Some(stack_size) = stack_size {
-                    thread_builder = thread_builder.stack_size(stack_size);
-                }
-
-                thread_builder
-                    .spawn(move || {
-                        let shutdown_future = ex.run(shutdown_rx.recv());
-                        // Use unwrap_err because we expect a Closed error
-                        future::block_on(shutdown_future).unwrap_err();
-                    })
-                    .expect("Failed to spawn thread.")
+                Self::spawn_thread_internal(
+                    stack_size,
+                    thread_name,
+                    executor.clone(),
+                    shutdown_rx.clone(),
+                )
             })
             .collect();
 
+        let panic_policy = panic_policy.unwrap_or(TaskPoolThreadPanicPolicy::Restart);
+
         Self {
             executor,
-            inner: Arc::new(TaskPoolInner {
+            inner: Arc::new(RwLock::new(TaskPoolInner {
                 threads,
                 shutdown_tx,
-            }),
+            })),
+            panic_policy,
+            shutdown_rx,
         }
     }
 
     /// Return the number of threads owned by the task pool
     pub fn thread_num(&self) -> usize {
-        self.inner.threads.len()
+        self.inner.read().threads.len()
     }
 
     /// Allows spawning non-`static futures on the thread pool. The function takes a callback,
@@ -250,6 +377,11 @@ impl Default for TaskPool {
     }
 }
 
+#[doc(hidden)]
+pub fn handle_task_pool_panicking_threads(task_pool: &TaskPool) {
+    task_pool.handle_panicking_threads();
+}
+
 #[derive(Debug)]
 pub struct Scope<'scope, T> {
     executor: &'scope async_executor::Executor<'scope>,
@@ -390,4 +522,59 @@ mod tests {
         assert!(!thread_check_failed.load(Ordering::Acquire));
         assert_eq!(count.load(Ordering::Acquire), 200);
     }
+
+    #[test]
+    fn test_restart_panic_policy() {
+        std::panic::set_hook(Box::new(|_| {}));
+
+        let pool = Arc::new(
+            TaskPoolBuilder::new()
+                .panic_policy(TaskPoolThreadPanicPolicy::Restart)
+                .num_threads(1)
+                .build(),
+        );
+
+        pool.spawn(async {
+            panic!("oh no!");
+        })
+        .detach();
+
+        while !pool.any_panicking_threads() {}
+
+        // even though the first thread panicked, we can still queue up a task
+        // as the next call to `handle_panicking_threads` will replace the panicked
+        // thread with a new once.
+        let success = Arc::new(AtomicBool::new(false));
+        let success_clone = success.clone();
+        pool.spawn(async move {
+            success_clone.store(true, Ordering::Release);
+        })
+        .detach();
+
+        pool.handle_panicking_threads();
+        assert!(!pool.any_panicking_threads());
+
+        while !success.load(Ordering::Acquire) {}
+    }
+
+    #[test]
+    #[should_panic = "bevy"]
+    fn test_propagate_panic_policy() {
+        std::panic::set_hook(Box::new(|_| {}));
+
+        let pool = Arc::new(
+            TaskPoolBuilder::new()
+                .panic_policy(TaskPoolThreadPanicPolicy::Propagate)
+                .num_threads(1)
+                .build(),
+        );
+
+        pool.spawn(async {
+            panic!("bevy");
+        })
+        .detach();
+
+        while !pool.any_panicking_threads() {}
+        pool.handle_panicking_threads();
+    }
 }
diff --git a/crates/bevy_tasks/src/task_pool_common.rs b/crates/bevy_tasks/src/task_pool_common.rs
new file mode 100644
index 0000000000000..d8f65902448e7
--- /dev/null
+++ b/crates/bevy_tasks/src/task_pool_common.rs
@@ -0,0 +1,10 @@
+/// The policy used when a task pool's internal thread panics
+#[derive(Copy, Clone, Debug)]
+pub enum TaskPoolThreadPanicPolicy {
+    /// Propagate the panic to the main thread, causing the main
+    /// thread to panic as well.
+    Propagate,
+    /// Restart the thread by joining the panicked thread and
+    /// spawning another one in it's place.
+    Restart,
+}
diff --git a/crates/bevy_tasks/src/usages.rs b/crates/bevy_tasks/src/usages.rs
index 7adf60639b116..9f33725af09ae 100644
--- a/crates/bevy_tasks/src/usages.rs
+++ b/crates/bevy_tasks/src/usages.rs
@@ -50,3 +50,7 @@ impl Deref for IoTaskPool {
         &self.0
     }
 }
+
+pub trait DerefTaskPool: Deref<Target = TaskPool> {}
+
+impl<T> DerefTaskPool for T where T: Deref<Target = TaskPool> {}