Alternative — Unix Datagram Servers

Note

This page uses two different API variants:

• Synchronous API with classic def functions, usable in any context.

• Asynchronous API with async def functions, using an asynchronous framework to perform I/O operations.

All asynchronous API examples assume that you are using either asyncio or trio, but you can use a different library thanks to the asynchronous backend engine API.

---

Introduction

Creating a Unix datagram server requires several steps:

1. Derive a class from AsyncDatagramRequestHandler and redefine its handle() method; this method will process incoming requests.

2. Instantiate the AsyncUnixDatagramServer class passing it the server’s address, the protocol object and the request handler instance.

3. Call serve_forever() to process requests.

Writing coroutine functions is mandatory to use this server.

See also

PEP 492 — Coroutines with async and await syntax

The proposal to introduce native coroutines in Python with async and await syntax.

asyncio — Asynchronous I/O

If you are not familiar with async/await syntax, you can use the standard library to get started with coroutines.

Request Handler Objects

Note

Unlike socketserver.BaseRequestHandler, there is only one AsyncDatagramRequestHandler instance for the entire service.

Here is a simple example:

from __future__ import annotations

from collections.abc import AsyncGenerator

from easynetwork.servers.handlers import AsyncDatagramClient, AsyncDatagramRequestHandler


class Request:
    """Object representing the client request."""

    ...


class Response:
    """Object representing the response to send to the client."""

    ...


class MyRequestHandler(AsyncDatagramRequestHandler[Request, Response]):
    """
    The request handler class for our server.

    It is instantiated once to the server, and must
    override the handle() method to implement communication to the
    client.
    """

    async def handle(
        self,
        client: AsyncDatagramClient[Response],
    ) -> AsyncGenerator[None, Request]:
        # "client" a placeholder to have a stream-like API.
        # All the datagrams sent by this client are sent
        # through the "yield" statement.
        request: Request = yield

        # Do some stuff
        ...

        response = Response()

        # The corresponding call is server_socket.sendto(data, remote_address)
        await client.send_packet(response)

Using handle() Generator

Important

There will always be only one active generator per client. All the pending datagrams received while the generator is running are queued.

This behavior is designed to act like a stream request handler.

Minimum Requirements

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    ### Before 'yield'
    # Initializes the generator.
    # This is the setup part before receiving a request.
    # Unlike the stream request handler, the generator is started
    # when the datagram is received (but is not parsed yet).
    ##################

    request: Request = yield

    ### After 'yield'
    # The received datagram is parsed.
    # you can do whatever you want with it and send responses back
    # to the client if necessary.
    await client.send_packet(Response())
    #################

    ### On a 'return'
    # When handle() returns, it means that this request handler is finished.
    # The server creates a new generator when a new datagram is received.
    #################
    return

Refuse datagrams

Your UDP socket can receive datagrams from anyone with permission to send them. You may want to control who can send you information.

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    if not self.should_handle(client):
        # By returning before the "yield" statement, you ask the server to discard
        # the received datagram.
        return

    request: Request = yield

Error Handling

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    try:
        # *All* exceptions are thrown through the "yield" statement
        # (including BaseException). But you should only catch Exception subclasses.
        request: Request = yield
    except DatagramProtocolParseError:
        await client.send_packet(BadRequest())
    except Exception:
        # Runtime error. Log the error.
        traceback.print_exc()

        await client.send_packet(InternalError())
    else:
        await client.send_packet(Response())

Warning

You should always log or re-raise a bare Exception thrown in your generator.

except Exception:
    # Runtime error. Log the error.
    traceback.print_exc()

    await client.send_packet(InternalError())

Having Multiple yield Statements

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    request: Request = yield

    ...

    await client.send_packet(Response())

    if self.need_something_else(request, client):
        additional_data: Request = yield

        ...

        await client.send_packet(Response())

Warning

Even if this feature is supported, it is not recommended to have more than one (unless you know what you are doing) for the following reasons:

• UDP does not guarantee ordered delivery. Packets are typically “sent” in order, but they may be received out of order. In large networks, it is reasonably common for some packets to arrive out of sequence (or not at all).

• The server has no way of knowing if this client has stopped sending you requests forever.

If you plan to use multiple yields in your request handler, you should always have a timeout applied. (See the section below.)

Cancellation And Timeouts

Using yield (Recommended)Using with

It is possible to send the timeout delay to the parent task:

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[RecvParams | None, Request]:
    # It is *never* useful to have a timeout for the 1st datagram
    # because the datagram is already in the queue.
    # The yielded value is simply ignored.
    request: Request = yield None

    ...

    await client.send_packet(Response())

    try:
        # The client has 30 seconds to send the 2nd request to the server.
        another_request: Request = yield RecvParams(timeout=30.0)
    except TimeoutError:
        await client.send_packet(TimedOut())
    else:
        await client.send_packet(Response())

Deprecated since version 1.2: Yielding a timeout without using RecvParams. Will be removed in 2.0.

try:
    # The client has 30 seconds to send the 2nd request to the server.
    another_request: Request = yield 30
except TimeoutError:
    await client.send_packet(TimedOut())
else:
    await client.send_packet(Response())

Sending Packets with socket control messages

By using SocketAncillary, you can send SCM data. See the Unix manual page sendmsg(2) for details.

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    request: Request = yield

    ancillary = SocketAncillary()
    ancillary.add_fds([4])
    await client.send_packet_with_ancillary(Response(), ancillary)

Receiving Packets with socket control messages

By using RecvAncillaryDataParams and SocketAncillary, you can receive SCM data. See the Unix manual page recvmsg(2) for details.

Warning

You must enable the feature in the server configuration to make this work.

server = AsyncUnixDatagramServer(
    "/var/run/app.sock",
    DatagramProtocol(JSONSerializer()),
    SCMRecvRequestHandler(),
    receive_ancillary_data=True,
)

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[RecvParams | None, Request]:
    ancillary = SocketAncillary()
    request: Request = yield RecvParams(recv_with_ancillary=RecvAncillaryDataParams(ancillary.update_from_raw))

    for message in ancillary.messages():
        match message:
            case SCMRights(fds):
                for fd in fds:
                    print(f"Received file descriptor: {fd}")
            case SCMCredentials(credentials):
                for ucred in credentials:
                    print(f"Received unix credential: {ucred}")

Tip

The default buffer size for this operation is approximately 8 KiB. However, you can customize this behavior.

from socket import CMSG_LEN

max_fds = 128
server = AsyncUnixDatagramServer(
    "/var/run/app.sock",
    DatagramProtocol(JSONSerializer()),
    SCMRecvRequestHandler(),
    receive_ancillary_data=True,
    ancillary_bufsize=CMSG_LEN(max_fds * 4),
)

Client Metadata

The client’s metadata are available via UNIXClientAttribute:

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    client_address = client.extra(UNIXClientAttribute.peer_name)

    request: Request = yield

    print(f"{client_address} sent {request}")

    await client.send_packet(Response())

Service Initialization

The server will call service_init() and pass it an AsyncExitStack at the beginning of the serve_forever() task to set up the global service.

This allows you to do something like this:

Using asyncioUsing trioUsing the AsyncBackend API

async def service_init(
    self,
    exit_stack: contextlib.AsyncExitStack,
    server: AsyncUnixDatagramServer[Request, Response],
) -> None:
    exit_stack.callback(self._service_quit)

    self.background_tasks = await exit_stack.enter_async_context(asyncio.TaskGroup())

    _ = self.background_tasks.create_task(self._service_actions())

async def _service_actions(self) -> None:
    while True:
        await asyncio.sleep(1)

        # Do some stuff each second in background
        ...

def _service_quit(self) -> None:
    print("Service stopped")

Low-Level Socket Operations

For low-level operations such as setsockopt(), the server object exposes the sockets through a SocketProxy:

async def service_init(
    self,
    exit_stack: contextlib.AsyncExitStack,
    server: AsyncUnixDatagramServer[Request, Response],
) -> None:
    for sock in server.get_sockets():
        # Enable SO_PASSCRED in order to use SCMCredentials
        sock.setsockopt(SOL_SOCKET, SO_PASSCRED, 1)

Per-client variables (contextvars integration)

If your asynchronous framework supports per-task context variables, you can use this feature in your request handler:

class ClientContextRequestHandler(AsyncDatagramRequestHandler[Request, Response]):
    client_addr_var: ClassVar[contextvars.ContextVar[UnixSocketAddress]]
    client_addr_var = contextvars.ContextVar("client_addr")

    @classmethod
    def client_log(cls, message: str) -> None:
        # The address of the currently handled client can be accessed
        # without passing it explicitly to this function.

        logger = logging.getLogger(cls.__name__)

        client_address = cls.client_addr_var.get()

        logger.info("From %s: %s", client_address, message)

    async def handle(
        self,
        client: AsyncDatagramClient[Response],
    ) -> AsyncGenerator[None, Request]:
        address = client.extra(UNIXClientAttribute.peer_name)
        self.client_addr_var.set(address)

        # In any code that we call within "handle()" is now possible to get
        # client's address by calling 'client_addr_var.get()'.

        request: Request = yield

        self.client_log(f"Received request: {request!r}")

        await client.send_packet(Response())

Tip

It is possible to initialize the context to be copied in service_init().

This means that the contextvars.ContextVar.set() calls made in service_init() will be applied to subsequent client tasks.

Server Object

A basic example of how to run the server:

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator

from easynetwork.protocol import DatagramProtocol
from easynetwork.servers.async_unix_datagram import AsyncUnixDatagramServer
from easynetwork.servers.handlers import AsyncDatagramClient, AsyncDatagramRequestHandler


class Request: ...


class Response: ...


class MyRequestHandler(AsyncDatagramRequestHandler[Request, Response]):
    async def handle(
        self,
        client: AsyncDatagramClient[Response],
    ) -> AsyncGenerator[None, Request]:
        request: Request = yield

        ...

        await client.send_packet(Response())


# NOTE: The sent packet is "Response" and the received packet is "Request"
class ServerProtocol(DatagramProtocol[Response, Request]):
    def __init__(self) -> None: ...


async def main() -> None:
    path = "/var/run/app/app.sock"
    protocol = ServerProtocol()
    handler = MyRequestHandler()

    # Create the server, binding to /var/run/app/app.sock
    async with AsyncUnixDatagramServer(path, protocol, handler) as server:
        # Activate the server; this will keep running until you
        # interrupt the program with Ctrl-C
        await server.serve_forever()


if __name__ == "__main__":
    asyncio.run(main())