loop: internal async referencing

tests/test_loop.py

@@ -162,12 +162,6 @@ def test_call_later(self):
         def callback():
             self.callback_called = True
 
-        def on_wakeup(async_handle):
-            async_handle.close()
-
-        self.async = uv.Async(on_wakeup=on_wakeup)
-        self.async.send()
-
         self.loop.call_later(callback)
         self.loop.run()
 

uv/base.py

@@ -194,6 +194,18 @@ def _destroy(self, _):
             self.run()
             self.close()
 
+    def reference_internal_async(self):
+        """
+        Reference the internal async handle used for wakeup.
+        """
+        lib.uv_ref(ffi.cast('uv_handle_t*', self.internal_uv_async))
+
+    def dereference_internal_async(self):
+        """
+        Dereference the internal async handle used for wakeup.
+        """
+        lib.uv_unref(ffi.cast('uv_handle_t*', self.internal_uv_async))
+
     @property
     def user_loop(self):
         """

uv/loop.py

@@ -373,6 +373,7 @@ def __init__(self, allocator=None, buffer_size=2**16, default=False):
         self.make_current()
         self.pending_structures = set()
         self.pending_callbacks = collections.deque()
+        self.pending_callbacks_lock = threading.RLock()
 
     @property
     def closed(self):
@@ -584,8 +585,10 @@ def call_later(self, callback, *arguments, **keywords):
         :type keywords:
             dict
         """
-        self.pending_callbacks.append((callback, arguments, keywords))
-        self.base_loop.wakeup()
+        with self.pending_callbacks_lock:
+            self.pending_callbacks.append((callback, arguments, keywords))
+            self.base_loop.wakeup()
+            self.base_loop.reference_internal_async()
 
     def on_wakeup(self):
         """
@@ -597,13 +600,17 @@ def on_wakeup(self):
         """
         try:
             while True:
-                callback, arguments, keywords = self.pending_callbacks.popleft()
+                with self.pending_callbacks_lock:
+                    callback, arguments, keywords = self.pending_callbacks.popleft()
                 try:
                     callback(*arguments, **keywords)
                 except BaseException:
                     self.handle_exception()
         except IndexError:
             pass
+        with self.pending_callbacks_lock:
+            if not self.pending_callbacks:
+                self.base_loop.dereference_internal_async()
 
     def handle_exception(self):
         """