Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tornado/process.py: 21%
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#
2# Copyright 2011 Facebook
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16"""Utilities for working with multiple processes, including both forking
17the server into multiple processes and managing subprocesses.
18"""
20import asyncio
21import multiprocessing
22import os
23import signal
24import subprocess
25import sys
26import time
27from binascii import hexlify
28from collections.abc import Callable
29from typing import Any
31from tornado import ioloop
32from tornado.concurrent import (
33 Future,
34 future_set_exception_unless_cancelled,
35 future_set_result_unless_cancelled,
36)
37from tornado.iostream import PipeIOStream
38from tornado.log import gen_log
40# Re-export this exception for convenience.
41CalledProcessError = subprocess.CalledProcessError
44def cpu_count() -> int:
45 """Returns the number of processors on this machine."""
46 if multiprocessing is None:
47 return 1
48 try:
49 return multiprocessing.cpu_count()
50 except NotImplementedError:
51 pass
52 try:
53 return os.sysconf("SC_NPROCESSORS_CONF") # type: ignore
54 except (AttributeError, ValueError):
55 pass
56 gen_log.error("Could not detect number of processors; assuming 1")
57 return 1
60def _reseed_random() -> None:
61 if "random" not in sys.modules:
62 return
63 import random
65 # If os.urandom is available, this method does the same thing as
66 # random.seed (at least as of python 2.6). If os.urandom is not
67 # available, we mix in the pid in addition to a timestamp.
68 try:
69 seed = int(hexlify(os.urandom(16)), 16)
70 except NotImplementedError:
71 seed = int(time.time() * 1000) ^ os.getpid()
72 random.seed(seed)
75_task_id = None
78def fork_processes(num_processes: int | None, max_restarts: int | None = None) -> int:
79 """Starts multiple worker processes.
81 If ``num_processes`` is None or <= 0, we detect the number of cores
82 available on this machine and fork that number of child
83 processes. If ``num_processes`` is given and > 0, we fork that
84 specific number of sub-processes.
86 Since we use processes and not threads, there is no shared memory
87 between any server code.
89 Note that multiple processes are not compatible with the autoreload
90 module (or the ``autoreload=True`` option to `tornado.web.Application`
91 which defaults to True when ``debug=True``).
92 When using multiple processes, no IOLoops can be created or
93 referenced until after the call to ``fork_processes``.
95 In each child process, ``fork_processes`` returns its *task id*, a
96 number between 0 and ``num_processes``. Processes that exit
97 abnormally (due to a signal or non-zero exit status) are restarted
98 with the same id (up to ``max_restarts`` times). In the parent
99 process, ``fork_processes`` calls ``sys.exit(0)`` after all child
100 processes have exited normally.
102 max_restarts defaults to 100.
104 Availability: Unix
105 """
106 if sys.platform == "win32":
107 # The exact form of this condition matters to mypy; it understands
108 # if but not assert in this context.
109 raise Exception("fork not available on windows")
110 if max_restarts is None:
111 max_restarts = 100
113 assert _task_id is None
114 if num_processes is None or num_processes <= 0:
115 num_processes = cpu_count()
116 gen_log.info("Starting %d processes", num_processes)
117 children = {}
119 def start_child(i: int) -> int | None:
120 pid = os.fork()
121 if pid == 0:
122 # child process
123 _reseed_random()
124 global _task_id
125 _task_id = i
126 return i
127 else:
128 children[pid] = i
129 return None
131 for i in range(num_processes):
132 id = start_child(i)
133 if id is not None:
134 return id
135 num_restarts = 0
136 while children:
137 pid, status = os.wait()
138 if pid not in children:
139 continue
140 id = children.pop(pid)
141 if os.WIFSIGNALED(status):
142 gen_log.warning(
143 "child %d (pid %d) killed by signal %d, restarting",
144 id,
145 pid,
146 os.WTERMSIG(status),
147 )
148 elif os.WEXITSTATUS(status) != 0:
149 gen_log.warning(
150 "child %d (pid %d) exited with status %d, restarting",
151 id,
152 pid,
153 os.WEXITSTATUS(status),
154 )
155 else:
156 gen_log.info("child %d (pid %d) exited normally", id, pid)
157 continue
158 num_restarts += 1
159 if num_restarts > max_restarts:
160 raise RuntimeError("Too many child restarts, giving up")
161 new_id = start_child(id)
162 if new_id is not None:
163 return new_id
164 # All child processes exited cleanly, so exit the master process
165 # instead of just returning to right after the call to
166 # fork_processes (which will probably just start up another IOLoop
167 # unless the caller checks the return value).
168 sys.exit(0)
171def task_id() -> int | None:
172 """Returns the current task id, if any.
174 Returns None if this process was not created by `fork_processes`.
175 """
176 return _task_id
179class Subprocess:
180 """Wraps ``subprocess.Popen`` with IOStream support.
182 The constructor is the same as ``subprocess.Popen`` with the following
183 additions:
185 * ``stdin``, ``stdout``, and ``stderr`` may have the value
186 ``tornado.process.Subprocess.STREAM``, which will make the corresponding
187 attribute of the resulting Subprocess a `.PipeIOStream`. If this option
188 is used, the caller is responsible for closing the streams when done
189 with them.
191 The ``Subprocess.STREAM`` option and the ``set_exit_callback`` and
192 ``wait_for_exit`` methods do not work on Windows. There is
193 therefore no reason to use this class instead of
194 ``subprocess.Popen`` on that platform.
196 .. versionchanged:: 5.0
197 The ``io_loop`` argument (deprecated since version 4.1) has been removed.
199 """
201 STREAM = object()
203 _initialized = False
204 _waiting = {} # type: ignore
206 def __init__(self, *args: Any, **kwargs: Any) -> None:
207 self.io_loop = ioloop.IOLoop.current()
208 # All FDs we create should be closed on error; those in to_close
209 # should be closed in the parent process on success.
210 pipe_fds: list[int] = []
211 to_close: list[int] = []
212 if kwargs.get("stdin") is Subprocess.STREAM:
213 in_r, in_w = os.pipe()
214 kwargs["stdin"] = in_r
215 pipe_fds.extend((in_r, in_w))
216 to_close.append(in_r)
217 self.stdin = PipeIOStream(in_w)
218 if kwargs.get("stdout") is Subprocess.STREAM:
219 out_r, out_w = os.pipe()
220 kwargs["stdout"] = out_w
221 pipe_fds.extend((out_r, out_w))
222 to_close.append(out_w)
223 self.stdout = PipeIOStream(out_r)
224 if kwargs.get("stderr") is Subprocess.STREAM:
225 err_r, err_w = os.pipe()
226 kwargs["stderr"] = err_w
227 pipe_fds.extend((err_r, err_w))
228 to_close.append(err_w)
229 self.stderr = PipeIOStream(err_r)
230 try:
231 self.proc = subprocess.Popen(*args, **kwargs)
232 except:
233 for fd in pipe_fds:
234 os.close(fd)
235 raise
236 for fd in to_close:
237 os.close(fd)
238 self.pid = self.proc.pid
239 for attr in ["stdin", "stdout", "stderr"]:
240 if not hasattr(self, attr): # don't clobber streams set above
241 setattr(self, attr, getattr(self.proc, attr))
242 self._exit_callback: Callable[[int], None] | None = None
243 self.returncode: int | None = None
245 def set_exit_callback(self, callback: Callable[[int], None]) -> None:
246 """Runs ``callback`` when this process exits.
248 The callback takes one argument, the return code of the process.
250 This method uses a ``SIGCHLD`` handler, which is a global setting
251 and may conflict if you have other libraries trying to handle the
252 same signal. If you are using more than one ``IOLoop`` it may
253 be necessary to call `Subprocess.initialize` first to designate
254 one ``IOLoop`` to run the signal handlers.
256 In many cases a close callback on the stdout or stderr streams
257 can be used as an alternative to an exit callback if the
258 signal handler is causing a problem.
260 Availability: Unix
261 """
262 self._exit_callback = callback
263 Subprocess.initialize()
264 Subprocess._waiting[self.pid] = self
265 Subprocess._try_cleanup_process(self.pid)
267 def wait_for_exit(self, raise_error: bool = True) -> "Future[int]":
268 """Returns a `.Future` which resolves when the process exits.
270 Usage::
272 ret = yield proc.wait_for_exit()
274 This is a coroutine-friendly alternative to `set_exit_callback`
275 (and a replacement for the blocking `subprocess.Popen.wait`).
277 By default, raises `subprocess.CalledProcessError` if the process
278 has a non-zero exit status. Use ``wait_for_exit(raise_error=False)``
279 to suppress this behavior and return the exit status without raising.
281 .. versionadded:: 4.2
283 Availability: Unix
284 """
285 future: Future[int] = Future()
287 def callback(ret: int) -> None:
288 if ret != 0 and raise_error:
289 # Unfortunately we don't have the original args any more.
290 future_set_exception_unless_cancelled(
291 future, CalledProcessError(ret, "unknown")
292 )
293 else:
294 future_set_result_unless_cancelled(future, ret)
296 self.set_exit_callback(callback)
297 return future
299 @classmethod
300 def initialize(cls) -> None:
301 """Initializes the ``SIGCHLD`` handler.
303 The signal handler is run on an `.IOLoop` to avoid locking issues.
304 Note that the `.IOLoop` used for signal handling need not be the
305 same one used by individual Subprocess objects (as long as the
306 ``IOLoops`` are each running in separate threads).
308 .. versionchanged:: 5.0
309 The ``io_loop`` argument (deprecated since version 4.1) has been
310 removed.
312 Availability: Unix
313 """
314 if cls._initialized:
315 return
316 loop = asyncio.get_event_loop()
317 loop.add_signal_handler(signal.SIGCHLD, cls._cleanup)
318 cls._initialized = True
320 @classmethod
321 def uninitialize(cls) -> None:
322 """Removes the ``SIGCHLD`` handler."""
323 if not cls._initialized:
324 return
325 loop = asyncio.get_event_loop()
326 loop.remove_signal_handler(signal.SIGCHLD)
327 cls._initialized = False
329 @classmethod
330 def _cleanup(cls) -> None:
331 for pid in list(cls._waiting.keys()): # make a copy
332 cls._try_cleanup_process(pid)
334 @classmethod
335 def _try_cleanup_process(cls, pid: int) -> None:
336 try:
337 ret_pid, status = os.waitpid(pid, os.WNOHANG) # type: ignore
338 except ChildProcessError:
339 return
340 if ret_pid == 0:
341 return
342 assert ret_pid == pid
343 subproc = cls._waiting.pop(pid)
344 subproc.io_loop.add_callback(subproc._set_returncode, status)
346 def _set_returncode(self, status: int) -> None:
347 if sys.platform == "win32":
348 self.returncode = -1
349 else:
350 if os.WIFSIGNALED(status):
351 self.returncode = -os.WTERMSIG(status)
352 else:
353 assert os.WIFEXITED(status)
354 self.returncode = os.WEXITSTATUS(status)
355 # We've taken over wait() duty from the subprocess.Popen
356 # object. If we don't inform it of the process's return code,
357 # it will log a warning at destruction in python 3.6+.
358 self.proc.returncode = self.returncode
359 if self._exit_callback:
360 callback = self._exit_callback
361 self._exit_callback = None
362 callback(self.returncode)