2
Self Samples
2
Cumulative
1
Lines Hit
0.01%
% of Total
2
Max Self
2
Max Total
Intensity:
Cold
Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
from __future__ import annotations
2
3
import asyncio
4
import contextlib
5
import functools
6
import logging
7
import os
8
import platform
9
import random
10
import signal
11
import socket
12
import sys
13
import threading
14
import time
15
from collections.abc import Generator, Sequence
16
from email.utils import formatdate
17
from types import FrameType
18
from typing import TYPE_CHECKING, TypeAlias
19
20
from uvicorn._ansi import style
21
from uvicorn._compat import asyncio_run
22
from uvicorn.config import STARTUP_FAILURE, Config
23
24
if TYPE_CHECKING:
25
from uvicorn.protocols.http.auto_zttp_impl import AutoZttpProtocol
26
from uvicorn.protocols.http.h11_impl import H11Protocol
27
from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
28
from uvicorn.protocols.http.zttp_h2_impl import ZttpH2Protocol
29
from uvicorn.protocols.http.zttp_impl import ZttpProtocol
30
from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
31
from uvicorn.protocols.websockets.websockets_sansio_impl import WebSocketsSansIOProtocol
32
from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
33
34
Protocols: TypeAlias = (
35
H11Protocol
36
| HttpToolsProtocol
37
| ZttpProtocol
38
| ZttpH2Protocol
39
| AutoZttpProtocol
40
| WSProtocol
41
| WebSocketProtocol
42
| WebSocketsSansIOProtocol
43
)
44
45
HANDLED_SIGNALS = (
46
signal.SIGINT, # Unix signal 2. Sent by Ctrl+C.
47
signal.SIGTERM, # Unix signal 15. Sent by `kill <pid>`.
48
)
49
if sys.platform == "win32": # pragma: py-not-win32
50
HANDLED_SIGNALS += (signal.SIGBREAK,) # Windows signal 21. Sent by Ctrl+Break.
51
52
logger = logging.getLogger("uvicorn.error")
53
54
55
class ServerState:
56
"""
57
Shared servers state that is available between all protocol instances.
58
"""
59
60
def __init__(self) -> None:
61
self.total_requests = 0
62
self.connections: set[Protocols] = set()
63
self.tasks: set[asyncio.Task[None]] = set()
64
self.default_headers: list[tuple[bytes, bytes]] = []
65
66
67
class Server:
68
def __init__(self, config: Config) -> None:
69
self.config = config
70
self.server_state = ServerState()
71
72
self.started = False
73
self.should_exit = False
74
self.force_exit = False
75
self.last_notified = 0.0
76
77
self._captured_signals: list[int] = []
78
79
@functools.cached_property
80
def limit_max_requests(self) -> int | None:
81
if self.config.limit_max_requests is None:
82
return None
83
return self.config.limit_max_requests + random.randint(0, self.config.limit_max_requests_jitter)
84
85
def run(self, sockets: list[socket.socket] | None = None) -> None:
86
return asyncio_run(self.serve(sockets=sockets), loop_factory=self.config.get_loop_factory())
87
88
async def serve(self, sockets: list[socket.socket] | None = None) -> None:
89
with self.capture_signals():
90
await self._serve(sockets)
91
92
async def _serve(self, sockets: list[socket.socket] | None = None) -> None:
93
process_id = os.getpid()
94
95
config = self.config
96
if not config.loaded:
97
config.load()
98
99
self.lifespan = config.lifespan_class(config)
100
101
message = "Started server process [%d]"
102
color_message = "Started server process [" + style("%d", fg="cyan") + "]"
103
logger.info(message, process_id, extra={"color_message": color_message})
104
105
await self.startup(sockets=sockets)
106
if not self.should_exit:
107
await self.main_loop()
108
if self.started:
109
await self.shutdown(sockets=sockets)
110
111
message = "Finished server process [%d]"
112
color_message = "Finished server process [" + style("%d", fg="cyan") + "]"
113
logger.info(message, process_id, extra={"color_message": color_message})
114
115
async def startup(self, sockets: list[socket.socket] | None = None) -> None:
116
await self.lifespan.startup()
117
if self.lifespan.should_exit:
118
sys.exit(STARTUP_FAILURE)
119
120
config = self.config
121
122
def create_protocol(
123
_loop: asyncio.AbstractEventLoop | None = None,
124
) -> asyncio.Protocol:
125
2
2
return config.http_protocol_class( # type: ignore[call-arg]
126
config=config,
127
server_state=self.server_state,
128
app_state=self.lifespan.state,
129
_loop=_loop,
130
)
131
132
loop = asyncio.get_running_loop()
133
134
listeners: Sequence[socket.SocketType]
135
if sockets is not None: # pragma: full coverage
136
# Explicitly passed a list of open sockets.
137
# We use this when the server is run from a Gunicorn worker.
138
139
def _share_socket(
140
sock: socket.SocketType,
141
) -> socket.SocketType: # pragma py-not-win32
142
# Windows requires the socket be explicitly shared across
143
# multiple workers (processes).
144
from socket import fromshare # type: ignore[attr-defined]
145
146
sock_data = sock.share(os.getpid()) # type: ignore[attr-defined]
147
return fromshare(sock_data)
148
149
self.servers: list[asyncio.base_events.Server] = []
150
for sock in sockets:
151
is_windows = platform.system() == "Windows"
152
if config.workers > 1 and is_windows: # pragma: py-not-win32
153
sock = _share_socket(sock) # type: ignore[assignment]
154
server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
155
self.servers.append(server)
156
listeners = sockets
157
158
elif config.fd is not None: # pragma: py-win32
159
# Use an existing socket, from a file descriptor.
160
sock = socket.fromfd(config.fd, socket.AF_UNIX, socket.SOCK_STREAM)
161
server = await loop.create_server(create_protocol, sock=sock, ssl=config.ssl, backlog=config.backlog)
162
assert server.sockets is not None # mypy
163
listeners = server.sockets
164
self.servers = [server]
165
166
elif config.uds is not None: # pragma: py-win32
167
# Create a socket using UNIX domain socket.
168
uds_perms = 0o666
169
if os.path.exists(config.uds):
170
uds_perms = os.stat(config.uds).st_mode # pragma: full coverage
171
server = await loop.create_unix_server(
172
create_protocol, path=config.uds, ssl=config.ssl, backlog=config.backlog
173
)
174
os.chmod(config.uds, uds_perms)
175
assert server.sockets is not None # mypy
176
listeners = server.sockets
177
self.servers = [server]
178
179
else:
180
# Standard case. Create a socket from a host/port pair.
181
try:
182
server = await loop.create_server(
183
create_protocol,
184
host=config.host,
185
port=config.port,
186
ssl=config.ssl,
187
backlog=config.backlog,
188
)
189
except OSError as exc:
190
logger.error(exc)
191
await self.lifespan.shutdown()
192
sys.exit(STARTUP_FAILURE)
193
194
assert server.sockets is not None
195
listeners = server.sockets
196
self.servers = [server]
197
198
if sockets is None:
199
self._log_started_message(listeners)
200
else:
201
# We're most likely running multiple workers, so a message has already been
202
# logged by `config.bind_socket()`.
203
pass # pragma: full coverage
204
205
self.started = True
206
207
def _log_started_message(self, listeners: Sequence[socket.SocketType]) -> None:
208
config = self.config
209
210
if config.fd is not None: # pragma: py-win32
211
sock = listeners[0]
212
logger.info(
213
"Uvicorn running on socket %s (Press CTRL+C to quit)",
214
sock.getsockname(),
215
)
216
217
elif config.uds is not None: # pragma: py-win32
218
logger.info("Uvicorn running on unix socket %s (Press CTRL+C to quit)", config.uds)
219
220
else:
221
addr_format = "%s://%s:%d"
222
host = "0.0.0.0" if config.host is None else config.host
223
if ":" in host:
224
# It's an IPv6 address.
225
addr_format = "%s://[%s]:%d"
226
227
port = config.port
228
if port == 0:
229
port = listeners[0].getsockname()[1]
230
231
protocol_name = "https" if config.ssl else "http"
232
message = f"Uvicorn running on {addr_format} (Press CTRL+C to quit)"
233
color_message = "Uvicorn running on " + style(addr_format, bold=True) + " (Press CTRL+C to quit)"
234
logger.info(
235
message,
236
protocol_name,
237
host,
238
port,
239
extra={"color_message": color_message},
240
)
241
242
async def main_loop(self) -> None:
243
counter = 0
244
should_exit = await self.on_tick(counter)
245
while not should_exit:
246
counter += 1
247
counter = counter % 864000
248
await asyncio.sleep(0.1)
249
should_exit = await self.on_tick(counter)
250
251
async def on_tick(self, counter: int) -> bool:
252
# Update the default headers, once per second.
253
if counter % 10 == 0:
254
current_time = time.time()
255
current_date = formatdate(current_time, usegmt=True).encode()
256
257
if self.config.date_header:
258
date_header = [(b"date", current_date)]
259
else:
260
date_header = []
261
262
self.server_state.default_headers = date_header + self.config.encoded_headers
263
264
# Callback to `callback_notify` once every `timeout_notify` seconds.
265
if self.config.callback_notify is not None:
266
if current_time - self.last_notified > self.config.timeout_notify: # pragma: full coverage
267
self.last_notified = current_time
268
await self.config.callback_notify()
269
270
# Determine if we should exit.
271
if self.should_exit:
272
return True
273
274
max_requests = self.limit_max_requests
275
if max_requests is not None and self.server_state.total_requests >= max_requests:
276
logger.info("Maximum request limit of %d exceeded. Terminating process.", max_requests)
277
return True
278
279
return False
280
281
async def shutdown(self, sockets: list[socket.socket] | None = None) -> None:
282
logger.info("Shutting down")
283
284
# Stop accepting new connections.
285
for server in self.servers:
286
server.close()
287
for sock in sockets or []:
288
sock.close() # pragma: full coverage
289
290
# Request shutdown on all existing connections.
291
for connection in list(self.server_state.connections):
292
connection.shutdown()
293
await asyncio.sleep(0.1)
294
295
# When 3.10 is not supported anymore, use `async with asyncio.timeout(...):`.
296
try:
297
await asyncio.wait_for(
298
self._wait_tasks_to_complete(),
299
timeout=self.config.timeout_graceful_shutdown,
300
)
301
except asyncio.TimeoutError:
302
logger.error(
303
"Cancel %s running task(s), timeout graceful shutdown exceeded",
304
len(self.server_state.tasks),
305
)
306
for t in self.server_state.tasks:
307
t.cancel(msg="Task cancelled, timeout graceful shutdown exceeded")
308
309
# Send the lifespan shutdown event, and wait for application shutdown.
310
if not self.force_exit:
311
await self.lifespan.shutdown()
312
313
async def _wait_tasks_to_complete(self) -> None:
314
# Wait for existing connections to finish sending responses.
315
if self.server_state.connections and not self.force_exit:
316
msg = "Waiting for connections to close. (CTRL+C to force quit)"
317
logger.info(msg)
318
while self.server_state.connections and not self.force_exit:
319
await asyncio.sleep(0.1)
320
321
# Wait for existing tasks to complete.
322
if self.server_state.tasks and not self.force_exit:
323
msg = "Waiting for background tasks to complete. (CTRL+C to force quit)"
324
logger.info(msg)
325
while self.server_state.tasks and not self.force_exit:
326
await asyncio.sleep(0.1)
327
328
for server in self.servers:
329
await server.wait_closed()
330
331
@contextlib.contextmanager
332
def capture_signals(self) -> Generator[None, None, None]:
333
# Signals can only be listened to from the main thread.
334
if threading.current_thread() is not threading.main_thread():
335
yield
336
return
337
# always use signal.signal, even if loop.add_signal_handler is available
338
# this allows to restore previous signal handlers later on
339
original_handlers = {sig: signal.signal(sig, self.handle_exit) for sig in HANDLED_SIGNALS}
340
try:
341
yield
342
finally:
343
for sig, handler in original_handlers.items():
344
signal.signal(sig, handler)
345
# If we did gracefully shut down due to a signal, try to
346
# trigger the expected behaviour now; multiple signals would be
347
# done LIFO, see https://stackoverflow.com/questions/48434964
348
for captured_signal in reversed(self._captured_signals):
349
signal.raise_signal(captured_signal)
350
351
def handle_exit(self, sig: int, frame: FrameType | None) -> None:
352
self._captured_signals.append(sig)
353
if self.should_exit and sig == signal.SIGINT:
354
self.force_exit = True # pragma: full coverage
355
else:
356
self.should_exit = True