gh-98896: Fix parsing issue in resource_tracker to allow shared memory names containing colons

Lib/multiprocessing/resource_tracker.py

@@ -15,6 +15,7 @@
 # this resource tracker process, "killall python" would probably leave unlinked
 # resources.
 
+import base64
 import os
 import signal
 import sys
@@ -248,6 +249,12 @@ def _write(self, msg):
         assert nbytes == len(msg), f"{nbytes=} != {len(msg)=}"
 
     def _send(self, cmd, name, rtype):
+        # Encode shared_memory names as they are created by the user and may contain
+        # colons or newlines.
+        if rtype == "shared_memory":
+            b = name.encode('utf-8', 'surrogateescape')
+            name = base64.urlsafe_b64encode(b).decode('ascii')
+
         msg = f"{cmd}:{name}:{rtype}\n".encode("ascii")
         if len(msg) > 512:
             # posix guarantees that writes to a pipe of less than PIPE_BUF
@@ -285,7 +292,13 @@ def main(fd):
         with open(fd, 'rb') as f:
             for line in f:
                 try:
-                    cmd, name, rtype = line.strip().decode('ascii').split(':')
+                    cmd, enc_name, rtype = line.rstrip(b'\n').decode('ascii').split(':', 2)
+                    if rtype == "shared_memory":
+                        name = base64.urlsafe_b64decode(enc_name.encode('ascii')).decode('utf-8', 'surrogateescape')
+                    else:
+                        # Semaphore names are generated internally, so no encoding is needed.
+                        name = enc_name
+
                     cleanup_func = _CLEANUP_FUNCS.get(rtype, None)
                     if cleanup_func is None:
                         raise ValueError(

Lib/test/_test_multiprocessing.py

@@ -7335,3 +7335,46 @@ def test_forkpty(self):
         res = assert_python_failure("-c", code, PYTHONWARNINGS='error')
         self.assertIn(b'DeprecationWarning', res.err)
         self.assertIn(b'is multi-threaded, use of forkpty() may lead to deadlocks in the child', res.err)
+
+@unittest.skipUnless(HAS_SHMEM, "requires multiprocessing.shared_memory")
+class TestSharedMemoryNames(unittest.TestCase):
+    def test_that_shared_memory_name_with_colons_has_no_resource_tracker_errors(self):
+        # Test script that creates and cleans up shared memory with colon in name
+        test_script = textwrap.dedent("""
+            import sys
+            from multiprocessing import shared_memory
+            import time
+
+            # Test various patterns of colons in names
+            test_names = [
+                "a:b",
+                "a:b:c",
+                "test:name:with:many:colons",
+                ":starts:with:colon",
+                "ends:with:colon:",
+                "::double::colons::",
+                "name\\nwithnewline",
+                "name-with-trailing-newline\\n",
+                "\\nname-starts-with-newline",
+                "colons:and\\nnewlines:mix",
+                "multi\\nline\\nname",
+            ]
+
+            for name in test_names:
+                try:
+                    shm = shared_memory.SharedMemory(create=True, size=100, name=name)
+                    shm.buf[:5] = b'hello'  # Write something to the shared memory
+                    shm.close()
+                    shm.unlink()
+
+                except Exception as e:
+                    print(f"Error with name '{name}': {e}", file=sys.stderr)
+                    sys.exit(1)
+
+            print("SUCCESS")
+        """)
+
+        rc, out, err = assert_python_ok("-c", test_script)
+        self.assertIn(b"SUCCESS", out)
+        self.assertNotIn(b"traceback", err.lower(), err)
+        self.assertNotIn(b"resource_tracker.py", err, err)

Misc/NEWS.d/next/Library/2025-09-03-20-18-39.gh-issue-98896.tjez89.rst

@@ -0,0 +1,2 @@
+Fix a failure in multiprocessing resource_tracker when SharedMemory names contain colons.
+Patch by Rani Pinchuk.