ENH: First pass free thread compatible fsenvents

Author: HaoZeke

src/watchdog_fsevents.c

@@ -15,6 +15,7 @@
 #include <stdlib.h>
 #include <signal.h>
 
+#include "pythoncapi_compat.h"
 
 /* Compatibility; since fsevents won't set these on earlier macOS versions the properties will always be False */
 #if MAC_OS_X_VERSION_MAX_ALLOWED < MAC_OS_X_VERSION_10_13
@@ -258,6 +259,14 @@ PyObject *thread_to_run_loop = NULL;
  */
 PyObject *watch_to_stream = NULL;
 
+#ifdef Py_GIL_DISABLED
+  static PyMutex g_watchdog_module_data_lock = {0};
+  #define LOCK_GDATA_MUTEX PyMutex_Lock(&g_watchdog_module_data_lock)
+  #define UNLOCK_GDATA_MUTEX PyMutex_Unlock(&g_watchdog_module_data_lock)
+#else
+  #define LOCK_GDATA_MUTEX ((void)0)
+  #define UNLOCK_GDATA_MUTEX ((void)0)
+#endif
 
 /**
  * PyCapsule destructor.
@@ -520,9 +529,10 @@ watchdog_CFMutableArrayRef_from_PyStringList(PyObject *py_string_list)
      * CFString array list. */
     for (i = 0; i < string_list_size; ++i)
     {
-        py_string = PyList_GetItem(py_string_list, i);
+        py_string = PyList_GetItemRef(py_string_list, i);
         G_RETURN_NULL_IF_NULL(py_string);
         cf_string = PyString_AsUTF8EncodedCFStringRef(py_string);
+        Py_DECREF(py_string);
         G_RETURN_NULL_IF_NULL(cf_string);
         CFArraySetValueAtIndex(array_of_cf_string, i, cf_string);
         CFRelease(cf_string);
@@ -622,16 +632,21 @@ watchdog_add_watch(PyObject *self, PyObject *args)
                                           &emitter_thread, &watch,
                                           &python_callback, &paths_to_watch));
 
+    /* Ensure only one thread tries to add the watch */
+    LOCK_GDATA_MUTEX;
+
     /* Watch must not already be scheduled. */
     if(PyDict_Contains(watch_to_stream, watch) == 1) {
         PyErr_Format(PyExc_RuntimeError, "Cannot add watch %S - it is already scheduled", watch);
+        UNLOCK_GDATA_MUTEX;
         return NULL;
     }
 
     /* Create an instance of the callback information structure. */
     stream_callback_info_ref = PyMem_New(StreamCallbackInfo, 1);
     if(stream_callback_info_ref == NULL) {
         PyErr_SetString(PyExc_SystemError, "Failed allocating stream callback info");
+        UNLOCK_GDATA_MUTEX;
         return NULL;
     }
 
@@ -643,13 +658,15 @@ watchdog_add_watch(PyObject *self, PyObject *args)
     if (!stream_ref) {
         PyMem_Free(stream_callback_info_ref);
         PyErr_SetString(PyExc_RuntimeError, "Failed creating fsevent stream");
+        UNLOCK_GDATA_MUTEX;
         return NULL;
     }
     value = PyCapsule_New(stream_ref, NULL, watchdog_pycapsule_destructor);
     if (!value || !PyCapsule_IsValid(value, NULL)) {
         PyMem_Free(stream_callback_info_ref);
         FSEventStreamInvalidate(stream_ref);
         FSEventStreamRelease(stream_ref);
+        UNLOCK_GDATA_MUTEX;
         return NULL;
     }
     PyDict_SetItem(watch_to_stream, watch, value);
@@ -666,6 +683,9 @@ watchdog_add_watch(PyObject *self, PyObject *args)
         run_loop_ref = PyCapsule_GetPointer(value, NULL);
     }
 
+    /* Finished working with global dictionaries */
+    UNLOCK_GDATA_MUTEX;
+
     /* Schedule the stream with the obtained runloop. */
     FSEventStreamScheduleWithRunLoop(stream_ref, run_loop_ref, kCFRunLoopDefaultMode);
 
@@ -707,6 +727,7 @@ watchdog_read_events(PyObject *self, PyObject *args)
     CFRunLoopRef run_loop_ref = NULL;
     PyObject *emitter_thread = NULL;
     PyObject *value = NULL;
+    int val_ind = 0;
 
     G_RETURN_NULL_IF_NOT(PyArg_ParseTuple(args, "O:loop", &emitter_thread));
 
@@ -716,27 +737,42 @@ watchdog_read_events(PyObject *self, PyObject *args)
     PyEval_InitThreads();
 #endif
 
+    LOCK_GDATA_MUTEX;
     /* Allocate information and store thread state. */
-    value = PyDict_GetItem(thread_to_run_loop, emitter_thread);
+    val_ind = PyDict_GetItemRef(thread_to_run_loop, emitter_thread, &value);
     if (G_IS_NULL(value))
     {
+        if (PyErr_Occurred()) { UNLOCK_GDATA_MUTEX; return NULL; }
         run_loop_ref = CFRunLoopGetCurrent();
         value = PyCapsule_New(run_loop_ref, NULL, watchdog_pycapsule_destructor);
-        PyDict_SetItem(thread_to_run_loop, emitter_thread, value);
+        if (!value) { UNLOCK_GDATA_MUTEX; return NULL; }
+        if (PyDict_SetItem(thread_to_run_loop, emitter_thread, value) < 0) {
+            Py_DECREF(value);
+            UNLOCK_GDATA_MUTEX;
+            return NULL;
+        }
         Py_INCREF(emitter_thread);
-        Py_INCREF(value);
     }
+    UNLOCK_GDATA_MUTEX;
 
     /* No timeout, block until events. */
     Py_BEGIN_ALLOW_THREADS;
     CFRunLoopRun();
     Py_END_ALLOW_THREADS;
 
     /* Clean up state information. */
-    if (PyDict_DelItem(thread_to_run_loop, emitter_thread) == 0)
+    LOCK_GDATA_MUTEX;
+    int del_result = PyDict_DelItem(thread_to_run_loop, emitter_thread);
+    UNLOCK_GDATA_MUTEX;
+    if (del_result == 0)
     {
         Py_DECREF(emitter_thread);
-        Py_INCREF(value);
+        Py_DECREF(value);
+    } else if (del_result < 0 && PyErr_Occurred()){
+        Py_DECREF(value);
+        return NULL;
+    } else {
+        Py_DECREF(value);
     }
 
     G_RETURN_NULL_IF(PyErr_Occurred());
@@ -754,11 +790,15 @@ static PyObject *
 watchdog_flush_events(PyObject *self, PyObject *watch)
 {
     UNUSED(self);
-    PyObject *value = PyDict_GetItem(watch_to_stream, watch);
+    PyObject *value = NULL;
+    LOCK_GDATA_MUTEX;
+    PyDict_GetItemRef(watch_to_stream, watch, &value);
 
     FSEventStreamRef stream_ref = PyCapsule_GetPointer(value, NULL);
 
     FSEventStreamFlushSync(stream_ref);
+    UNLOCK_GDATA_MUTEX;
+    Py_DECREF(value);
 
     Py_INCREF(Py_None);
     return Py_None;
@@ -773,19 +813,23 @@ static PyObject *
 watchdog_remove_watch(PyObject *self, PyObject *watch)
 {
     UNUSED(self);
-    PyObject *streamref_capsule = PyDict_GetItem(watch_to_stream, watch);
+    PyObject *streamref_capsule = NULL;
+    LOCK_GDATA_MUTEX;
+    PyDict_GetItemRef(watch_to_stream, watch, &streamref_capsule);
     if (!streamref_capsule) {
         // A watch might have been removed explicitly before, in which case we can simply early out.
         Py_RETURN_NONE;
     }
     PyDict_DelItem(watch_to_stream, watch);
+    UNLOCK_GDATA_MUTEX;
 
     FSEventStreamRef stream_ref = PyCapsule_GetPointer(streamref_capsule, NULL);
 
     FSEventStreamStop(stream_ref);
     FSEventStreamInvalidate(stream_ref);
     FSEventStreamRelease(stream_ref);
 
+    Py_DECREF(streamref_capsule);
     Py_RETURN_NONE;
 }
 
@@ -798,7 +842,11 @@ static PyObject *
 watchdog_stop(PyObject *self, PyObject *emitter_thread)
 {
     UNUSED(self);
-    PyObject *value = PyDict_GetItem(thread_to_run_loop, emitter_thread);
+    PyObject *value = NULL;
+    int val_ind = 0;
+    LOCK_GDATA_MUTEX;
+    val_ind = PyDict_GetItemRef(thread_to_run_loop, emitter_thread, &value);
+    UNLOCK_GDATA_MUTEX;
     if (G_IS_NULL(value)) {
       goto success;
     }
@@ -813,6 +861,7 @@ watchdog_stop(PyObject *self, PyObject *emitter_thread)
     }
 
  success:
+    Py_DECREF(value);
     Py_INCREF(Py_None);
     return Py_None;
 }