refactor semlock_acquire and semlock_release to make them atomic

Author: YvesDup

Modules/_multiprocessing/semaphore.c

@@ -376,78 +376,102 @@ shm_semlock_counters = {
     .handle_glock = (SEM_HANDLE)0,
     .header = (HeaderObject *)NULL,
     .counters = (CounterObject *)NULL,
+    .uuid = 0L
 };
 
 /* ---------------------------------*/
 /* unlink sem and shared mem from multiprocessing.resource_tracker
 module */
 
 static int
-_register_resource_tracker_sem(void)
+_resource_tracker_getfd(void)
 {
-    PyObject *f_register = PyImport_ImportModuleAttrString("multiprocessing.resource_tracker",\
-                                                         "register");
-    if (f_register == NULL) {
-        PyErr_Clear();
-        puts("no import");
-        return -1;
-    }
-    PyObject *psem_name = PyUnicode_FromString(GLOCK_NAME);
-    if (psem_name == NULL) {
-        PyErr_Clear();
-        Py_DECREF(f_register);
-        puts("no sem name");
-        return -1;
-    }
-    PyObject *ptype_sem = PyUnicode_FromString("semaphore");
-    if (psem_name == NULL) {
-        Py_DECREF(f_register);
-        Py_DECREF(psem_name);
+    const char *operation = "getfd";
+    PyObject *f_operation = PyImport_ImportModuleAttrString("multiprocessing.resource_tracker",
+                                                        operation);
+    if (f_operation == NULL) {
         PyErr_Clear();
-        puts("no type name");
+        printf("no call to %s\n", operation);
         return -1;
     }
-
-    PyObject *ret = PyObject_CallFunctionObjArgs(f_register, psem_name, ptype_sem, NULL);
-    Py_DECREF(psem_name);
-    Py_DECREF(ptype_sem);
-    Py_DECREF(f_register);
+    PyObject *ret = PyObject_CallNoArgs(f_operation);
+    Py_DECREF(f_operation);
     if (ret == NULL) {
         PyErr_Clear();
-        puts("no call to register");
+        printf("no call to %s()\n", operation);
         return -1;
     }
     Py_DECREF(ret);
-    DEBUG_PID_FUNC("register SEMAPHORE", 0, 0, "O");
+    DEBUG_PID_FUNC(operation, 0, 0, "");
     return 0;
 }
 
+/* 
+static long
+_get_GC(void)
+{
+    PyObject *module = PyImport_ImportModule("gc");
+    if (module == NULL) {
+        return -1;
+    }
+    Py_DECREF(module);
+    DEBUG_PID_FUNC("import gc", 0, 0, "");
+    return (long)module;
+}
+*/
+
 static int
-_register_resource_tracker_mem(void)
+_resource_tracker(const char *operation, const char *name, const char *rtype)
 {
-    PyObject *f_register = PyImport_ImportModuleAttrString("multiprocessing.resource_tracker",
-                                                        "register");
-    if (f_register == NULL) {
+    PyObject *f_operation = PyImport_ImportModuleAttrString("multiprocessing.resource_tracker",
+                                                        operation);
+    if (f_operation == NULL) {
+        PyErr_Clear();
+        printf("no call to %s\n", operation);
+        return -1;
+    }
+    PyObject *p_name = PyUnicode_FromString(name);
+    if (p_name == NULL) {
         PyErr_Clear();
-        puts("no call to register");
+        Py_DECREF(f_operation);
+        puts("no name");
         return -1;
     }
-    PyObject *pmem_name = PyUnicode_FromString(SHAREDMEM_NAME);
-    PyObject *ptype_mem = PyUnicode_FromString("shared_memory");
-    PyObject *ret = PyObject_CallFunctionObjArgs(f_register, pmem_name, ptype_mem, NULL);
-    Py_DECREF(pmem_name);
-    Py_DECREF(ptype_mem);
-    Py_DECREF(f_register);
+    PyObject *p_type = PyUnicode_FromString(rtype);
+    if (p_type == NULL) {
+        Py_DECREF(f_operation);
+        Py_DECREF(p_name);
+        PyErr_Clear();
+        puts("no rtype");
+        return -1;
+    }
+
+    PyObject *ret = PyObject_CallFunctionObjArgs(f_operation, p_name, p_type, NULL);
+    Py_DECREF(p_name);
+    Py_DECREF(p_type);
+    Py_DECREF(f_operation);
     if (ret == NULL) {
         PyErr_Clear();
-        puts("no call to register");
+        printf("no call to %s(%s, %s)\n", operation, name, rtype);
         return -1;
     }
     Py_DECREF(ret);
-    DEBUG_PID_FUNC("register MEMORY", 0, 0, "Okay");
+    DEBUG_PID_FUNC(operation, 0, 0, name);
     return 0;
 }
 
+static int
+_register_resource_tracker_sem(void)
+{
+    return _resource_tracker("register", GLOCK_NAME, "semaphore") ;
+}
+
+static int
+_register_resource_tracker_mem(void)
+{
+    return _resource_tracker("register", SHAREDMEM_NAME, "shared_memory") ;
+}
+
 /*-- */
 
 /*
@@ -635,7 +659,6 @@ connect_glock_and_lock(const char *sem_name)
     return 0;
     -- */
 
-
     SEM_HANDLE sem = SEM_FAILED;
     errno = 0;
    // create and lock the semaphore, via initial value set 0)
@@ -685,6 +708,21 @@ _posixshmem_shm_open(char *name, int flags, int mode)
     return fd;
 }
 
+
+static char *
+_build_glock_name(char *buf, int size, const char *name)
+{
+    // get uuid if not set.
+    if (shm_semlock_counters.uuid == 0L) {
+        shm_semlock_counters.uuid = 100;
+    }
+
+    // make name with glock sem name or shm name copncatenate with addr of GC
+    // PyOS_snprintf(buf, size, "%s_%012ld", name, shm_semlock_counters.uuid);
+    PyOS_snprintf(buf, size, "%s", name);
+    return buf;
+}
+
 static int
 create_shm_semlock_counters(const char *from_sem_name)
 {
@@ -695,20 +733,26 @@ create_shm_semlock_counters(const char *from_sem_name)
     char *datas = NULL;
     HeaderObject *header = NULL;
     long size_shm = ALIGN_SHM_PAGE(CALC_SIZE_SHM);
+    char glock_name[SIZE_GLOCK_NAME+16] ;
+
+    // Build glock name
+    _build_glock_name(glock_name, SIZE_GLOCK_NAME, shm_semlock_counters.name_glock);
 
     // Link to semaphore and lock immediatly.
-    if (connect_glock_and_lock(shm_semlock_counters.name_glock) < 0) {
+    if (connect_glock_and_lock(glock_name) < 0) {
         DEBUG_PID_FUNC(shm_semlock_counters.name_shm, shm, 0, "connect or create GLOCK Failed");
         return -1;
     };
 
     // We are alone to create/connect to the shared memory.
     if (shm_semlock_counters.handle_shm == (MEMORY_HANDLE)0) {
-        shm = _posixshmem_shm_open(shm_semlock_counters.name_shm, oflag, mode);
+            // Link to semaphore and lock immediatly.
+        _build_glock_name(glock_name, SIZE_GLOCK_NAME, shm_semlock_counters.name_shm);
+        shm = _posixshmem_shm_open(glock_name, oflag, mode);
         res = 0;
         if (shm == -1) {
             oflag |= O_CREAT;
-            shm = _posixshmem_shm_open(shm_semlock_counters.name_shm, oflag, mode);
+            shm = _posixshmem_shm_open(glock_name, oflag, mode);
             if (shm > 0) {
                 DEBUG_PID_FUNC(shm_semlock_counters.name_shm, shm, 0, "***create shm***");
                 // Set size.
@@ -834,15 +878,17 @@ _connect_shm_semlock_counters(const char *from_sem_name)
     int res = -1;
     char *datas = NULL;
     long size_shm = CALC_SIZE_SHM;
+    char glock_name[SIZE_GLOCK_NAME] ;
 
     // Link to semaphore and lock immediatly.
-    if (_connect_glock_and_lock(shm_semlock_counters.name_glock) < 0) {
+    _build_glock_name(glock_name, SIZE_GLOCK_NAME, shm_semlock_counters.name_glock);
+    if (_connect_glock_and_lock(glock_name) < 0) {
         DEBUG_PID_FUNC("", 0, 0, "error connect glock and lock");
         return -1;
     };
-    // The global lock is now acquired,
-    // connect to the shared memory.
-    shm = _posixshmem_shm_open(shm_semlock_counters.name_shm, oflag, S_IRUSR | S_IWUSR);
+    // The global lock is now acquired, connect to the shared memory.
+    _build_glock_name(glock_name, SIZE_GLOCK_NAME, shm_semlock_counters.name_shm);
+    shm = _posixshmem_shm_open(glock_name, oflag, S_IRUSR | S_IWUSR);
     if (shm == -1) {
         shm_semlock_counters.handle_shm = (MEMORY_HANDLE)0;
         return -1;
@@ -1030,13 +1076,46 @@ _multiprocessing_SemLock_acquire_impl(SemLockObject *self, int blocking,
         deadline.tv_nsec %= 1000000000;
     }
 
+#ifdef HAVE_BROKEN_SEM_GETVALUE
+    /* For counting semaphores: acquire the counter mutex before sem_trywait so
+       that the trywait and the internal_value decrement are atomic with respect
+       to release_impl.  We must NOT hold this mutex while blocking on the
+       semaphore (see blocking path below), because release_impl holds it while
+       calling sem_post — keeping it here would deadlock. */
+    if (ISSEMAPHORE(self) && !ACQUIRE_COUNTER_MUTEX(self->handle_mutex)) {
+        return NULL;
+    }
+#endif
+
     /* Check whether we can acquire without releasing the GIL and blocking */
     do {
         res = sem_trywait(self->handle);
         err = errno;
     } while (res < 0 && errno == EINTR && !PyErr_CheckSignals());
-    errno = err;
+    //errno = err;
 
+#ifdef HAVE_BROKEN_SEM_GETVALUE
+    if (ISSEMAPHORE(self)) {
+        if (res >= 0) {
+            /* Atomically decrement internal_value while the mutex is held */
+            --self->counter->internal_value;
+            if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
+                return NULL;
+            }
+            // acquire done
+            ++self->count;
+            self->last_tid = PyThread_get_thread_ident();
+            Py_RETURN_TRUE;
+        } else {
+            /* sem_trywait failed: release the mutex before potentially blocking */
+            if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
+                return NULL;
+            }
+        }
+    }
+    #endif
+
+    errno = err;
     if (res < 0 && errno == EAGAIN && blocking) {
         /* Couldn't acquire immediately, need to block */
         do {
@@ -1052,6 +1131,27 @@ _multiprocessing_SemLock_acquire_impl(SemLockObject *self, int blocking,
             if (res == MP_EXCEPTION_HAS_BEEN_SET)
                 break;
         } while (res < 0 && errno == EINTR && !PyErr_CheckSignals());
+
+#ifdef HAVE_BROKEN_SEM_GETVALUE
+        /* After the blocking wait: acquire the mutex and decrement
+           internal_value.  A brief non-atomic window exists between
+           sem_(timed)wait returning and acquiring the mutex here.  This is
+           unavoidable: release_impl holds the mutex across sem_post,
+           so we cannot hold it during sem_(timed)wait without deadlocking. */
+        if (ISSEMAPHORE(self) && res >= 0) {
+            if (ACQUIRE_COUNTER_MUTEX(self->handle_mutex)) {
+                --self->counter->internal_value;
+                if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
+                    return NULL;
+                }
+                ++self->count;
+                self->last_tid = PyThread_get_thread_ident();
+                Py_RETURN_TRUE;
+            } else {
+                return NULL;
+            }
+        }
+#endif
     }
     if (res < 0) {
         errno = err;
@@ -1063,25 +1163,13 @@ _multiprocessing_SemLock_acquire_impl(SemLockObject *self, int blocking,
         }
         return PyErr_SetFromErrno(PyExc_OSError);
     }
-#ifdef HAVE_BROKEN_SEM_GETVALUE
-    if (ISSEMAPHORE(self)) {
-        // error is set in ACQUIRE/RELEASE_* macros.
-        if (ACQUIRE_COUNTER_MUTEX(self->handle_mutex)) {
-            --self->counter->internal_value;
-            if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
-                return NULL;
-            }
-        } else {
-            return NULL;
-        }
-    }
-#endif
 
     ++self->count;
     self->last_tid = PyThread_get_thread_ident();
     Py_RETURN_TRUE;
 }
 
+
 /*[clinic input]
 @critical_section
 _multiprocessing.SemLock.release
@@ -1132,16 +1220,17 @@ _multiprocessing_SemLock_release_impl(SemLockObject *self)
             if (ISSEMAPHORE(self)) {
                 if (ACQUIRE_COUNTER_MUTEX(self->handle_mutex)) {
                     sval = self->counter->internal_value;
-                    if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
-                        return NULL;
-                    }
+                    //if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
+                    //    return NULL;
+                    //}
                 } else {
                     return NULL;
                 }
 
                 if (sval >= self->maxvalue) {
                     PyErr_SetString(PyExc_ValueError, "semaphore or lock "
                             "released too many times");
+                    RELEASE_COUNTER_MUTEX(self->handle_mutex);
                     return NULL;
                 }
             }
@@ -1165,13 +1254,8 @@ _multiprocessing_SemLock_release_impl(SemLockObject *self)
 
 #ifdef HAVE_BROKEN_SEM_GETVALUE
     if (ISSEMAPHORE(self)) {
-        // error is set in ACQUIRE/RELEASE_* macros.
-        if (ACQUIRE_COUNTER_MUTEX(self->handle_mutex)) {
-            ++self->counter->internal_value;
-            if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
-                return NULL;
-            }
-        } else {
+        ++self->counter->internal_value;
+        if (!RELEASE_COUNTER_MUTEX(self->handle_mutex)) {
             return NULL;
         }
     }

Modules/_multiprocessing/semaphore_macosx.h

@@ -28,6 +28,7 @@ typedef struct {
 
 #define SIZE_SEM_NAME 16
 #define SIZE_MUTEX_NAME (SIZE_SEM_NAME<<1)
+#define SIZE_GLOCK_NAME SIZE_MUTEX_NAME
 
 typedef struct {
     char sem_name[SIZE_SEM_NAME];  // Name of semaphore.
@@ -55,6 +56,9 @@ struct _CountersWorkaround{
     /*-- Pointers to shared memory --*/
     HeaderObject *header;     // Pointer to header (shared memory).
     CounterObject*counters;   // Pointer to the first item of fixed array (shared memory).
+
+    /* ---uuid link to main process- */
+    long uuid;
 };
 
 #define ISSEMAPHORE(o)                  ((o)->maxvalue > 1 && (o)->kind == SEMAPHORE)