Make JoinError Sync (#1888)

* Make JoinError Sync

* Move Mutex inside JoinError internals, hide its constructors

* Deprecate JoinError constructors, fix internal usages

tokio/src/task/error.rs

@@ -1,6 +1,7 @@
 use std::any::Any;
 use std::fmt;
 use std::io;
+use std::sync::Mutex;
 
 doc_rt_core! {
     /// Task failed to execute to completion.
@@ -11,21 +12,31 @@ doc_rt_core! {
 
 enum Repr {
     Cancelled,
-    Panic(Box<dyn Any + Send + 'static>),
+    Panic(Mutex<Box<dyn Any + Send + 'static>>),
 }
 
 impl JoinError {
-    /// Create a new `cancelled` error
+    #[doc(hidden)]
+    #[deprecated]
     pub fn cancelled() -> JoinError {
+        Self::cancelled2()
+    }
+
+    pub(crate) fn cancelled2() -> JoinError {
         JoinError {
             repr: Repr::Cancelled,
         }
     }
 
-    /// Create a new `panic` error
+    #[doc(hidden)]
+    #[deprecated]
     pub fn panic(err: Box<dyn Any + Send + 'static>) -> JoinError {
+        Self::panic2(err)
+    }
+
+    pub(crate) fn panic2(err: Box<dyn Any + Send + 'static>) -> JoinError {
         JoinError {
-            repr: Repr::Panic(err),
+            repr: Repr::Panic(Mutex::new(err)),
         }
     }
 }

tokio/src/task/harness.rs

@@ -142,7 +142,7 @@ where
                 }
             }
             Err(err) => {
-                self.complete(executor, join_interest, Err(JoinError::panic(err)));
+                self.complete(executor, join_interest, Err(JoinError::panic2(err)));
                 false
             }
         }
@@ -192,7 +192,7 @@ where
         state: Snapshot,
     ) {
         if state.is_canceled() {
-            dst.write(Track::new(Err(JoinError::cancelled())));
+            dst.write(Track::new(Err(JoinError::cancelled2())));
         } else {
             self.core().read_output(dst);
         }