add ThreadSafeHandle

Author: kumaraditya303

Lib/asyncio/base_events.py

@@ -873,7 +873,10 @@ def call_soon_threadsafe(self, callback, *args, context=None):
         self._check_closed()
         if self._debug:
             self._check_callback(callback, 'call_soon_threadsafe')
-        handle = self._call_soon(callback, args, context)
+        handle = events._ThreadSafeHandle(callback, args, self, context)
+        self._ready.append(handle)
+        if handle._source_traceback:
+            del handle._source_traceback[-1]
         if handle._source_traceback:
             del handle._source_traceback[-1]
         self._write_to_self()
@@ -1937,7 +1940,7 @@ def call_exception_handler(self, context):
 
     def _add_callback(self, handle):
         """Add a Handle to _ready."""
-        if not handle._cancelled:
+        if not handle.cancelled():
             self._ready.append(handle)
 
     def _add_callback_signalsafe(self, handle):
@@ -1966,7 +1969,7 @@ def _run_once(self):
             # is too high
             new_scheduled = []
             for handle in self._scheduled:
-                if handle._cancelled:
+                if handle.cancelled():
                     handle._scheduled = False
                 else:
                     new_scheduled.append(handle)
@@ -2016,7 +2019,7 @@ def _run_once(self):
         ntodo = len(self._ready)
         for i in range(ntodo):
             handle = self._ready.popleft()
-            if handle._cancelled:
+            if handle.cancelled():
                 continue
             if self._debug:
                 try:

Lib/asyncio/events.py

@@ -59,7 +59,7 @@ def __init__(self, callback, args, loop, context=None):
 
     def _repr_info(self):
         info = [self.__class__.__name__]
-        if self._cancelled:
+        if self.cancelled():
             info.append('cancelled')
         if self._callback is not None:
             info.append(format_helpers._format_callback_source(
@@ -113,6 +113,24 @@ def _run(self):
             self._loop.call_exception_handler(context)
         self = None  # Needed to break cycles when an exception occurs.
 
+class _ThreadSafeHandle(Handle):
+
+    def __init__(self, callback, args, loop, context=None):
+        super().__init__(callback, args, loop, context)
+        self._lock = threading.RLock()
+
+    def cancel(self):
+        with self._lock:
+            return super().cancel()
+
+    def cancelled(self):
+        with self._lock:
+            return super().cancelled()
+
+    def _run(self):
+        with self._lock:
+            return super()._run()
+
 
 class TimerHandle(Handle):
     """Object returned by timed callback registration methods."""

Lib/asyncio/unix_events.py

@@ -138,7 +138,7 @@ def _handle_signal(self, sig):
         handle = self._signal_handlers.get(sig)
         if handle is None:
             return  # Assume it's some race condition.
-        if handle._cancelled:
+        if handle.cancelled():
             self.remove_signal_handler(sig)  # Remove it properly.
         else:
             self._add_callback_signalsafe(handle)

Lib/test/test_asyncio/test_tasks.py

@@ -1810,11 +1810,11 @@ def call_later(delay, callback, *args):
         loop.call_later = call_later
         test_utils.run_briefly(loop)
 
-        self.assertFalse(handle._cancelled)
+        self.assertFalse(handle.cancelled())
 
         t.cancel()
         test_utils.run_briefly(loop)
-        self.assertTrue(handle._cancelled)
+        self.assertTrue(handle.cancelled())
 
     def test_task_cancel_sleeping_task(self):