patx/veela
Fix worker startup without main guard and clean shutdown
Commit a7cc9d7 · patx · 2026-10-13T23:15:54-04:00
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index 48ffb60..118e1c3 100644
--- a/README.md
+++ b/README.md
@@ -34,8 +34,7 @@ app = App()
async def index(request):
return {"hello": "world"}
-if __name__ == '__main__':
- app.run()
+app.run()
```
```bash
@@ -147,14 +146,19 @@ your director's cut.
## Advanced enterprise configuration
```python
-if __name__ == '__main__':
- app.run(host='0.0.0.0', port=8080, workers=4)
+app.run(host='0.0.0.0', port=8080, workers=4)
```
Four workers. One file. Please allow the procurement department a moment.
-The default is `127.0.0.1:8000` with one worker. Multiple workers need a
-module-level app and the `if __name__ == '__main__':` guard shown above.
+The default is `127.0.0.1:8000` with one worker. No
+`if __name__ == '__main__':` guard is needed, including with multiple workers.
+An existing guard around `app.run()` is fine too. Multiple workers need an app
+defined at module level outside that guard, and a Python script or importable
+module (not a REPL or `python -c`). Start them from the main thread.
+Workers re-execute application setup and have separate memory.
+Ctrl+C and SIGTERM stop the server, join workers, and remove the temporary
+server archive before `run()` returns.
The bundled server uses asyncio and h11, with WebSockets and lifespan disabled.
`app.run()` also needs `uvicorn`, `h11`, and `click` not to be imported already.
The magic prefers to make its own introductions.
diff --git a/source/wred.py b/source/wred.py
index 4546a7e..e2507de 100644
--- a/source/wred.py
+++ b/source/wred.py
@@ -26,6 +26,26 @@ from urllib.parse import parse_qsl
import json
+_loading_worker_app = False
+
+
+def _load_worker_app(module, name):
+ """Load an app without starting another server during its import."""
+ from importlib import import_module
+
+ global _loading_worker_app
+ _loading_worker_app = True
+ try:
+ return getattr(import_module(module), name)
+ except Exception as error:
+ import logging
+ from uvicorn.config import STARTUP_FAILURE
+ logging.getLogger('uvicorn.error').error('Cannot load application %s:%s: %s', module, name, error)
+ raise SystemExit(STARTUP_FAILURE) from None
+ finally:
+ _loading_worker_app = False
+
+
class WredResponseError(Exception):
"""Raised when a handler returns a non-JSON response."""
pass
@@ -85,22 +105,39 @@ class App:
self.pattern_routes: dict[str, dict[str, Callable]] = {}
def run(self, host="127.0.0.1", port=8000, workers=1):
- """Run the bundled HTTP-only Uvicorn server; load it only on demand."""
+ """Serve HTTP until interrupted. Workers need a module-level app, no main guard."""
import base64
+ import multiprocessing
import signal
import sys
import tempfile
import threading
+ from functools import partial
- if not isinstance(workers, int) or workers < 1:
+ # Spawn re-executes the entry module before loading the worker target.
+ # Python exposes no public bootstrap-state check; this is the same flag
+ # multiprocessing uses to detect an unsafe recursive start.
+ if _loading_worker_app or getattr(multiprocessing.current_process(), '_inheriting', False):
+ return
+ if type(workers) is not int or workers < 1:
raise ValueError('workers must be a positive integer')
application = self
if workers > 1:
+ if threading.current_thread() is not threading.main_thread():
+ raise RuntimeError('Multiple workers must be started from the main thread')
namespace = sys._getframe(1).f_globals
name = next((name for name, value in namespace.items() if value is self), None)
if name is None:
raise ValueError('Multiple workers require a module-level application')
- application = f"{namespace['__name__']}:{name}"
+ module = namespace['__name__']
+ if module == '__main__':
+ spec = namespace.get('__spec__')
+ if spec is not None and spec.name.endswith('.__main__'):
+ # multiprocessing does not re-execute package __main__ modules.
+ module = spec.name
+ elif not Path(namespace.get('__file__', '')).is_file():
+ raise ValueError('Multiple workers require a Python script or importable module')
+ application = partial(_load_worker_app, module, name)
if any(name in sys.modules for name in ('uvicorn', 'h11', 'click')):
raise RuntimeError('App.run() requires uvicorn, h11, and click not to be imported already')
pending_signals = []
@@ -115,24 +152,43 @@ class App:
sys.path.insert(0, archive)
try:
import uvicorn
+ if pending_signals:
+ return
if workers > 1:
- from uvicorn.supervisors.multiprocess import SIGNALS
+ from uvicorn.supervisors.multiprocess import Multiprocess, SIGNALS
for sig in SIGNALS:
handlers.setdefault(sig, signal.getsignal(sig))
- uvicorn.run(application, host=host, port=port, http="h11", ws="none",
- lifespan="off", loop="asyncio", interface="asgi3",
- workers=workers, access_log=False)
+ config = uvicorn.Config(application, host=host, port=port, http="h11", ws="none",
+ lifespan="off", loop="asyncio", interface="asgi3",
+ workers=workers, access_log=False, factory=True)
+ with config.bind_socket() as sock:
+ supervisor = Multiprocess(config, sockets=[sock])
+ supervisor.signal_queue.extend(pending_signals)
+ try:
+ supervisor.run()
+ except BaseException:
+ # Uvicorn's supervisor does not clean up a partial start.
+ supervisor.terminate_all()
+ supervisor.join_all()
+ raise
+ else:
+ uvicorn.run(application, host=host, port=port, http="h11", ws="none",
+ lifespan="off", loop="asyncio", interface="asgi3",
+ workers=1, access_log=False)
finally:
sys.path.remove(archive)
for name, module in list(sys.modules.items()):
if (getattr(module, '__file__', None) or '').startswith((archive + '/', archive + '\\')):
del sys.modules[name]
- sys.path_importer_cache.pop(archive, None)
+ for path in list(sys.path_importer_cache):
+ if path == archive or path.startswith((archive + '/', archive + '\\')):
+ del sys.path_importer_cache[path]
+ except KeyboardInterrupt:
+ # Also cover Ctrl+C during imports and startup, before Uvicorn owns it.
+ pass
finally:
for sig, handler in handlers.items():
signal.signal(sig, handler)
- for sig in reversed(pending_signals):
- signal.raise_signal(sig)
async def __call__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]) -> None:
"""Handle ASGI requests."""
diff --git a/tests/test_embedded.py b/tests/test_embedded.py
index 3c2d452..572325d 100644
--- a/tests/test_embedded.py
+++ b/tests/test_embedded.py
@@ -1,12 +1,14 @@
"""Exercise the generated single file using an empty venv and real HTTP sockets."""
import http.client
import json
+import os
from pathlib import Path
import shutil
import signal
import socket
import subprocess
import tempfile
+import textwrap
import time
import unittest
import venv
@@ -45,12 +47,11 @@ async def modules(request):
@app.route('/pid')
async def pid(request):
return {'pid': os.getpid()}
-if __name__ == '__main__':
- if os.environ.get('WRED_DEFAULTS'):
- app.run()
- else:
- app.run(host='0.0.0.0', port=int(os.environ['WRED_TEST_PORT']),
- workers=int(os.environ.get('WRED_TEST_WORKERS', '1')))
+if os.environ.get('WRED_DEFAULTS'):
+ app.run()
+else:
+ app.run(host='0.0.0.0', port=int(os.environ['WRED_TEST_PORT']),
+ workers=int(os.environ.get('WRED_TEST_WORKERS', '1')))
'''
@@ -63,13 +64,20 @@ class StandaloneTests(unittest.TestCase):
cls.python = cls.directory / 'venv' / ('Scripts/python.exe' if __import__('os').name == 'nt' else 'bin/python')
shutil.copyfile(ROOT / 'wred.py', cls.directory / 'wred.py')
(cls.directory / 'app.py').write_text(APPLICATION)
+ setup, startup = APPLICATION.split("if os.environ.get('WRED_DEFAULTS'):", 1)
+ guarded = setup + "if __name__ == '__main__':\n" + textwrap.indent(
+ "if os.environ.get('WRED_DEFAULTS'):" + startup, ' ')
+ (cls.directory / 'guarded.py').write_text(guarded)
+ package = cls.directory / 'demo'
+ package.mkdir()
+ (package / '__init__.py').write_text('')
+ (package / '__main__.py').write_text(APPLICATION)
@classmethod
def tearDownClass(cls):
cls.temporary.cleanup()
- def start(self, default=False, workers=1):
- import os
+ def start(self, default=False, workers=1, command=('app.py',)):
if default:
port = 8000
else:
@@ -83,8 +91,9 @@ class StandaloneTests(unittest.TestCase):
environment.pop('WRED_DEFAULTS', None)
if default:
environment['WRED_DEFAULTS'] = '1'
- process = subprocess.Popen([str(self.python), 'app.py'], cwd=self.directory,
- env=environment, stdout=logfile, stderr=logfile)
+ process = subprocess.Popen([str(self.python), *command], cwd=self.directory,
+ env=environment, stdout=logfile, stderr=logfile,
+ start_new_session=os.name == 'posix')
def stop():
if process.poll() is None:
process.terminate()
@@ -109,7 +118,8 @@ class StandaloneTests(unittest.TestCase):
return process, port, logfile
except OSError:
time.sleep(0.02)
- self.fail('Server did not start within 15 seconds')
+ logfile.seek(0)
+ self.fail('Server did not start within 15 seconds:\n' + logfile.read()[-8000:])
def test_import_has_no_dependencies_and_does_not_load_or_decode_server(self):
program = '''import base64, importlib.metadata, sys, time
@@ -178,9 +188,10 @@ tempfile.tempdir = str(pathlib.Path('scratch').resolve())
pathlib.Path(tempfile.tempdir).mkdir()
original_path = sys.path.copy()
handlers = {sig: signal.getsignal(sig) for sig in (signal.SIGINT, signal.SIGTERM)}
-for attempt in range(2):
+app = App()
+for workers in (1, 2, 1, 2):
try:
- App().run(port=65536)
+ app.run(port=65536, workers=workers)
except OverflowError:
pass
else:
@@ -190,13 +201,14 @@ for attempt in range(2):
assert not any(n in sys.modules for n in ('uvicorn', 'h11', 'click'))
assert all(signal.getsignal(sig) == handler for sig, handler in handlers.items())
'''
- subprocess.run([str(self.python), '-c', program], cwd=self.directory,
+ (self.directory / 'startup_failure.py').write_text(program)
+ subprocess.run([str(self.python), 'startup_failure.py'], cwd=self.directory,
text=True, capture_output=True, check=True)
def test_invalid_worker_counts_and_unbound_apps_fail_before_server_loading(self):
program = '''from wred import App
import sys
-for workers in (0, -1, 1.5, None, '2'):
+for workers in (0, -1, 1.5, None, '2', True):
try:
App().run(workers=workers)
except ValueError:
@@ -219,10 +231,11 @@ assert not any(n in sys.modules for n in ('uvicorn', 'h11', 'click'))
self.shutdown(process, port, logfile, signal.SIGINT)
def test_multiple_workers_serve_requests_and_clean_up_on_both_signals(self):
- import os
- for sig in (signal.SIGINT, signal.SIGTERM):
- with self.subTest(signal=sig):
- process, port, logfile = self.start(workers=2)
+ cases = [(('app.py',), signal.SIGINT), (('-m', 'app'), signal.SIGTERM),
+ (('-m', 'demo'), signal.SIGINT), (('guarded.py',), signal.SIGTERM)]
+ for command, sig in cases:
+ with self.subTest(command=command, signal=sig):
+ process, port, logfile = self.start(workers=2, command=command)
pids = set()
deadline = time.monotonic() + 10
while len(pids) < 2 and time.monotonic() < deadline:
@@ -254,13 +267,19 @@ assert not any(n in sys.modules for n in ('uvicorn', 'h11', 'click'))
response = connection.getresponse()
archive = Path(json.loads(response.read())['archive'])
connection.close()
- process.send_signal(sig)
+ if sig == signal.SIGINT and os.name == 'posix':
+ os.killpg(process.pid, sig)
+ else:
+ process.send_signal(sig)
process.wait(timeout=10)
+ self.assertEqual(process.returncode, 0)
self.assertFalse(archive.parent.exists())
logfile.seek(0)
log = logfile.read()
self.assertIn('Finished server process', log)
self.assertNotIn('Waiting for application startup', log)
+ self.assertNotIn('Traceback', log)
+ self.assertNotIn('Exception in thread', log)
if __name__ == '__main__':
diff --git a/wred.py b/wred.py
index b42f7e2..745b620 100644
--- a/wred.py
+++ b/wred.py
@@ -26,6 +26,26 @@ from urllib.parse import parse_qsl
import json
+_loading_worker_app = False
+
+
+def _load_worker_app(module, name):
+ """Load an app without starting another server during its import."""
+ from importlib import import_module
+
+ global _loading_worker_app
+ _loading_worker_app = True
+ try:
+ return getattr(import_module(module), name)
+ except Exception as error:
+ import logging
+ from uvicorn.config import STARTUP_FAILURE
+ logging.getLogger('uvicorn.error').error('Cannot load application %s:%s: %s', module, name, error)
+ raise SystemExit(STARTUP_FAILURE) from None
+ finally:
+ _loading_worker_app = False
+
+
class WredResponseError(Exception):
"""Raised when a handler returns a non-JSON response."""
pass
@@ -85,22 +105,39 @@ class App:
self.pattern_routes: dict[str, dict[str, Callable]] = {}
def run(self, host="127.0.0.1", port=8000, workers=1):
- """Run the bundled HTTP-only Uvicorn server; load it only on demand."""
+ """Serve HTTP until interrupted. Workers need a module-level app, no main guard."""
import base64
+ import multiprocessing
import signal
import sys
import tempfile
import threading
-
- if not isinstance(workers, int) or workers < 1:
+ from functools import partial
+
+ # Spawn re-executes the entry module before loading the worker target.
+ # Python exposes no public bootstrap-state check; this is the same flag
+ # multiprocessing uses to detect an unsafe recursive start.
+ if _loading_worker_app or getattr(multiprocessing.current_process(), '_inheriting', False):
+ return
+ if type(workers) is not int or workers < 1:
raise ValueError('workers must be a positive integer')
application = self
if workers > 1:
+ if threading.current_thread() is not threading.main_thread():
+ raise RuntimeError('Multiple workers must be started from the main thread')
namespace = sys._getframe(1).f_globals
name = next((name for name, value in namespace.items() if value is self), None)
if name is None:
raise ValueError('Multiple workers require a module-level application')
- application = f"{namespace['__name__']}:{name}"
+ module = namespace['__name__']
+ if module == '__main__':
+ spec = namespace.get('__spec__')
+ if spec is not None and spec.name.endswith('.__main__'):
+ # multiprocessing does not re-execute package __main__ modules.
+ module = spec.name
+ elif not Path(namespace.get('__file__', '')).is_file():
+ raise ValueError('Multiple workers require a Python script or importable module')
+ application = partial(_load_worker_app, module, name)
if any(name in sys.modules for name in ('uvicorn', 'h11', 'click')):
raise RuntimeError('App.run() requires uvicorn, h11, and click not to be imported already')
pending_signals = []
@@ -115,24 +152,43 @@ class App:
sys.path.insert(0, archive)
try:
import uvicorn
+ if pending_signals:
+ return
if workers > 1:
- from uvicorn.supervisors.multiprocess import SIGNALS
+ from uvicorn.supervisors.multiprocess import Multiprocess, SIGNALS
for sig in SIGNALS:
handlers.setdefault(sig, signal.getsignal(sig))
- uvicorn.run(application, host=host, port=port, http="h11", ws="none",
- lifespan="off", loop="asyncio", interface="asgi3",
- workers=workers, access_log=False)
+ config = uvicorn.Config(application, host=host, port=port, http="h11", ws="none",
+ lifespan="off", loop="asyncio", interface="asgi3",
+ workers=workers, access_log=False, factory=True)
+ with config.bind_socket() as sock:
+ supervisor = Multiprocess(config, sockets=[sock])
+ supervisor.signal_queue.extend(pending_signals)
+ try:
+ supervisor.run()
+ except BaseException:
+ # Uvicorn's supervisor does not clean up a partial start.
+ supervisor.terminate_all()
+ supervisor.join_all()
+ raise
+ else:
+ uvicorn.run(application, host=host, port=port, http="h11", ws="none",
+ lifespan="off", loop="asyncio", interface="asgi3",
+ workers=1, access_log=False)
finally:
sys.path.remove(archive)
for name, module in list(sys.modules.items()):
if (getattr(module, '__file__', None) or '').startswith((archive + '/', archive + '\\')):
del sys.modules[name]
- sys.path_importer_cache.pop(archive, None)
+ for path in list(sys.path_importer_cache):
+ if path == archive or path.startswith((archive + '/', archive + '\\')):
+ del sys.path_importer_cache[path]
+ except KeyboardInterrupt:
+ # Also cover Ctrl+C during imports and startup, before Uvicorn owns it.
+ pass
finally:
for sig, handler in handlers.items():
signal.signal(sig, handler)
- for sig in reversed(pending_signals):
- signal.raise_signal(sig)
async def __call__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]) -> None:
"""Handle ASGI requests."""