Fixes on server startup and shutdown with Sanic internals. Parallel shutdown tasks as non-blocking async tasks to avoid blocking the event loop and causing a warning of that.
This commit is contained in:
1 parent
c7ba0d5a04
commit
1dc0c4441a
4 files changed
+77
-11
No files matched your search
+10
-6
@@ -148,12 +148,16 @@ async def main_after_start(app):
|
||||
# Sanic sometimes fails to execute after_server_stop, so we do it before instead (potentially interrupting handlers)
|
||||
@app.before_server_stop
|
||||
async def main_stop(app):
|
||||
watching.stop(app)
|
||||
await onlyoffice.close_oo_client()
|
||||
await shutdown_preview_workers()
|
||||
app.ctx.threadexec.shutdown()
|
||||
app.ctx.zipexec.shutdown(cancel_futures=True)
|
||||
await sso.close_client()
|
||||
async with asyncio.TaskGroup() as tg:
|
||||
tg.create_task(asyncio.to_thread(watching.stop, app))
|
||||
tg.create_task(onlyoffice.close_oo_client())
|
||||
tg.create_task(shutdown_preview_workers())
|
||||
tg.create_task(sso.close_client())
|
||||
|
||||
async with asyncio.TaskGroup() as tg:
|
||||
tg.create_task(asyncio.to_thread(app.ctx.threadexec.shutdown))
|
||||
tg.create_task(asyncio.to_thread(app.ctx.zipexec.shutdown, cancel_futures=True))
|
||||
|
||||
logger.debug("Cista worker threads all finished")
|
||||
|
||||
|
||||
|
||||
+40
-1
@@ -1,6 +1,7 @@
|
||||
"""Custom access logging middleware for Sanic."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import unicodedata
|
||||
from ipaddress import IPv6Address
|
||||
@@ -9,6 +10,40 @@ from sanic.log import LOGGING_CONFIG_DEFAULTS
|
||||
|
||||
logger = logging.getLogger("cista.access")
|
||||
|
||||
|
||||
class ReentrantSafeStreamHandler(logging.StreamHandler):
|
||||
"""Stream handler that degrades gracefully on signal-time reentrant writes.
|
||||
|
||||
Python's buffered text streams are not reentrant. If a signal handler logs
|
||||
while another log write is in progress, StreamHandler.emit can raise:
|
||||
RuntimeError("reentrant call inside <_io.BufferedWriter ...>")
|
||||
|
||||
Instead of letting logging emit a long "--- Logging error ---" traceback,
|
||||
we fall back to a best-effort os.write to the same file descriptor.
|
||||
"""
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
msg = ""
|
||||
try:
|
||||
msg = self.format(record)
|
||||
stream = self.stream
|
||||
stream.write(msg + self.terminator)
|
||||
self.flush()
|
||||
except RuntimeError as exc:
|
||||
if "reentrant call inside" not in str(exc):
|
||||
self.handleError(record)
|
||||
return
|
||||
stream = self.stream
|
||||
fd = stream.fileno()
|
||||
encoding = getattr(stream, "encoding", None) or "utf-8"
|
||||
data = (msg + self.terminator).encode(encoding, errors="replace")
|
||||
os.write(fd, data)
|
||||
except RecursionError:
|
||||
raise
|
||||
except Exception:
|
||||
self.handleError(record)
|
||||
|
||||
|
||||
_RESET = "\033[0m"
|
||||
_STATUS_INFO = "\033[32m" # 1xx (green)
|
||||
_STATUS_OK = "\033[1;92m" # 2xx (bright green)
|
||||
@@ -236,7 +271,7 @@ def log_ws_close(
|
||||
|
||||
def configure_access_logging() -> None:
|
||||
"""Configure the cista.access logger to output to stderr."""
|
||||
handler = logging.StreamHandler(sys.stderr)
|
||||
handler = ReentrantSafeStreamHandler(sys.stderr)
|
||||
handler.setFormatter(logging.Formatter("%(message)s"))
|
||||
logger.addHandler(handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
@@ -271,6 +306,10 @@ def configure_main_logging() -> None:
|
||||
Patches LOGGING_CONFIG_DEFAULTS so the formatter survives every dictConfig
|
||||
call Sanic makes during serve_single() / serve().
|
||||
"""
|
||||
for handler_name in ("console", "error_console", "access_console"):
|
||||
LOGGING_CONFIG_DEFAULTS["handlers"][handler_name]["class"] = (
|
||||
"cista.sanic_logging.ReentrantSafeStreamHandler"
|
||||
)
|
||||
LOGGING_CONFIG_DEFAULTS["formatters"]["generic"] = {
|
||||
"class": "cista.sanic_logging._EmojiFormatter",
|
||||
}
|
||||
|
||||
+7
-1
@@ -4,11 +4,17 @@ from pathlib import Path
|
||||
|
||||
from fastapi_vue.hostutil import parse_endpoint
|
||||
from sanic import Sanic
|
||||
from sanic.worker.loader import AppLoader
|
||||
|
||||
from cista import config, server80
|
||||
from cista.app import app
|
||||
|
||||
|
||||
def load_app() -> Sanic:
|
||||
"""Return the app instance for spawned Sanic worker/reloader processes."""
|
||||
return app
|
||||
|
||||
|
||||
def run(*, dev=False):
|
||||
"""Run Sanic main process that spawns worker processes to serve HTTP requests."""
|
||||
_url, opts = parse_listen(config.config.listen)
|
||||
@@ -29,7 +35,7 @@ def run(*, dev=False):
|
||||
access_log=False,
|
||||
) # type: ignore[call-arg]
|
||||
if dev:
|
||||
Sanic.serve()
|
||||
Sanic.serve(app_loader=AppLoader(factory=load_app))
|
||||
else:
|
||||
Sanic.serve_single()
|
||||
|
||||
|
||||
+20
-3
@@ -55,6 +55,10 @@ class FormatUpdateLoopError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class _WatcherStoppingError(Exception):
|
||||
"""Internal control-flow exception for quick watcher shutdown."""
|
||||
|
||||
|
||||
class State:
|
||||
def __init__(self):
|
||||
self.lock = threading.RLock()
|
||||
@@ -215,7 +219,7 @@ def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]
|
||||
li = []
|
||||
for f in path.iterdir():
|
||||
if stop_event.is_set():
|
||||
raise SystemExit("quit")
|
||||
raise _WatcherStoppingError
|
||||
if f.name.startswith("."):
|
||||
continue # No dotfiles
|
||||
with suppress(FileNotFoundError):
|
||||
@@ -227,7 +231,11 @@ def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]
|
||||
li.append((int(isfile), f.name, s))
|
||||
# Build the tree as a list of FileEntries
|
||||
for [_, name, s] in humansorted(li):
|
||||
if stop_event.is_set():
|
||||
raise _WatcherStoppingError
|
||||
sub = walk(rel / name, stat=s)
|
||||
if not sub:
|
||||
continue
|
||||
child = sub[0]
|
||||
entry = FileEntry(
|
||||
level=entry.level,
|
||||
@@ -679,7 +687,10 @@ def watcher(loop):
|
||||
inotify_tree = inotify.adapters.InotifyTree(rootpath.as_posix())
|
||||
|
||||
# Initialize the tree from filesystem
|
||||
update_root(loop)
|
||||
try:
|
||||
update_root(loop)
|
||||
except _WatcherStoppingError:
|
||||
return
|
||||
path_index = PathIndex(state.root[:])
|
||||
|
||||
trefresh = time.monotonic() + 300.0
|
||||
@@ -763,7 +774,10 @@ def watcher(loop):
|
||||
# Process each collapsed path
|
||||
new_root = path_index.root
|
||||
for path in collapsed:
|
||||
new_entries = walk(path)
|
||||
try:
|
||||
new_entries = walk(path)
|
||||
except _WatcherStoppingError:
|
||||
return
|
||||
new_root = path_index.apply_update(path, new_entries)
|
||||
|
||||
# Broadcast if changed
|
||||
@@ -782,6 +796,8 @@ def watcher(loop):
|
||||
with state.lock:
|
||||
broadcast(update_msg, loop)
|
||||
state.root = fresh
|
||||
except _WatcherStoppingError:
|
||||
return
|
||||
except Exception:
|
||||
logger.exception("Fallback failed; sending full root")
|
||||
with state.lock:
|
||||
@@ -870,6 +886,7 @@ def start(app):
|
||||
global rootpath
|
||||
config.load_config()
|
||||
rootpath = config.config.path
|
||||
stop_event.clear()
|
||||
app.ctx.watcher = threading.Thread(
|
||||
target=watcher,
|
||||
args=[app.loop],
|
||||
|
||||
Reference in new issue
Block a user