Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiohttp/worker.py: 7%

1"""Async gunicorn worker for aiohttp.web""" 

2 

3import asyncio 

4import inspect 

5import os 

6import re 

7import signal 

8import sys 

9from types import FrameType 

10from typing import TYPE_CHECKING, Any, Optional 

11 

12from gunicorn.config import AccessLogFormat as GunicornAccessLogFormat 

13from gunicorn.workers import base 

14 

15from aiohttp import web 

16 

17from .helpers import set_result 

18from .web_app import Application 

19from .web_log import AccessLogger 

20 

21if TYPE_CHECKING: 

22 import ssl 

23 

24 SSLContext = ssl.SSLContext 

25else: 

26 try: 

27 import ssl 

28 

29 SSLContext = ssl.SSLContext 

30 except ImportError: # pragma: no cover 

31 ssl = None # type: ignore[assignment] 

32 SSLContext = object # type: ignore[misc,assignment] 

33 

34 

35__all__ = ("GunicornWebWorker", "GunicornUVLoopWebWorker") 

36 

37 

38class GunicornWebWorker(base.Worker): # type: ignore[misc,no-any-unimported] 

39 DEFAULT_AIOHTTP_LOG_FORMAT = AccessLogger.LOG_FORMAT 

40 DEFAULT_GUNICORN_LOG_FORMAT = GunicornAccessLogFormat.default 

41 

42 def __init__(self, *args: Any, **kw: Any) -> None: # pragma: no cover 

43 super().__init__(*args, **kw) 

44 

45 self._task: asyncio.Task[None] | None = None 

46 self.exit_code = 0 

47 self._notify_waiter: asyncio.Future[bool] | None = None 

48 

49 def init_process(self) -> None: 

50 # create new event_loop after fork 

51 try: 

52 asyncio.get_event_loop().close() 

53 except RuntimeError: 

54 # No loop was running 

55 pass 

56 

57 self.loop = asyncio.new_event_loop() 

58 asyncio.set_event_loop(self.loop) 

59 

60 super().init_process() 

61 

62 def run(self) -> None: 

63 # base.Worker.init_process() sets self.booted = True before 

64 # invoking run(), but for the aiohttp worker the real boot work 

65 # (factory call, runner setup, binding sockets) happens here. 

66 # Reset until _run() reaches the serve loop so that the arbiter 

67 # can tell a startup failure from a normal worker exit and 

68 # halt instead of endlessly respawning workers. 

69 self.booted = False 

70 

71 self._task = self.loop.create_task(self._run()) 

72 try: 

73 self.loop.run_until_complete(self._task) 

74 finally: 

75 self.loop.run_until_complete(self.loop.shutdown_asyncgens()) 

76 self.loop.close() 

77 

78 sys.exit(self.exit_code) 

79 

80 async def _run(self) -> None: 

81 runner = None 

82 if isinstance(self.wsgi, Application): 

83 app = self.wsgi 

84 elif inspect.iscoroutinefunction(self.wsgi) or ( 

85 sys.version_info < (3, 14) and asyncio.iscoroutinefunction(self.wsgi) 

86 ): 

87 wsgi = await self.wsgi() 

88 if isinstance(wsgi, web.AppRunner): 

89 runner = wsgi 

90 app = runner.app 

91 else: 

92 app = wsgi 

93 else: 

94 raise RuntimeError( 

95 "wsgi app should be either Application or " 

96 f"async function returning Application, got {self.wsgi}" 

97 ) 

98 

99 if runner is None: 

100 access_log = self.log.access_log if self.cfg.accesslog else None 

101 runner = web.AppRunner( 

102 app, 

103 logger=self.log, 

104 keepalive_timeout=self.cfg.keepalive, 

105 access_log=access_log, 

106 access_log_format=self._get_valid_log_format( 

107 self.cfg.access_log_format 

108 ), 

109 shutdown_timeout=self.cfg.graceful_timeout / 100 * 95, 

110 ) 

111 await runner.setup() 

112 

113 ctx = self._create_ssl_context(self.cfg) if self.cfg.is_ssl else None 

114 

115 runner = runner 

116 assert runner is not None 

117 server = runner.server 

118 assert server is not None 

119 for sock in self.sockets: 

120 site = web.SockSite( 

121 runner, 

122 sock, 

123 ssl_context=ctx, 

124 ) 

125 await site.start() 

126 

127 # Sockets are bound; tell the arbiter the worker is ready to 

128 # accept requests. Any failure before this point propagates out 

129 # of run() with self.booted=False so the arbiter exits with 

130 # WORKER_BOOT_ERROR instead of treating this as a clean exit. 

131 self.booted = True 

132 

133 # If our parent changed then we shut down. 

134 pid = os.getpid() 

135 try: 

136 while self.alive: # type: ignore[has-type] 

137 self.notify() 

138 

139 cnt = server.requests_count 

140 if self.max_requests and cnt > self.max_requests: 

141 self.alive = False 

142 self.log.info("Max requests, shutting down: %s", self) 

143 

144 elif pid == os.getpid() and self.ppid != os.getppid(): 

145 self.alive = False 

146 self.log.info("Parent changed, shutting down: %s", self) 

147 else: 

148 await self._wait_next_notify() 

149 except Exception: 

150 pass 

151 

152 await runner.cleanup() 

153 

154 def _wait_next_notify(self) -> "asyncio.Future[bool]": 

155 self._notify_waiter_done() 

156 

157 loop = self.loop 

158 assert loop is not None 

159 self._notify_waiter = waiter = loop.create_future() 

160 self.loop.call_later(1.0, self._notify_waiter_done, waiter) 

161 

162 return waiter 

163 

164 def _notify_waiter_done( 

165 self, waiter: Optional["asyncio.Future[bool]"] = None 

166 ) -> None: 

167 if waiter is None: 

168 waiter = self._notify_waiter 

169 if waiter is not None: 

170 set_result(waiter, True) 

171 

172 if waiter is self._notify_waiter: 

173 self._notify_waiter = None 

174 

175 def init_signals(self) -> None: 

176 # Set up signals through the event loop API. 

177 

178 self.loop.add_signal_handler( 

179 signal.SIGQUIT, self.handle_quit, signal.SIGQUIT, None 

180 ) 

181 

182 self.loop.add_signal_handler( 

183 signal.SIGTERM, self.handle_exit, signal.SIGTERM, None 

184 ) 

185 

186 self.loop.add_signal_handler( 

187 signal.SIGINT, self.handle_quit, signal.SIGINT, None 

188 ) 

189 

190 self.loop.add_signal_handler( 

191 signal.SIGWINCH, self.handle_winch, signal.SIGWINCH, None 

192 ) 

193 

194 self.loop.add_signal_handler( 

195 signal.SIGUSR1, self.handle_usr1, signal.SIGUSR1, None 

196 ) 

197 

198 self.loop.add_signal_handler( 

199 signal.SIGABRT, self.handle_abort, signal.SIGABRT, None 

200 ) 

201 

202 # Don't let SIGTERM and SIGUSR1 disturb active requests 

203 # by interrupting system calls 

204 signal.siginterrupt(signal.SIGTERM, False) 

205 signal.siginterrupt(signal.SIGUSR1, False) 

206 

207 # Reset SIGCHLD to default so Gunicorn doesn't swallow subprocess 

208 # return codes. Without this, workers inherit the master arbiter's 

209 # SIGCHLD handler, causing spurious "Worker exited" errors when 

210 # application code spawns subprocesses. 

211 signal.signal(signal.SIGCHLD, signal.SIG_DFL) 

212 

213 def handle_quit(self, sig: int, frame: FrameType | None) -> None: 

214 self.alive = False 

215 

216 # worker_int callback 

217 self.cfg.worker_int(self) 

218 

219 # wakeup closing process 

220 self._notify_waiter_done() 

221 

222 def handle_abort(self, sig: int, frame: FrameType | None) -> None: 

223 self.alive = False 

224 self.exit_code = 1 

225 self.cfg.worker_abort(self) 

226 sys.exit(1) 

227 

228 @staticmethod 

229 def _create_ssl_context(cfg: Any) -> "SSLContext": 

230 """Creates SSLContext instance for usage in asyncio.create_server. 

231 

232 See ssl.SSLSocket.__init__ for more details. 

233 """ 

234 if ssl is None: # pragma: no cover 

235 raise RuntimeError("SSL is not supported.") 

236 

237 ctx = ssl.SSLContext(cfg.ssl_version) 

238 ctx.load_cert_chain(cfg.certfile, cfg.keyfile) 

239 ctx.verify_mode = cfg.cert_reqs 

240 if cfg.ca_certs: 

241 ctx.load_verify_locations(cfg.ca_certs) 

242 if cfg.ciphers: 

243 ctx.set_ciphers(cfg.ciphers) 

244 return ctx 

245 

246 def _get_valid_log_format(self, source_format: str) -> str: 

247 if source_format == self.DEFAULT_GUNICORN_LOG_FORMAT: 

248 return self.DEFAULT_AIOHTTP_LOG_FORMAT 

249 elif re.search(r"%\([^\)]+\)", source_format): 

250 raise ValueError( 

251 "Gunicorn's style options in form of `%(name)s` are not " 

252 "supported for the log formatting. Please use aiohttp's " 

253 "format specification to configure access log formatting: " 

254 "http://docs.aiohttp.org/en/stable/logging.html" 

255 "#format-specification" 

256 ) 

257 else: 

258 return source_format 

259 

260 

261class GunicornUVLoopWebWorker(GunicornWebWorker): 

262 def init_process(self) -> None: 

263 import uvloop 

264 

265 # Close any existing event loop before setting a 

266 # new policy. 

267 try: 

268 asyncio.get_event_loop().close() 

269 except RuntimeError: 

270 # No loop was running 

271 pass 

272 

273 # Setup uvloop policy, so that every 

274 # asyncio.get_event_loop() will create an instance 

275 # of uvloop event loop. 

276 asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) 

277 

278 super().init_process()