Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Portions of this file are derived from CPython's asyncio sources
2# (notably asyncio.base_events and asyncio.selector_events).
3# Copyright (c) Python Software Foundation.
4# Licensed under the Python Software Foundation License Version 2.
5# See LICENSES/PSF-2.0.txt and THIRD_PARTY_NOTICES for details.
7import collections
8import errno
9import itertools
10import os
11import socket
12import asyncio
13import sys
15from .api_utils import _is_asyncio_loop, _check_ssl_socket, _logger, _HAS_IPv6, _ensure_resolved, \
16 _validate_ssl_timeout, _validate_bio_size, Server
17from .ssl_transport import SSLTransport_Transport
18from .transport import (aiofn_is_buffered_protocol)
19from .wrapped_transport import (
20 _WrappedProtocol, _WrappedBufferedProtocol,
21 _should_fallback_to_asyncio, _get_original_loop_method
22)
25async def create_server(
26 loop: asyncio.AbstractEventLoop,
27 protocol_factory, host=None, port=None,
28 *,
29 family=socket.AF_UNSPEC,
30 flags=socket.AI_PASSIVE,
31 sock=None,
32 backlog=100,
33 ssl=None,
34 reuse_address=None,
35 reuse_port=None,
36 keep_alive=None,
37 ssl_handshake_timeout=None,
38 ssl_shutdown_timeout=None,
39 ssl_incoming_bio_size=None,
40 ssl_outgoing_bio_size=None,
41 start_serving=True):
42 """Create a TCP server.
44 The host parameter can be a string, in that case the TCP server is
45 bound to host and port.
47 The host parameter can also be a sequence of strings and in that case
48 the TCP server is bound to all hosts of the sequence. If a host
49 appears multiple times (possibly indirectly e.g. when hostnames
50 resolve to the same IP address), the server is only bound once to that
51 host.
53 Return a Server object which can be used to stop the service.
55 This method is a coroutine.
56 """
57 if isinstance(ssl, bool):
58 raise TypeError('ssl argument must be an SSLContext or None')
60 ssl_handshake_timeout = _validate_ssl_timeout("ssl_handshake_timeout", ssl_handshake_timeout, ssl)
61 ssl_shutdown_timeout = _validate_ssl_timeout("ssl_shutdown_timeout", ssl_shutdown_timeout, ssl)
62 ssl_incoming_bio_size = _validate_bio_size("ssl_incoming_bio_size", ssl_incoming_bio_size, ssl)
63 ssl_outgoing_bio_size = _validate_bio_size("ssl_outgoing_bio_size", ssl_outgoing_bio_size, ssl)
65 if _should_fallback_to_asyncio(loop):
66 kwargs = {
67 'host': host,
68 'port': port,
69 'family': family,
70 'flags': flags,
71 'sock': sock,
72 'backlog': backlog,
73 'reuse_address': reuse_address,
74 'reuse_port': reuse_port,
75 'start_serving': start_serving
76 }
77 if sys.version_info >= (3, 13) and _is_asyncio_loop(loop):
78 kwargs['keep_alive'] = keep_alive
80 return await _create_server_fallback(
81 loop, protocol_factory, ssl,
82 ssl_handshake_timeout, ssl_shutdown_timeout,
83 ssl_incoming_bio_size, ssl_outgoing_bio_size,
84 **kwargs)
86 if sock is not None:
87 _check_ssl_socket(sock)
89 if host is not None or port is not None:
90 if sock is not None:
91 raise ValueError(
92 'host/port and sock can not be specified at the same time')
94 if reuse_address is None:
95 reuse_address = os.name == "posix" and sys.platform != "cygwin"
96 sockets = []
97 if host == '':
98 hosts = [None]
99 elif (isinstance(host, str) or
100 not isinstance(host, collections.abc.Iterable)):
101 hosts = [host]
102 else:
103 hosts = host
105 fs = [_create_server_getaddrinfo(loop, host, port, family=family, flags=flags)
106 for host in hosts]
107 infos = await asyncio.gather(*fs)
108 infos = set(itertools.chain.from_iterable(infos))
110 completed = False
111 try:
112 for res in infos:
113 af, socktype, proto, canonname, sa = res
114 try:
115 sock = socket.socket(af, socktype, proto)
116 except socket.error:
117 # Assume it's a bad family/type/protocol combination.
118 if loop.get_debug():
119 _logger.warning('create_server() failed to create '
120 'socket.socket(%r, %r, %r)',
121 af, socktype, proto, exc_info=True)
122 continue
123 sockets.append(sock)
124 if reuse_address:
125 sock.setsockopt(
126 socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
127 # Since Linux 6.12.9, SO_REUSEPORT is not allowed
128 # on other address families than AF_INET/AF_INET6.
129 if reuse_port and af in (socket.AF_INET, socket.AF_INET6):
130 _set_reuseport(sock)
131 if keep_alive:
132 sock.setsockopt(
133 socket.SOL_SOCKET, socket.SO_KEEPALIVE, True)
134 # Disable IPv4/IPv6 dual stack support (enabled by
135 # default on Linux) which makes a single socket
136 # listen on both address families.
137 if (_HAS_IPv6 and
138 af == socket.AF_INET6 and
139 hasattr(socket, 'IPPROTO_IPV6')):
140 sock.setsockopt(socket.IPPROTO_IPV6,
141 socket.IPV6_V6ONLY,
142 True)
143 try:
144 sock.bind(sa)
145 except OSError as err:
146 msg = ('error while attempting '
147 'to bind on address %r: %s'
148 % (sa, str(err).lower()))
149 if err.errno == errno.EADDRNOTAVAIL:
150 # Assume the family is not enabled (bpo-30945)
151 sockets.pop()
152 sock.close()
153 if loop.get_debug():
154 _logger.warning(msg)
155 continue
156 raise OSError(err.errno, msg) from None
158 if not sockets:
159 raise OSError('could not bind on any address out of %r'
160 % ([info[4] for info in infos],))
162 completed = True
163 finally:
164 if not completed:
165 for sock in sockets:
166 sock.close()
167 else:
168 if sock is None:
169 raise ValueError('Neither host/port nor sock were specified')
170 if sock.type != socket.SOCK_STREAM:
171 raise ValueError(f'A Stream Socket was expected, got {sock!r}')
172 sockets = [sock]
174 for sock in sockets:
175 sock.setblocking(False)
177 server = Server(
178 loop, sockets, protocol_factory,
179 ssl, backlog,
180 ssl_handshake_timeout,
181 ssl_shutdown_timeout,
182 ssl_incoming_bio_size,
183 ssl_outgoing_bio_size,
184 )
185 if start_serving:
186 server._start_serving()
187 # Skip one loop iteration so that all 'loop.add_reader'
188 # go through.
189 await asyncio.sleep(0)
191 if loop.get_debug():
192 _logger.info("%r is serving", server)
193 return server
196async def _create_server_fallback(loop,
197 protocol_factory,
198 ssl,
199 ssl_handshake_timeout,
200 ssl_shutdown_timeout,
201 ssl_incoming_bio_size,
202 ssl_outgoing_bio_size,
203 **kwargs
204):
205 if ssl:
206 sslcontext = None if isinstance(ssl, bool) else ssl
208 def ssl_protocol_factory():
209 protocol = protocol_factory()
210 server_side = True
211 tls_transport = SSLTransport_Transport(
212 loop, protocol, sslcontext,
213 server_side,
214 ssl_handshake_timeout,
215 ssl_shutdown_timeout,
216 ssl_incoming_bio_size,
217 ssl_outgoing_bio_size
218 )
219 return tls_transport.get_tls_protocol()
221 create_server = _get_original_loop_method(loop, "create_server")
222 return await create_server(ssl_protocol_factory, **kwargs)
223 else:
224 def wrapped_protocol_factory():
225 user_protocol = protocol_factory()
226 if aiofn_is_buffered_protocol(user_protocol):
227 return _WrappedBufferedProtocol(user_protocol)
228 else:
229 return _WrappedProtocol(user_protocol)
231 create_server = _get_original_loop_method(loop, "create_server")
232 return await create_server(wrapped_protocol_factory, **kwargs)
235def _set_reuseport(sock):
236 if not hasattr(socket, 'SO_REUSEPORT'):
237 raise ValueError('reuse_port not supported by socket module')
238 else:
239 try:
240 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
241 except OSError:
242 raise ValueError('reuse_port not supported by socket module, '
243 'SO_REUSEPORT defined but not implemented.')
246async def _create_server_getaddrinfo(loop, host, port, family, flags):
247 infos = await _ensure_resolved((host, port), family=family,
248 type=socket.SOCK_STREAM,
249 flags=flags, loop=loop)
250 if not infos:
251 raise OSError(f'getaddrinfo({host!r}) returned empty list')
252 return infos