Deflake the channel_argument_test & improve socket error handling

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/iomgr.pyx.pxi

@@ -151,6 +151,7 @@ cdef grpc_error* asyncio_socket_bind(
         _asyncio_apply_socket_options(socket, flags)
         socket.bind((host, port))
     except IOError as io_error:
+        socket.close()
         return socket_error("bind", str(io_error))
     else:
         aio_socket = _AsyncioSocket.create_with_py_socket(grpc_socket, socket)

src/python/grpcio/grpc/_cython/_cygrpc/aio/iomgr/socket.pyx.pxi

@@ -16,6 +16,8 @@ import socket as native_socket
 
 from libc cimport string
 
+cdef int _ASYNCIO_STREAM_DEFAULT_SOCKET_BACKLOG = 100
+
 
 # TODO(https://github.com/grpc/grpc/issues/21348) Better flow control needed.
 cdef class _AsyncioSocket:
@@ -182,6 +184,7 @@ cdef class _AsyncioSocket:
         self._grpc_accept_cb(self._grpc_socket, self._grpc_client_socket, grpc_error_none())
 
     cdef listen(self):
+        self._py_socket.listen(_ASYNCIO_STREAM_DEFAULT_SOCKET_BACKLOG)
         async def create_asyncio_server():
             self._server = await asyncio.start_server(
                 self._new_connection_callback,

src/python/grpcio_tests/tests_aio/unit/channel_argument_test.py

@@ -17,6 +17,7 @@ import asyncio
 import logging
 import platform
 import random
+import errno
 import unittest
 
 import grpc
@@ -37,11 +38,13 @@ _OPTIONS = (
     (_DISABLE_REUSE_PORT, ((_SOCKET_OPT_SO_REUSEPORT, 0),)),
 )
 
-_NUM_SERVER_CREATED = 100
+_NUM_SERVER_CREATED = 10
 
 _GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH = 'grpc.max_receive_message_length'
 _MAX_MESSAGE_LENGTH = 1024
 
+_ADDRESS_TOKEN_ERRNO = errno.EADDRINUSE, errno.ENOSR
+
 
 class _TestPointerWrapper(object):
 
@@ -78,8 +81,11 @@ async def test_if_reuse_port_enabled(server: aio.Server):
         ) as (unused_host, bound_port):
             assert bound_port == port
     except OSError as e:
-        assert 'Address already in use' in str(e)
+        if e.errno in _ADDRESS_TOKEN_ERRNO:
             return False
+        else:
+            logging.exception(e)
+            raise
     else:
         return True