Initial commit

Author: YvesDup

Lib/test/_test_multiprocessing.py

@@ -137,8 +137,12 @@ def _resource_unlink(name, rtype):
 
 WAIT_ACTIVE_CHILDREN_TIMEOUT = 5.0
 
-HAVE_GETVALUE = not getattr(_multiprocessing,
-                            'HAVE_BROKEN_SEM_GETVALUE', False)
+# Since gh-125828, we no longer need HAVE_GETVALUE.
+# This value should be remove from Modules/_multiprocessing/multiprocessing.c.
+# when cleanup is complete.
+# -------------------
+# HAVE_GETVALUE = not getattr(_multiprocessing,
+#                            'HAVE_BROKEN_SEM_GETVALUE', False)
 
 WIN32 = (sys.platform == "win32")
 
@@ -6741,6 +6745,61 @@ def f(x): return x*x
         self.assertEqual("332833500", out.decode('utf-8').strip())
         self.assertFalse(err, msg=err.decode('utf-8'))
 
+#
+# Tests for workaround macOSX Semaphore
+#
+
+ACQUIRE, RELEASE = range(2)
+@unittest.skipIf(sys.platform != "darwin", "MacOSX only")
+class _TestMacOSXSemaphore(BaseTestCase):
+    ALLOWED_TYPES = ('processes',)
+    @classmethod
+    def _run_thread(cls, sem, meth, ntime, delay):
+        if meth == ACQUIRE:
+            for _ in range(ntime):
+                sem.acquire()
+                time.sleep(delay)
+        else:
+            for _ in range(ntime):
+                sem.release()
+                time.sleep(delay)
+
+    @classmethod
+    def _run_process(cls, sem, sem_meth, nthread=1, ntime=10, delay=0.1):
+        ts = []
+        for _ in range(nthread):
+            t = threading.Thread(target=cls._run_thread,
+                            args=(sem, sem_meth, ntime, delay))
+            ts.append(t)
+        for t in ts:
+            t.start()
+        for t in ts:
+            t.join()
+
+    def test_mix_several_acquire_release(self):
+        # n processes, threads per process and loops per threads
+        n_p_acq, n_th_acq, n_loop_acq = 15, 5, 20
+        n_p_rel, n_th_rel, n_loop_rel = 8, 8, 8
+
+        n_acq = n_p_acq*n_th_acq*n_loop_acq
+        n_rel = n_p_rel*n_th_rel*n_loop_rel
+        sem = self.Semaphore(n_acq)
+        ps = []
+        for _ in range(n_p_acq):
+            p = self.Process(target=self._run_process,
+                             args=(sem, ACQUIRE, n_th_acq, n_loop_acq, 0.01))
+            ps.append(p)
+
+        for _ in range(n_p_rel):
+            p = self.Process(target=self._run_process,
+                             args=(sem, RELEASE, n_th_rel, n_loop_rel, 0.005))
+            ps.append(p)
+
+        for p in ps:
+            p.start()
+        for p in ps:
+            p.join()
+        self.assertEqual(sem.get_value(), n_rel)
 
 #
 # Mixins

Modules/_multiprocessing/dump_shm_macosx/dump_shared_mem.c

@@ -0,0 +1,102 @@
+#include <unistd.h>
+#include <stdio.h>  // puts, printf, scanf
+#include <time.h>   // ctime, time
+#include <string.h> // memcpy, memcmp
+
+#include <semaphore.h> // sem_t
+typedef sem_t *SEM_HANDLE;
+
+#define MAX_SEMAPHORES_SHOW  32
+
+#include "../semaphore_macosx.h"
+#include "shared_mem.h"
+
+// Static datas for each process.
+CountersWorkaround shm_semlock_counters = {
+    .state_this = THIS_NOT_OPEN,
+    .name_shm = "/shm_gh125828",
+    .handle_shm = (MEMORY_HANDLE)0,
+    .create_shm = 0,
+    .name_shm_lock = "/mp_gh125828",
+    .handle_shm_lock = (SEM_HANDLE)0,
+    .header = (HeaderObject *)NULL,
+    .counters = (CounterObject *)NULL,
+};
+
+HeaderObject *header = NULL;
+CounterObject *counter =  NULL;
+
+static char *show_counter(char *p, CounterObject *counter) {
+    sprintf(p, "p:%p, n:%s, v:%d, u:%d, t:%s", counter,
+                                           counter->sem_name,
+                                           counter->internal_value,
+                                           counter->unlink_done,
+                                           ctime(&counter->ctimestamp));
+    return p;
+}
+
+static void dump_shm_semlock_counters(void) {
+puts(__func__);
+
+    char buf[256];
+    int i = 0, j = 0;
+
+    if (shm_semlock_counters.state_this == THIS_AVAILABLE) {
+        CounterObject *counter = shm_semlock_counters.counters;
+        HeaderObject *header = shm_semlock_counters.header;
+        dump_shm_semlock_header_counters();
+        dump_shm_semlock_header();
+        int show_max = header->n_semlocks > MAX_SEMAPHORES_SHOW ? MAX_SEMAPHORES_SHOW : header->n_semlocks;
+        for(; i < header->n_slots && j < show_max; i++, counter++ ) {
+            if (counter->sem_name[0] != 0) {
+                printf("%s", show_counter(buf, counter));
+                ++j;
+            }
+        }
+        if (show_max < header->n_semlocks) {
+            printf("......\n--------- More %d Semphores ---------\n", header->n_semlocks-show_max);
+        }
+    }
+}
+
+int main(int argc, char *argv[]) {
+    int repeat = 0;
+    long udelay = 5000;
+    HeaderObject save = {0};
+    int unlink = 0;
+    int force_open = 1;
+    int release_lock = 1;
+
+    puts("--------");
+    printf("PID:%d, PPID:%d\n", getpid(), getppid());
+    connect_shm_semlock_counters(unlink, force_open, release_lock);
+    puts("+++++++++");
+    if (argc > 1) {
+        sscanf(argv[1], "%d", &repeat);
+        if (argc >= 2) {
+            puts(argv[2]);
+            sscanf(argv[2], "%lu", &udelay);
+        }
+    } else {
+        puts("dump_shared_mem <repeat> <delay> where:\n repeat (-1 "
+             "is infinite) and a delay (us) between two dumps \n");
+        return 1;
+    }
+
+    printf("Repeat:%d, udelay:%lu\n", repeat, udelay);
+
+    if (shm_semlock_counters.state_this == THIS_AVAILABLE) {
+        memset(&save, '\0', sizeof(save));
+        do {
+            if (memcmp(&save, shm_semlock_counters.header, sizeof(HeaderObject)) ) {
+                time_t timestamp = time(NULL);
+                puts(ctime(&timestamp));
+                dump_shm_semlock_counters();
+                memcpy(&save, shm_semlock_counters.header, sizeof(HeaderObject));
+                puts("==========");
+            }
+            usleep(udelay);
+        } while(repeat--);
+    }
+    return 1;
+}

Modules/_multiprocessing/dump_shm_macosx/make_all.sh

@@ -0,0 +1,2 @@
+gcc -o ./dump_shm ./dump_shared_mem.c ./shared_mem.c
+gcc -o ./reset_shm ./reset_shared_mem.c ./shared_mem.c

Modules/_multiprocessing/dump_shm_macosx/readme.md

@@ -0,0 +1,41 @@
+** For MacOSX only **
+---
+
+This directory contains 2 programs :
+
+* dump_shared_mem: view content of shared memory.
+* reset_shared_mem: erase all stored datas of shared memory.
+
+the `make_all.sh` batch builds these 2 programs.
+
+# dump_shm.
+
+`dump_shm` tries to connect to the shared memory only if its exists.
+This program doesn't use synchronization primitive to read the shared memory.
+To quit this program, press `Ctrl+C`.
+
+```zsh
+dump_shm -1 300
+```
+Executes this program forever, and check all 300 *us* if shared memory changes.
+
+When there are changes in the shared memory (only about sempahore count), program prints the new content of shared memory as below:
+
+```zsh
+==========
+Tue Feb 25 17:04:05 2025
+
+dump_shm_semlock_counters
+header:0x1022b4000 - counter array:0x1022b4010
+n sems:2 - n sem_slots:87551, n procs:1, size_shm:2801664
+p:0x1022b4010, n:/mp-fwl20ahw, v:6, r:0, t:Tue Feb 25 17:04:05 2025
+p:0x1022b4030, n:/mp-z3635cdr, v:6, r:0, t:Tue Feb 25 17:04:04 2025
+
+```
+
+# reset_shm.
+
+`reset_shm` tries to connect to the shared memory only if its exists.
+This program uses synchronization primitive to read the shared memory.
+
+When exits, this program calls `shm_unlink`.

Modules/_multiprocessing/dump_shm_macosx/reset_shared_mem.c

@@ -0,0 +1,84 @@
+#include <unistd.h>
+#include <stdio.h>  // puts, printf, scanf
+#include <string.h> // memcpy, memcmp, memset
+
+#include <semaphore.h>
+typedef sem_t *SEM_HANDLE;
+
+#include "../semaphore_macosx.h"
+#include "shared_mem.h"
+
+// Static datas for each process.
+CountersWorkaround shm_semlock_counters = {
+    .state_this = THIS_NOT_OPEN,
+    .name_shm = "/shm_gh125828",
+    .handle_shm = (MEMORY_HANDLE)0,
+    .create_shm = 0,
+    .name_shm_lock = "/mp_gh125828",
+    .handle_shm_lock = (SEM_HANDLE)0,
+    .header = (HeaderObject *)NULL,
+    .counters = (CounterObject *)NULL,
+};
+
+HeaderObject *header = NULL;
+CounterObject *counter =  NULL;
+
+static void reset_shm_semlock_counters(int size, int nb_slots) {
+puts(__func__);
+
+    if (shm_semlock_counters.state_this == THIS_AVAILABLE) {
+        if (ACQUIRE_SHM_LOCK) {
+            CounterObject *counter = shm_semlock_counters.counters;
+            HeaderObject *header = shm_semlock_counters.header;
+            dump_shm_semlock_header_counters();
+            dump_shm_semlock_header();
+            long size_to_reset = header->size_shm-sizeof(HeaderObject);
+            printf("1 - size to reset:%lu\n", size_to_reset);
+            if (size && size <= size_to_reset) {
+                memset(counter, 0, size);
+
+            } else {
+                memset(counter, 0, size_to_reset);
+            }
+            puts("2 - Reset all header parameters");
+            if (nb_slots) {
+                header->n_slots = nb_slots;
+            }
+            header->n_semlocks = 0;
+            header->n_slots = CALC_NB_SLOTS(header->size_shm);
+            header->n_procs = 0;
+            dump_shm_semlock_header();
+            RELEASE_SHM_LOCK;
+        }
+    } else {
+        puts("No datas");
+    }
+}
+
+int main(int argc, char *argv[]) {
+    char c;
+    int size = CALC_SIZE_SHM;
+    int nb_slots = CALC_NB_SLOTS(size);
+    int unlink = 1;
+    int force_open = 1;
+    int release_lock = 1;
+
+    if (argc >= 2) {
+        sscanf(argv[2], "%d", &size);
+        nb_slots = CALC_NB_SLOTS(size);
+    }
+    puts("--------");
+    printf("size:%d, sem slots:%d\n", size, nb_slots);
+    connect_shm_semlock_counters(unlink, force_open, release_lock);
+    puts("+++++++++");
+    dump_shm_semlock_header_counters();
+    dump_shm_semlock_header();
+    if (shm_semlock_counters.state_this == THIS_AVAILABLE) {
+        puts("confirm (Y/N):");
+        c = getchar();
+        if ( c == 'Y' || c == 'y') {
+            reset_shm_semlock_counters(size, nb_slots);
+        }
+    }
+    return 1;
+}