Skip to content

feat: Add support for simultaneous TCP and UDS binding #805

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
IngmarStein wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat: Add support for simultaneous TCP and UDS binding #805

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions granian/_granian.pyi
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ class ASGIWorker:
cls,
worker_id: int,
sock: Any,
uds_sock: Any,
ipc: Any,
threads: int,
blocking_threads: int,
py_threads: int,
Expand All @@ -71,7 +73,7 @@ class ASGIWorker:
http1_opts: HTTP1Settings | None,
http2_opts: HTTP2Settings | None,
websockets_enabled: bool,
static_files: tuple[str, str, str | None, str | None] | None,
static_files: tuple[list[tuple[str, str]], str | None, str | None] | None,
ssl_enabled: bool,
ssl_cert: str | None,
ssl_key: str | None,
Expand All @@ -80,13 +82,16 @@ class ASGIWorker:
ssl_ca: str | None,
ssl_crl: list[str],
ssl_client_verify: bool,
metrics: Any,
) -> ASGIWorker: ...

class WSGIWorker:
def __new__(
cls,
worker_id: int,
sock: Any,
uds_sock: Any,
ipc: Any,
threads: int,
blocking_threads: int,
py_threads: int,
Expand All @@ -95,7 +100,7 @@ class WSGIWorker:
http_mode: str,
http1_opts: HTTP1Settings | None,
http2_opts: HTTP2Settings | None,
static_files: tuple[str, str, str | None, str | None] | None,
static_files: tuple[list[tuple[str, str]], str | None, str | None] | None,
ssl_enabled: bool,
ssl_cert: str | None,
ssl_key: str | None,
Expand All @@ -104,13 +109,16 @@ class WSGIWorker:
ssl_ca: str | None,
ssl_crl: list[str],
ssl_client_verify: bool,
metrics: Any,
) -> WSGIWorker: ...

class RSGIWorker:
def __new__(
cls,
worker_id: int,
sock: Any,
uds_sock: Any,
ipc: Any,
threads: int,
blocking_threads: int,
py_threads: int,
Expand All @@ -120,7 +128,7 @@ class RSGIWorker:
http1_opts: HTTP1Settings | None,
http2_opts: HTTP2Settings | None,
websockets_enabled: bool,
static_files: tuple[str, str, str | None, str | None] | None,
static_files: tuple[list[tuple[str, str]], str | None, str | None] | None,
ssl_enabled: bool,
ssl_cert: str | None,
ssl_key: str | None,
Expand All @@ -129,6 +137,7 @@ class RSGIWorker:
ssl_ca: str | None,
ssl_crl: list[str],
ssl_client_verify: bool,
metrics: Any,
) -> RSGIWorker: ...

class SocketHolder:
Expand Down
6 changes: 4 additions & 2 deletions granian/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,6 @@ def option(*param_decls: str, cls: type[click.Option] | None = None, **attrs: An
@click.argument('app', required=True)
@option(
'--host',
default='127.0.0.1',
help='Host address to bind to',
)
@option('--port', type=int, default=8000, help='Port to bind to.')
Expand Down Expand Up @@ -434,7 +433,7 @@ def option(*param_decls: str, cls: type[click.Option] | None = None, **attrs: An
@click.version_option(message='%(prog)s %(version)s')
def cli(
app: str,
host: str,
host: str | None,
port: int,
uds: pathlib.Path | None,
uds_permissions: int | None,
Expand Down Expand Up @@ -518,6 +517,9 @@ def cli(

patch_pypath(working_dir)

if not host and not uds:
host = '127.0.0.1'

server = Server(
app,
address=host,
Expand Down
25 changes: 17 additions & 8 deletions granian/server/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,8 @@
WT = TypeVar('WT')

WORKERS_METHODS = {
RuntimeModes.mt: {False: 'serve_mtr', True: 'serve_mtr_uds'},
RuntimeModes.st: {False: 'serve_str', True: 'serve_str_uds'},
RuntimeModes.mt: {False: 'serve_mtr', True: 'serve_mtr_uds', 'dual': 'serve_mtr_dual'},
RuntimeModes.st: {False: 'serve_str', True: 'serve_str_uds', 'dual': 'serve_str_dual'},
}


Expand Down Expand Up @@ -83,7 +83,7 @@ class AbstractServer(Generic[WT]):
def __init__(
self,
target: str,
address: str = '127.0.0.1',
address: str | None = '127.0.0.1',
port: int = 8000,
uds: Path | None = None,
uds_permissions: int | None = None,
Expand Down Expand Up @@ -220,6 +220,9 @@ def __init__(
self._ssp = None
self._shd = None
self._sfd = None
self._ssp_uds = None
self._shd_uds = None
self._sfd_uds = None
self._metrics = MetricsAggregator(self.workers)
self._metrics_exporter = MetricsExporter(self._metrics)
self.wrks: list[WT] = []
Expand Down Expand Up @@ -291,7 +294,11 @@ def build_ssl_context(

@property
def _bind_addr_fmt(self):
return f'unix:{self.bind_uds}' if self.bind_uds else f'{self.bind_addr}:{self.bind_port}'
if self.bind_uds and self.bind_addr:
return f'unix:{self.bind_uds} and {self.bind_addr}:{self.bind_port}'
if self.bind_uds:
return f'unix:{self.bind_uds}'
return f'{self.bind_addr}:{self.bind_port}'

@staticmethod
def _call_hooks(hooks):
Expand All @@ -312,11 +319,13 @@ def on_shutdown(self, hook: Callable[[], Any]) -> Callable[[], Any]:

def _init_shared_socket(self):
if self.bind_uds:
self._ssp = UnixSocketSpec(str(self.bind_uds), self.backlog, self.uds_permissions)
else:
self._ssp_uds = UnixSocketSpec(str(self.bind_uds), self.backlog, self.uds_permissions)
self._shd_uds = self._ssp_uds.build()
self._sfd_uds = self._shd_uds.get_fd()
if self.bind_addr:
self._ssp = SocketSpec(self.bind_addr, self.bind_port, self.backlog)
self._shd = self._ssp.build()
self._sfd = self._shd.get_fd()
self._shd = self._ssp.build()
self._sfd = self._shd.get_fd()

def signal_handler_interrupt(self, *args, **kwargs):
self.interrupt_signal = True
Expand Down
77 changes: 63 additions & 14 deletions granian/server/mp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,17 +78,25 @@ def wrapped(
load_env(env_files)

_ipc_handle = None
sock, _sso = sock
if sys.platform == 'win32':
sock = SocketHolder(_sso.fileno())
elif ipc:
sock_tcp, sock_uds = sock
_sock_tcp = _sock_uds = None

if sock_tcp[0]:
_sock_tcp, _sso = sock_tcp
if sys.platform == 'win32':
_sock_tcp = SocketHolder(_sso.fileno())

if sock_uds[0]:
_sock_uds, _ = sock_uds

if ipc:
_ipc_fd = os.dup(ipc.fileno())
os.set_blocking(_ipc_fd, False)
_ipc_handle = IPCSenderHandle(_ipc_fd)

loop = loops.get(loop_impl)
callback = callback_loader()
return target(worker_id, callback, sock, _ipc_handle, loop, *args, **kwargs)
return target(worker_id, callback, _sock_tcp, _sock_uds, _ipc_handle, loop, *args, **kwargs)

return wrapped

Expand Down Expand Up @@ -116,6 +124,7 @@ def _spawn_asgi_worker(
worker_id: int,
callback: Any,
sock: Any,
uds_sock: Any,
ipc: Any,
loop: Any,
runtime_mode: RuntimeModes,
Expand Down Expand Up @@ -143,6 +152,7 @@ def _spawn_asgi_worker(
worker = ASGIWorker(
worker_id,
sock,
uds_sock,
ipc,
runtime_threads,
runtime_blocking_threads,
Expand All @@ -157,7 +167,12 @@ def _spawn_asgi_worker(
*ssl_ctx,
metrics,
)
serve = getattr(worker, WORKERS_METHODS[runtime_mode][sock.is_uds()])
if sock and uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode]['dual'])
elif uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][True])
else:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][False])
scheduler = _new_cbscheduler(loop, wcallback, impl_asyncio=task_impl == TaskImpl.asyncio)
serve(scheduler, loop, shutdown_event)

Expand All @@ -167,6 +182,7 @@ def _spawn_asgi_lifespan_worker(
worker_id: int,
callback: Any,
sock: Any,
uds_sock: Any,
ipc: Any,
loop: Any,
runtime_mode: RuntimeModes,
Expand Down Expand Up @@ -202,6 +218,7 @@ def _spawn_asgi_lifespan_worker(
worker = ASGIWorker(
worker_id,
sock,
uds_sock,
ipc,
runtime_threads,
runtime_blocking_threads,
Expand All @@ -216,7 +233,12 @@ def _spawn_asgi_lifespan_worker(
*ssl_ctx,
metrics,
)
serve = getattr(worker, WORKERS_METHODS[runtime_mode][sock.is_uds()])
if sock and uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode]['dual'])
elif uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][True])
else:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][False])
scheduler = _new_cbscheduler(loop, wcallback, impl_asyncio=task_impl == TaskImpl.asyncio)
serve(scheduler, loop, shutdown_event)
loop.run_until_complete(lifespan_handler.shutdown())
Expand All @@ -227,6 +249,7 @@ def _spawn_rsgi_worker(
worker_id: int,
callback: Any,
sock: Any,
uds_sock: Any,
ipc: Any,
loop: Any,
runtime_mode: RuntimeModes,
Expand Down Expand Up @@ -256,6 +279,7 @@ def _spawn_rsgi_worker(
worker = RSGIWorker(
worker_id,
sock,
uds_sock,
ipc,
runtime_threads,
runtime_blocking_threads,
Expand All @@ -270,7 +294,12 @@ def _spawn_rsgi_worker(
*ssl_ctx,
metrics,
)
serve = getattr(worker, WORKERS_METHODS[runtime_mode][sock.is_uds()])
if sock and uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode]['dual'])
elif uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][True])
else:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][False])
scheduler = _new_cbscheduler(loop, wcallback, impl_asyncio=task_impl == TaskImpl.asyncio)
serve(scheduler, loop, shutdown_event)
callback_del(loop)
Expand All @@ -281,6 +310,7 @@ def _spawn_wsgi_worker(
worker_id: int,
callback: Any,
sock: Any,
uds_sock: Any,
ipc: Any,
loop: Any,
runtime_mode: RuntimeModes,
Expand Down Expand Up @@ -308,6 +338,7 @@ def _spawn_wsgi_worker(
worker = WSGIWorker(
worker_id,
sock,
uds_sock,
ipc,
runtime_threads,
runtime_blocking_threads,
Expand All @@ -321,22 +352,40 @@ def _spawn_wsgi_worker(
*ssl_ctx,
metrics,
)
serve = getattr(worker, WORKERS_METHODS[runtime_mode][sock.is_uds()])
if sock and uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode]['dual'])
elif uds_sock:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][True])
else:
serve = getattr(worker, WORKERS_METHODS[runtime_mode][False])
scheduler = _new_cbscheduler(loop, wcallback, impl_asyncio=task_impl == TaskImpl.asyncio)
serve(scheduler, loop, shutdown_event)

def _init_shared_socket(self):
super()._init_shared_socket()
sock = socket.socket(fileno=self._sfd)
sock.set_inheritable(True)
self._sso = sock
if self._sfd:
sock = socket.socket(fileno=self._sfd)
sock.set_inheritable(True)
self._sso = sock
else:
self._sso = None

if self._sfd_uds:
sock = socket.socket(fileno=self._sfd_uds)
sock.set_inheritable(True)
self._sso_uds = sock
else:
self._sso_uds = None

def _write_pidfile(self):
super()._write_pidfile()
self._rss_collector = ProcInfoCollector()

def _unlink_pidfile(self):
self._sso.detach()
if self._sso:
self._sso.detach()
if self._sso_uds:
self._sso_uds.detach()
super()._unlink_pidfile()

def _start_ipc(self):
Expand Down Expand Up @@ -407,7 +456,7 @@ def _spawn_worker(self, idx, target, callback_loader) -> WorkerProcess:
idx + 1,
self.process_name,
callback_loader,
(self._shd, self._sso),
((self._shd, self._sso), (self._shd_uds, self._sso_uds)),
# NOTE: given we use IPC only for metrics right now, let's share the pipe
# only if metrics collection is actually enabled.
self._ipc[idx][1] if self.metrics_enabled else None,
Expand Down
Loading