pythongh-111246: Add opt-out of Unix socket cleanup (python#111246)

Allow the implicit cleanup to be disabled in case there are some odd
corner cases where this causes issues.

Doc/library/asyncio-eventloop.rst

@@ -776,7 +776,7 @@ Creating network servers
                           *, sock=None, backlog=100, ssl=None, \
                           ssl_handshake_timeout=None, \
                           ssl_shutdown_timeout=None, \
-                          start_serving=True)
+                          start_serving=True, cleanup_socket=True)
 
    Similar to :meth:`loop.create_server` but works with the
    :py:const:`~socket.AF_UNIX` socket family.
@@ -786,6 +786,10 @@ Creating network servers
    :class:`str`, :class:`bytes`, and :class:`~pathlib.Path` paths
    are supported.
 
+   If *cleanup_socket* is True then the Unix socket will automatically
+   be removed from the filesystem when the server is closed, unless the
+   socket has been replaced after the server has been created.
+
    See the documentation of the :meth:`loop.create_server` method
    for information about arguments to this method.
 
@@ -802,9 +806,7 @@ Creating network servers
 
    .. versionchanged:: 3.13
 
-      The Unix socket will automatically be removed from the filesystem
-      when the server is closed, unless the socket has been replaced
-      after the server has been created.
+      Added the *cleanup_socket* parameter.
 
 
 .. coroutinemethod:: loop.connect_accepted_socket(protocol_factory, \

Lib/asyncio/unix_events.py

@@ -285,7 +285,7 @@ async def create_unix_server(
             sock=None, backlog=100, ssl=None,
             ssl_handshake_timeout=None,
             ssl_shutdown_timeout=None,
-            start_serving=True):
+            start_serving=True, cleanup_socket=True):
         if isinstance(ssl, bool):
             raise TypeError('ssl argument must be an SSLContext or None')
 
@@ -341,13 +341,14 @@ async def create_unix_server(
                 raise ValueError(
                     f'A UNIX Domain Stream Socket was expected, got {sock!r}')
 
-        path = sock.getsockname()
-        # Check for abstract socket. `str` and `bytes` paths are supported.
-        if path[0] not in (0, '\x00'):
-            try:
-                self._unix_server_sockets[sock] = os.stat(path).st_ino
-            except FileNotFoundError:
-                pass
+        if cleanup_socket:
+            path = sock.getsockname()
+            # Check for abstract socket. `str` and `bytes` paths are supported.
+            if path[0] not in (0, '\x00'):
+                try:
+                    self._unix_server_sockets[sock] = os.stat(path).st_ino
+                except FileNotFoundError:
+                    pass
 
         sock.setblocking(False)
         server = base_events.Server(self, [sock], protocol_factory,

Lib/test/test_asyncio/test_server.py

@@ -205,6 +205,17 @@ async def serve(*args):
             srv.close()
             self.assertTrue(os.path.exists(addr))
 
+    @socket_helper.skip_unless_bind_unix_socket
+    async def test_unix_server_cleanup_prevented(self):
+        with test_utils.unix_socket_path() as addr:
+            async def serve(*args):
+                pass
+
+            srv = await asyncio.start_unix_server(serve, addr, cleanup_socket=False)
+
+            srv.close()
+            self.assertTrue(os.path.exists(addr))
+
 
 @unittest.skipUnless(hasattr(asyncio, 'ProactorEventLoop'), 'Windows only')
 class ProactorStartServerTests(BaseStartServer, unittest.TestCase):