Cleanup fixes, and-- cleanup.
This commit is contained in:
@@ -3,6 +3,7 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
import subprocess
|
import subprocess
|
||||||
from collections.abc import Coroutine
|
from collections.abc import Coroutine
|
||||||
|
from contextlib import suppress
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
@@ -58,14 +59,7 @@ class ProcessGroup:
|
|||||||
|
|
||||||
async def __aexit__(self, exc_type, *_):
|
async def __aexit__(self, exc_type, *_):
|
||||||
"""Wait for one process to exit, terminate others, then wait for all."""
|
"""Wait for one process to exit, terminate others, then wait for all."""
|
||||||
cleanup_task = asyncio.create_task(
|
await self._cleanup(immediate=exc_type is not None)
|
||||||
self._cleanup(immediate=exc_type is not None)
|
|
||||||
)
|
|
||||||
try:
|
|
||||||
await asyncio.shield(cleanup_task)
|
|
||||||
except asyncio.CancelledError:
|
|
||||||
# Shield was cancelled but cleanup_task continues - wait for it
|
|
||||||
await cleanup_task
|
|
||||||
|
|
||||||
async def _cleanup(self, immediate: bool = False):
|
async def _cleanup(self, immediate: bool = False):
|
||||||
running = [p for p in self._procs if p.returncode is None]
|
running = [p for p in self._procs if p.returncode is None]
|
||||||
@@ -74,51 +68,49 @@ class ProcessGroup:
|
|||||||
|
|
||||||
if not immediate:
|
if not immediate:
|
||||||
# Wait for any one process to exit
|
# Wait for any one process to exit
|
||||||
await asyncio.wait(
|
with suppress(asyncio.CancelledError):
|
||||||
[asyncio.create_task(p.wait()) for p in running],
|
await asyncio.wait(
|
||||||
return_when=asyncio.FIRST_COMPLETED,
|
[asyncio.create_task(p.wait()) for p in running],
|
||||||
)
|
return_when=asyncio.FIRST_COMPLETED,
|
||||||
|
)
|
||||||
|
|
||||||
# Terminate remaining processes
|
# Terminate remaining processes
|
||||||
for p in self._procs:
|
for p in self._procs:
|
||||||
if p.returncode is None:
|
if p.returncode is None:
|
||||||
try:
|
with suppress(ProcessLookupError):
|
||||||
p.terminate()
|
p.terminate()
|
||||||
except ProcessLookupError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
# Wait for all to finish (with overall timeout)
|
# Wait for all to finish (with overall timeout), shielded from cancellation
|
||||||
still_running = [p for p in self._procs if p.returncode is None]
|
still_running = [p for p in self._procs if p.returncode is None]
|
||||||
if still_running:
|
if still_running:
|
||||||
try:
|
with suppress(asyncio.CancelledError):
|
||||||
await asyncio.wait_for(
|
try:
|
||||||
asyncio.gather(*[p.wait() for p in still_running]),
|
await asyncio.shield(
|
||||||
timeout=10,
|
asyncio.wait_for(
|
||||||
)
|
asyncio.gather(*[p.wait() for p in still_running]),
|
||||||
except TimeoutError:
|
timeout=10,
|
||||||
for p in self._procs:
|
)
|
||||||
if p.returncode is None:
|
)
|
||||||
try:
|
except TimeoutError:
|
||||||
p.kill()
|
for p in self._procs:
|
||||||
except ProcessLookupError:
|
if p.returncode is None:
|
||||||
pass
|
with suppress(ProcessLookupError):
|
||||||
await p.wait()
|
p.kill()
|
||||||
|
await p.wait()
|
||||||
|
|
||||||
|
|
||||||
async def check_ports_free(*urls: str) -> None:
|
async def check_ports_free(*urls: str) -> None:
|
||||||
"""Verify URLs are not responding (ports are free). Raise SystemExit if any respond."""
|
"""Verify URLs are not responding (ports are free). Raise SystemExit if any respond."""
|
||||||
|
|
||||||
|
async def check(client: httpx.AsyncClient, url: str) -> None:
|
||||||
|
with suppress(httpx.RequestError):
|
||||||
|
res = await client.get(url, timeout=0.1)
|
||||||
|
server = res.headers.get("server", "server")
|
||||||
|
logger.warning("Conflicting %s already running at %s", server, url)
|
||||||
|
raise SystemExit(1)
|
||||||
|
|
||||||
async with httpx.AsyncClient() as client:
|
async with httpx.AsyncClient() as client:
|
||||||
for url in urls:
|
await asyncio.gather(*[check(client, url) for url in urls])
|
||||||
try:
|
|
||||||
res = await client.get(url, timeout=0.1)
|
|
||||||
logger.warning(
|
|
||||||
"Conflicting %s already running at %s",
|
|
||||||
res.headers.get("server", "server"),
|
|
||||||
url,
|
|
||||||
)
|
|
||||||
raise SystemExit(1)
|
|
||||||
except httpx.RequestError:
|
|
||||||
pass # Expected - port is free
|
|
||||||
|
|
||||||
|
|
||||||
async def ready(url: str, path: str = "") -> None:
|
async def ready(url: str, path: str = "") -> None:
|
||||||
|
|||||||
Reference in New Issue
Block a user