Kernel

kernel@openeuler.org

20 participants
25390 discussions

[PATCH openEuler-26.09] xsched/famem: introduce lightweight NPU HBM management infrastructure FaMem
by Gu Bowen 12 Nov '26

12 Nov '26

hulk inclusion category: feature bugzilla: https://atomgit.com/openeuler/kernel/issues/10313 -------------------------------- FaMem (Fast-Map Memory) provides NPU HBM management for multi-model time-division multiplexing. It establishes a stable virtual address Arena per NPU to hold model weights and KV Cache; a native HBM server owns the physical pages and manages cross-model leases. Suspend releases physical HBM while preserving virtual addresses; resume uses a dedicated Copier process to restore weights layer-by-layer (H2D), overlapping weight loading with inference computation to reduce model switching latency. Signed-off-by: Alekseev Dmitry <alekseev.dmitry(a)huawei.com> Signed-off-by: Meskhidze Konstantin <konstantin.meskhidze(a)huawei.com> Signed-off-by: Mingrui Liu <liumingrui(a)huawei.com> Signed-off-by: Gu Bowen <gubowen5(a)huawei.com> --- arch/arm64/configs/openeuler_defconfig | 1 + arch/x86/configs/openeuler_defconfig | 1 + kernel/xsched/Kconfig | 18 + kernel/xsched/Makefile | 2 + kernel/xsched/famem_hbm_server.c | 1457 ++++++++++++++++++++++++ 5 files changed, 1479 insertions(+) create mode 100644 kernel/xsched/famem_hbm_server.c diff --git a/arch/arm64/configs/openeuler_defconfig b/arch/arm64/configs/openeuler_defconfig index 476efcfb46fe..b223548f3408 100644 --- a/arch/arm64/configs/openeuler_defconfig +++ b/arch/arm64/configs/openeuler_defconfig @@ -193,6 +193,7 @@ CONFIG_QOS_SCHED_SMART_GRID=y CONFIG_CGROUP_PIDS=y CONFIG_CGROUP_RDMA=y # CONFIG_CGROUP_DMEM is not set +# CONFIG_FAMEM is not set CONFIG_CGROUP_FREEZER=y CONFIG_CGROUP_HUGETLB=y CONFIG_CPUSETS=y diff --git a/arch/x86/configs/openeuler_defconfig b/arch/x86/configs/openeuler_defconfig index d4d392b59ab7..6ca79286fea8 100644 --- a/arch/x86/configs/openeuler_defconfig +++ b/arch/x86/configs/openeuler_defconfig @@ -210,6 +210,7 @@ CONFIG_QOS_SCHED_DYNAMIC_AFFINITY=y CONFIG_CGROUP_PIDS=y CONFIG_CGROUP_RDMA=y # CONFIG_CGROUP_DMEM is not set +# CONFIG_FAMEM is not set CONFIG_CGROUP_FREEZER=y CONFIG_CGROUP_HUGETLB=y CONFIG_CPUSETS=y diff --git a/kernel/xsched/Kconfig b/kernel/xsched/Kconfig index df6576da88fd..805b04de68f8 100644 --- a/kernel/xsched/Kconfig +++ b/kernel/xsched/Kconfig @@ -91,3 +91,21 @@ config CGROUP_XCU Required for systems requiring fine-grained resource control in cgroups. If unsure, say N. + +config FAMEM + bool "Enable FaMem (Fast-Map Memory) NPU HBM management" + default n + help + This option enables the FaMem (Fast-Map Memory) feature for managing + NPU HBM (High Bandwidth Memory) across multiple model instances. + + FaMem establishes a stable virtual address Arena per NPU to hold model + weights and KV Cache. A native HBM server owns the physical pages and + manages cross-model leases. Suspend releases physical HBM while + preserving virtual addresses; resume uses a dedicated Copier process to + restore weights layer-by-layer (H2D), overlapping weight loading with + inference computation to significantly reduce model-switching latency + and enable time-division multiplexing of HBM across model instances. + + This code is userspace-oriented and is isolated from the kernel build + by this config option. If unsure, say N. diff --git a/kernel/xsched/Makefile b/kernel/xsched/Makefile index 3e23012ea298..af788b279ed1 100644 --- a/kernel/xsched/Makefile +++ b/kernel/xsched/Makefile @@ -8,3 +8,5 @@ obj-$(CONFIG_XCU_SCHED_CFS) += cfs.o cfs_quota.o obj-$(CONFIG_CGROUP_XCU) += cgroup.o obj-$(CONFIG_CGROUP_DMEM) += dmem.o endif + +obj- += famem_hbm_server.o diff --git a/kernel/xsched/famem_hbm_server.c b/kernel/xsched/famem_hbm_server.c new file mode 100644 index 000000000000..fef2d86e5f96 --- /dev/null +++ b/kernel/xsched/famem_hbm_server.c @@ -0,0 +1,1457 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * SPDX-FileCopyrightText: Copyright (c) 2026 Huawei Technologies Co., Ltd. All Rights Reserved. + * + * famem_hbm_server.c — Native Famem HBM server for one Ascend NPU. + * + * This binary owns the physical HBM extents of a single device and serves + * the Famem control protocol (famem_protocol.h) on a Unix stream socket + * named after the device UUID. Clients (vLLM Worker processes and, for + * handle authorization, their Copier subprocesses) connect, claim a lease + * and drive the sleep/wake cycle; see FamemClient in + * vllm_ascend/device_allocator/famem_client.py for the client side. + * + * Server model + * ------------ + * - One thread per client connection; a mutex (server_state.mutex) + * serializes all state transitions. + * - One pool of at most FAMEM_MAX_EXTENTS physical extents, sized either + * by --pool-size at startup or by the first ACQUIRE request (pool_size + * == 0 defers sizing). Later ACQUIRE requests must request the exact + * same size or they are rejected — multi-instance time-sharing + * therefore requires identical arena sizes across instances. + * - At most one ACTIVE lease per NPU ("active" = extents currently + * allocated and mapped). Other connected leases may exist in the + * sleeping state; ACQUIRE/WAKE from a second lease gets + * FAMEM_STATUS_BUSY until the active lease SLEEPs or RELEASEs. + * + * Lease lifecycle + * --------------- + * HELLO create lease; record Worker + Copier identities (pidfd or + * pid+start_time) for liveness checking + * ACQUIRE allocate pool extents; authorize shareable handles for the + * Worker and Copier TGIDs; lease becomes active + * SLEEP free the physical extents (clients must have unmapped first); + * lease stays; epoch unchanged on the wire (client keeps it) + * WAKE re-allocate the same extent layout; epoch += 1; fresh + * shareable handles are exported and authorized + * RELEASE destroy the lease (frees extents if still active) + * + * Orphan reclamation + * ------------------ + * If a client disconnects while its lease is still active, the server + * blocks in the connection thread until the Worker and Copier processes + * exit (pidfd poll, or pid liveness + start_time comparison when pidfd + * is unavailable), then frees the extents. This guarantees CANN + * physical memory is never freed while a mapping in another process + * still references it. + * + * Poisoning + * --------- + * Any failure to free or allocate physical extents marks the server + * poisoned: it refuses new leases (FAMEM_STATUS_INTERNAL) until + * restarted, because the physical pool state is no longer trustworthy. + * + * Flags: --device (required), --socket-dir, --socket-name, --pool-size, + * --log-level, --skip-tgid-check (containers without SO_PEERCRED). + */ + +#define _GNU_SOURCE + +#include <errno.h> +#include <fcntl.h> +#include <getopt.h> +#include <limits.h> +#include <poll.h> +#include <pthread.h> +#include <signal.h> +#include <stdbool.h> +#include <stdarg.h> +#include <stddef.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/file.h> +#include <sys/prctl.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/syscall.h> +#include <sys/types.h> +#include <sys/un.h> +#include <time.h> +#include <unistd.h> + +#define FAMEM_DEFAULT_SOCKET_DIR "/run/vllm-ascend/famem" +#define FAMEM_SOCKET_PATH_MAX 103U +#define FAMEM_UUID_SIZE 32U +#define FAMEM_SESSION_SIZE 32U +#define FAMEM_MAX_EXTENTS 2U +#define FAMEM_PAGE_HUGE_1G 1 +#define FAMEM_PAGE_HUGE_2M 2 +#define FAMEM_HUGE_1G_BYTES (UINT64_C(1) << 30) +#define FAMEM_HUGE_2M_BYTES (UINT64_C(2) << 20) +#define FAMEM_MAX_POOL_GIB UINT64_C(4096) + +enum log_level { + LOG_DEBUG = 0, + LOG_INFO = 1, + LOG_WARNING = 2, + LOG_ERROR = 3, +}; + +struct request_frame { + uint16_t operation; + uint32_t request_id; + uint32_t payload_size; + uint8_t payload[FAMEM_PROTOCOL_MAX_PAYLOAD]; +}; +struct physical_extent { + int page_type; + uint64_t size; + uint64_t physical_handle; + uint64_t shareable_handle; +}; +struct process_identity { + pid_t pid; + int pidfd; + uint64_t start_time; + bool has_start_time; +}; +struct lease { + bool acquired; + bool active; + char session_id[FAMEM_SESSION_SIZE + 1U]; + struct process_identity worker_process; + struct process_identity copier_process; + int32_t bare_tgid; + int32_t copier_bare_tgid; + uint64_t epoch; +}; +struct server_state; +struct client_context { + int socket_fd; + pthread_t thread; + bool thread_finished; + struct lease lease; + struct server_state *server; + struct client_context *next; +}; +struct server_state { + int device; + char device_uuid[FAMEM_UUID_SIZE + 1U]; + pthread_mutex_t mutex; + struct client_context *clients; + struct lease *active_lease; + uint64_t pool_size; + uint64_t epoch; + size_t pool_extent_count; + struct physical_extent pool_extents[FAMEM_MAX_EXTENTS]; + bool poisoned; +}; + +static volatile sig_atomic_t g_stop; +static enum log_level g_log_level = LOG_INFO; +static int g_skip_tgid_check; + +/* Level-filtered stderr logger with a fixed [Famem HBM Server] prefix. */ +static void log_message(enum log_level level, const char *format, ...) +{ + static const char * const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + va_list arguments; + if (level < g_log_level) { + return; + } + fprintf(stderr, "[Famem HBM Server] %s: ", names[level]); + va_start(arguments, format); + vfprintf(stderr, format, arguments); + va_end(arguments); + fputc('\n', stderr); +} + +/* SIGTERM/SIGINT handler: set the global stop flag (checked by accept/poll loops). */ +static void stop_server(int signal_number) +{ + (void)signal_number; + g_stop = 1; +} + +/* Read one big-endian uint16 (wire protocol; see famem_protocol.h). */ +static uint16_t load_u16(const uint8_t *input) +{ + return (uint16_t)(((uint16_t)input[0] << 8U) | (uint16_t)input[1]); +} +/* Read one big-endian uint32. */ +static uint32_t load_u32(const uint8_t *input) +{ + return ((uint32_t)input[0] << 24U) | ((uint32_t)input[1] << 16U) | ((uint32_t)input[2] << 8U) | (uint32_t)input[3]; +} +/* Read one big-endian uint64. */ +static uint64_t load_u64(const uint8_t *input) +{ + return ((uint64_t)load_u32(input) << 32U) | (uint64_t)load_u32(input + 4U); +} +/* Write one big-endian uint16. */ +static void store_u16(uint8_t *output, uint16_t value) +{ + output[0] = (uint8_t)(value >> 8U); + output[1] = (uint8_t)value; +} +/* Write one big-endian uint32. */ +static void store_u32(uint8_t *output, uint32_t value) +{ + output[0] = (uint8_t)(value >> 24U); + output[1] = (uint8_t)(value >> 16U); + output[2] = (uint8_t)(value >> 8U); + output[3] = (uint8_t)value; +} +/* Write one big-endian uint64. */ +static void store_u64(uint8_t *output, uint64_t value) +{ + store_u32(output, (uint32_t)(value >> 32U)); + store_u32(output + 4U, (uint32_t)value); +} + +/* Read exactly *size* bytes; *clean_eof* distinguishes EOF-before-any-data from a mid-frame close. */ +static int receive_exact(int socket_fd, uint8_t *output, size_t size, bool *clean_eof) +{ + size_t offset = 0; + *clean_eof = false; + while (offset < size) { + const ssize_t received = recv(socket_fd, output + offset, size - offset, 0); + if (received > 0) { + offset += (size_t)received; + continue; + } + if (received == 0) { + *clean_eof = offset == 0; + return -1; + } + if (errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Write exactly *size* bytes; retries on EINTR, fails on short writes. */ +static int send_exact(int socket_fd, const uint8_t *input, size_t size) +{ + size_t offset = 0; + while (offset < size) { + const ssize_t sent = send(socket_fd, input + offset, size - offset, MSG_NOSIGNAL); + if (sent > 0) { + offset += (size_t)sent; + continue; + } + if (sent < 0 && errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Read + validate one request frame into *request*; on error fills *error* and returns -1. */ +static int receive_request(int socket_fd, struct request_frame *request, char *error, size_t error_size) +{ + uint8_t header[FAMEM_PROTOCOL_HEADER_SIZE]; + bool clean_eof = false; + + memset(request, 0, sizeof(*request)); + if (receive_exact(socket_fd, header, sizeof(header), &clean_eof) != 0) { + if (clean_eof) { + return 0; + } + snprintf(error, error_size, "connection closed while reading a frame header"); + return -1; + } + request->operation = load_u16(header + 8U); + request->request_id = load_u32(header + 12U); + request->payload_size = load_u32(header + 20U); + + if (load_u32(header) != FAMEM_PROTOCOL_MAGIC || load_u16(header + 4U) != FAMEM_PROTOCOL_VERSION || + load_u16(header + 6U) != FAMEM_MESSAGE_REQUEST || load_u16(header + 10U) != 0U || + load_u32(header + 16U) != FAMEM_STATUS_OK || request->operation < FAMEM_OP_HELLO || + request->operation > FAMEM_OP_STATS || request->request_id == 0U || + request->payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + snprintf(error, error_size, "invalid protocol header"); + return -2; + } + if (request->operation > FAMEM_OP_STATS) { + snprintf(error, error_size, "unsupported operation %u", request->operation); + return -2; + } + if (request->payload_size != 0U && + receive_exact(socket_fd, request->payload, request->payload_size, &clean_eof) != 0) { + snprintf(error, error_size, "connection closed while reading a frame payload"); + return -1; + } + return 1; +} +/* Frame and send one response (payload optional). */ +static int send_response(int socket_fd, uint16_t operation, uint32_t request_id, uint32_t status, + const uint8_t *payload, uint32_t payload_size) +{ + uint8_t frame[FAMEM_PROTOCOL_HEADER_SIZE + FAMEM_PROTOCOL_MAX_PAYLOAD]; + if (payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + errno = EMSGSIZE; + return -1; + } + store_u32(frame, FAMEM_PROTOCOL_MAGIC); + store_u16(frame + 4U, FAMEM_PROTOCOL_VERSION); + store_u16(frame + 6U, FAMEM_MESSAGE_RESPONSE); + store_u16(frame + 8U, operation); + store_u16(frame + 10U, 0U); + store_u32(frame + 12U, request_id); + store_u32(frame + 16U, status); + store_u32(frame + 20U, payload_size); + if (payload_size != 0U) { + memcpy(frame + FAMEM_PROTOCOL_HEADER_SIZE, payload, payload_size); + } + return send_exact(socket_fd, frame, FAMEM_PROTOCOL_HEADER_SIZE + payload_size); +} +/* send_response wrapper encoding a UTF-8 error message payload. */ +static int send_error_response(int socket_fd, const struct request_frame *request, uint32_t status, + const char *message) +{ + const size_t message_size = strnlen(message, FAMEM_PROTOCOL_MAX_PAYLOAD); + return send_response(socket_fd, request->operation, request->request_id, status, (const uint8_t *)message, + (uint32_t)message_size); +} +/* Send an error response and log the rejection at WARNING. */ +static int reject_client(int socket_fd, const struct request_frame *request, uint32_t status, const char *message) +{ + (void)send_error_response(socket_fd, request, status, message); + return -1; +} + +/* True when every byte is a lowercase hex digit (session/UUID validation). */ +static bool is_lower_hex(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (!((value[index] >= (uint8_t)'0' && value[index] <= (uint8_t)'9') || + (value[index] >= (uint8_t)'a' && value[index] <= (uint8_t)'f'))) { + return false; + } + } + return true; +} + +/* True when the buffer is all zero bytes (UUID-unsupported detection). */ +static bool is_all_zero(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (value[index] != 0) { + return false; + } + } + return true; +} + +/* Parse /proc/<pid>/stat for the start time (field 22) and parent PID (field 4). */ +static bool read_process_stat(pid_t pid, uint64_t *start_time, pid_t *parent_pid) +{ + char path[64]; + char buffer[4096]; + char *closing_parenthesis; + char *save_pointer = NULL; + char *token; + unsigned int field = 4U; + FILE *process_stat; + char *end = NULL; + unsigned long long parsed; + + if (pid <= 0 || start_time == NULL || snprintf(path, sizeof(path), "/proc/%ld/stat", (long)pid) < 0) { + return false; + } + process_stat = fopen(path, "r"); + if (process_stat == NULL) { + return false; + } + if (fgets(buffer, sizeof(buffer), process_stat) == NULL) { + (void)fclose(process_stat); + return false; + } + (void)fclose(process_stat); + closing_parenthesis = strrchr(buffer, ')'); + if (closing_parenthesis == NULL) { + return false; + } + token = strtok_r(closing_parenthesis + 1, " \t\r\n", &save_pointer); + token = token == NULL ? NULL : strtok_r(NULL, " \t\r\n", &save_pointer); + if (token == NULL) { + return false; + } + if (parent_pid != NULL) { + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0' || parsed > INT_MAX) { + return false; + } + *parent_pid = (pid_t)parsed; + } + while (token != NULL && field < 22U) { + token = strtok_r(NULL, " \t\r\n", &save_pointer); + ++field; + } + if (token == NULL || field != 22U) { + return false; + } + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0') { + return false; + } + *start_time = (uint64_t)parsed; + return true; +} + +/* Open a pidfd for *pid* (0 on kernels without pidfd_open). */ +static int open_process_pidfd(pid_t pid) +{ +#ifdef SYS_pidfd_open + const long descriptor = syscall(SYS_pidfd_open, pid, 0U); + if (descriptor >= 0 && descriptor <= INT_MAX) { + return (int)descriptor; + } +#else + (void)pid; +#endif + return -1; +} +/* Snapshot (pid, pidfd, start_time) for liveness tracking of a lease peer. */ +static void capture_process_identity(struct process_identity *identity, pid_t pid, pid_t *parent_pid) +{ + memset(identity, 0, sizeof(*identity)); + identity->pid = pid; + identity->pidfd = open_process_pidfd(pid); + identity->has_start_time = read_process_stat(pid, &identity->start_time, parent_pid); +} +/* Liveness check: pidfd poll or pid signal-probe plus start_time comparison (PID reuse guard). */ +static bool process_exists(const struct process_identity *identity) +{ + uint64_t current_start_time = 0U; + if (identity->pid <= 0) { + return false; + } + if (kill(identity->pid, 0) == 0) { + if (identity->has_start_time && read_process_stat(identity->pid, &current_start_time, NULL) && + current_start_time != identity->start_time) { + return false; + } + return true; + } + return errno == EPERM; +} +/* Block until the tracked process exits (pidfd poll or 200ms polling loop). */ +static void wait_for_process_exit(const struct process_identity *identity) +{ + struct timespec interval = {.tv_sec = 0, .tv_nsec = 200000000L}; + if (identity->pid <= 0) { + return; + } + if (identity->pidfd >= 0) { + struct pollfd descriptor = {.fd = identity->pidfd, .events = POLLIN | POLLHUP | POLLERR, .revents = 0}; + while (poll(&descriptor, 1U, -1) < 0) { + if (errno != EINTR) { + break; + } + } + if (descriptor.revents != 0) { + return; + } + } + while (process_exists(identity)) { + (void)nanosleep(&interval, NULL); + } +} +/* Close one identity's pidfd. */ +static void close_process_identity(struct process_identity *identity) +{ + if (identity->pid > 0 && identity->pidfd >= 0) { + (void)close(identity->pidfd); + } + memset(identity, 0, sizeof(*identity)); + identity->pidfd = -1; +} +/* Close the Worker and Copier identity pidfds of one lease. */ +static void close_lease_process_identities(struct lease *lease) +{ + close_process_identity(&lease->worker_process); + close_process_identity(&lease->copier_process); +} + +/* Famem last_error() string, or a placeholder when the library is silent. */ +static const char *native_error(void) +{ + const char *error = famem_last_error(); + return error == NULL || error[0] == '\0' ? "unknown native allocator error" : error; +} + +/* Release N physical extents via famem_server_free; any failure poisons the server. */ +static int free_extents(struct physical_extent *extents, size_t extent_count) +{ + int result = 0; + while (extent_count > 0U) { + struct physical_extent *extent = &extents[extent_count - 1U]; + if (extent->physical_handle != 0U) { + if (famem_server_free(extent->physical_handle) != 0) { + log_message(LOG_ERROR, "failed to free physical extent: %s", native_error()); + result = -1; + } else { + extent->physical_handle = 0U; + extent->shareable_handle = 0U; + } + } + --extent_count; + } + return result; +} + +enum { EXTENT_ALLOCATION_ROLLBACK_FAILED = -2 }; +/* Allocate N extents and authorize the lease's TGIDs on each shareable handle. */ +static int allocate_extents(int device, struct physical_extent *extents, size_t extent_count) +{ + size_t index; + for (index = 0; index < extent_count; ++index) { + if (famem_server_allocate_export(device, extents[index].size, extents[index].page_type, + &extents[index].physical_handle, &extents[index].shareable_handle) != 0) { + log_message(LOG_WARNING, "failed to allocate extent %zu: %s", index, native_error()); + if (free_extents(extents, index) != 0) { + return EXTENT_ALLOCATION_ROLLBACK_FAILED; + } + return -1; + } + } + return 0; +} + +/* (Re)authorize both lease TGIDs on every pool shareable handle (called after WAKE too). */ +static int authorize_pool(struct server_state *server, const struct lease *lease) +{ + const int32_t bare_tgids[] = {lease->bare_tgid, lease->copier_bare_tgid}; + size_t index; + for (index = 0; index < server->pool_extent_count; ++index) { + if (famem_server_authorize(server->pool_extents[index].shareable_handle, bare_tgids, + sizeof(bare_tgids) / sizeof(bare_tgids[0])) != 0) { + log_message(LOG_ERROR, "failed to authorize extent %zu: %s", index, native_error()); + server->poisoned = true; + log_message(LOG_ERROR, "pool authorization state is uncertain; server poisoned"); + return -1; + } + } + return 0; +} + +/* Choose the extent layout: fill with 1G pages up to availability, remainder 2M pages. */ +static int plan_extents(int device, uint64_t size, struct physical_extent *extents, size_t *extent_count) +{ + uint64_t small_granularity = 0U; + uint64_t large_granularity = 0U; + uint64_t free_large = 0U; + uint64_t large_size = 0U; + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_2M, &small_granularity) != 0 || + small_granularity != FAMEM_HUGE_2M_BYTES) { + return -1; + } + if (size == 0U || size % small_granularity != 0U) { + return -1; + } + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_1G, &large_granularity) == 0 && + large_granularity == FAMEM_HUGE_1G_BYTES && + famem_get_page_free_memory(device, FAMEM_PAGE_HUGE_1G, &free_large) == 0) { + large_size = size < free_large ? size : free_large; + large_size -= large_size % large_granularity; + } else { + log_message(LOG_WARNING, "1 GiB pages are unavailable; using 2 MiB pages only: %s", native_error()); + } + memset(extents, 0, sizeof(*extents) * FAMEM_MAX_EXTENTS); + *extent_count = 0U; + if (large_size != 0U) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_1G; + extents[*extent_count].size = large_size; + ++(*extent_count); + } + if (large_size != size) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_2M; + extents[*extent_count].size = size - large_size; + ++(*extent_count); + } + return 0; +} + +/* Allocate the pool extents once (first ACQUIRE) and fix server->pool_size. */ +static int initialize_pool(struct server_state *server, uint64_t size) +{ + int result; + if (plan_extents(server->device, size, server->pool_extents, &server->pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot plan the physical pool layout: %s", native_error()); + return -1; + } + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + if (result == -1 && server->pool_extents[0].page_type == FAMEM_PAGE_HUGE_1G) { + log_message(LOG_WARNING, "preferred pool layout failed; retrying with 2 MiB pages"); + memset(server->pool_extents, 0, sizeof(server->pool_extents)); + server->pool_extents[0].page_type = FAMEM_PAGE_HUGE_2M; + server->pool_extents[0].size = size; + server->pool_extent_count = 1U; + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + } + if (result != 0) { + log_message(LOG_ERROR, "cannot allocate the physical pool%s", + result == EXTENT_ALLOCATION_ROLLBACK_FAILED ? "; rollback left HBM allocated" : ""); + return -1; + } + server->pool_size = size; + log_message(LOG_INFO, "allocated persistent pool of %llu bytes in %zu extent(s)", (unsigned long long)size, + server->pool_extent_count); + return 0; +} + +/* ACQUIRE/WAKE success response: extent page types, sizes, and shareable handles. */ +static int send_mapping_response(int socket_fd, const struct request_frame *request, const struct server_state *server, + const struct lease *lease) +{ + uint8_t payload[FAMEM_MAPPING_PREFIX_SIZE + FAMEM_MAX_EXTENTS * FAMEM_MAPPING_EXTENT_SIZE]; + size_t offset = FAMEM_MAPPING_PREFIX_SIZE; + size_t index; + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, (uint32_t)server->pool_extent_count); + for (index = 0; index < server->pool_extent_count; ++index) { + store_u32(payload + offset, (uint32_t)server->pool_extents[index].page_type); + store_u64(payload + offset + 4U, server->pool_extents[index].size); + store_u64(payload + offset + 12U, server->pool_extents[index].shareable_handle); + offset += FAMEM_MAPPING_EXTENT_SIZE; + } + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, (uint32_t)offset); +} + +/* Mark *lease *the single active lease and authorize its TGIDs on the pool. */ +static int activate_lease(struct server_state *server, struct lease *lease) +{ + if (server->epoch == UINT64_MAX) { + server->poisoned = true; + } + if (server->poisoned) { + log_message(LOG_DEBUG, "activate_lease: server is poisoned"); + return -1; + } + if (!g_skip_tgid_check && !process_exists(&lease->worker_process)) { + log_message(LOG_DEBUG, "activate_lease: worker process (pid=%d) does not exist", lease->worker_process.pid); + return -1; + } + if (authorize_pool(server, lease) != 0) { + log_message(LOG_DEBUG, "activate_lease: authorize_pool failed"); + return -1; + } + lease->epoch = ++server->epoch; + lease->active = true; + server->active_lease = lease; + return 0; +} + +/* ACQUIRE: fix pool size on first use, verify capacity match, activate the lease. */ +static int handle_acquire(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) +{ + log_message(LOG_INFO, "ACQUIRE: lease->acquired=%d, poisoned=%d, worker_pid=%d, copier_pid=%d", + lease->acquired, server->poisoned, lease->worker_process.pid, lease->copier_process.pid); + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid ACQUIRE session or payload size"); + } + if (lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the lease has already acquired an arena"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (load_u64(request->payload + FAMEM_SESSION_SIZE) != server->pool_size) { + if (server->pool_size == 0U) { + /* Pool not yet allocated; use the client's requested size. */ + uint64_t requested_size = load_u64(request->payload + FAMEM_SESSION_SIZE); + log_message(LOG_INFO, "allocating pool on first ACQUIRE: %llu bytes", (unsigned long long)requested_size); + if (initialize_pool(server, requested_size) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "failed to allocate Famem pool"); + } + } else { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, + "requested arena size does not match the configured pool"); + } + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + lease->acquired = true; + return send_mapping_response(socket_fd, request, server, lease); +} + +/* Reject SLEEP/WAKE/RELEASE with a stale epoch (PROTOCOL error). */ +static int validate_epoch_request(int socket_fd, const struct request_frame *request, const struct lease *lease) +{ + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE || !lease->acquired || + load_u64(request->payload + FAMEM_SESSION_SIZE) != lease->epoch) { + (void)send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid lease epoch"); + return -1; + } + return 0; +} + +/* SLEEP: free the physical extents; the lease stays and the client keeps the epoch. */ +static int handle_sleep(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) +{ + uint8_t payload[FAMEM_EPOCH_RESPONSE_SIZE]; + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (!lease->active || server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is already sleeping"); + } + lease->active = false; + server->active_lease = NULL; + store_u64(payload, lease->epoch); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, sizeof(payload)); +} + +/* WAKE: re-allocate the same extent plan, bump the epoch, re-authorize, reply with mapping. */ +static int handle_wake(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) +{ + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (lease->active) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is not sleeping"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + return send_mapping_response(socket_fd, request, server, lease); +} + +/* RELEASE: free extents if still active and destroy the lease. */ +static int handle_release(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease, bool *released) +{ + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + *released = true; + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, NULL, 0U); +} + +/* STATS: read-only pool/lease diagnostics. */ +static int handle_stats(int socket_fd, const struct request_frame *request, struct server_state *server, + const struct lease *lease) +{ + uint8_t payload[FAMEM_STATS_RESPONSE_SIZE]; + uint64_t huge_1g_bytes = 0U; + uint64_t huge_2m_bytes = 0U; + size_t index; + if (request->payload_size != FAMEM_SESSION_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid STATS payload size"); + } + if (!lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "session has not acquired an arena"); + } + for (index = 0; index < server->pool_extent_count; ++index) { + if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_1G) { + huge_1g_bytes += server->pool_extents[index].size; + } else if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_2M) { + huge_2m_bytes += server->pool_extents[index].size; + } + } + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, lease->active ? 1U : 0U); + store_u32(payload + 20U, (uint32_t)server->pool_extent_count); + store_u64(payload + 24U, huge_1g_bytes); + store_u64(payload + 32U, huge_2m_bytes); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, + sizeof(payload)); +} + +/* True when a connected client already holds this session_id (duplicate HELLO guard). */ +static bool session_exists(const struct server_state *server, const uint8_t session_id[FAMEM_SESSION_SIZE]) +{ + const struct client_context *client = server->clients; + while (client != NULL) { + if (client->lease.session_id[0] != '\0' && memcmp(client->lease.session_id, session_id, FAMEM_SESSION_SIZE) == 0) { + return true; + } + client = client->next; + } + return false; +} + +/* One client's whole lifecycle: HELLO handshake, request loop, disconnect reclamation. */ +static int handle_client(struct client_context *client) +{ + struct server_state *server = client->server; + struct lease *lease = &client->lease; + const int socket_fd = client->socket_fd; + struct ucred credentials; + socklen_t credentials_size = sizeof(credentials); + struct timeval hello_timeout = {.tv_sec = 10, .tv_usec = 0}; + struct timeval no_timeout = {.tv_sec = 0, .tv_usec = 0}; + struct request_frame request; + uint32_t worker_bare_tgid; + uint32_t copier_bare_tgid; + pid_t copier_parent; + char error[256]; + int receive_result; + int response_result; + bool released = false; + + if (getsockopt(socket_fd, SOL_SOCKET, SO_PEERCRED, &credentials, &credentials_size) != 0 || + credentials_size != sizeof(credentials)) { + log_message(LOG_WARNING, "cannot read client credentials: %s", strerror(errno)); + return -1; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &hello_timeout, sizeof(hello_timeout)); + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result <= 0) { + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid HELLO frame: %s", error); + } + return -1; + } + if (request.operation != FAMEM_OP_HELLO || request.payload_size != FAMEM_HELLO_REQUEST_SIZE) { + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "the first request must be a valid HELLO"); + } + worker_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE); + copier_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE + 4U); + log_message(LOG_INFO, "HELLO: uuid_cmp=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE), + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid); + if ((!g_skip_tgid_check && memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) != 0) || + !is_lower_hex(request.payload, FAMEM_UUID_SIZE) || + !is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) || + copier_bare_tgid == 0U || copier_bare_tgid > INT32_MAX || + copier_bare_tgid == worker_bare_tgid) { + log_message(LOG_WARNING, "HELLO rejected: uuid_match=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u, skip_tgid=%d", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) == 0 ? 1 : 0, + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid, g_skip_tgid_check); + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "invalid Famem HELLO identity"); + } + pthread_mutex_lock(&server->mutex); + if (server->poisoned) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_INTERNAL, + "the Famem server has uncertain authorization state and must restart"); + } + if (session_exists(server, request.payload + FAMEM_UUID_SIZE)) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_BUSY, "this Famem session is already connected"); + } + + memset(lease, 0, sizeof(*lease)); + memcpy(lease->session_id, request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE); + lease->session_id[FAMEM_SESSION_SIZE] = '\0'; + capture_process_identity(&lease->worker_process, credentials.pid, NULL); + capture_process_identity(&lease->copier_process, (pid_t)copier_bare_tgid, &copier_parent); + lease->bare_tgid = (int32_t)worker_bare_tgid; + lease->copier_bare_tgid = (int32_t)copier_bare_tgid; + pthread_mutex_unlock(&server->mutex); + if (send_response(socket_fd, request.operation, request.request_id, FAMEM_STATUS_OK, + (const uint8_t *)server->device_uuid, FAMEM_UUID_SIZE) != 0) { + goto disconnect; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &no_timeout, sizeof(no_timeout)); + + while (!g_stop && !released) { + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result == 0) { + break; + } + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid request frame: %s", error); + if (receive_result == -2 && request.request_id != 0U && request.operation >= FAMEM_OP_HELLO && + request.operation <= FAMEM_OP_STATS) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, error); + } + break; + } + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] == '\0' || request.payload_size < FAMEM_SESSION_SIZE || + memcmp(request.payload, lease->session_id, FAMEM_SESSION_SIZE) != 0) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, + "the request session does not belong to this connection"); + pthread_mutex_unlock(&server->mutex); + continue; + } + switch (request.operation) { + case FAMEM_OP_ACQUIRE: + response_result = handle_acquire(socket_fd, &request, server, lease); + break; + case FAMEM_OP_SLEEP: + response_result = handle_sleep(socket_fd, &request, server, lease); + break; + case FAMEM_OP_WAKE: + response_result = handle_wake(socket_fd, &request, server, lease); + break; + case FAMEM_OP_RELEASE: + response_result = handle_release(socket_fd, &request, server, lease, &released); + break; + case FAMEM_OP_STATS: + response_result = handle_stats(socket_fd, &request, server, lease); + break; + case FAMEM_OP_HELLO: + default: + response_result = + send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "HELLO is only valid as the first request"); + break; + } + pthread_mutex_unlock(&server->mutex); + if (response_result != 0) { + goto disconnect; + } + } + +disconnect: + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] != '\0') { + if (lease->active) { + log_message(LOG_WARNING, "client disconnected with mapped HBM; waiting for Worker PID %ld and Copier PID %ld", + (long)lease->worker_process.pid, (long)lease->copier_process.pid); + pthread_mutex_unlock(&server->mutex); + if (!g_skip_tgid_check) { + wait_for_process_exit(&lease->worker_process); + } + wait_for_process_exit(&lease->copier_process); + pthread_mutex_lock(&server->mutex); + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + lease->active = false; + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + } + pthread_mutex_unlock(&server->mutex); + return 0; +} + +/* Per-connection thread entry: run handle_client, then flag the context finished. */ +static void *client_thread_main(void *argument) +{ + struct client_context *client = (struct client_context *)argument; + (void)handle_client(client); + close(client->socket_fd); + pthread_mutex_lock(&client->server->mutex); + client->socket_fd = -1; + client->thread_finished = true; + pthread_mutex_unlock(&client->server->mutex); + return NULL; +} + +/* Join finished client threads and unlink their contexts from the list. */ +static void reap_finished_clients(struct server_state *server) +{ + while (true) { + struct client_context **link; + struct client_context *client = NULL; + pthread_mutex_lock(&server->mutex); + link = &server->clients; + while (*link != NULL) { + if ((*link)->thread_finished && (*link)->lease.session_id[0] == '\0') { + client = *link; + *link = client->next; + break; + } + link = &(*link)->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + return; + } + (void)pthread_join(client->thread, NULL); + free(client); + } +} + +/* On shutdown: shutdown() every socket, join threads, reclaim still-active leases. */ +static void shutdown_clients(struct server_state *server) +{ + struct client_context *client; + pthread_mutex_lock(&server->mutex); + for (client = server->clients; client != NULL; client = client->next) { + if (client->socket_fd >= 0) { + (void)shutdown(client->socket_fd, SHUT_RDWR); + } + } + pthread_mutex_unlock(&server->mutex); + + while (true) { + pthread_mutex_lock(&server->mutex); + client = server->clients; + if (client != NULL) { + server->clients = client->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + break; + } + (void)pthread_join(client->thread, NULL); + close_lease_process_identities(&client->lease); + free(client); + } +} + +/* mkdir -p the socket directory with 0750, validating each created component. */ +static int make_directories(const char *directory) +{ + char path[PATH_MAX]; + size_t length; + size_t index; + struct stat directory_stat; + length = strlen(directory); + if (length >= sizeof(path)) { + errno = ENAMETOOLONG; + return -1; + } + memcpy(path, directory, length + 1U); + while (length > 1U && path[length - 1U] == '/') { + path[--length] = '\0'; + } + for (index = 1U; index < length; ++index) { + if (path[index] != '/') { + continue; + } + path[index] = '\0'; + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + path[index] = '/'; + } + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + if (stat(path, &directory_stat) != 0 || !S_ISDIR(directory_stat.st_mode)) { + errno = ENOTDIR; + return -1; + } + if (directory_stat.st_uid != getuid() || (directory_stat.st_mode & S_IWOTH) != 0U) { + errno = EACCES; + return -1; + } + return 0; +} + +/* Unlink a dead socket (ECONNREFUSED probe); a live one aborts startup. */ +static int remove_stale_socket(const char *socket_path) +{ + struct stat socket_stat; + struct stat current_socket_stat; + struct sockaddr_un address; + int probe_fd; + int connect_error; + if (lstat(socket_path, &socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(socket_stat.st_mode) || socket_stat.st_uid != getuid()) { + errno = EACCES; + return -1; + } + probe_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (probe_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, strlen(socket_path) + 1U); + if (connect(probe_fd, (struct sockaddr *)&address, sizeof(address)) == 0) { + close(probe_fd); + errno = EADDRINUSE; + return -1; + } + connect_error = errno; + close(probe_fd); + if (connect_error == ENOENT) { + return 0; + } + if (connect_error != ECONNREFUSED) { + errno = connect_error; + return -1; + } + if (lstat(socket_path, &current_socket_stat) != 0 || !S_ISSOCK(current_socket_stat.st_mode) || + current_socket_stat.st_dev != socket_stat.st_dev || current_socket_stat.st_ino != socket_stat.st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path) == 0 || errno == ENOENT ? 0 : -1; +} + +/* Unlink our socket at exit only if (dev,ino) still matches what we bound. */ +static int unlink_owned_socket(const char *socket_path, const struct stat *bound_socket_stat) +{ + struct stat current_socket_stat; + if (lstat(socket_path, &current_socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(current_socket_stat.st_mode) || current_socket_stat.st_dev != bound_socket_stat->st_dev || + current_socket_stat.st_ino != bound_socket_stat->st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path); +} + +/* Listener cleanup path shared between fatal startup errors. */ +static int discard_listener(int listener_fd, const char *socket_path, const struct stat *bound_socket_stat, + int saved_errno) +{ + close(listener_fd); + if (bound_socket_stat != NULL) { + (void)unlink_owned_socket(socket_path, bound_socket_stat); + } + errno = saved_errno; + return -1; +} + +/* bind()+listen() the server socket, chmod 0600, record (dev,ino) identity. */ +static int create_listener(const char *socket_path, struct stat *bound_socket_stat) +{ + struct sockaddr_un address; + const size_t path_length = strlen(socket_path); + int listener_fd; + listener_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (listener_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, path_length + 1U); + if (bind(listener_fd, (struct sockaddr *)&address, + offsetof(struct sockaddr_un, sun_path) + path_length + 1U) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (lstat(socket_path, bound_socket_stat) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (!S_ISSOCK(bound_socket_stat->st_mode) || bound_socket_stat->st_uid != getuid()) { + return discard_listener(listener_fd, socket_path, NULL, EACCES); + } + if (chmod(socket_path, 0600) != 0 || listen(listener_fd, 8) != 0) { + return discard_listener(listener_fd, socket_path, bound_socket_stat, errno); + } + return listener_fd; +} + +/* Print the --help text. */ +static void print_usage(FILE *output, const char *program) +{ + fprintf(output, + "Usage: %s --device DEVICE [--size-gib GIB | --size-bytes BYTES] [--socket-dir DIR] [--socket-name NAME] [--skip-tgid-check] [--log-level LEVEL]\n" + "Serve Famem physical HBM for one Ascend NPU.\n", + program); +} + +/* Map --log-level names to the enum. */ +static int parse_log_level(const char *value, enum log_level *output) +{ + static const char *const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + size_t index; + for (index = 0; index < sizeof(names) / sizeof(names[0]); ++index) { + if (strcmp(value, names[index]) == 0) { + *output = (enum log_level)index; + return 0; + } + } + return -1; +} + +/* Parse --device as a non-negative integer. */ +static int parse_device(const char *value, int *output) +{ + char *end = NULL; + long device; + errno = 0; + device = strtol(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || device < 0 || device > INT_MAX) { + return -1; + } + *output = (int)device; + return 0; +} + +/* Parse --pool-size in GiB (0 = defer to first ACQUIRE). */ +static int parse_pool_size(const char *value, uint64_t *output) +{ + char *end = NULL; + unsigned long long size_gib; + errno = 0; + size_gib = strtoull(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || size_gib == 0U || size_gib > FAMEM_MAX_POOL_GIB) { + return -1; + } + *output = (uint64_t)size_gib * FAMEM_HUGE_1G_BYTES; + return 0; +} + +/* Entry point: parse flags, bind the socket, serve until SIGTERM, reclaim everything. */ +int main(int argc, char **argv) +{ + static const struct option options[] = { + {"device", required_argument, NULL, 'd'}, + {"size-gib", required_argument, NULL, 'm'}, + {"size-bytes", required_argument, NULL, 'b'}, + {"socket-dir", required_argument, NULL, 's'}, + {"socket-name", required_argument, NULL, 'n'}, + {"skip-tgid-check", no_argument, NULL, 'k'}, + {"log-level", required_argument, NULL, 'l'}, + {"help", no_argument, NULL, 'h'}, + {NULL, 0, NULL, 0}, + }; + const char *socket_dir = NULL; + const char *socket_name = NULL; + char socket_path[PATH_MAX] = {0}; + struct sigaction action; + struct stat bound_socket_stat; + struct server_state server; + int device = -1; + int listener_fd = -1; + int socket_dir_fd = -1; + uint64_t pool_size = 0U; + int option; + int exit_code = EXIT_FAILURE; + while ((option = getopt_long(argc, argv, "d:m:b:s:n:kl:h", options, NULL)) != -1) { + switch (option) { + case 'd': + if (parse_device(optarg, &device) != 0) { + fprintf(stderr, "Invalid --device value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'm': + if (parse_pool_size(optarg, &pool_size) != 0) { + fprintf(stderr, "Invalid --size-gib value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'b': { + char *end = NULL; + errno = 0; + pool_size = strtoull(optarg, &end, 10); + if (errno != 0 || end == optarg || *end != '\0' || pool_size == 0U) { + fprintf(stderr, "Invalid --size-bytes value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + } + case 's': + socket_dir = optarg; + break; + case 'n': + socket_name = optarg; + break; + case 'k': + /* --skip-tgid-check: skip PID/TGID verification (needed in containers + where the PID namespace differs from the host) */ + g_skip_tgid_check = 1; + break; + case 'l': + if (parse_log_level(optarg, &g_log_level) != 0) { + fprintf(stderr, "Invalid --log-level value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'h': + print_usage(stdout, argv[0]); + return EXIT_SUCCESS; + default: + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + } + if (device < 0 || optind != argc) { + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + prctl(PR_SET_NAME, "hbm_server", 0, 0, 0); + /* Pool size will be determined by the first ACQUIRE request when not + explicitly set via --size-gib / --size-bytes. This allows a single + server binary to serve models with different per-NPU arena sizes + (e.g. 27 GiB for TP1-7B, 54 GiB for TP4-72B) without changing the + launch command. */ + server.pool_size = pool_size; /* 0 means "defer to first ACQUIRE" */ + if (socket_dir == NULL) { + socket_dir = getenv("VLLM_ASCEND_FAMEM_SOCKET_DIR"); + } + if (socket_dir == NULL || socket_dir[0] == '\0') { + socket_dir = FAMEM_DEFAULT_SOCKET_DIR; + } + memset(&action, 0, sizeof(action)); + action.sa_handler = stop_server; + sigemptyset(&action.sa_mask); + if (sigaction(SIGTERM, &action, NULL) != 0 || sigaction(SIGINT, &action, NULL) != 0) { + log_message(LOG_ERROR, "cannot install signal handlers: %s", strerror(errno)); + return EXIT_FAILURE; + } + signal(SIGPIPE, SIG_IGN); + if (famem_server_initialize(device) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem backend: %s", native_error()); + return EXIT_FAILURE; + } + memset(&server, 0, sizeof(server)); + server.device = device; + if (famem_get_device_uuid(device, server.device_uuid, sizeof(server.device_uuid)) != 0 || + !is_lower_hex((const uint8_t *)server.device_uuid, FAMEM_UUID_SIZE) || + is_all_zero((const uint8_t *)server.device_uuid, FAMEM_UUID_SIZE)) { + /* Some NPU models (e.g. Ascend 910B3) do not provide a valid UUID via + aclrtDeviceGetUuid (returns ACL_ERROR_RT_DEVICE_UNAVAILABLE or an + all-zero UUID). In that case, synthesize a deterministic UUID from + the device index so that the server can still operate. */ + log_message(LOG_WARNING, + "NPU UUID not available (device %d), synthesizing from device index", + device); + option = snprintf(server.device_uuid, sizeof(server.device_uuid), + "d%031d", device); + if (option < 0 || (size_t)option != FAMEM_UUID_SIZE) { + log_message(LOG_ERROR, "cannot synthesize device UUID"); + goto cleanup_backend; + } + } + server.device_uuid[FAMEM_UUID_SIZE] = '\0'; + if (pthread_mutex_init(&server.mutex, NULL) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem server mutex"); + goto cleanup_backend; + } + if (make_directories(socket_dir) != 0) { + log_message(LOG_ERROR, "cannot prepare socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + socket_dir_fd = open(socket_dir, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + if (socket_dir_fd < 0 || flock(socket_dir_fd, LOCK_EX) != 0) { + log_message(LOG_ERROR, "cannot lock Famem socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + /* Build socket path: use --socket-name if provided, otherwise device_uuid */ + if (socket_name != NULL) { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, socket_name); + } else { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, server.device_uuid); + } + if (option < 0 || (size_t)option >= sizeof(socket_path) || (size_t)option > FAMEM_SOCKET_PATH_MAX) { + errno = ENAMETOOLONG; + log_message(LOG_ERROR, "Famem socket path is too long"); + goto cleanup_pool; + } + if (remove_stale_socket(socket_path) != 0) { + log_message(LOG_ERROR, "cannot prepare Famem socket %s: %s", socket_path, strerror(errno)); + goto cleanup_pool; + } + if (pool_size != 0U && initialize_pool(&server, pool_size) != 0) { + goto cleanup_pool; + } + if (pool_size == 0U) { + log_message(LOG_INFO, "pool size not specified; will be determined by the first ACQUIRE request"); + } + listener_fd = create_listener(socket_path, &bound_socket_stat); + option = errno; + /* When --socket-name overrides the socket filename, create a UUID-named + hard link so that clients using famem_get_device_uuid() can find the socket. + A hard link (not symlink) is required because the client checks S_ISSOCK + with follow_symlinks=False, and only a hard link preserves the socket mode. */ + if (socket_name != NULL) { + char uuid_sock[PATH_MAX]; + option = snprintf(uuid_sock, sizeof(uuid_sock), "%s/%s.sock", socket_dir, server.device_uuid); + if (option > 0 && (size_t)option < sizeof(uuid_sock)) { + (void)unlink(uuid_sock); + if (link(socket_path, uuid_sock) != 0) { + log_message(LOG_WARNING, "cannot create UUID hardlink %s: %s", uuid_sock, strerror(errno)); + } + } + } + close(socket_dir_fd); + socket_dir_fd = -1; + errno = option; + if (listener_fd < 0) { + log_message(LOG_ERROR, "cannot create Famem listener: %s", strerror(errno)); + goto cleanup_pool; + } + log_message(LOG_INFO, "pool ready; listening on %s for NPU UUID %s", socket_path, server.device_uuid); + + while (!g_stop) { + struct client_context *client; + int client_fd = accept(listener_fd, NULL, NULL); + if (client_fd < 0) { + if (errno == EINTR) { + continue; + } + log_message(LOG_ERROR, "accept failed: %s", strerror(errno)); + goto cleanup; + } + client = (struct client_context *)calloc(1U, sizeof(*client)); + if (client == NULL) { + log_message(LOG_ERROR, "cannot allocate Famem client context"); + close(client_fd); + continue; + } + client->socket_fd = client_fd; + client->server = &server; + pthread_mutex_lock(&server.mutex); + client->next = server.clients; + server.clients = client; + pthread_mutex_unlock(&server.mutex); + if (pthread_create(&client->thread, NULL, client_thread_main, client) != 0) { + pthread_mutex_lock(&server.mutex); + server.clients = client->next; + pthread_mutex_unlock(&server.mutex); + log_message(LOG_ERROR, "cannot start Famem client thread"); + close(client_fd); + free(client); + continue; + } + reap_finished_clients(&server); + } + exit_code = EXIT_SUCCESS; + +cleanup: + close(listener_fd); + shutdown_clients(&server); + if (unlink_owned_socket(socket_path, &bound_socket_stat) != 0 && errno != ENOENT) { + log_message(LOG_WARNING, "cannot remove owned socket %s: %s", socket_path, strerror(errno)); + } +cleanup_pool: + if (free_extents(server.pool_extents, server.pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot release every persistent pool extent"); + exit_code = EXIT_FAILURE; + } +cleanup_mutex: + if (socket_dir_fd >= 0) { + close(socket_dir_fd); + } + (void)pthread_mutex_destroy(&server.mutex); +cleanup_backend: + if (famem_server_finalize() != 0) { + log_message(LOG_ERROR, "cannot finalize Famem backend: %s", native_error()); + exit_code = EXIT_FAILURE; + } + return exit_code; +} -- 2.43.0

2 4

[PATCH OLK-5.10] ocfs2: validate DIO orphan slot during inode read
by Jiacheng Yu 12 Nov '26

12 Nov '26

From: ZhengYuan Huang <gality369(a)gmail.com> mainline inclusion from mainline-v7.3-rc1 commit bb88131c9831075b8dc08cdd375743e5d44c7ca2 category: bugfix bugzilla: https://atomgit.com/src-openeuler/kernel/issues/19263 CVE: CVE-2026-90204 Reference: https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/?… -------------------------------- [BUG] A corrupted append-DIO dinode (high byte at offset 0xa1 corrupted from 0 to 1) can carry an i_dio_orphaned_slot outside the mounted filesystem slot range and trigger a use-after-free error: BUG: KASAN: slab-use-after-free in ocfs2_get_system_file_inode+0x780/0x820 fs/ocfs2/sysfile.c:102 Read of size 8 at addr ffff88800b767c00 by task kworker/u8:3/85 Call Trace: ... ocfs2_get_system_file_inode+0x780/0x820 fs/ocfs2/sysfile.c:102 ocfs2_wipe_inode+0x292/0xf70 fs/ocfs2/inode.c:840 ocfs2_delete_inode fs/ocfs2/inode.c:1155 [inline] ocfs2_evict_inode+0x6c9/0x1170 fs/ocfs2/inode.c:1295 evict+0x38e/0x8f0 fs/inode.c:810 iput_final fs/inode.c:1914 [inline] iput fs/inode.c:1966 [inline] iput+0x55b/0x8b0 fs/inode.c:1926 ocfs2_recover_orphans+0x610/0xe40 fs/ocfs2/journal.c:2374 ocfs2_complete_recovery+0x5af/0xd00 fs/ocfs2/journal.c:1373 ... [CAUSE] ocfs2_del_inode_from_orphan() uses i_dio_orphaned_slot to index the slot-local system inode cache. The dinode validator does not check this active slot, so an out-of-range value produces an invalid cache entry pointer that is dereferenced as an inode pointer. [FIX] Reject an active i_dio_orphaned_slot outside the slot range during dinode validation, before DIO orphan recovery can consume it. Link: https://lore.kernel.org/20260803030007.3993199-3-gality369@gmail.com Fixes: 06ee5c75b575 ("ocfs2: add functions to add and remove inode in orphan dir") Signed-off-by: ZhengYuan Huang <gality369(a)gmail.com> Reviewed-by: Joseph Qi <joseph.qi(a)linux.alibaba.com> Cc: Mark Fasheh <mark(a)fasheh.com> Cc: Joel Becker <jlbec(a)evilplan.org> Cc: Junxiao Bi <junxiao.bi(a)oracle.com> Cc: Changwei Ge <gechangwei(a)live.cn> Cc: Jun Piao <piaojun(a)huawei.com> Cc: Heming Zhao <heming.zhao(a)suse.com> Signed-off-by: Andrew Morton <akpm(a)linux-foundation.org> Conflicts: fs/ocfs2/inode.c [Context conflicts.] Signed-off-by: Jiacheng Yu <yujiacheng3(a)huawei.com> --- fs/ocfs2/inode.c | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/fs/ocfs2/inode.c b/fs/ocfs2/inode.c index 7d8df85e7e48..be8d669ce0c1 100644 --- a/fs/ocfs2/inode.c +++ b/fs/ocfs2/inode.c @@ -1426,6 +1426,14 @@ int ocfs2_validate_inode_block(struct super_block *sb, goto bail; } + if ((le32_to_cpu(di->i_flags) & OCFS2_DIO_ORPHANED_FL) && + le16_to_cpu(di->i_dio_orphaned_slot) >= OCFS2_SB(sb)->max_slots) { + rc = ocfs2_error(sb, "Invalid dinode %llu: DIO orphaned slot %u\n", + (unsigned long long)bh->b_blocknr, + le16_to_cpu(di->i_dio_orphaned_slot)); + goto bail; + } + if (S_ISLNK(le16_to_cpu(di->i_mode)) && !le32_to_cpu(di->i_clusters)) { int max_inline = ocfs2_fast_symlink_chars(sb); -- 2.34.1

2 1

[PATCH OLK-6.6 0/3] net: stop napi kthreads when THREADED napi is disabled
by JiangJieHua 12 Nov '26

12 Nov '26

Maxime Chevallier (1): net: Use a dedicated flag to track disabled napi instances Samiullah Khawaja (2): net: stop napi kthreads when THREADED napi is disabled net: Use NAPI_* in test_bit when stopping napi kthread include/linux/netdevice.h | 2 ++ net/core/dev.c | 59 ++++++++++++++++++++++++++++++++++++--- 2 files changed, 57 insertions(+), 4 deletions(-) -- 2.33.8

2 5

[PATCH OLK-6.6] sched/fair: Remove spurious shorter slice preemption
by Chen Jinghuang 12 Nov '26

12 Nov '26

From: Vincent Guittot <vincent.guittot(a)linaro.org> mainline inclusion from mainline-v6.17-rc1 commit 9de74a9850b9468ac2f515bfbe0844e0bfae869d category: bugfix bugzilla: https://atomgit.com/openeuler/kernel/issues/8903 Reference: https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/?… -------------------------------- Even if the waking task can preempt current, it might not be the one selected by pick_task_fair. Check that the waking task will be selected if we cancel the slice protection before doing so. Signed-off-by: Vincent Guittot <vincent.guittot(a)linaro.org> Signed-off-by: Peter Zijlstra (Intel) <peterz(a)infradead.org> Link: https://lkml.kernel.org/r/20250708165630.1948751-4-vincent.guittot@linaro.o… Conflicts: kernel/sched/fair.c [Fixes: 71db1ca0d0e6 ("sched/fair: Separate se->vlag from se->vprot"), this commit broke the RUN_TO_PARITY_WAKEUP feature, causing throughput degradation in high-concurrency scenarios; this patch restores the expected behavior.] Signed-off-by: Chen Jinghuang <chenjinghuang2(a)huawei.com> --- kernel/sched/fair.c | 48 +++++++++++++++------------------------------ 1 file changed, 16 insertions(+), 32 deletions(-) diff --git a/kernel/sched/fair.c b/kernel/sched/fair.c index 69176c7134bb..bf0eeaacd797 100644 --- a/kernel/sched/fair.c +++ b/kernel/sched/fair.c @@ -1250,7 +1250,7 @@ static inline void cancel_protect_slice(struct sched_entity *se) * * Which allows tree pruning through eligibility. */ -static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq, bool wakeup_preempt) +static struct sched_entity *__pick_eevdf(struct cfs_rq *cfs_rq, bool protect) { struct rb_node *node = cfs_rq->tasks_timeline.rb_root.rb_node; struct sched_entity *se = __pick_first_entity(cfs_rq); @@ -1279,11 +1279,11 @@ static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq, bool wakeup_preemp * elapsed. */ if (curr && - !(sched_feat(RUN_TO_PARITY_WAKEUP) && wakeup_preempt && curr->vlag == curr->deadline) && + !(sched_feat(RUN_TO_PARITY_WAKEUP) && protect && protect_slice(curr)) && !entity_eligible(cfs_rq, curr)) curr = NULL; - if (curr && protect_slice(curr)) + if (curr && protect && protect_slice(curr)) return curr; /* Pick the leftmost entity if it's eligible */ @@ -1327,6 +1327,11 @@ static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq, bool wakeup_preemp return best; } +static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq) +{ + return __pick_eevdf(cfs_rq, true); +} + #ifdef CONFIG_SCHED_DEBUG struct sched_entity *__pick_last_entity(struct cfs_rq *cfs_rq) { @@ -1527,27 +1532,6 @@ static inline bool resched_next_slice(struct cfs_rq *cfs_rq, struct sched_entity return !entity_eligible(cfs_rq, curr); } -static inline bool do_preempt_short(struct cfs_rq *cfs_rq, - struct sched_entity *pse, struct sched_entity *se) -{ - if (!sched_feat(PREEMPT_SHORT)) - return false; - - if (pse->slice >= se->slice) - return false; - - if (!entity_eligible(cfs_rq, pse)) - return false; - - if (entity_before(pse, se)) - return true; - - if (!entity_eligible(cfs_rq, se)) - return true; - - return false; -} - /* * Used by other classes to account runtime. */ @@ -5930,7 +5914,7 @@ pick_next_entity(struct cfs_rq *cfs_rq) cfs_rq->next && entity_eligible(cfs_rq, cfs_rq->next)) return cfs_rq->next; - return pick_eevdf(cfs_rq, false); + return pick_eevdf(cfs_rq); } static bool check_cfs_rq_runtime(struct cfs_rq *cfs_rq); @@ -9861,6 +9845,7 @@ static void check_preempt_wakeup_fair(struct rq *rq, struct task_struct *p, int struct cfs_rq *cfs_rq = task_cfs_rq(curr); int next_buddy_marked = 0; int cse_is_idle, pse_is_idle; + bool do_preempt_short = false; bool ret = 0; if (unlikely(se == pse)) @@ -9916,7 +9901,7 @@ static void check_preempt_wakeup_fair(struct rq *rq, struct task_struct *p, int * When non-idle entity preempt an idle entity, * don't give idle entity slice protection. */ - cancel_protect_slice(se); + do_preempt_short = true; goto preempt; } @@ -9939,22 +9924,21 @@ static void check_preempt_wakeup_fair(struct rq *rq, struct task_struct *p, int /* * If @p has a shorter slice than current and @p is eligible, override * current's slice protection in order to allow preemption. - * - * Note that even if @p does not turn out to be the most eligible - * task at this moment, current's slice protection will be lost. */ - if (do_preempt_short(cfs_rq, pse, se)) - cancel_protect_slice(se); + do_preempt_short = sched_feat(PREEMPT_SHORT) && (pse->slice < se->slice); /* * If @p has become the most eligible task, force preemption. */ - if (pick_eevdf(cfs_rq, true) == pse) + if (__pick_eevdf(cfs_rq, !do_preempt_short) == pse) goto preempt; return; preempt: + if (do_preempt_short) + cancel_protect_slice(se); + resched_curr(rq); } -- 2.34.1

2 1

[PATCH OLK-6.6] net: always dissect GSO packets in __virtio_net_hdr_to_skb()
by Hongtao Zhang 12 Nov '26

12 Nov '26

From: Eric Dumazet <edumazet(a)kernel.org> mainline inclusion from mainline-v7.3-rc7 commit 44378d02c5aaae24b60a75fbbb539b4aa4db503d category: bugfix bugzilla: https://atomgit.com/src-openeuler/kernel/issues/20744 CVE: CVE-2026-98954 Reference: https://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git/commit/?id… -------------------------------- Commit 9e8db5913264 ("net: avoid false positives in untrusted gso validation") added a '&& skb->network_header' check before flow-dissecting GSO packets without VIRTIO_NET_HDR_F_NEEDS_CSUM in __virtio_net_hdr_to_skb(), because some callers (such as tun_get_user(), tun_xdp_one(), virtnet_receive_done(), and raw_verify_header()) called virtio_net_hdr_*_to_skb() before initializing skb->network_header and skb->dev. Because __alloc_skb() and __build_skb_around() zero-initialize skb->network_header to 0 (unlike mac_header and transport_header which are initialized to ~0U), those four callers always had skb->network_header == 0 and bypassed flow dissection in __virtio_net_hdr_to_skb(). More generally, skb->network_header is an offset from skb->head (where 0 is also a valid offset whenever skb_headroom(skb) is 0), not a boolean flag. Whenever the 'if (gso_type && skb->network_header)' branch was skipped, the fallback 'else if (gso_type)' only pulled nh_min_len + thlen (40 bytes for TCPv4) without dissecting the packet, without validating ip_proto or n_proto, and without setting skb->transport_header. If the packet has a malformed network header, it is not rejected and a subsequent skb_probe_transport_header() also fails, leaving skb->transport_header at ~0U (0xffff). Similarly, if an IPv4 packet carries IP options (ihl > 5) or an IPv6 packet carries extension headers, pulling only nh_min_len + thlen can leave the TCP header outside skb->head. In both cases, tcp_hdrlen(skb) in skb_gso_transport_seglen() reads out-of-bounds: BUG: KASAN: slab-out-of-bounds in skb_gso_transport_seglen Read of size 2 by task poc/133 skb_gso_transport_seglen (net/core/gso.c:155) skb_gso_validate_mac_len (net/core/gso.c:270) tbf_enqueue (net/sched/sch_tbf.c:260) dev_qdisc_enqueue (net/core/dev.c:4227) __dev_queue_xmit (net/core/dev.c:4884) In addition, checking virtio_net_hdr_match_proto() only inside 'if (!skb->protocol)' before flow dissection both skipped validation when skb->protocol was pre-set by the caller and rejected VLAN-tagged frames whose outer L2 protocol is ETH_P_8021Q or ETH_P_8021AD. Fix this by: 1. Initializing skb->dev and skb->network_header (plus skb->protocol for IFF_TUN) before virtio_net_hdr_*_to_skb() in tun_get_user(), tun_xdp_one(), virtnet_receive_done(), and raw_verify_header(). In tun_get_user(), drop the redundant skb_reset_mac_header(skb) in the IFF_TUN case since __virtio_net_hdr_to_skb() unconditionally resets mac_header. 2. Removing '&& skb->network_header' and the unvalidated 'else if (gso_type)' fallback in __virtio_net_hdr_to_skb() so all GSO packets without VIRTIO_NET_HDR_F_NEEDS_CSUM are flow-dissected, have their transport header pulled into linear data, and have skb->transport_header set. 3. Moving the virtio_net_hdr_match_proto() check to after skb_flow_dissect_flow_keys_basic(), validating keys.basic.n_proto against hdr_gso_type. Fixes: 9e8db5913264 ("net: avoid false positives in untrusted gso validation") Fixes: d5be7f632bad ("net: validate untrusted gso packets without csum offload") Fixes: 924a9bc362a5 ("net: check if protocol extracted by virtio_net_hdr_set_proto is correct") Reported-by: Weiming Shi <bestswngs(a)gmail.com> Closes: https://lore.kernel.org/netdev/20260927163117.746432-2-bestswngs@gmail.com/ Signed-off-by: Eric Dumazet <edumazet(a)kernel.org> Reviewed-by: Willem de Bruijn <willemb(a)google.com> Cc: Michael S. Tsirkin <mst(a)redhat.com> Link: https://patch.msgid.link/20261001191140.2818991-3-edumazet@kernel.org Signed-off-by: Jakub Kicinski <kuba(a)kernel.org> Conflicts: drivers/net/tun.c drivers/net/virtio_net.c include/linux/virtio_net.h [The tunnel-aware helpers (tun_vnet_hdr_tnl_to_skb(), virtio_net_hdr_tnl_to_skb(), __virtio_net_hdr_to_skb()) and virtnet_receive_done() don't exist in 6.6. Keep using virtio_net_hdr_to_skb() and hdr->gso_type, initialize skb->dev and the network header before it in tun_get_user(), tun_xdp_one() and receive_buf(), and drop the xdp metadata context that is not present in this tree. In tun_get_user(), move virtio_net_hdr_to_skb() after the TUN/TAP switch and call eth_type_trans() for IFF_TAP afterwards.] Signed-off-by: Zhang Hongtao <zhanghongtao35(a)huawei.com> --- arch/um/drivers/vector_transports.c | 1 + drivers/net/tun.c | 23 ++++++----- drivers/net/virtio_net.c | 2 + include/linux/virtio_net.h | 59 ++++++++++++----------------- 4 files changed, 42 insertions(+), 43 deletions(-) diff --git a/arch/um/drivers/vector_transports.c b/arch/um/drivers/vector_transports.c index 0794d23f07cbc..92a127b5d54cb 100644 --- a/arch/um/drivers/vector_transports.c +++ b/arch/um/drivers/vector_transports.c @@ -209,6 +209,7 @@ static int raw_verify_header( if ((vheader->flags & VIRTIO_NET_HDR_F_DATA_VALID) > 0) return 1; + skb_set_network_header(skb, ETH_HLEN); virtio_net_hdr_to_skb(skb, vheader, virtio_legacy_is_little_endian()); return 0; } diff --git a/drivers/net/tun.c b/drivers/net/tun.c index 98711c4e12d1c..55a53feabdcb4 100644 --- a/drivers/net/tun.c +++ b/drivers/net/tun.c @@ -1875,12 +1875,7 @@ static ssize_t tun_get_user(struct tun_struct *tun, struct tun_file *tfile, } } - if (virtio_net_hdr_to_skb(skb, &gso, tun_is_little_endian(tun))) { - atomic_long_inc(&tun->rx_frame_errors); - err = -EINVAL; - goto free_skb; - } - + skb->dev = tun->dev; switch (tun->flags & TUN_TYPE_MASK) { case IFF_TUN: if (tun->flags & IFF_NO_PI) { @@ -1905,9 +1900,8 @@ static ssize_t tun_get_user(struct tun_struct *tun, struct tun_file *tfile, } } - skb_reset_mac_header(skb); + skb_reset_network_header(skb); skb->protocol = pi.proto; - skb->dev = tun->dev; break; case IFF_TAP: if (!pskb_may_pull(skb, ETH_HLEN)) { @@ -1915,10 +1909,19 @@ static ssize_t tun_get_user(struct tun_struct *tun, struct tun_file *tfile, drop_reason = SKB_DROP_REASON_HDR_TRUNC; goto drop; } - skb->protocol = eth_type_trans(skb, tun->dev); + skb_set_network_header(skb, ETH_HLEN); break; } + if (virtio_net_hdr_to_skb(skb, &gso, tun_is_little_endian(tun))) { + atomic_long_inc(&tun->rx_frame_errors); + err = -EINVAL; + goto free_skb; + } + + if ((tun->flags & TUN_TYPE_MASK) == IFF_TAP) + skb->protocol = eth_type_trans(skb, tun->dev); + /* copy skb_ubuf_info for callback when skb has no error */ if (zerocopy) { skb_zcopy_init(skb, msg_control); @@ -2524,6 +2527,8 @@ static int tun_xdp_one(struct tun_struct *tun, skb_reserve(skb, xdp->data - xdp->data_hard_start); skb_put(skb, xdp->data_end - xdp->data); + skb->dev = tun->dev; + skb_set_network_header(skb, ETH_HLEN); if (virtio_net_hdr_to_skb(skb, gso, tun_is_little_endian(tun))) { atomic_long_inc(&tun->rx_frame_errors); kfree_skb(skb); diff --git a/drivers/net/virtio_net.c b/drivers/net/virtio_net.c index 4352c190ce002..4a94d96fa754a 100644 --- a/drivers/net/virtio_net.c +++ b/drivers/net/virtio_net.c @@ -1857,6 +1857,8 @@ static void receive_buf(struct virtnet_info *vi, struct receive_queue *rq, if (flags & VIRTIO_NET_HDR_F_DATA_VALID) skb->ip_summed = CHECKSUM_UNNECESSARY; + skb->dev = dev; + skb_set_network_header(skb, ETH_HLEN); if (virtio_net_hdr_to_skb(skb, &hdr->hdr, virtio_is_little_endian(vi->vdev))) { net_warn_ratelimited("%s: bad gso: type: %u, size: %u\n", diff --git a/include/linux/virtio_net.h b/include/linux/virtio_net.h index d2cab10569644..497583676e53a 100644 --- a/include/linux/virtio_net.h +++ b/include/linux/virtio_net.h @@ -111,47 +111,38 @@ static inline int virtio_net_hdr_to_skb(struct sk_buff *skb, p_off = nh_min_len + thlen; if (!pskb_may_pull(skb, p_off)) return -EINVAL; - } else { + } else if (gso_type) { /* gso packets without NEEDS_CSUM do not set transport_offset. * probe and drop if does not match one of the above types. */ - if (gso_type && skb->network_header) { - struct flow_keys_basic keys; - - if (!skb->protocol) { - __be16 protocol = dev_parse_header_protocol(skb); - - if (!protocol) - virtio_net_hdr_set_proto(skb, hdr); - else if (!virtio_net_hdr_match_proto(protocol, hdr->gso_type)) - return -EINVAL; - else - skb->protocol = protocol; - } + struct flow_keys_basic keys; + + if (!skb->protocol) { + skb->protocol = dev_parse_header_protocol(skb); + if (!skb->protocol) + virtio_net_hdr_set_proto(skb, hdr); + } retry: - if (!skb_flow_dissect_flow_keys_basic(NULL, skb, &keys, - NULL, 0, 0, 0, - 0)) { - /* UFO does not specify ipv4 or 6: try both */ - if (gso_type & SKB_GSO_UDP && - skb->protocol == htons(ETH_P_IP)) { - skb->protocol = htons(ETH_P_IPV6); - goto retry; - } - return -EINVAL; + if (!skb_flow_dissect_flow_keys_basic(NULL, skb, &keys, + NULL, 0, 0, 0, + 0)) { + /* UFO does not specify ipv4 or 6: try both */ + if (gso_type & SKB_GSO_UDP && + skb->protocol == htons(ETH_P_IP)) { + skb->protocol = htons(ETH_P_IPV6); + goto retry; } + return -EINVAL; + } - p_off = keys.control.thoff + thlen; - if (!pskb_may_pull(skb, p_off) || - keys.basic.ip_proto != ip_proto) - return -EINVAL; + p_off = keys.control.thoff + thlen; + if (!pskb_may_pull(skb, p_off) || + keys.basic.ip_proto != ip_proto || + !virtio_net_hdr_match_proto(keys.basic.n_proto, + hdr->gso_type)) + return -EINVAL; - skb_set_transport_header(skb, keys.control.thoff); - } else if (gso_type) { - p_off = nh_min_len + thlen; - if (!pskb_may_pull(skb, p_off)) - return -EINVAL; - } + skb_set_transport_header(skb, keys.control.thoff); } if (hdr->gso_type != VIRTIO_NET_HDR_GSO_NONE) { -- 2.43.0

2 1

[PATCH OLK-6.6] sched/fair: Remove spurious shorter slice preemption
by Chen Jinghuang 12 Nov '26

12 Nov '26

From: Vincent Guittot <vincent.guittot(a)linaro.org> mainline inclusion from mainline-v6.17-rc1 commit 9de74a9850b9468ac2f515bfbe0844e0bfae869d category: bugfix bugzilla: https://atomgit.com/openeuler/kernel/issues/8903 Reference: https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/commit/?… -------------------------------- Even if the waking task can preempt current, it might not be the one selected by pick_task_fair. Check that the waking task will be selected if we cancel the slice protection before doing so. Signed-off-by: Vincent Guittot <vincent.guittot(a)linaro.org> Signed-off-by: Peter Zijlstra (Intel) <peterz(a)infradead.org> Link: https://lkml.kernel.org/r/20250708165630.1948751-4-vincent.guittot@linaro.o… Conflicts: kernel/sched/fair.c [Fixes: 71db1ca0d0e6 ("sched/fair: Separate se->vlag from se->vprot"), this commit broke the RUN_TO_PARITY_WAKEUP feature, causing throughput degradation in high-concurrency scenarios; this patch restores the expected behavior.] Signed-off-by: Chen Jinghuang <chenjinghuang2(a)huawei.com> --- kernel/sched/fair.c | 48 +++++++++++++++------------------------------ 1 file changed, 16 insertions(+), 32 deletions(-) diff --git a/kernel/sched/fair.c b/kernel/sched/fair.c index 69176c7134bb..e7d05661bafb 100644 --- a/kernel/sched/fair.c +++ b/kernel/sched/fair.c @@ -1250,7 +1250,7 @@ static inline void cancel_protect_slice(struct sched_entity *se) * * Which allows tree pruning through eligibility. */ -static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq, bool wakeup_preempt) +static struct sched_entity *__pick_eevdf(struct cfs_rq *cfs_rq, bool protect, bool wakeup_preempt) { struct rb_node *node = cfs_rq->tasks_timeline.rb_root.rb_node; struct sched_entity *se = __pick_first_entity(cfs_rq); @@ -1279,11 +1279,11 @@ static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq, bool wakeup_preemp * elapsed. */ if (curr && - !(sched_feat(RUN_TO_PARITY_WAKEUP) && wakeup_preempt && curr->vlag == curr->deadline) && + !(sched_feat(RUN_TO_PARITY_WAKEUP) && wakeup_preempt && protect_slice(curr)) && !entity_eligible(cfs_rq, curr)) curr = NULL; - if (curr && protect_slice(curr)) + if (curr && protect && protect_slice(curr)) return curr; /* Pick the leftmost entity if it's eligible */ @@ -1327,6 +1327,11 @@ static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq, bool wakeup_preemp return best; } +static struct sched_entity *pick_eevdf(struct cfs_rq *cfs_rq) +{ + return __pick_eevdf(cfs_rq, true, false); +} + #ifdef CONFIG_SCHED_DEBUG struct sched_entity *__pick_last_entity(struct cfs_rq *cfs_rq) { @@ -1527,27 +1532,6 @@ static inline bool resched_next_slice(struct cfs_rq *cfs_rq, struct sched_entity return !entity_eligible(cfs_rq, curr); } -static inline bool do_preempt_short(struct cfs_rq *cfs_rq, - struct sched_entity *pse, struct sched_entity *se) -{ - if (!sched_feat(PREEMPT_SHORT)) - return false; - - if (pse->slice >= se->slice) - return false; - - if (!entity_eligible(cfs_rq, pse)) - return false; - - if (entity_before(pse, se)) - return true; - - if (!entity_eligible(cfs_rq, se)) - return true; - - return false; -} - /* * Used by other classes to account runtime. */ @@ -5930,7 +5914,7 @@ pick_next_entity(struct cfs_rq *cfs_rq) cfs_rq->next && entity_eligible(cfs_rq, cfs_rq->next)) return cfs_rq->next; - return pick_eevdf(cfs_rq, false); + return pick_eevdf(cfs_rq); } static bool check_cfs_rq_runtime(struct cfs_rq *cfs_rq); @@ -9861,6 +9845,7 @@ static void check_preempt_wakeup_fair(struct rq *rq, struct task_struct *p, int struct cfs_rq *cfs_rq = task_cfs_rq(curr); int next_buddy_marked = 0; int cse_is_idle, pse_is_idle; + bool do_preempt_short = false; bool ret = 0; if (unlikely(se == pse)) @@ -9916,7 +9901,7 @@ static void check_preempt_wakeup_fair(struct rq *rq, struct task_struct *p, int * When non-idle entity preempt an idle entity, * don't give idle entity slice protection. */ - cancel_protect_slice(se); + do_preempt_short = true; goto preempt; } @@ -9939,22 +9924,21 @@ static void check_preempt_wakeup_fair(struct rq *rq, struct task_struct *p, int /* * If @p has a shorter slice than current and @p is eligible, override * current's slice protection in order to allow preemption. - * - * Note that even if @p does not turn out to be the most eligible - * task at this moment, current's slice protection will be lost. */ - if (do_preempt_short(cfs_rq, pse, se)) - cancel_protect_slice(se); + do_preempt_short = sched_feat(PREEMPT_SHORT) && (pse->slice < se->slice); /* * If @p has become the most eligible task, force preemption. */ - if (pick_eevdf(cfs_rq, true) == pse) + if (__pick_eevdf(cfs_rq, !do_preempt_short, true) == pse) goto preempt; return; preempt: + if (do_preempt_short) + cancel_protect_slice(se); + resched_curr(rq); } -- 2.34.1

2 2

[PATCH openEuler-26.09] xsched/famem: introduce lightweight NPU HBM management infrastructure FaMem
by Gu Bowen 12 Nov '26

12 Nov '26

hulk inclusion category: feature bugzilla: https://gitcode.com/openeuler/kernel/issues/10313 -------------------------------- FaMem (Fast-Map Memory) provides NPU HBM management for multi-model time-division multiplexing. It establishes a stable virtual address Arena per NPU to hold model weights and KV Cache; a native HBM server owns the physical pages and manages cross-model leases. Suspend releases physical HBM while preserving virtual addresses; resume uses a dedicated Copier process to restore weights layer-by-layer (H2D), overlapping weight loading with inference computation to reduce model switching latency. Signed-off-by: Alekseev Dmitry <alekseev.dmitry(a)huawei.com> Signed-off-by: Meskhidze Konstantin <konstantin.meskhidze(a)huawei.com> Signed-off-by: Mingrui Liu <liumingrui(a)huawei.com> Signed-off-by: Gu Bowen <gubowen5(a)huawei.com> --- arch/arm64/configs/openeuler_defconfig | 1 + arch/x86/configs/openeuler_defconfig | 1 + kernel/xsched/Kconfig | 18 + kernel/xsched/Makefile | 2 + kernel/xsched/famem_hbm_server.c | 1457 ++++++++++++++++++++++++ 5 files changed, 1479 insertions(+) create mode 100644 kernel/xsched/famem_hbm_server.c diff --git a/arch/arm64/configs/openeuler_defconfig b/arch/arm64/configs/openeuler_defconfig index 476efcfb46fe..b223548f3408 100644 --- a/arch/arm64/configs/openeuler_defconfig +++ b/arch/arm64/configs/openeuler_defconfig @@ -193,6 +193,7 @@ CONFIG_QOS_SCHED_SMART_GRID=y CONFIG_CGROUP_PIDS=y CONFIG_CGROUP_RDMA=y # CONFIG_CGROUP_DMEM is not set +# CONFIG_FAMEM is not set CONFIG_CGROUP_FREEZER=y CONFIG_CGROUP_HUGETLB=y CONFIG_CPUSETS=y diff --git a/arch/x86/configs/openeuler_defconfig b/arch/x86/configs/openeuler_defconfig index d4d392b59ab7..6ca79286fea8 100644 --- a/arch/x86/configs/openeuler_defconfig +++ b/arch/x86/configs/openeuler_defconfig @@ -210,6 +210,7 @@ CONFIG_QOS_SCHED_DYNAMIC_AFFINITY=y CONFIG_CGROUP_PIDS=y CONFIG_CGROUP_RDMA=y # CONFIG_CGROUP_DMEM is not set +# CONFIG_FAMEM is not set CONFIG_CGROUP_FREEZER=y CONFIG_CGROUP_HUGETLB=y CONFIG_CPUSETS=y diff --git a/kernel/xsched/Kconfig b/kernel/xsched/Kconfig index df6576da88fd..805b04de68f8 100644 --- a/kernel/xsched/Kconfig +++ b/kernel/xsched/Kconfig @@ -91,3 +91,21 @@ config CGROUP_XCU Required for systems requiring fine-grained resource control in cgroups. If unsure, say N. + +config FAMEM + bool "Enable FaMem (Fast-Map Memory) NPU HBM management" + default n + help + This option enables the FaMem (Fast-Map Memory) feature for managing + NPU HBM (High Bandwidth Memory) across multiple model instances. + + FaMem establishes a stable virtual address Arena per NPU to hold model + weights and KV Cache. A native HBM server owns the physical pages and + manages cross-model leases. Suspend releases physical HBM while + preserving virtual addresses; resume uses a dedicated Copier process to + restore weights layer-by-layer (H2D), overlapping weight loading with + inference computation to significantly reduce model-switching latency + and enable time-division multiplexing of HBM across model instances. + + This code is userspace-oriented and is isolated from the kernel build + by this config option. If unsure, say N. diff --git a/kernel/xsched/Makefile b/kernel/xsched/Makefile index 3e23012ea298..af788b279ed1 100644 --- a/kernel/xsched/Makefile +++ b/kernel/xsched/Makefile @@ -8,3 +8,5 @@ obj-$(CONFIG_XCU_SCHED_CFS) += cfs.o cfs_quota.o obj-$(CONFIG_CGROUP_XCU) += cgroup.o obj-$(CONFIG_CGROUP_DMEM) += dmem.o endif + +obj- += famem_hbm_server.o diff --git a/kernel/xsched/famem_hbm_server.c b/kernel/xsched/famem_hbm_server.c new file mode 100644 index 000000000000..89a6300022f0 --- /dev/null +++ b/kernel/xsched/famem_hbm_server.c @@ -0,0 +1,1457 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * SPDX-FileCopyrightText: Copyright (c) 2026 Huawei Technologies Co., Ltd. All Rights Reserved. + * + * famem_hbm_server.c — Native Famem HBM server for one Ascend NPU. + * + * This binary owns the physical HBM extents of a single device and serves + * the Famem control protocol (famem_protocol.h) on a Unix stream socket + * named after the device UUID. Clients (vLLM Worker processes and, for + * handle authorization, their Copier subprocesses) connect, claim a lease + * and drive the sleep/wake cycle; see FamemClient in + * vllm_ascend/device_allocator/famem_client.py for the client side. + * + * Server model + * ------------ + * - One thread per client connection; a mutex (server_state.mutex) + * serializes all state transitions. + * - One pool of at most FAMEM_MAX_EXTENTS physical extents, sized either + * by --pool-size at startup or by the first ACQUIRE request (pool_size + * == 0 defers sizing). Later ACQUIRE requests must request the exact + * same size or they are rejected — multi-instance time-sharing + * therefore requires identical arena sizes across instances. + * - At most one ACTIVE lease per NPU ("active" = extents currently + * allocated and mapped). Other connected leases may exist in the + * sleeping state; ACQUIRE/WAKE from a second lease gets + * FAMEM_STATUS_BUSY until the active lease SLEEPs or RELEASEs. + * + * Lease lifecycle + * --------------- + * HELLO create lease; record Worker + Copier identities (pidfd or + * pid+start_time) for liveness checking + * ACQUIRE allocate pool extents; authorize shareable handles for the + * Worker and Copier TGIDs; lease becomes active + * SLEEP free the physical extents (clients must have unmapped first); + * lease stays; epoch unchanged on the wire (client keeps it) + * WAKE re-allocate the same extent layout; epoch += 1; fresh + * shareable handles are exported and authorized + * RELEASE destroy the lease (frees extents if still active) + * + * Orphan reclamation + * ------------------ + * If a client disconnects while its lease is still active, the server + * blocks in the connection thread until the Worker and Copier processes + * exit (pidfd poll, or pid liveness + start_time comparison when pidfd + * is unavailable), then frees the extents. This guarantees CANN + * physical memory is never freed while a mapping in another process + * still references it. + * + * Poisoning + * --------- + * Any failure to free or allocate physical extents marks the server + * poisoned: it refuses new leases (FAMEM_STATUS_INTERNAL) until + * restarted, because the physical pool state is no longer trustworthy. + * + * Flags: --device (required), --socket-dir, --socket-name, --pool-size, + * --log-level, --skip-tgid-check (containers without SO_PEERCRED). + */ + +#define _GNU_SOURCE + +#include <errno.h> +#include <fcntl.h> +#include <getopt.h> +#include <limits.h> +#include <poll.h> +#include <pthread.h> +#include <signal.h> +#include <stdbool.h> +#include <stdarg.h> +#include <stddef.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/file.h> +#include <sys/prctl.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/syscall.h> +#include <sys/types.h> +#include <sys/un.h> +#include <time.h> +#include <unistd.h> + +#define FAMEM_DEFAULT_SOCKET_DIR "/run/vllm-ascend/famem" +#define FAMEM_SOCKET_PATH_MAX 103U +#define FAMEM_UUID_SIZE 32U +#define FAMEM_SESSION_SIZE 32U +#define FAMEM_MAX_EXTENTS 2U +#define FAMEM_PAGE_HUGE_1G 1 +#define FAMEM_PAGE_HUGE_2M 2 +#define FAMEM_HUGE_1G_BYTES (UINT64_C(1) << 30) +#define FAMEM_HUGE_2M_BYTES (UINT64_C(2) << 20) +#define FAMEM_MAX_POOL_GIB UINT64_C(4096) + +enum log_level { + LOG_DEBUG = 0, + LOG_INFO = 1, + LOG_WARNING = 2, + LOG_ERROR = 3, +}; + +struct request_frame { + uint16_t operation; + uint32_t request_id; + uint32_t payload_size; + uint8_t payload[FAMEM_PROTOCOL_MAX_PAYLOAD]; +}; +struct physical_extent { + int page_type; + uint64_t size; + uint64_t physical_handle; + uint64_t shareable_handle; +}; +struct process_identity { + pid_t pid; + int pidfd; + uint64_t start_time; + bool has_start_time; +}; +struct lease { + bool acquired; + bool active; + char session_id[FAMEM_SESSION_SIZE + 1U]; + struct process_identity worker_process; + struct process_identity copier_process; + int32_t bare_tgid; + int32_t copier_bare_tgid; + uint64_t epoch; +}; +struct server_state; +struct client_context { + int socket_fd; + pthread_t thread; + bool thread_finished; + struct lease lease; + struct server_state *server; + struct client_context *next; +}; +struct server_state { + int device; + char device_uuid[FAMEM_UUID_SIZE + 1U]; + pthread_mutex_t mutex; + struct client_context *clients; + struct lease *active_lease; + uint64_t pool_size; + uint64_t epoch; + size_t pool_extent_count; + struct physical_extent pool_extents[FAMEM_MAX_EXTENTS]; + bool poisoned; +}; + +static volatile sig_atomic_t g_stop; +static enum log_level g_log_level = LOG_INFO; +static int g_skip_tgid_check; + +/* Level-filtered stderr logger with a fixed [Famem HBM Server] prefix. */ +static void log_message(enum log_level level, const char *format, ...) +{ + static const char * const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + va_list arguments; + if (level < g_log_level) { + return; + } + fprintf(stderr, "[Famem HBM Server] %s: ", names[level]); + va_start(arguments, format); + vfprintf(stderr, format, arguments); + va_end(arguments); + fputc('\n', stderr); +} + +/* SIGTERM/SIGINT handler: set the global stop flag (checked by accept/poll loops). */ +static void stop_server(int signal_number) +{ + (void)signal_number; + g_stop = 1; +} + +/* Read one big-endian uint16 (wire protocol; see famem_protocol.h). */ +static uint16_t load_u16(const uint8_t *input) +{ + return (uint16_t)(((uint16_t)input[0] << 8U) | (uint16_t)input[1]); +} +/* Read one big-endian uint32. */ +static uint32_t load_u32(const uint8_t *input) +{ + return ((uint32_t)input[0] << 24U) | ((uint32_t)input[1] << 16U) | ((uint32_t)input[2] << 8U) | (uint32_t)input[3]; +} +/* Read one big-endian uint64. */ +static uint64_t load_u64(const uint8_t *input) +{ + return ((uint64_t)load_u32(input) << 32U) | (uint64_t)load_u32(input + 4U); +} +/* Write one big-endian uint16. */ +static void store_u16(uint8_t *output, uint16_t value) +{ + output[0] = (uint8_t)(value >> 8U); + output[1] = (uint8_t)value; +} +/* Write one big-endian uint32. */ +static void store_u32(uint8_t *output, uint32_t value) +{ + output[0] = (uint8_t)(value >> 24U); + output[1] = (uint8_t)(value >> 16U); + output[2] = (uint8_t)(value >> 8U); + output[3] = (uint8_t)value; +} +/* Write one big-endian uint64. */ +static void store_u64(uint8_t *output, uint64_t value) +{ + store_u32(output, (uint32_t)(value >> 32U)); + store_u32(output + 4U, (uint32_t)value); +} + +/* Read exactly *size* bytes; *clean_eof* distinguishes EOF-before-any-data from a mid-frame close. */ +static int receive_exact(int socket_fd, uint8_t *output, size_t size, bool *clean_eof) +{ + size_t offset = 0; + *clean_eof = false; + while (offset < size) { + const ssize_t received = recv(socket_fd, output + offset, size - offset, 0); + if (received > 0) { + offset += (size_t)received; + continue; + } + if (received == 0) { + *clean_eof = offset == 0; + return -1; + } + if (errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Write exactly *size* bytes; retries on EINTR, fails on short writes. */ +static int send_exact(int socket_fd, const uint8_t *input, size_t size) +{ + size_t offset = 0; + while (offset < size) { + const ssize_t sent = send(socket_fd, input + offset, size - offset, MSG_NOSIGNAL); + if (sent > 0) { + offset += (size_t)sent; + continue; + } + if (sent < 0 && errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Read + validate one request frame into *request*; on error fills *error* and returns -1. */ +static int receive_request(int socket_fd, struct request_frame *request, char *error, size_t error_size) +{ + uint8_t header[FAMEM_PROTOCOL_HEADER_SIZE]; + bool clean_eof = false; + + memset(request, 0, sizeof(*request)); + if (receive_exact(socket_fd, header, sizeof(header), &clean_eof) != 0) { + if (clean_eof) { + return 0; + } + snprintf(error, error_size, "connection closed while reading a frame header"); + return -1; + } + request->operation = load_u16(header + 8U); + request->request_id = load_u32(header + 12U); + request->payload_size = load_u32(header + 20U); + + if (load_u32(header) != FAMEM_PROTOCOL_MAGIC || load_u16(header + 4U) != FAMEM_PROTOCOL_VERSION || + load_u16(header + 6U) != FAMEM_MESSAGE_REQUEST || load_u16(header + 10U) != 0U || + load_u32(header + 16U) != FAMEM_STATUS_OK || request->operation < FAMEM_OP_HELLO || + request->operation > FAMEM_OP_STATS || request->request_id == 0U || + request->payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + snprintf(error, error_size, "invalid protocol header"); + return -2; + } + if (request->operation > FAMEM_OP_STATS) { + snprintf(error, error_size, "unsupported operation %u", request->operation); + return -2; + } + if (request->payload_size != 0U && + receive_exact(socket_fd, request->payload, request->payload_size, &clean_eof) != 0) { + snprintf(error, error_size, "connection closed while reading a frame payload"); + return -1; + } + return 1; +} +/* Frame and send one response (payload optional). */ +static int send_response(int socket_fd, uint16_t operation, uint32_t request_id, uint32_t status, + const uint8_t *payload, uint32_t payload_size) +{ + uint8_t frame[FAMEM_PROTOCOL_HEADER_SIZE + FAMEM_PROTOCOL_MAX_PAYLOAD]; + if (payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + errno = EMSGSIZE; + return -1; + } + store_u32(frame, FAMEM_PROTOCOL_MAGIC); + store_u16(frame + 4U, FAMEM_PROTOCOL_VERSION); + store_u16(frame + 6U, FAMEM_MESSAGE_RESPONSE); + store_u16(frame + 8U, operation); + store_u16(frame + 10U, 0U); + store_u32(frame + 12U, request_id); + store_u32(frame + 16U, status); + store_u32(frame + 20U, payload_size); + if (payload_size != 0U) { + memcpy(frame + FAMEM_PROTOCOL_HEADER_SIZE, payload, payload_size); + } + return send_exact(socket_fd, frame, FAMEM_PROTOCOL_HEADER_SIZE + payload_size); +} +/* send_response wrapper encoding a UTF-8 error message payload. */ +static int send_error_response(int socket_fd, const struct request_frame *request, uint32_t status, + const char *message) +{ + const size_t message_size = strnlen(message, FAMEM_PROTOCOL_MAX_PAYLOAD); + return send_response(socket_fd, request->operation, request->request_id, status, (const uint8_t *)message, + (uint32_t)message_size); +} +/* Send an error response and log the rejection at WARNING. */ +static int reject_client(int socket_fd, const struct request_frame *request, uint32_t status, const char *message) +{ + (void)send_error_response(socket_fd, request, status, message); + return -1; +} + +/* True when every byte is a lowercase hex digit (session/UUID validation). */ +static bool is_lower_hex(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (!((value[index] >= (uint8_t)'0' && value[index] <= (uint8_t)'9') || + (value[index] >= (uint8_t)'a' && value[index] <= (uint8_t)'f'))) { + return false; + } + } + return true; +} + +/* True when the buffer is all zero bytes (UUID-unsupported detection). */ +static bool is_all_zero(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (value[index] != 0) { + return false; + } + } + return true; +} + +/* Parse /proc/<pid>/stat for the start time (field 22) and parent PID (field 4). */ +static bool read_process_stat(pid_t pid, uint64_t *start_time, pid_t *parent_pid) +{ + char path[64]; + char buffer[4096]; + char *closing_parenthesis; + char *save_pointer = NULL; + char *token; + unsigned int field = 4U; + FILE *process_stat; + char *end = NULL; + unsigned long long parsed; + + if (pid <= 0 || start_time == NULL || snprintf(path, sizeof(path), "/proc/%ld/stat", (long)pid) < 0) { + return false; + } + process_stat = fopen(path, "r"); + if (process_stat == NULL) { + return false; + } + if (fgets(buffer, sizeof(buffer), process_stat) == NULL) { + (void)fclose(process_stat); + return false; + } + (void)fclose(process_stat); + closing_parenthesis = strrchr(buffer, ')'); + if (closing_parenthesis == NULL) { + return false; + } + token = strtok_r(closing_parenthesis + 1, " \t\r\n", &save_pointer); + token = token == NULL ? NULL : strtok_r(NULL, " \t\r\n", &save_pointer); + if (token == NULL) { + return false; + } + if (parent_pid != NULL) { + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0' || parsed > INT_MAX) { + return false; + } + *parent_pid = (pid_t)parsed; + } + while (token != NULL && field < 22U) { + token = strtok_r(NULL, " \t\r\n", &save_pointer); + ++field; + } + if (token == NULL || field != 22U) { + return false; + } + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0') { + return false; + } + *start_time = (uint64_t)parsed; + return true; +} + +/* Open a pidfd for *pid* (0 on kernels without pidfd_open). */ +static int open_process_pidfd(pid_t pid) +{ +#ifdef SYS_pidfd_open + const long descriptor = syscall(SYS_pidfd_open, pid, 0U); + if (descriptor >= 0 && descriptor <= INT_MAX) { + return (int)descriptor; + } +#else + (void)pid; +#endif + return -1; +} +/* Snapshot (pid, pidfd, start_time) for liveness tracking of a lease peer. */ +static void capture_process_identity(struct process_identity *identity, pid_t pid, pid_t *parent_pid) +{ + memset(identity, 0, sizeof(*identity)); + identity->pid = pid; + identity->pidfd = open_process_pidfd(pid); + identity->has_start_time = read_process_stat(pid, &identity->start_time, parent_pid); +} +/* Liveness check: pidfd poll or pid signal-probe plus start_time comparison (PID reuse guard). */ +static bool process_exists(const struct process_identity *identity) +{ + uint64_t current_start_time = 0U; + if (identity->pid <= 0) { + return false; + } + if (kill(identity->pid, 0) == 0) { + if (identity->has_start_time && read_process_stat(identity->pid, &current_start_time, NULL) && + current_start_time != identity->start_time) { + return false; + } + return true; + } + return errno == EPERM; +} +/* Block until the tracked process exits (pidfd poll or 200ms polling loop). */ +static void wait_for_process_exit(const struct process_identity *identity) +{ + struct timespec interval = {.tv_sec = 0, .tv_nsec = 200000000L}; + if (identity->pid <= 0) { + return; + } + if (identity->pidfd >= 0) { + struct pollfd descriptor = {.fd = identity->pidfd, .events = POLLIN | POLLHUP | POLLERR, .revents = 0}; + while (poll(&descriptor, 1U, -1) < 0) { + if (errno != EINTR) { + break; + } + } + if (descriptor.revents != 0) { + return; + } + } + while (process_exists(identity)) { + (void)nanosleep(&interval, NULL); + } +} +/* Close one identity's pidfd. */ +static void close_process_identity(struct process_identity *identity) +{ + if (identity->pid > 0 && identity->pidfd >= 0) { + (void)close(identity->pidfd); + } + memset(identity, 0, sizeof(*identity)); + identity->pidfd = -1; +} +/* Close the Worker and Copier identity pidfds of one lease. */ +static void close_lease_process_identities(struct lease *lease) +{ + close_process_identity(&lease->worker_process); + close_process_identity(&lease->copier_process); +} + +/* Famem last_error() string, or a placeholder when the library is silent. */ +static const char *native_error(void) +{ + const char *error = famem_last_error(); + return error == NULL || error[0] == '\0' ? "unknown native allocator error" : error; +} + +/* Release N physical extents via famem_server_free; any failure poisons the server. */ +static int free_extents(struct physical_extent *extents, size_t extent_count) +{ + int result = 0; + while (extent_count > 0U) { + struct physical_extent *extent = &extents[extent_count - 1U]; + if (extent->physical_handle != 0U) { + if (famem_server_free(extent->physical_handle) != 0) { + log_message(LOG_ERROR, "failed to free physical extent: %s", native_error()); + result = -1; + } else { + extent->physical_handle = 0U; + extent->shareable_handle = 0U; + } + } + --extent_count; + } + return result; +} + +enum { EXTENT_ALLOCATION_ROLLBACK_FAILED = -2 }; +/* Allocate N extents and authorize the lease's TGIDs on each shareable handle. */ +static int allocate_extents(int device, struct physical_extent *extents, size_t extent_count) +{ + size_t index; + for (index = 0; index < extent_count; ++index) { + if (famem_server_allocate_export(device, extents[index].size, extents[index].page_type, + &extents[index].physical_handle, &extents[index].shareable_handle) != 0) { + log_message(LOG_WARNING, "failed to allocate extent %zu: %s", index, native_error()); + if (free_extents(extents, index) != 0) { + return EXTENT_ALLOCATION_ROLLBACK_FAILED; + } + return -1; + } + } + return 0; +} + +/* (Re)authorize both lease TGIDs on every pool shareable handle (called after WAKE too). */ +static int authorize_pool(struct server_state *server, const struct lease *lease) +{ + const int32_t bare_tgids[] = {lease->bare_tgid, lease->copier_bare_tgid}; + size_t index; + for (index = 0; index < server->pool_extent_count; ++index) { + if (famem_server_authorize(server->pool_extents[index].shareable_handle, bare_tgids, + sizeof(bare_tgids) / sizeof(bare_tgids[0])) != 0) { + log_message(LOG_ERROR, "failed to authorize extent %zu: %s", index, native_error()); + server->poisoned = true; + log_message(LOG_ERROR, "pool authorization state is uncertain; server poisoned"); + return -1; + } + } + return 0; +} + +/* Choose the extent layout: fill with 1G pages up to availability, remainder 2M pages. */ +static int plan_extents(int device, uint64_t size, struct physical_extent *extents, size_t *extent_count) +{ + uint64_t small_granularity = 0U; + uint64_t large_granularity = 0U; + uint64_t free_large = 0U; + uint64_t large_size = 0U; + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_2M, &small_granularity) != 0 || + small_granularity != FAMEM_HUGE_2M_BYTES) { + return -1; + } + if (size == 0U || size % small_granularity != 0U) { + return -1; + } + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_1G, &large_granularity) == 0 && + large_granularity == FAMEM_HUGE_1G_BYTES && + famem_get_page_free_memory(device, FAMEM_PAGE_HUGE_1G, &free_large) == 0) { + large_size = size < free_large ? size : free_large; + large_size -= large_size % large_granularity; + } else { + log_message(LOG_WARNING, "1 GiB pages are unavailable; using 2 MiB pages only: %s", native_error()); + } + memset(extents, 0, sizeof(*extents) * FAMEM_MAX_EXTENTS); + *extent_count = 0U; + if (large_size != 0U) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_1G; + extents[*extent_count].size = large_size; + ++(*extent_count); + } + if (large_size != size) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_2M; + extents[*extent_count].size = size - large_size; + ++(*extent_count); + } + return 0; +} + +/* Allocate the pool extents once (first ACQUIRE) and fix server->pool_size. */ +static int initialize_pool(struct server_state *server, uint64_t size) +{ + int result; + if (plan_extents(server->device, size, server->pool_extents, &server->pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot plan the physical pool layout: %s", native_error()); + return -1; + } + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + if (result == -1 && server->pool_extents[0].page_type == FAMEM_PAGE_HUGE_1G) { + log_message(LOG_WARNING, "preferred pool layout failed; retrying with 2 MiB pages"); + memset(server->pool_extents, 0, sizeof(server->pool_extents)); + server->pool_extents[0].page_type = FAMEM_PAGE_HUGE_2M; + server->pool_extents[0].size = size; + server->pool_extent_count = 1U; + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + } + if (result != 0) { + log_message(LOG_ERROR, "cannot allocate the physical pool%s", + result == EXTENT_ALLOCATION_ROLLBACK_FAILED ? "; rollback left HBM allocated" : ""); + return -1; + } + server->pool_size = size; + log_message(LOG_INFO, "allocated persistent pool of %llu bytes in %zu extent(s)", (unsigned long long)size, + server->pool_extent_count); + return 0; +} + +/* ACQUIRE/WAKE success response: extent page types, sizes, and shareable handles. */ +static int send_mapping_response(int socket_fd, const struct request_frame *request, const struct server_state *server, + const struct lease *lease) +{ + uint8_t payload[FAMEM_MAPPING_PREFIX_SIZE + FAMEM_MAX_EXTENTS * FAMEM_MAPPING_EXTENT_SIZE]; + size_t offset = FAMEM_MAPPING_PREFIX_SIZE; + size_t index; + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, (uint32_t)server->pool_extent_count); + for (index = 0; index < server->pool_extent_count; ++index) { + store_u32(payload + offset, (uint32_t)server->pool_extents[index].page_type); + store_u64(payload + offset + 4U, server->pool_extents[index].size); + store_u64(payload + offset + 12U, server->pool_extents[index].shareable_handle); + offset += FAMEM_MAPPING_EXTENT_SIZE; + } + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, (uint32_t)offset); +} + +/* Mark *lease *the single active lease and authorize its TGIDs on the pool. */ +static int activate_lease(struct server_state *server, struct lease *lease) +{ + if (server->epoch == UINT64_MAX) { + server->poisoned = true; + } + if (server->poisoned) { + log_message(LOG_DEBUG, "activate_lease: server is poisoned"); + return -1; + } + if (!g_skip_tgid_check && !process_exists(&lease->worker_process)) { + log_message(LOG_DEBUG, "activate_lease: worker process (pid=%d) does not exist", lease->worker_process.pid); + return -1; + } + if (authorize_pool(server, lease) != 0) { + log_message(LOG_DEBUG, "activate_lease: authorize_pool failed"); + return -1; + } + lease->epoch = ++server->epoch; + lease->active = true; + server->active_lease = lease; + return 0; +} + +/* ACQUIRE: fix pool size on first use, verify capacity match, activate the lease. */ +static int handle_acquire(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) +{ + log_message(LOG_INFO, "ACQUIRE: lease->acquired=%d, poisoned=%d, worker_pid=%d, copier_pid=%d", + lease->acquired, server->poisoned, lease->worker_process.pid, lease->copier_process.pid); + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid ACQUIRE session or payload size"); + } + if (lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the lease has already acquired an arena"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (load_u64(request->payload + FAMEM_SESSION_SIZE) != server->pool_size) { + if (server->pool_size == 0U) { + /* Pool not yet allocated; use the client's requested size. */ + uint64_t requested_size = load_u64(request->payload + FAMEM_SESSION_SIZE); + log_message(LOG_INFO, "allocating pool on first ACQUIRE: %llu bytes", (unsigned long long)requested_size); + if (initialize_pool(server, requested_size) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "failed to allocate Famem pool"); + } + } else { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, + "requested arena size does not match the configured pool"); + } + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + lease->acquired = true; + return send_mapping_response(socket_fd, request, server, lease); +} + +/* Reject SLEEP/WAKE/RELEASE with a stale epoch (PROTOCOL error). */ +static int validate_epoch_request(int socket_fd, const struct request_frame *request, const struct lease *lease) +{ + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE || !lease->acquired || + load_u64(request->payload + FAMEM_SESSION_SIZE) != lease->epoch) { + (void)send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid lease epoch"); + return -1; + } + return 0; +} + +/* SLEEP: free the physical extents; the lease stays and the client keeps the epoch. */ +static int handle_sleep(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) +{ + uint8_t payload[FAMEM_EPOCH_RESPONSE_SIZE]; + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (!lease->active || server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is already sleeping"); + } + lease->active = false; + server->active_lease = NULL; + store_u64(payload, lease->epoch); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, sizeof(payload)); +} + +/* WAKE: re-allocate the same extent plan, bump the epoch, re-authorize, reply with mapping. */ +static int handle_wake(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) +{ + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (lease->active) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is not sleeping"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + return send_mapping_response(socket_fd, request, server, lease); +} + +/* RELEASE: free extents if still active and destroy the lease. */ +static int handle_release(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease, bool *released) +{ + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + *released = true; + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, NULL, 0U); +} + +/* STATS: read-only pool/lease diagnostics. */ +static int handle_stats(int socket_fd, const struct request_frame *request, struct server_state *server, + const struct lease *lease) +{ + uint8_t payload[FAMEM_STATS_RESPONSE_SIZE]; + uint64_t huge_1g_bytes = 0U; + uint64_t huge_2m_bytes = 0U; + size_t index; + if (request->payload_size != FAMEM_SESSION_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid STATS payload size"); + } + if (!lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "session has not acquired an arena"); + } + for (index = 0; index < server->pool_extent_count; ++index) { + if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_1G) { + huge_1g_bytes += server->pool_extents[index].size; + } else if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_2M) { + huge_2m_bytes += server->pool_extents[index].size; + } + } + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, lease->active ? 1U : 0U); + store_u32(payload + 20U, (uint32_t)server->pool_extent_count); + store_u64(payload + 24U, huge_1g_bytes); + store_u64(payload + 32U, huge_2m_bytes); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, + sizeof(payload)); +} + +/* True when a connected client already holds this session_id (duplicate HELLO guard). */ +static bool session_exists(const struct server_state *server, const uint8_t session_id[FAMEM_SESSION_SIZE]) +{ + const struct client_context *client = server->clients; + while (client != NULL) { + if (client->lease.session_id[0] != '\0' && memcmp(client->lease.session_id, session_id, FAMEM_SESSION_SIZE) == 0) { + return true; + } + client = client->next; + } + return false; +} + +/* One client's whole lifecycle: HELLO handshake, request loop, disconnect reclamation. */ +static int handle_client(struct client_context *client) +{ + struct server_state *server = client->server; + struct lease *lease = &client->lease; + const int socket_fd = client->socket_fd; + struct ucred credentials; + socklen_t credentials_size = sizeof(credentials); + struct timeval hello_timeout = {.tv_sec = 10, .tv_usec = 0}; + struct timeval no_timeout = {.tv_sec = 0, .tv_usec = 0}; + struct request_frame request; + uint32_t worker_bare_tgid; + uint32_t copier_bare_tgid; + pid_t copier_parent; + char error[256]; + int receive_result; + int response_result; + bool released = false; + + if (getsockopt(socket_fd, SOL_SOCKET, SO_PEERCRED, &credentials, &credentials_size) != 0 || + credentials_size != sizeof(credentials)) { + log_message(LOG_WARNING, "cannot read client credentials: %s", strerror(errno)); + return -1; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &hello_timeout, sizeof(hello_timeout)); + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result <= 0) { + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid HELLO frame: %s", error); + } + return -1; + } + if (request.operation != FAMEM_OP_HELLO || request.payload_size != FAMEM_HELLO_REQUEST_SIZE) { + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "the first request must be a valid HELLO"); + } + worker_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE); + copier_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE + 4U); + log_message(LOG_INFO, "HELLO: uuid_cmp=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE), + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid); + if ((!g_skip_tgid_check && memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) != 0) || + !is_lower_hex(request.payload, FAMEM_UUID_SIZE) || + !is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) || + copier_bare_tgid == 0U || copier_bare_tgid > INT32_MAX || + copier_bare_tgid == worker_bare_tgid) { + log_message(LOG_WARNING, "HELLO rejected: uuid_match=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u, skip_tgid=%d", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) == 0 ? 1 : 0, + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid, g_skip_tgid_check); + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "invalid Famem HELLO identity"); + } + pthread_mutex_lock(&server->mutex); + if (server->poisoned) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_INTERNAL, + "the Famem server has uncertain authorization state and must restart"); + } + if (session_exists(server, request.payload + FAMEM_UUID_SIZE)) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_BUSY, "this Famem session is already connected"); + } + + memset(lease, 0, sizeof(*lease)); + memcpy(lease->session_id, request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE); + lease->session_id[FAMEM_SESSION_SIZE] = '\0'; + capture_process_identity(&lease->worker_process, credentials.pid, NULL); + capture_process_identity(&lease->copier_process, (pid_t)copier_bare_tgid, &copier_parent); + lease->bare_tgid = (int32_t)worker_bare_tgid; + lease->copier_bare_tgid = (int32_t)copier_bare_tgid; + pthread_mutex_unlock(&server->mutex); + if (send_response(socket_fd, request.operation, request.request_id, FAMEM_STATUS_OK, + (const uint8_t *)server->device_uuid, FAMEM_UUID_SIZE) != 0) { + goto disconnect; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &no_timeout, sizeof(no_timeout)); + + while (!g_stop && !released) { + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result == 0) { + break; + } + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid request frame: %s", error); + if (receive_result == -2 && request.request_id != 0U && request.operation >= FAMEM_OP_HELLO && + request.operation <= FAMEM_OP_STATS) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, error); + } + break; + } + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] == '\0' || request.payload_size < FAMEM_SESSION_SIZE || + memcmp(request.payload, lease->session_id, FAMEM_SESSION_SIZE) != 0) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, + "the request session does not belong to this connection"); + pthread_mutex_unlock(&server->mutex); + continue; + } + switch (request.operation) { + case FAMEM_OP_ACQUIRE: + response_result = handle_acquire(socket_fd, &request, server, lease); + break; + case FAMEM_OP_SLEEP: + response_result = handle_sleep(socket_fd, &request, server, lease); + break; + case FAMEM_OP_WAKE: + response_result = handle_wake(socket_fd, &request, server, lease); + break; + case FAMEM_OP_RELEASE: + response_result = handle_release(socket_fd, &request, server, lease, &released); + break; + case FAMEM_OP_STATS: + response_result = handle_stats(socket_fd, &request, server, lease); + break; + case FAMEM_OP_HELLO: + default: + response_result = + send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "HELLO is only valid as the first request"); + break; + } + pthread_mutex_unlock(&server->mutex); + if (response_result != 0) { + goto disconnect; + } + } + +disconnect: + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] != '\0') { + if (lease->active) { + log_message(LOG_WARNING, "client disconnected with mapped HBM; waiting for Worker PID %ld and Copier PID %ld", + (long)lease->worker_process.pid, (long)lease->copier_process.pid); + pthread_mutex_unlock(&server->mutex); + if (!g_skip_tgid_check) { + wait_for_process_exit(&lease->worker_process); + } + wait_for_process_exit(&lease->copier_process); + pthread_mutex_lock(&server->mutex); + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + lease->active = false; + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + } + pthread_mutex_unlock(&server->mutex); + return 0; +} + +/* Per-connection thread entry: run handle_client, then flag the context finished. */ +static void *client_thread_main(void *argument) +{ + struct client_context *client = (struct client_context *)argument; + (void)handle_client(client); + close(client->socket_fd); + pthread_mutex_lock(&client->server->mutex); + client->socket_fd = -1; + client->thread_finished = true; + pthread_mutex_unlock(&client->server->mutex); + return NULL; +} + +/* Join finished client threads and unlink their contexts from the list. */ +static void reap_finished_clients(struct server_state *server) +{ + while (true) { + struct client_context **link; + struct client_context *client = NULL; + pthread_mutex_lock(&server->mutex); + link = &server->clients; + while (*link != NULL) { + if ((*link)->thread_finished && (*link)->lease.session_id[0] == '\0') { + client = *link; + *link = client->next; + break; + } + link = &(*link)->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + return; + } + (void)pthread_join(client->thread, NULL); + free(client); + } +} + +/* On shutdown: shutdown() every socket, join threads, reclaim still-active leases. */ +static void shutdown_clients(struct server_state *server) +{ + struct client_context *client; + pthread_mutex_lock(&server->mutex); + for (client = server->clients; client != NULL; client = client->next) { + if (client->socket_fd >= 0) { + (void)shutdown(client->socket_fd, SHUT_RDWR); + } + } + pthread_mutex_unlock(&server->mutex); + + while (true) { + pthread_mutex_lock(&server->mutex); + client = server->clients; + if (client != NULL) { + server->clients = client->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + break; + } + (void)pthread_join(client->thread, NULL); + close_lease_process_identities(&client->lease); + free(client); + } +} + +/* mkdir -p the socket directory with 0750, validating each created component. */ +static int make_directories(const char *directory) +{ + char path[PATH_MAX]; + size_t length; + size_t index; + struct stat directory_stat; + length = strlen(directory); + if (length >= sizeof(path)) { + errno = ENAMETOOLONG; + return -1; + } + memcpy(path, directory, length + 1U); + while (length > 1U && path[length - 1U] == '/') { + path[--length] = '\0'; + } + for (index = 1U; index < length; ++index) { + if (path[index] != '/') { + continue; + } + path[index] = '\0'; + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + path[index] = '/'; + } + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + if (stat(path, &directory_stat) != 0 || !S_ISDIR(directory_stat.st_mode)) { + errno = ENOTDIR; + return -1; + } + if (directory_stat.st_uid != getuid() || (directory_stat.st_mode & S_IWOTH) != 0U) { + errno = EACCES; + return -1; + } + return 0; +} + +/* Unlink a dead socket (ECONNREFUSED probe); a live one aborts startup. */ +static int remove_stale_socket(const char *socket_path) +{ + struct stat socket_stat; + struct stat current_socket_stat; + struct sockaddr_un address; + int probe_fd; + int connect_error; + if (lstat(socket_path, &socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(socket_stat.st_mode) || socket_stat.st_uid != getuid()) { + errno = EACCES; + return -1; + } + probe_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (probe_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, strlen(socket_path) + 1U); + if (connect(probe_fd, (struct sockaddr *)&address, sizeof(address)) == 0) { + close(probe_fd); + errno = EADDRINUSE; + return -1; + } + connect_error = errno; + close(probe_fd); + if (connect_error == ENOENT) { + return 0; + } + if (connect_error != ECONNREFUSED) { + errno = connect_error; + return -1; + } + if (lstat(socket_path, &current_socket_stat) != 0 || !S_ISSOCK(current_socket_stat.st_mode) || + current_socket_stat.st_dev != socket_stat.st_dev || current_socket_stat.st_ino != socket_stat.st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path) == 0 || errno == ENOENT ? 0 : -1; +} + +/* Unlink our socket at exit only if (dev,ino) still matches what we bound. */ +static int unlink_owned_socket(const char *socket_path, const struct stat *bound_socket_stat) +{ + struct stat current_socket_stat; + if (lstat(socket_path, &current_socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(current_socket_stat.st_mode) || current_socket_stat.st_dev != bound_socket_stat->st_dev || + current_socket_stat.st_ino != bound_socket_stat->st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path); +} + +/* Listener cleanup path shared between fatal startup errors. */ +static int discard_listener(int listener_fd, const char *socket_path, const struct stat *bound_socket_stat, + int saved_errno) +{ + close(listener_fd); + if (bound_socket_stat != NULL) { + (void)unlink_owned_socket(socket_path, bound_socket_stat); + } + errno = saved_errno; + return -1; +} + +/* bind()+listen() the server socket, chmod 0600, record (dev,ino) identity. */ +static int create_listener(const char *socket_path, struct stat *bound_socket_stat) +{ + struct sockaddr_un address; + const size_t path_length = strlen(socket_path); + int listener_fd; + listener_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (listener_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, path_length + 1U); + if (bind(listener_fd, (struct sockaddr *)&address, + offsetof(struct sockaddr_un, sun_path) + path_length + 1U) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (lstat(socket_path, bound_socket_stat) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (!S_ISSOCK(bound_socket_stat->st_mode) || bound_socket_stat->st_uid != getuid()) { + return discard_listener(listener_fd, socket_path, NULL, EACCES); + } + if (chmod(socket_path, 0600) != 0 || listen(listener_fd, 8) != 0) { + return discard_listener(listener_fd, socket_path, bound_socket_stat, errno); + } + return listener_fd; +} + +/* Print the --help text. */ +static void print_usage(FILE *output, const char *program) +{ + fprintf(output, + "Usage: %s --device DEVICE [--size-gib GIB | --size-bytes BYTES] [--socket-dir DIR] [--socket-name NAME] [--skip-tgid-check] [--log-level LEVEL]\n" + "Serve Famem physical HBM for one Ascend NPU.\n", + program); +} + +/* Map --log-level names to the enum. */ +static int parse_log_level(const char *value, enum log_level *output) +{ + static const char *const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + size_t index; + for (index = 0; index < sizeof(names) / sizeof(names[0]); ++index) { + if (strcmp(value, names[index]) == 0) { + *output = (enum log_level)index; + return 0; + } + } + return -1; +} + +/* Parse --device as a non-negative integer. */ +static int parse_device(const char *value, int *output) +{ + char *end = NULL; + long device; + errno = 0; + device = strtol(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || device < 0 || device > INT_MAX) { + return -1; + } + *output = (int)device; + return 0; +} + +/* Parse --pool-size in GiB (0 = defer to first ACQUIRE). */ +static int parse_pool_size(const char *value, uint64_t *output) +{ + char *end = NULL; + unsigned long long size_gib; + errno = 0; + size_gib = strtoull(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || size_gib == 0U || size_gib > FAMEM_MAX_POOL_GIB) { + return -1; + } + *output = (uint64_t)size_gib * FAMEM_HUGE_1G_BYTES; + return 0; +} + +/* Entry point: parse flags, bind the socket, serve until SIGTERM, reclaim everything. */ +int main(int argc, char **argv) +{ + static const struct option options[] = { + {"device", required_argument, NULL, 'd'}, + {"size-gib", required_argument, NULL, 'm'}, + {"size-bytes", required_argument, NULL, 'b'}, + {"socket-dir", required_argument, NULL, 's'}, + {"socket-name", required_argument, NULL, 'n'}, + {"skip-tgid-check", no_argument, NULL, 'k'}, + {"log-level", required_argument, NULL, 'l'}, + {"help", no_argument, NULL, 'h'}, + {NULL, 0, NULL, 0}, + }; + const char *socket_dir = NULL; + const char *socket_name = NULL; + char socket_path[PATH_MAX] = {0}; + struct sigaction action; + struct stat bound_socket_stat; + struct server_state server; + int device = -1; + int listener_fd = -1; + int socket_dir_fd = -1; + uint64_t pool_size = 0U; + int option; + int exit_code = EXIT_FAILURE; + while ((option = getopt_long(argc, argv, "d:m:b:s:n:kl:h", options, NULL)) != -1) { + switch (option) { + case 'd': + if (parse_device(optarg, &device) != 0) { + fprintf(stderr, "Invalid --device value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'm': + if (parse_pool_size(optarg, &pool_size) != 0) { + fprintf(stderr, "Invalid --size-gib value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'b': { + char *end = NULL; + errno = 0; + pool_size = strtoull(optarg, &end, 10); + if (errno != 0 || end == optarg || *end != '\0' || pool_size == 0U) { + fprintf(stderr, "Invalid --size-bytes value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + } + case 's': + socket_dir = optarg; + break; + case 'n': + socket_name = optarg; + break; + case 'k': + /* --skip-tgid-check: skip PID/TGID verification (needed in containers + where the PID namespace differs from the host) */ + g_skip_tgid_check = 1; + break; + case 'l': + if (parse_log_level(optarg, &g_log_level) != 0) { + fprintf(stderr, "Invalid --log-level value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'h': + print_usage(stdout, argv[0]); + return EXIT_SUCCESS; + default: + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + } + if (device < 0 || optind != argc) { + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + prctl(PR_SET_NAME, "hbm_server", 0, 0, 0); + /* Pool size will be determined by the first ACQUIRE request when not + explicitly set via --size-gib / --size-bytes. This allows a single + server binary to serve models with different per-NPU arena sizes + (e.g. 27 GiB for TP1-7B, 54 GiB for TP4-72B) without changing the + launch command. */ + server.pool_size = pool_size; /* 0 means "defer to first ACQUIRE" */ + if (socket_dir == NULL) { + socket_dir = getenv("VLLM_ASCEND_FAMEM_SOCKET_DIR"); + } + if (socket_dir == NULL || socket_dir[0] == '\0') { + socket_dir = FAMEM_DEFAULT_SOCKET_DIR; + } + memset(&action, 0, sizeof(action)); + action.sa_handler = stop_server; + sigemptyset(&action.sa_mask); + if (sigaction(SIGTERM, &action, NULL) != 0 || sigaction(SIGINT, &action, NULL) != 0) { + log_message(LOG_ERROR, "cannot install signal handlers: %s", strerror(errno)); + return EXIT_FAILURE; + } + signal(SIGPIPE, SIG_IGN); + if (famem_server_initialize(device) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem backend: %s", native_error()); + return EXIT_FAILURE; + } + memset(&server, 0, sizeof(server)); + server.device = device; + if (famem_get_device_uuid(device, server.device_uuid, sizeof(server.device_uuid)) != 0 || + !is_lower_hex((const uint8_t *)server.device_uuid, FAMEM_UUID_SIZE) || + is_all_zero((const uint8_t *)server.device_uuid, FAMEM_UUID_SIZE)) { + /* Some NPU models (e.g. Ascend 910B3) do not provide a valid UUID via + aclrtDeviceGetUuid (returns ACL_ERROR_RT_DEVICE_UNAVAILABLE or an + all-zero UUID). In that case, synthesize a deterministic UUID from + the device index so that the server can still operate. */ + log_message(LOG_WARNING, + "NPU UUID not available (device %d), synthesizing from device index", + device); + option = snprintf(server.device_uuid, sizeof(server.device_uuid), + "d%031d", device); + if (option < 0 || (size_t)option != FAMEM_UUID_SIZE) { + log_message(LOG_ERROR, "cannot synthesize device UUID"); + goto cleanup_backend; + } + } + server.device_uuid[FAMEM_UUID_SIZE] = '\0'; + if (pthread_mutex_init(&server.mutex, NULL) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem server mutex"); + goto cleanup_backend; + } + if (make_directories(socket_dir) != 0) { + log_message(LOG_ERROR, "cannot prepare socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + socket_dir_fd = open(socket_dir, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + if (socket_dir_fd < 0 || flock(socket_dir_fd, LOCK_EX) != 0) { + log_message(LOG_ERROR, "cannot lock Famem socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + /* Build socket path: use --socket-name if provided, otherwise device_uuid */ + if (socket_name != NULL) { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, socket_name); + } else { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, server.device_uuid); + } + if (option < 0 || (size_t)option >= sizeof(socket_path) || (size_t)option > FAMEM_SOCKET_PATH_MAX) { + errno = ENAMETOOLONG; + log_message(LOG_ERROR, "Famem socket path is too long"); + goto cleanup_pool; + } + if (remove_stale_socket(socket_path) != 0) { + log_message(LOG_ERROR, "cannot prepare Famem socket %s: %s", socket_path, strerror(errno)); + goto cleanup_pool; + } + if (pool_size != 0U && initialize_pool(&server, pool_size) != 0) { + goto cleanup_pool; + } + if (pool_size == 0U) { + log_message(LOG_INFO, "pool size not specified; will be determined by the first ACQUIRE request"); + } + listener_fd = create_listener(socket_path, &bound_socket_stat); + option = errno; + /* When --socket-name overrides the socket filename, create a UUID-named + hard link so that clients using famem_get_device_uuid() can find the socket. + A hard link (not symlink) is required because the client checks S_ISSOCK + with follow_symlinks=False, and only a hard link preserves the socket mode. */ + if (socket_name != NULL) { + char uuid_sock[PATH_MAX]; + option = snprintf(uuid_sock, sizeof(uuid_sock), "%s/%s.sock", socket_dir, server.device_uuid); + if (option > 0 && (size_t)option < sizeof(uuid_sock)) { + (void)unlink(uuid_sock); + if (link(socket_path, uuid_sock) != 0) { + log_message(LOG_WARNING, "cannot create UUID hardlink %s: %s", uuid_sock, strerror(errno)); + } + } + } + close(socket_dir_fd); + socket_dir_fd = -1; + errno = option; + if (listener_fd < 0) { + log_message(LOG_ERROR, "cannot create Famem listener: %s", strerror(errno)); + goto cleanup_pool; + } + log_message(LOG_INFO, "pool ready; listening on %s for NPU UUID %s", socket_path, server.device_uuid); + + while (!g_stop) { + struct client_context *client; + int client_fd = accept(listener_fd, NULL, NULL); + if (client_fd < 0) { + if (errno == EINTR) { + continue; + } + log_message(LOG_ERROR, "accept failed: %s", strerror(errno)); + goto cleanup; + } + client = (struct client_context *)calloc(1U, sizeof(*client)); + if (client == NULL) { + log_message(LOG_ERROR, "cannot allocate Famem client context"); + close(client_fd); + continue; + } + client->socket_fd = client_fd; + client->server = &server; + pthread_mutex_lock(&server.mutex); + client->next = server.clients; + server.clients = client; + pthread_mutex_unlock(&server.mutex); + if (pthread_create(&client->thread, NULL, client_thread_main, client) != 0) { + pthread_mutex_lock(&server.mutex); + server.clients = client->next; + pthread_mutex_unlock(&server.mutex); + log_message(LOG_ERROR, "cannot start Famem client thread"); + close(client_fd); + free(client); + continue; + } + reap_finished_clients(&server); + } + exit_code = EXIT_SUCCESS; + +cleanup: + close(listener_fd); + shutdown_clients(&server); + if (unlink_owned_socket(socket_path, &bound_socket_stat) != 0 && errno != ENOENT) { + log_message(LOG_WARNING, "cannot remove owned socket %s: %s", socket_path, strerror(errno)); + } +cleanup_pool: + if (free_extents(server.pool_extents, server.pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot release every persistent pool extent"); + exit_code = EXIT_FAILURE; + } +cleanup_mutex: + if (socket_dir_fd >= 0) { + close(socket_dir_fd); + } + (void)pthread_mutex_destroy(&server.mutex); +cleanup_backend: + if (famem_server_finalize() != 0) { + log_message(LOG_ERROR, "cannot finalize Famem backend: %s", native_error()); + exit_code = EXIT_FAILURE; + } + return exit_code; +} -- 2.43.0

2 3

[PATCH OLK-6.6 V5 00/10] P2PDMA PATCHSET PART 1
by Zizhi Wo 12 Nov '26

12 Nov '26

To support iostat statistics of passthrough and support dm/md/nvme-multipath p2pdma. Bart Van Assche (1): block: Fix queue_iostats_passthrough_show() Chaitanya Kulkarni (1): block: clear BLK_FEAT_PCI_P2PDMA in blk_stack_limits() for non-supporting devices Keith Busch (3): block: enable passthrough command statistics block: export passthrough stats enabled nvme: add support multipath passthrough iostats Kiran Kumar Modukuri (2): md: propagate BLK_FEAT_PCI_P2PDMA from member devices to RAID device nvme-multipath: enable PCI P2PDMA for multipath devices Zizhi Wo (2): block: withhold P2PDMA from integrity-bearing stacks nvme: clear PCI P2PDMA for integrity-bearing namespaces hotforest (1): dm: pass PCI P2PDMA through linear and striped targets Documentation/ABI/stable/sysfs-block | 8 ++++++ block/blk-mq-debugfs.c | 1 + block/blk-mq.c | 8 ++++++ block/blk-settings.c | 5 ++++ block/blk-sysfs.c | 26 ++++++++++++++++++ block/blk.h | 2 +- drivers/md/dm-linear.c | 3 +- drivers/md/dm-stripe.c | 3 +- drivers/md/dm-table.c | 27 ++++++++++++++++++ drivers/md/raid0.c | 2 ++ drivers/md/raid1.c | 1 + drivers/md/raid10.c | 1 + drivers/md/raid5.c | 6 ++++ drivers/nvme/host/core.c | 13 +++++++++ drivers/nvme/host/ioctl.c | 9 ++++++ drivers/nvme/host/multipath.c | 6 +++- include/linux/blk-mq.h | 41 ++++++++++++++++++++++++++++ include/linux/blkdev.h | 3 ++ include/linux/device-mapper.h | 5 ++++ 19 files changed, 166 insertions(+), 4 deletions(-) -- 2.52.0

2 11

[PATCH openEuler-26.09] xsched/famem: introduce lightweight NPU HBM management infrastructure FaMem
by Gu Bowen 12 Nov '26

12 Nov '26

hulk inclusion category: feature bugzilla: https://gitcode.com/openeuler/kernel/issues/10313 -------------------------------- FaMem (Fast-Map Memory) provides NPU HBM management for multi-model time-division multiplexing. It establishes a stable virtual address Arena per NPU to hold model weights and KV Cache; a native HBM server owns the physical pages and manages cross-model leases. Suspend releases physical HBM while preserving virtual addresses; resume uses a dedicated Copier process to restore weights layer-by-layer (H2D), overlapping weight loading with inference computation to reduce model switching latency. Signed-off-by: Alekseev Dmitry <alekseev.dmitry(a)huawei.com> Signed-off-by: Meskhidze Konstantin <konstantin.meskhidze(a)huawei.com> Signed-off-by: Mingrui Liu <liumingrui(a)huawei.com> Signed-off-by: Gu Bowen <gubowen5(a)huawei.com> --- arch/arm64/configs/openeuler_defconfig | 1 + kernel/xsched/Kconfig | 18 + kernel/xsched/Makefile | 4 + kernel/xsched/famem_hbm_server.c | 1418 ++++++++++++++++++++++++ 4 files changed, 1441 insertions(+) create mode 100644 kernel/xsched/famem_hbm_server.c diff --git a/arch/arm64/configs/openeuler_defconfig b/arch/arm64/configs/openeuler_defconfig index 476efcfb46fe..8c176902cacb 100644 --- a/arch/arm64/configs/openeuler_defconfig +++ b/arch/arm64/configs/openeuler_defconfig @@ -103,6 +103,7 @@ CONFIG_XCU_VSTREAM=y CONFIG_XSCHED_NR_CUS=128 CONFIG_XCU_SCHED_RT=y # CONFIG_XCU_SCHED_CFS is not set +CONFIG_FAMEM=n # # CPU/Task time and stats accounting diff --git a/kernel/xsched/Kconfig b/kernel/xsched/Kconfig index df6576da88fd..805b04de68f8 100644 --- a/kernel/xsched/Kconfig +++ b/kernel/xsched/Kconfig @@ -91,3 +91,21 @@ config CGROUP_XCU Required for systems requiring fine-grained resource control in cgroups. If unsure, say N. + +config FAMEM + bool "Enable FaMem (Fast-Map Memory) NPU HBM management" + default n + help + This option enables the FaMem (Fast-Map Memory) feature for managing + NPU HBM (High Bandwidth Memory) across multiple model instances. + + FaMem establishes a stable virtual address Arena per NPU to hold model + weights and KV Cache. A native HBM server owns the physical pages and + manages cross-model leases. Suspend releases physical HBM while + preserving virtual addresses; resume uses a dedicated Copier process to + restore weights layer-by-layer (H2D), overlapping weight loading with + inference computation to significantly reduce model-switching latency + and enable time-division multiplexing of HBM across model instances. + + This code is userspace-oriented and is isolated from the kernel build + by this config option. If unsure, say N. diff --git a/kernel/xsched/Makefile b/kernel/xsched/Makefile index 3e23012ea298..ae19dcc6a86e 100644 --- a/kernel/xsched/Makefile +++ b/kernel/xsched/Makefile @@ -8,3 +8,7 @@ obj-$(CONFIG_XCU_SCHED_CFS) += cfs.o cfs_quota.o obj-$(CONFIG_CGROUP_XCU) += cgroup.o obj-$(CONFIG_CGROUP_DMEM) += dmem.o endif + +ifdef CONFIG_FAMEM +obj-y += famem_hbm_server.o +endif diff --git a/kernel/xsched/famem_hbm_server.c b/kernel/xsched/famem_hbm_server.c new file mode 100644 index 000000000000..dd5e895ec500 --- /dev/null +++ b/kernel/xsched/famem_hbm_server.c @@ -0,0 +1,1418 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * SPDX-FileCopyrightText: Copyright (c) 2026 Huawei Technologies Co., Ltd. All Rights Reserved. + * + * famem_hbm_server.c — Native Famem HBM server for one Ascend NPU. + * + * This binary owns the physical HBM extents of a single device and serves + * the Famem control protocol (famem_protocol.h) on a Unix stream socket + * named after the device UUID. Clients (vLLM Worker processes and, for + * handle authorization, their Copier subprocesses) connect, claim a lease + * and drive the sleep/wake cycle; see FamemClient in + * vllm_ascend/device_allocator/famem_client.py for the client side. + * + * Server model + * ------------ + * - One thread per client connection; a mutex (server_state.mutex) + * serializes all state transitions. + * - One pool of at most FAMEM_MAX_EXTENTS physical extents, sized either + * by --pool-size at startup or by the first ACQUIRE request (pool_size + * == 0 defers sizing). Later ACQUIRE requests must request the exact + * same size or they are rejected — multi-instance time-sharing + * therefore requires identical arena sizes across instances. + * - At most one ACTIVE lease per NPU ("active" = extents currently + * allocated and mapped). Other connected leases may exist in the + * sleeping state; ACQUIRE/WAKE from a second lease gets + * FAMEM_STATUS_BUSY until the active lease SLEEPs or RELEASEs. + * + * Lease lifecycle + * --------------- + * HELLO create lease; record Worker + Copier identities (pidfd or + * pid+start_time) for liveness checking + * ACQUIRE allocate pool extents; authorize shareable handles for the + * Worker and Copier TGIDs; lease becomes active + * SLEEP free the physical extents (clients must have unmapped first); + * lease stays; epoch unchanged on the wire (client keeps it) + * WAKE re-allocate the same extent layout; epoch += 1; fresh + * shareable handles are exported and authorized + * RELEASE destroy the lease (frees extents if still active) + * + * Orphan reclamation + * ------------------ + * If a client disconnects while its lease is still active, the server + * blocks in the connection thread until the Worker and Copier processes + * exit (pidfd poll, or pid liveness + start_time comparison when pidfd + * is unavailable), then frees the extents. This guarantees CANN + * physical memory is never freed while a mapping in another process + * still references it. + * + * Poisoning + * --------- + * Any failure to free or allocate physical extents marks the server + * poisoned: it refuses new leases (FAMEM_STATUS_INTERNAL) until + * restarted, because the physical pool state is no longer trustworthy. + * + * Flags: --device (required), --socket-dir, --socket-name, --pool-size, + * --log-level, --skip-tgid-check (containers without SO_PEERCRED). + */ + +#define _GNU_SOURCE + +#include <errno.h> +#include <fcntl.h> +#include <getopt.h> +#include <limits.h> +#include <poll.h> +#include <pthread.h> +#include <signal.h> +#include <stdbool.h> +#include <stdarg.h> +#include <stddef.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/file.h> +#include <sys/prctl.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/syscall.h> +#include <sys/types.h> +#include <sys/un.h> +#include <time.h> +#include <unistd.h> + +#define FAMEM_DEFAULT_SOCKET_DIR "/run/vllm-ascend/famem" +#define FAMEM_SOCKET_PATH_MAX 103U +#define FAMEM_UUID_SIZE 32U +#define FAMEM_SESSION_SIZE 32U +#define FAMEM_MAX_EXTENTS 2U +#define FAMEM_PAGE_HUGE_1G 1 +#define FAMEM_PAGE_HUGE_2M 2 +#define FAMEM_HUGE_1G_BYTES (UINT64_C(1) << 30) +#define FAMEM_HUGE_2M_BYTES (UINT64_C(2) << 20) +#define FAMEM_MAX_POOL_GIB UINT64_C(4096) + +enum log_level { + LOG_DEBUG = 0, + LOG_INFO = 1, + LOG_WARNING = 2, + LOG_ERROR = 3, +}; + +struct request_frame { + uint16_t operation; + uint32_t request_id; + uint32_t payload_size; + uint8_t payload[FAMEM_PROTOCOL_MAX_PAYLOAD]; +}; +struct physical_extent { + int page_type; + uint64_t size; + uint64_t physical_handle; + uint64_t shareable_handle; +}; +struct process_identity { + pid_t pid; + int pidfd; + uint64_t start_time; + bool has_start_time; +}; +struct lease { + bool acquired; + bool active; + char session_id[FAMEM_SESSION_SIZE + 1U]; + struct process_identity worker_process; + struct process_identity copier_process; + int32_t bare_tgid; + int32_t copier_bare_tgid; + uint64_t epoch; +}; +struct server_state; +struct client_context { + int socket_fd; + pthread_t thread; + bool thread_finished; + struct lease lease; + struct server_state *server; + struct client_context *next; +}; +struct server_state { + int device; + char device_uuid[FAMEM_UUID_SIZE + 1U]; + pthread_mutex_t mutex; + struct client_context *clients; + struct lease *active_lease; + uint64_t pool_size; + uint64_t epoch; + size_t pool_extent_count; + struct physical_extent pool_extents[FAMEM_MAX_EXTENTS]; + bool poisoned; +}; + +static volatile sig_atomic_t g_stop; +static enum log_level g_log_level = LOG_INFO; +static int g_skip_tgid_check; + +/* Level-filtered stderr logger with a fixed [Famem HBM Server] prefix. */ +static void log_message(enum log_level level, const char *format, ...) +{ + static const char * const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + va_list arguments; + if (level < g_log_level) { + return; + } + fprintf(stderr, "[Famem HBM Server] %s: ", names[level]); + va_start(arguments, format); + vfprintf(stderr, format, arguments); + va_end(arguments); + fputc('\n', stderr); +} + +/* SIGTERM/SIGINT handler: set the global stop flag (checked by accept/poll loops). */ +static void stop_server(int signal_number) +{ + (void)signal_number; + g_stop = 1; +} + +/* Read one big-endian uint16 (wire protocol; see famem_protocol.h). */ +static uint16_t load_u16(const uint8_t *input) { return (uint16_t)(((uint16_t)input[0] << 8U) | (uint16_t)input[1]); } +/* Read one big-endian uint32. */ +static uint32_t load_u32(const uint8_t *input) { + return ((uint32_t)input[0] << 24U) | ((uint32_t)input[1] << 16U) | ((uint32_t)input[2] << 8U) | (uint32_t)input[3]; +} +/* Read one big-endian uint64. */ +static uint64_t load_u64(const uint8_t *input) { + return ((uint64_t)load_u32(input) << 32U) | (uint64_t)load_u32(input + 4U); +} +/* Write one big-endian uint16. */ +static void store_u16(uint8_t *output, uint16_t value) +{ + output[0] = (uint8_t)(value >> 8U); + output[1] = (uint8_t)value; +} +/* Write one big-endian uint32. */ +static void store_u32(uint8_t *output, uint32_t value) +{ + output[0] = (uint8_t)(value >> 24U); + output[1] = (uint8_t)(value >> 16U); + output[2] = (uint8_t)(value >> 8U); + output[3] = (uint8_t)value; +} +/* Write one big-endian uint64. */ +static void store_u64(uint8_t *output, uint64_t value) +{ + store_u32(output, (uint32_t)(value >> 32U)); + store_u32(output + 4U, (uint32_t)value); +} + +/* Read exactly *size* bytes; *clean_eof* distinguishes EOF-before-any-data from a mid-frame close. */ +static int receive_exact(int socket_fd, uint8_t *output, size_t size, bool *clean_eof) +{ + size_t offset = 0; + *clean_eof = false; + while (offset < size) { + const ssize_t received = recv(socket_fd, output + offset, size - offset, 0); + if (received > 0) { + offset += (size_t)received; + continue; + } + if (received == 0) { + *clean_eof = offset == 0; + return -1; + } + if (errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Write exactly *size* bytes; retries on EINTR, fails on short writes. */ +static int send_exact(int socket_fd, const uint8_t *input, size_t size) +{ + size_t offset = 0; + while (offset < size) { + const ssize_t sent = send(socket_fd, input + offset, size - offset, MSG_NOSIGNAL); + if (sent > 0) { + offset += (size_t)sent; + continue; + } + if (sent < 0 && errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Read + validate one request frame into *request*; on error fills *error* and returns -1. */ +static int receive_request(int socket_fd, struct request_frame *request, char *error, size_t error_size) +{ + uint8_t header[FAMEM_PROTOCOL_HEADER_SIZE]; + bool clean_eof = false; + + memset(request, 0, sizeof(*request)); + if (receive_exact(socket_fd, header, sizeof(header), &clean_eof) != 0) { + if (clean_eof) { + return 0; + } + snprintf(error, error_size, "connection closed while reading a frame header"); + return -1; + } + request->operation = load_u16(header + 8U); + request->request_id = load_u32(header + 12U); + request->payload_size = load_u32(header + 20U); + + if (load_u32(header) != FAMEM_PROTOCOL_MAGIC || load_u16(header + 4U) != FAMEM_PROTOCOL_VERSION || + load_u16(header + 6U) != FAMEM_MESSAGE_REQUEST || load_u16(header + 10U) != 0U || + load_u32(header + 16U) != FAMEM_STATUS_OK || request->operation < FAMEM_OP_HELLO || + request->operation > FAMEM_OP_STATS || request->request_id == 0U || + request->payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + snprintf(error, error_size, "invalid protocol header"); + return -2; + } + if (request->operation > FAMEM_OP_STATS) { + snprintf(error, error_size, "unsupported operation %u", request->operation); + return -2; + } + if (request->payload_size != 0U && + receive_exact(socket_fd, request->payload, request->payload_size, &clean_eof) != 0) { + snprintf(error, error_size, "connection closed while reading a frame payload"); + return -1; + } + return 1; +} +/* Frame and send one response (payload optional). */ +static int send_response(int socket_fd, uint16_t operation, uint32_t request_id, uint32_t status, + const uint8_t *payload, uint32_t payload_size) { + uint8_t frame[FAMEM_PROTOCOL_HEADER_SIZE + FAMEM_PROTOCOL_MAX_PAYLOAD]; + if (payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + errno = EMSGSIZE; + return -1; + } + store_u32(frame, FAMEM_PROTOCOL_MAGIC); + store_u16(frame + 4U, FAMEM_PROTOCOL_VERSION); + store_u16(frame + 6U, FAMEM_MESSAGE_RESPONSE); + store_u16(frame + 8U, operation); + store_u16(frame + 10U, 0U); + store_u32(frame + 12U, request_id); + store_u32(frame + 16U, status); + store_u32(frame + 20U, payload_size); + if (payload_size != 0U) { + memcpy(frame + FAMEM_PROTOCOL_HEADER_SIZE, payload, payload_size); + } + return send_exact(socket_fd, frame, FAMEM_PROTOCOL_HEADER_SIZE + payload_size); +} +/* send_response wrapper encoding a UTF-8 error message payload. */ +static int send_error_response(int socket_fd, const struct request_frame *request, uint32_t status, + const char *message) { + const size_t message_size = strnlen(message, FAMEM_PROTOCOL_MAX_PAYLOAD); + return send_response(socket_fd, request->operation, request->request_id, status, (const uint8_t*)message, + (uint32_t)message_size); +} +/* Send an error response and log the rejection at WARNING. */ +static int reject_client(int socket_fd, const struct request_frame *request, uint32_t status, const char *message) +{ + (void)send_error_response(socket_fd, request, status, message); + return -1; +} + +/* True when every byte is a lowercase hex digit (session/UUID validation). */ +static bool is_lower_hex(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (!((value[index] >= (uint8_t)'0' && value[index] <= (uint8_t)'9') || + (value[index] >= (uint8_t)'a' && value[index] <= (uint8_t)'f'))) { + return false; + } + } + return true; +} + +/* True when the buffer is all zero bytes (UUID-unsupported detection). */ +static bool is_all_zero(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (value[index] != 0) { + return false; + } + } + return true; +} + +/* Parse /proc/<pid>/stat for the start time (field 22) and parent PID (field 4). */ +static bool read_process_stat(pid_t pid, uint64_t *start_time, pid_t *parent_pid) +{ + char path[64]; + char buffer[4096]; + char *closing_parenthesis; + char *save_pointer = NULL; + char *token; + unsigned int field = 4U; + FILE *process_stat; + char *end = NULL; + unsigned long long parsed; + + if (pid <= 0 || start_time == NULL || snprintf(path, sizeof(path), "/proc/%ld/stat", (long)pid) < 0) { + return false; + } + process_stat = fopen(path, "r"); + if (process_stat == NULL) { + return false; + } + if (fgets(buffer, sizeof(buffer), process_stat) == NULL) { + (void)fclose(process_stat); + return false; + } + (void)fclose(process_stat); + closing_parenthesis = strrchr(buffer, ')'); + if (closing_parenthesis == NULL) { + return false; + } + token = strtok_r(closing_parenthesis + 1, " \t\r\n", &save_pointer); + token = token == NULL ? NULL : strtok_r(NULL, " \t\r\n", &save_pointer); + if (token == NULL) { + return false; + } + if (parent_pid != NULL) { + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0' || parsed > INT_MAX) { + return false; + } + *parent_pid = (pid_t)parsed; + } + while (token != NULL && field < 22U) { + token = strtok_r(NULL, " \t\r\n", &save_pointer); + ++field; + } + if (token == NULL || field != 22U) { + return false; + } + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0') { + return false; + } + *start_time = (uint64_t)parsed; + return true; +} + +/* Open a pidfd for *pid* (0 on kernels without pidfd_open). */ +static int open_process_pidfd(pid_t pid) +{ +#ifdef SYS_pidfd_open + const long descriptor = syscall(SYS_pidfd_open, pid, 0U); + if (descriptor >= 0 && descriptor <= INT_MAX) { + return (int)descriptor; + } +#else + (void)pid; +#endif + return -1; +} +/* Snapshot (pid, pidfd, start_time) for liveness tracking of a lease peer. */ +static void capture_process_identity(struct process_identity *identity, pid_t pid, pid_t *parent_pid) +{ + memset(identity, 0, sizeof(*identity)); + identity->pid = pid; + identity->pidfd = open_process_pidfd(pid); + identity->has_start_time = read_process_stat(pid, &identity->start_time, parent_pid); +} +/* Liveness check: pidfd poll or pid signal-probe plus start_time comparison (PID reuse guard). */ +static bool process_exists(const struct process_identity *identity) +{ + uint64_t current_start_time = 0U; + if (identity->pid <= 0) { + return false; + } + if (kill(identity->pid, 0) == 0) { + if (identity->has_start_time && read_process_stat(identity->pid, &current_start_time, NULL) && + current_start_time != identity->start_time) { + return false; + } + return true; + } + return errno == EPERM; +} +/* Block until the tracked process exits (pidfd poll or 200ms polling loop). */ +static void wait_for_process_exit(const struct process_identity *identity) { + struct timespec interval = {.tv_sec = 0, .tv_nsec = 200000000L}; + if (identity->pid <= 0) { + return; + } + if (identity->pidfd >= 0) { + struct pollfd descriptor = {.fd = identity->pidfd, .events = POLLIN | POLLHUP | POLLERR, .revents = 0}; + while (poll(&descriptor, 1U, -1) < 0) { + if (errno != EINTR) { + break; + } + } + if (descriptor.revents != 0) { + return; + } + } + while (process_exists(identity)) { + (void)nanosleep(&interval, NULL); + } +} +/* Close one identity's pidfd. */ +static void close_process_identity(struct process_identity *identity) { + if (identity->pid > 0 && identity->pidfd >= 0) { + (void)close(identity->pidfd); + } + memset(identity, 0, sizeof(*identity)); + identity->pidfd = -1; +} +/* Close the Worker and Copier identity pidfds of one lease. */ +static void close_lease_process_identities(struct lease *lease) { + close_process_identity(&lease->worker_process); + close_process_identity(&lease->copier_process); +} + +/* Famem last_error() string, or a placeholder when the library is silent. */ +static const char *native_error(void) { + const char *error = famem_last_error(); + return error == NULL || error[0] == '\0' ? "unknown native allocator error" : error; +} + +/* Release N physical extents via famem_server_free; any failure poisons the server. */ +static int free_extents(struct physical_extent *extents, size_t extent_count) { + int result = 0; + while (extent_count > 0U) { + struct physical_extent *extent = &extents[extent_count - 1U]; + if (extent->physical_handle != 0U) { + if (famem_server_free(extent->physical_handle) != 0) { + log_message(LOG_ERROR, "failed to free physical extent: %s", native_error()); + result = -1; + } else { + extent->physical_handle = 0U; + extent->shareable_handle = 0U; + } + } + --extent_count; + } + return result; +} + +enum { EXTENT_ALLOCATION_ROLLBACK_FAILED = -2 }; +/* Allocate N extents and authorize the lease's TGIDs on each shareable handle. */ +static int allocate_extents(int device, struct physical_extent *extents, size_t extent_count) { + size_t index; + for (index = 0; index < extent_count; ++index) { + if (famem_server_allocate_export(device, extents[index].size, extents[index].page_type, + &extents[index].physical_handle, &extents[index].shareable_handle) != 0) { + log_message(LOG_WARNING, "failed to allocate extent %zu: %s", index, native_error()); + if (free_extents(extents, index) != 0) { + return EXTENT_ALLOCATION_ROLLBACK_FAILED; + } + return -1; + } + } + return 0; +} + +/* (Re)authorize both lease TGIDs on every pool shareable handle (called after WAKE too). */ +static int authorize_pool(struct server_state *server, const struct lease *lease) { + const int32_t bare_tgids[] = {lease->bare_tgid, lease->copier_bare_tgid}; + size_t index; + for (index = 0; index < server->pool_extent_count; ++index) { + if (famem_server_authorize(server->pool_extents[index].shareable_handle, bare_tgids, + sizeof(bare_tgids) / sizeof(bare_tgids[0])) != 0) { + log_message(LOG_ERROR, "failed to authorize extent %zu: %s", index, native_error()); + server->poisoned = true; + log_message(LOG_ERROR, "pool authorization state is uncertain; server poisoned"); + return -1; + } + } + return 0; +} + +/* Choose the extent layout: fill with 1G pages up to availability, remainder 2M pages. */ +static int plan_extents(int device, uint64_t size, struct physical_extent *extents, size_t *extent_count) { + uint64_t small_granularity = 0U; + uint64_t large_granularity = 0U; + uint64_t free_large = 0U; + uint64_t large_size = 0U; + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_2M, &small_granularity) != 0 || + small_granularity != FAMEM_HUGE_2M_BYTES) { + return -1; + } + if (size == 0U || size % small_granularity != 0U) { + return -1; + } + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_1G, &large_granularity) == 0 && + large_granularity == FAMEM_HUGE_1G_BYTES && + famem_get_page_free_memory(device, FAMEM_PAGE_HUGE_1G, &free_large) == 0) { + large_size = size < free_large ? size : free_large; + large_size -= large_size % large_granularity; + } else { + log_message(LOG_WARNING, "1 GiB pages are unavailable; using 2 MiB pages only: %s", native_error()); + } + memset(extents, 0, sizeof(*extents) * FAMEM_MAX_EXTENTS); + *extent_count = 0U; + if (large_size != 0U) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_1G; + extents[*extent_count].size = large_size; + ++(*extent_count); + } + if (large_size != size) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_2M; + extents[*extent_count].size = size - large_size; + ++(*extent_count); + } + return 0; +} + +/* Allocate the pool extents once (first ACQUIRE) and fix server->pool_size. */ +static int initialize_pool(struct server_state *server, uint64_t size) { + int result; + if (plan_extents(server->device, size, server->pool_extents, &server->pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot plan the physical pool layout: %s", native_error()); + return -1; + } + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + if (result == -1 && server->pool_extents[0].page_type == FAMEM_PAGE_HUGE_1G) { + log_message(LOG_WARNING, "preferred pool layout failed; retrying with 2 MiB pages"); + memset(server->pool_extents, 0, sizeof(server->pool_extents)); + server->pool_extents[0].page_type = FAMEM_PAGE_HUGE_2M; + server->pool_extents[0].size = size; + server->pool_extent_count = 1U; + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + } + if (result != 0) { + log_message(LOG_ERROR, "cannot allocate the physical pool%s", + result == EXTENT_ALLOCATION_ROLLBACK_FAILED ? "; rollback left HBM allocated" : ""); + return -1; + } + server->pool_size = size; + log_message(LOG_INFO, "allocated persistent pool of %llu bytes in %zu extent(s)", (unsigned long long)size, + server->pool_extent_count); + return 0; +} + +/* ACQUIRE/WAKE success response: extent page types, sizes, and shareable handles. */ +static int send_mapping_response(int socket_fd, const struct request_frame *request, const struct server_state *server, + const struct lease *lease) { + uint8_t payload[FAMEM_MAPPING_PREFIX_SIZE + FAMEM_MAX_EXTENTS * FAMEM_MAPPING_EXTENT_SIZE]; + size_t offset = FAMEM_MAPPING_PREFIX_SIZE; + size_t index; + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, (uint32_t)server->pool_extent_count); + for (index = 0; index < server->pool_extent_count; ++index) { + store_u32(payload + offset, (uint32_t)server->pool_extents[index].page_type); + store_u64(payload + offset + 4U, server->pool_extents[index].size); + store_u64(payload + offset + 12U, server->pool_extents[index].shareable_handle); + offset += FAMEM_MAPPING_EXTENT_SIZE; + } + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, (uint32_t)offset); +} + +/* Mark *lease *the single active lease and authorize its TGIDs on the pool. */ +static int activate_lease(struct server_state *server, struct lease *lease) { + if (server->epoch == UINT64_MAX) { + server->poisoned = true; + } + if (server->poisoned) { + log_message(LOG_DEBUG, "activate_lease: server is poisoned"); + return -1; + } + if (!g_skip_tgid_check && !process_exists(&lease->worker_process)) { + log_message(LOG_DEBUG, "activate_lease: worker process (pid=%d) does not exist", lease->worker_process.pid); + return -1; + } + if (authorize_pool(server, lease) != 0) { + log_message(LOG_DEBUG, "activate_lease: authorize_pool failed"); + return -1; + } + lease->epoch = ++server->epoch; + lease->active = true; + server->active_lease = lease; + return 0; +} + +/* ACQUIRE: fix pool size on first use, verify capacity match, activate the lease. */ +static int handle_acquire(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) { + log_message(LOG_INFO, "ACQUIRE: lease->acquired=%d, poisoned=%d, worker_pid=%d, copier_pid=%d", + lease->acquired, server->poisoned, lease->worker_process.pid, lease->copier_process.pid); + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid ACQUIRE session or payload size"); + } + if (lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the lease has already acquired an arena"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (load_u64(request->payload + FAMEM_SESSION_SIZE) != server->pool_size) { + if (server->pool_size == 0U) { + /* Pool not yet allocated; use the client's requested size. */ + uint64_t requested_size = load_u64(request->payload + FAMEM_SESSION_SIZE); + log_message(LOG_INFO, "allocating pool on first ACQUIRE: %llu bytes", (unsigned long long)requested_size); + if (initialize_pool(server, requested_size) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "failed to allocate Famem pool"); + } + } else { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, + "requested arena size does not match the configured pool"); + } + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + lease->acquired = true; + return send_mapping_response(socket_fd, request, server, lease); +} + +/* Reject SLEEP/WAKE/RELEASE with a stale epoch (PROTOCOL error). */ +static int validate_epoch_request(int socket_fd, const struct request_frame *request, const struct lease *lease) { + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE || !lease->acquired || + load_u64(request->payload + FAMEM_SESSION_SIZE) != lease->epoch) { + (void)send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid lease epoch"); + return -1; + } + return 0; +} + +/* SLEEP: free the physical extents; the lease stays and the client keeps the epoch. */ +static int handle_sleep(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) { + uint8_t payload[FAMEM_EPOCH_RESPONSE_SIZE]; + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (!lease->active || server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is already sleeping"); + } + lease->active = false; + server->active_lease = NULL; + store_u64(payload, lease->epoch); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, sizeof(payload)); +} + +/* WAKE: re-allocate the same extent plan, bump the epoch, re-authorize, reply with mapping. */ +static int handle_wake(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) { + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (lease->active) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is not sleeping"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + return send_mapping_response(socket_fd, request, server, lease); +} + +/* RELEASE: free extents if still active and destroy the lease. */ +static int handle_release(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease, bool *released) { + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + *released = true; + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, NULL, 0U); +} + +/* STATS: read-only pool/lease diagnostics. */ +static int handle_stats(int socket_fd, const struct request_frame *request, struct server_state *server, + const struct lease *lease) { + uint8_t payload[FAMEM_STATS_RESPONSE_SIZE]; + uint64_t huge_1g_bytes = 0U; + uint64_t huge_2m_bytes = 0U; + size_t index; + if (request->payload_size != FAMEM_SESSION_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid STATS payload size"); + } + if (!lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "session has not acquired an arena"); + } + for (index = 0; index < server->pool_extent_count; ++index) { + if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_1G) { + huge_1g_bytes += server->pool_extents[index].size; + } else if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_2M) { + huge_2m_bytes += server->pool_extents[index].size; + } + } + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, lease->active ? 1U : 0U); + store_u32(payload + 20U, (uint32_t)server->pool_extent_count); + store_u64(payload + 24U, huge_1g_bytes); + store_u64(payload + 32U, huge_2m_bytes); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, + sizeof(payload)); +} + +/* True when a connected client already holds this session_id (duplicate HELLO guard). */ +static bool session_exists(const struct server_state *server, const uint8_t session_id[FAMEM_SESSION_SIZE]) { + const struct client_context *client = server->clients; + while (client != NULL) { + if (client->lease.session_id[0] != '\0' && memcmp(client->lease.session_id, session_id, FAMEM_SESSION_SIZE) == 0) { + return true; + } + client = client->next; + } + return false; +} + +/* One client's whole lifecycle: HELLO handshake, request loop, disconnect reclamation. */ +static int handle_client(struct client_context *client) { + struct server_state *server = client->server; + struct lease *lease = &client->lease; + const int socket_fd = client->socket_fd; + struct ucred credentials; + socklen_t credentials_size = sizeof(credentials); + struct timeval hello_timeout = {.tv_sec = 10, .tv_usec = 0}; + struct timeval no_timeout = {.tv_sec = 0, .tv_usec = 0}; + struct request_frame request; + uint32_t worker_bare_tgid; + uint32_t copier_bare_tgid; + pid_t copier_parent; + char error[256]; + int receive_result; + int response_result; + bool released = false; + + if (getsockopt(socket_fd, SOL_SOCKET, SO_PEERCRED, &credentials, &credentials_size) != 0 || + credentials_size != sizeof(credentials)) { + log_message(LOG_WARNING, "cannot read client credentials: %s", strerror(errno)); + return -1; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &hello_timeout, sizeof(hello_timeout)); + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result <= 0) { + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid HELLO frame: %s", error); + } + return -1; + } + if (request.operation != FAMEM_OP_HELLO || request.payload_size != FAMEM_HELLO_REQUEST_SIZE) { + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "the first request must be a valid HELLO"); + } + worker_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE); + copier_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE + 4U); + log_message(LOG_INFO, "HELLO: uuid_cmp=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE), + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid); + if ((!g_skip_tgid_check && memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) != 0) || + !is_lower_hex(request.payload, FAMEM_UUID_SIZE) || + !is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) || + copier_bare_tgid == 0U || copier_bare_tgid > INT32_MAX || + copier_bare_tgid == worker_bare_tgid) { + log_message(LOG_WARNING, "HELLO rejected: uuid_match=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u, skip_tgid=%d", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) == 0 ? 1 : 0, + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid, g_skip_tgid_check); + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "invalid Famem HELLO identity"); + } + pthread_mutex_lock(&server->mutex); + if (server->poisoned) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_INTERNAL, + "the Famem server has uncertain authorization state and must restart"); + } + if (session_exists(server, request.payload + FAMEM_UUID_SIZE)) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_BUSY, "this Famem session is already connected"); + } + + memset(lease, 0, sizeof(*lease)); + memcpy(lease->session_id, request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE); + lease->session_id[FAMEM_SESSION_SIZE] = '\0'; + capture_process_identity(&lease->worker_process, credentials.pid, NULL); + capture_process_identity(&lease->copier_process, (pid_t)copier_bare_tgid, &copier_parent); + lease->bare_tgid = (int32_t)worker_bare_tgid; + lease->copier_bare_tgid = (int32_t)copier_bare_tgid; + pthread_mutex_unlock(&server->mutex); + if (send_response(socket_fd, request.operation, request.request_id, FAMEM_STATUS_OK, + (const uint8_t*)server->device_uuid, FAMEM_UUID_SIZE) != 0) { + goto disconnect; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &no_timeout, sizeof(no_timeout)); + + while (!g_stop && !released) { + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result == 0) { + break; + } + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid request frame: %s", error); + if (receive_result == -2 && request.request_id != 0U && request.operation >= FAMEM_OP_HELLO && + request.operation <= FAMEM_OP_STATS) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, error); + } + break; + } + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] == '\0' || request.payload_size < FAMEM_SESSION_SIZE || + memcmp(request.payload, lease->session_id, FAMEM_SESSION_SIZE) != 0) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, + "the request session does not belong to this connection"); + pthread_mutex_unlock(&server->mutex); + continue; + } + switch (request.operation) { + case FAMEM_OP_ACQUIRE: + response_result = handle_acquire(socket_fd, &request, server, lease); + break; + case FAMEM_OP_SLEEP: + response_result = handle_sleep(socket_fd, &request, server, lease); + break; + case FAMEM_OP_WAKE: + response_result = handle_wake(socket_fd, &request, server, lease); + break; + case FAMEM_OP_RELEASE: + response_result = handle_release(socket_fd, &request, server, lease, &released); + break; + case FAMEM_OP_STATS: + response_result = handle_stats(socket_fd, &request, server, lease); + break; + case FAMEM_OP_HELLO: + default: + response_result = + send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "HELLO is only valid as the first request"); + break; + } + pthread_mutex_unlock(&server->mutex); + if (response_result != 0) { + goto disconnect; + } + } + +disconnect: + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] != '\0') { + if (lease->active) { + log_message(LOG_WARNING, "client disconnected with mapped HBM; waiting for Worker PID %ld and Copier PID %ld", + (long)lease->worker_process.pid, (long)lease->copier_process.pid); + pthread_mutex_unlock(&server->mutex); + if (!g_skip_tgid_check) { + wait_for_process_exit(&lease->worker_process); + } + wait_for_process_exit(&lease->copier_process); + pthread_mutex_lock(&server->mutex); + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + lease->active = false; + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + } + pthread_mutex_unlock(&server->mutex); + return 0; +} + +/* Per-connection thread entry: run handle_client, then flag the context finished. */ +static void *client_thread_main(void *argument) { + struct client_context *client = (struct client_context*)argument; + (void)handle_client(client); + close(client->socket_fd); + pthread_mutex_lock(&client->server->mutex); + client->socket_fd = -1; + client->thread_finished = true; + pthread_mutex_unlock(&client->server->mutex); + return NULL; +} + +/* Join finished client threads and unlink their contexts from the list. */ +static void reap_finished_clients(struct server_state *server) { + while (true) { + struct client_context **link; + struct client_context *client = NULL; + pthread_mutex_lock(&server->mutex); + link = &server->clients; + while (*link != NULL) { + if ((*link)->thread_finished && (*link)->lease.session_id[0] == '\0') { + client = *link; + *link = client->next; + break; + } + link = &(*link)->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + return; + } + (void)pthread_join(client->thread, NULL); + free(client); + } +} + +/* On shutdown: shutdown() every socket, join threads, reclaim still-active leases. */ +static void shutdown_clients(struct server_state *server) { + struct client_context *client; + pthread_mutex_lock(&server->mutex); + for (client = server->clients; client != NULL; client = client->next) { + if (client->socket_fd >= 0) { + (void)shutdown(client->socket_fd, SHUT_RDWR); + } + } + pthread_mutex_unlock(&server->mutex); + + while (true) { + pthread_mutex_lock(&server->mutex); + client = server->clients; + if (client != NULL) { + server->clients = client->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + break; + } + (void)pthread_join(client->thread, NULL); + close_lease_process_identities(&client->lease); + free(client); + } +} + +/* mkdir -p the socket directory with 0750, validating each created component. */ +static int make_directories(const char *directory) { + char path[PATH_MAX]; + size_t length; + size_t index; + struct stat directory_stat; + length = strlen(directory); + if (length >= sizeof(path)) { + errno = ENAMETOOLONG; + return -1; + } + memcpy(path, directory, length + 1U); + while (length > 1U && path[length - 1U] == '/') { + path[--length] = '\0'; + } + for (index = 1U; index < length; ++index) { + if (path[index] != '/') { + continue; + } + path[index] = '\0'; + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + path[index] = '/'; + } + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + if (stat(path, &directory_stat) != 0 || !S_ISDIR(directory_stat.st_mode)) { + errno = ENOTDIR; + return -1; + } + if (directory_stat.st_uid != getuid() || (directory_stat.st_mode & S_IWOTH) != 0U) { + errno = EACCES; + return -1; + } + return 0; +} + +/* Unlink a dead socket (ECONNREFUSED probe); a live one aborts startup. */ +static int remove_stale_socket(const char *socket_path) { + struct stat socket_stat; + struct stat current_socket_stat; + struct sockaddr_un address; + int probe_fd; + int connect_error; + if (lstat(socket_path, &socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(socket_stat.st_mode) || socket_stat.st_uid != getuid()) { + errno = EACCES; + return -1; + } + probe_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (probe_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, strlen(socket_path) + 1U); + if (connect(probe_fd, (struct sockaddr*)&address, sizeof(address)) == 0) { + close(probe_fd); + errno = EADDRINUSE; + return -1; + } + connect_error = errno; + close(probe_fd); + if (connect_error == ENOENT) { + return 0; + } + if (connect_error != ECONNREFUSED) { + errno = connect_error; + return -1; + } + if (lstat(socket_path, &current_socket_stat) != 0 || !S_ISSOCK(current_socket_stat.st_mode) || + current_socket_stat.st_dev != socket_stat.st_dev || current_socket_stat.st_ino != socket_stat.st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path) == 0 || errno == ENOENT ? 0 : -1; +} + +/* Unlink our socket at exit only if (dev,ino) still matches what we bound. */ +static int unlink_owned_socket(const char *socket_path, const struct stat *bound_socket_stat) { + struct stat current_socket_stat; + if (lstat(socket_path, &current_socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(current_socket_stat.st_mode) || current_socket_stat.st_dev != bound_socket_stat->st_dev || + current_socket_stat.st_ino != bound_socket_stat->st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path); +} + +/* Listener cleanup path shared between fatal startup errors. */ +static int discard_listener(int listener_fd, const char *socket_path, const struct stat *bound_socket_stat, + int saved_errno) { + close(listener_fd); + if (bound_socket_stat != NULL) { + (void)unlink_owned_socket(socket_path, bound_socket_stat); + } + errno = saved_errno; + return -1; +} + +/* bind()+listen() the server socket, chmod 0600, record (dev,ino) identity. */ +static int create_listener(const char *socket_path, struct stat *bound_socket_stat) { + struct sockaddr_un address; + const size_t path_length = strlen(socket_path); + int listener_fd; + listener_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (listener_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, path_length + 1U); + if (bind(listener_fd, (struct sockaddr*)&address, + offsetof(struct sockaddr_un, sun_path) + path_length + 1U) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (lstat(socket_path, bound_socket_stat) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (!S_ISSOCK(bound_socket_stat->st_mode) || bound_socket_stat->st_uid != getuid()) { + return discard_listener(listener_fd, socket_path, NULL, EACCES); + } + if (chmod(socket_path, 0600) != 0 || listen(listener_fd, 8) != 0) { + return discard_listener(listener_fd, socket_path, bound_socket_stat, errno); + } + return listener_fd; +} + +/* Print the --help text. */ +static void print_usage(FILE *output, const char *program) { + fprintf(output, + "Usage: %s --device DEVICE [--size-gib GIB | --size-bytes BYTES] [--socket-dir DIR] [--socket-name NAME] [--skip-tgid-check] [--log-level LEVEL]\n" + "Serve Famem physical HBM for one Ascend NPU.\n", + program); +} + +/* Map --log-level names to the enum. */ +static int parse_log_level(const char *value, enum log_level *output) { + static const char *const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + size_t index; + for (index = 0; index < sizeof(names) / sizeof(names[0]); ++index) { + if (strcmp(value, names[index]) == 0) { + *output = (enum log_level)index; + return 0; + } + } + return -1; +} + +/* Parse --device as a non-negative integer. */ +static int parse_device(const char *value, int *output) { + char *end = NULL; + long device; + errno = 0; + device = strtol(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || device < 0 || device > INT_MAX) { + return -1; + } + *output = (int)device; + return 0; +} + +/* Parse --pool-size in GiB (0 = defer to first ACQUIRE). */ +static int parse_pool_size(const char *value, uint64_t *output) { + char *end = NULL; + unsigned long long size_gib; + errno = 0; + size_gib = strtoull(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || size_gib == 0U || size_gib > FAMEM_MAX_POOL_GIB) { + return -1; + } + *output = (uint64_t)size_gib * FAMEM_HUGE_1G_BYTES; + return 0; +} + +/* Entry point: parse flags, bind the socket, serve until SIGTERM, reclaim everything. */ +int main(int argc, char **argv) { + static const struct option options[] = { + {"device", required_argument, NULL, 'd'}, + {"size-gib", required_argument, NULL, 'm'}, + {"size-bytes", required_argument, NULL, 'b'}, + {"socket-dir", required_argument, NULL, 's'}, + {"socket-name", required_argument, NULL, 'n'}, + {"skip-tgid-check", no_argument, NULL, 'k'}, + {"log-level", required_argument, NULL, 'l'}, + {"help", no_argument, NULL, 'h'}, + {NULL, 0, NULL, 0}, + }; + const char *socket_dir = NULL; + const char *socket_name = NULL; + char socket_path[PATH_MAX] = {0}; + struct sigaction action; + struct stat bound_socket_stat; + struct server_state server; + int device = -1; + int listener_fd = -1; + int socket_dir_fd = -1; + uint64_t pool_size = 0U; + int option; + int exit_code = EXIT_FAILURE; + while ((option = getopt_long(argc, argv, "d:m:b:s:n:kl:h", options, NULL)) != -1) { + switch (option) { + case 'd': + if (parse_device(optarg, &device) != 0) { + fprintf(stderr, "Invalid --device value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'm': + if (parse_pool_size(optarg, &pool_size) != 0) { + fprintf(stderr, "Invalid --size-gib value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'b': { + char *end = NULL; + errno = 0; + pool_size = strtoull(optarg, &end, 10); + if (errno != 0 || end == optarg || *end != '\0' || pool_size == 0U) { + fprintf(stderr, "Invalid --size-bytes value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + } + case 's': + socket_dir = optarg; + break; + case 'n': + socket_name = optarg; + break; + case 'k': + /* --skip-tgid-check: skip PID/TGID verification (needed in containers + where the PID namespace differs from the host) */ + g_skip_tgid_check = 1; + break; + case 'l': + if (parse_log_level(optarg, &g_log_level) != 0) { + fprintf(stderr, "Invalid --log-level value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'h': + print_usage(stdout, argv[0]); + return EXIT_SUCCESS; + default: + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + } + if (device < 0 || optind != argc) { + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + prctl(PR_SET_NAME, "hbm_server", 0, 0, 0); + /* Pool size will be determined by the first ACQUIRE request when not + explicitly set via --size-gib / --size-bytes. This allows a single + server binary to serve models with different per-NPU arena sizes + (e.g. 27 GiB for TP1-7B, 54 GiB for TP4-72B) without changing the + launch command. */ + server.pool_size = pool_size; /* 0 means "defer to first ACQUIRE" */ + if (socket_dir == NULL) { + socket_dir = getenv("VLLM_ASCEND_FAMEM_SOCKET_DIR"); + } + if (socket_dir == NULL || socket_dir[0] == '\0') { + socket_dir = FAMEM_DEFAULT_SOCKET_DIR; + } + memset(&action, 0, sizeof(action)); + action.sa_handler = stop_server; + sigemptyset(&action.sa_mask); + if (sigaction(SIGTERM, &action, NULL) != 0 || sigaction(SIGINT, &action, NULL) != 0) { + log_message(LOG_ERROR, "cannot install signal handlers: %s", strerror(errno)); + return EXIT_FAILURE; + } + signal(SIGPIPE, SIG_IGN); + if (famem_server_initialize(device) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem backend: %s", native_error()); + return EXIT_FAILURE; + } + memset(&server, 0, sizeof(server)); + server.device = device; + if (famem_get_device_uuid(device, server.device_uuid, sizeof(server.device_uuid)) != 0 || + !is_lower_hex((const uint8_t*)server.device_uuid, FAMEM_UUID_SIZE) || + is_all_zero((const uint8_t*)server.device_uuid, FAMEM_UUID_SIZE)) { + /* Some NPU models (e.g. Ascend 910B3) do not provide a valid UUID via + aclrtDeviceGetUuid (returns ACL_ERROR_RT_DEVICE_UNAVAILABLE or an + all-zero UUID). In that case, synthesize a deterministic UUID from + the device index so that the server can still operate. */ + log_message(LOG_WARNING, + "NPU UUID not available (device %d), synthesizing from device index", + device); + option = snprintf(server.device_uuid, sizeof(server.device_uuid), + "d%031d", device); + if (option < 0 || (size_t)option != FAMEM_UUID_SIZE) { + log_message(LOG_ERROR, "cannot synthesize device UUID"); + goto cleanup_backend; + } + } + server.device_uuid[FAMEM_UUID_SIZE] = '\0'; + if (pthread_mutex_init(&server.mutex, NULL) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem server mutex"); + goto cleanup_backend; + } + if (make_directories(socket_dir) != 0) { + log_message(LOG_ERROR, "cannot prepare socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + socket_dir_fd = open(socket_dir, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + if (socket_dir_fd < 0 || flock(socket_dir_fd, LOCK_EX) != 0) { + log_message(LOG_ERROR, "cannot lock Famem socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + /* Build socket path: use --socket-name if provided, otherwise device_uuid */ + if (socket_name != NULL) { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, socket_name); + } else { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, server.device_uuid); + } + if (option < 0 || (size_t)option >= sizeof(socket_path) || (size_t)option > FAMEM_SOCKET_PATH_MAX) { + errno = ENAMETOOLONG; + log_message(LOG_ERROR, "Famem socket path is too long"); + goto cleanup_pool; + } + if (remove_stale_socket(socket_path) != 0) { + log_message(LOG_ERROR, "cannot prepare Famem socket %s: %s", socket_path, strerror(errno)); + goto cleanup_pool; + } + if (pool_size != 0U && initialize_pool(&server, pool_size) != 0) { + goto cleanup_pool; + } + if (pool_size == 0U) { + log_message(LOG_INFO, "pool size not specified; will be determined by the first ACQUIRE request"); + } + listener_fd = create_listener(socket_path, &bound_socket_stat); + option = errno; + /* When --socket-name overrides the socket filename, create a UUID-named + hard link so that clients using famem_get_device_uuid() can find the socket. + A hard link (not symlink) is required because the client checks S_ISSOCK + with follow_symlinks=False, and only a hard link preserves the socket mode. */ + if (socket_name != NULL) { + char uuid_sock[PATH_MAX]; + option = snprintf(uuid_sock, sizeof(uuid_sock), "%s/%s.sock", socket_dir, server.device_uuid); + if (option > 0 && (size_t)option < sizeof(uuid_sock)) { + (void)unlink(uuid_sock); + if (link(socket_path, uuid_sock) != 0) { + log_message(LOG_WARNING, "cannot create UUID hardlink %s: %s", uuid_sock, strerror(errno)); + } + } + } + close(socket_dir_fd); + socket_dir_fd = -1; + errno = option; + if (listener_fd < 0) { + log_message(LOG_ERROR, "cannot create Famem listener: %s", strerror(errno)); + goto cleanup_pool; + } + log_message(LOG_INFO, "pool ready; listening on %s for NPU UUID %s", socket_path, server.device_uuid); + + while (!g_stop) { + struct client_context *client; + int client_fd = accept(listener_fd, NULL, NULL); + if (client_fd < 0) { + if (errno == EINTR) { + continue; + } + log_message(LOG_ERROR, "accept failed: %s", strerror(errno)); + goto cleanup; + } + client = (struct client_context*)calloc(1U, sizeof(*client)); + if (client == NULL) { + log_message(LOG_ERROR, "cannot allocate Famem client context"); + close(client_fd); + continue; + } + client->socket_fd = client_fd; + client->server = &server; + pthread_mutex_lock(&server.mutex); + client->next = server.clients; + server.clients = client; + pthread_mutex_unlock(&server.mutex); + if (pthread_create(&client->thread, NULL, client_thread_main, client) != 0) { + pthread_mutex_lock(&server.mutex); + server.clients = client->next; + pthread_mutex_unlock(&server.mutex); + log_message(LOG_ERROR, "cannot start Famem client thread"); + close(client_fd); + free(client); + continue; + } + reap_finished_clients(&server); + } + exit_code = EXIT_SUCCESS; + +cleanup: + close(listener_fd); + shutdown_clients(&server); + if (unlink_owned_socket(socket_path, &bound_socket_stat) != 0 && errno != ENOENT) { + log_message(LOG_WARNING, "cannot remove owned socket %s: %s", socket_path, strerror(errno)); + } +cleanup_pool: + if (free_extents(server.pool_extents, server.pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot release every persistent pool extent"); + exit_code = EXIT_FAILURE; + } +cleanup_mutex: + if (socket_dir_fd >= 0) { + close(socket_dir_fd); + } + (void)pthread_mutex_destroy(&server.mutex); +cleanup_backend: + if (famem_server_finalize() != 0) { + log_message(LOG_ERROR, "cannot finalize Famem backend: %s", native_error()); + exit_code = EXIT_FAILURE; + } + return exit_code; +} -- 2.43.0

2 1

[PATCH openEuler-26.09] xsched/famem: introduce lightweight NPU HBM management infrastructure FaMem
by Gu Bowen 11 Nov '26

11 Nov '26

hulk inclusion category: feature bugzilla: https://gitcode.com/openeuler/kernel/issues/10313 -------------------------------- FaMem (Fast-Map Memory) provides NPU HBM management for multi-model time-division multiplexing. It establishes a stable virtual address Arena per NPU to hold model weights and KV Cache; a native HBM server owns the physical pages and manages cross-model leases. Suspend releases physical HBM while preserving virtual addresses; resume uses a dedicated Copier process to restore weights layer-by-layer (H2D), overlapping weight loading with inference computation to reduce model switching latency. Signed-off-by: Alekseev Dmitry <alekseev.dmitry(a)huawei.com> Signed-off-by: Meskhidze Konstantin <konstantin.meskhidze(a)huawei.com> Signed-off-by: Mingrui Liu <liumingrui(a)huawei.com> Signed-off-by: Gu Bowen <gubowen5(a)huawei.com> --- arch/arm64/configs/openeuler_defconfig | 1 + kernel/xsched/Kconfig | 18 + kernel/xsched/Makefile | 4 + kernel/xsched/famem_hbm_server.c | 1418 ++++++++++++++++++++++++ 4 files changed, 1441 insertions(+) create mode 100644 kernel/xsched/famem_hbm_server.c diff --git a/arch/arm64/configs/openeuler_defconfig b/arch/arm64/configs/openeuler_defconfig index 1561eb379cc4..8e791c988848 100644 --- a/arch/arm64/configs/openeuler_defconfig +++ b/arch/arm64/configs/openeuler_defconfig @@ -104,6 +104,7 @@ CONFIG_XSCHED_NR_CUS=128 CONFIG_XCU_SCHED_RT=y CONFIG_XCU_SCHED_CFS=y CONFIG_CGROUP_XCU=y +CONFIG_FAMEM=n # # CPU/Task time and stats accounting diff --git a/kernel/xsched/Kconfig b/kernel/xsched/Kconfig index df6576da88fd..805b04de68f8 100644 --- a/kernel/xsched/Kconfig +++ b/kernel/xsched/Kconfig @@ -91,3 +91,21 @@ config CGROUP_XCU Required for systems requiring fine-grained resource control in cgroups. If unsure, say N. + +config FAMEM + bool "Enable FaMem (Fast-Map Memory) NPU HBM management" + default n + help + This option enables the FaMem (Fast-Map Memory) feature for managing + NPU HBM (High Bandwidth Memory) across multiple model instances. + + FaMem establishes a stable virtual address Arena per NPU to hold model + weights and KV Cache. A native HBM server owns the physical pages and + manages cross-model leases. Suspend releases physical HBM while + preserving virtual addresses; resume uses a dedicated Copier process to + restore weights layer-by-layer (H2D), overlapping weight loading with + inference computation to significantly reduce model-switching latency + and enable time-division multiplexing of HBM across model instances. + + This code is userspace-oriented and is isolated from the kernel build + by this config option. If unsure, say N. diff --git a/kernel/xsched/Makefile b/kernel/xsched/Makefile index 3e23012ea298..ae19dcc6a86e 100644 --- a/kernel/xsched/Makefile +++ b/kernel/xsched/Makefile @@ -8,3 +8,7 @@ obj-$(CONFIG_XCU_SCHED_CFS) += cfs.o cfs_quota.o obj-$(CONFIG_CGROUP_XCU) += cgroup.o obj-$(CONFIG_CGROUP_DMEM) += dmem.o endif + +ifdef CONFIG_FAMEM +obj-y += famem_hbm_server.o +endif diff --git a/kernel/xsched/famem_hbm_server.c b/kernel/xsched/famem_hbm_server.c new file mode 100644 index 000000000000..dd5e895ec500 --- /dev/null +++ b/kernel/xsched/famem_hbm_server.c @@ -0,0 +1,1418 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * SPDX-FileCopyrightText: Copyright (c) 2026 Huawei Technologies Co., Ltd. All Rights Reserved. + * + * famem_hbm_server.c — Native Famem HBM server for one Ascend NPU. + * + * This binary owns the physical HBM extents of a single device and serves + * the Famem control protocol (famem_protocol.h) on a Unix stream socket + * named after the device UUID. Clients (vLLM Worker processes and, for + * handle authorization, their Copier subprocesses) connect, claim a lease + * and drive the sleep/wake cycle; see FamemClient in + * vllm_ascend/device_allocator/famem_client.py for the client side. + * + * Server model + * ------------ + * - One thread per client connection; a mutex (server_state.mutex) + * serializes all state transitions. + * - One pool of at most FAMEM_MAX_EXTENTS physical extents, sized either + * by --pool-size at startup or by the first ACQUIRE request (pool_size + * == 0 defers sizing). Later ACQUIRE requests must request the exact + * same size or they are rejected — multi-instance time-sharing + * therefore requires identical arena sizes across instances. + * - At most one ACTIVE lease per NPU ("active" = extents currently + * allocated and mapped). Other connected leases may exist in the + * sleeping state; ACQUIRE/WAKE from a second lease gets + * FAMEM_STATUS_BUSY until the active lease SLEEPs or RELEASEs. + * + * Lease lifecycle + * --------------- + * HELLO create lease; record Worker + Copier identities (pidfd or + * pid+start_time) for liveness checking + * ACQUIRE allocate pool extents; authorize shareable handles for the + * Worker and Copier TGIDs; lease becomes active + * SLEEP free the physical extents (clients must have unmapped first); + * lease stays; epoch unchanged on the wire (client keeps it) + * WAKE re-allocate the same extent layout; epoch += 1; fresh + * shareable handles are exported and authorized + * RELEASE destroy the lease (frees extents if still active) + * + * Orphan reclamation + * ------------------ + * If a client disconnects while its lease is still active, the server + * blocks in the connection thread until the Worker and Copier processes + * exit (pidfd poll, or pid liveness + start_time comparison when pidfd + * is unavailable), then frees the extents. This guarantees CANN + * physical memory is never freed while a mapping in another process + * still references it. + * + * Poisoning + * --------- + * Any failure to free or allocate physical extents marks the server + * poisoned: it refuses new leases (FAMEM_STATUS_INTERNAL) until + * restarted, because the physical pool state is no longer trustworthy. + * + * Flags: --device (required), --socket-dir, --socket-name, --pool-size, + * --log-level, --skip-tgid-check (containers without SO_PEERCRED). + */ + +#define _GNU_SOURCE + +#include <errno.h> +#include <fcntl.h> +#include <getopt.h> +#include <limits.h> +#include <poll.h> +#include <pthread.h> +#include <signal.h> +#include <stdbool.h> +#include <stdarg.h> +#include <stddef.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/file.h> +#include <sys/prctl.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/syscall.h> +#include <sys/types.h> +#include <sys/un.h> +#include <time.h> +#include <unistd.h> + +#define FAMEM_DEFAULT_SOCKET_DIR "/run/vllm-ascend/famem" +#define FAMEM_SOCKET_PATH_MAX 103U +#define FAMEM_UUID_SIZE 32U +#define FAMEM_SESSION_SIZE 32U +#define FAMEM_MAX_EXTENTS 2U +#define FAMEM_PAGE_HUGE_1G 1 +#define FAMEM_PAGE_HUGE_2M 2 +#define FAMEM_HUGE_1G_BYTES (UINT64_C(1) << 30) +#define FAMEM_HUGE_2M_BYTES (UINT64_C(2) << 20) +#define FAMEM_MAX_POOL_GIB UINT64_C(4096) + +enum log_level { + LOG_DEBUG = 0, + LOG_INFO = 1, + LOG_WARNING = 2, + LOG_ERROR = 3, +}; + +struct request_frame { + uint16_t operation; + uint32_t request_id; + uint32_t payload_size; + uint8_t payload[FAMEM_PROTOCOL_MAX_PAYLOAD]; +}; +struct physical_extent { + int page_type; + uint64_t size; + uint64_t physical_handle; + uint64_t shareable_handle; +}; +struct process_identity { + pid_t pid; + int pidfd; + uint64_t start_time; + bool has_start_time; +}; +struct lease { + bool acquired; + bool active; + char session_id[FAMEM_SESSION_SIZE + 1U]; + struct process_identity worker_process; + struct process_identity copier_process; + int32_t bare_tgid; + int32_t copier_bare_tgid; + uint64_t epoch; +}; +struct server_state; +struct client_context { + int socket_fd; + pthread_t thread; + bool thread_finished; + struct lease lease; + struct server_state *server; + struct client_context *next; +}; +struct server_state { + int device; + char device_uuid[FAMEM_UUID_SIZE + 1U]; + pthread_mutex_t mutex; + struct client_context *clients; + struct lease *active_lease; + uint64_t pool_size; + uint64_t epoch; + size_t pool_extent_count; + struct physical_extent pool_extents[FAMEM_MAX_EXTENTS]; + bool poisoned; +}; + +static volatile sig_atomic_t g_stop; +static enum log_level g_log_level = LOG_INFO; +static int g_skip_tgid_check; + +/* Level-filtered stderr logger with a fixed [Famem HBM Server] prefix. */ +static void log_message(enum log_level level, const char *format, ...) +{ + static const char * const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + va_list arguments; + if (level < g_log_level) { + return; + } + fprintf(stderr, "[Famem HBM Server] %s: ", names[level]); + va_start(arguments, format); + vfprintf(stderr, format, arguments); + va_end(arguments); + fputc('\n', stderr); +} + +/* SIGTERM/SIGINT handler: set the global stop flag (checked by accept/poll loops). */ +static void stop_server(int signal_number) +{ + (void)signal_number; + g_stop = 1; +} + +/* Read one big-endian uint16 (wire protocol; see famem_protocol.h). */ +static uint16_t load_u16(const uint8_t *input) { return (uint16_t)(((uint16_t)input[0] << 8U) | (uint16_t)input[1]); } +/* Read one big-endian uint32. */ +static uint32_t load_u32(const uint8_t *input) { + return ((uint32_t)input[0] << 24U) | ((uint32_t)input[1] << 16U) | ((uint32_t)input[2] << 8U) | (uint32_t)input[3]; +} +/* Read one big-endian uint64. */ +static uint64_t load_u64(const uint8_t *input) { + return ((uint64_t)load_u32(input) << 32U) | (uint64_t)load_u32(input + 4U); +} +/* Write one big-endian uint16. */ +static void store_u16(uint8_t *output, uint16_t value) +{ + output[0] = (uint8_t)(value >> 8U); + output[1] = (uint8_t)value; +} +/* Write one big-endian uint32. */ +static void store_u32(uint8_t *output, uint32_t value) +{ + output[0] = (uint8_t)(value >> 24U); + output[1] = (uint8_t)(value >> 16U); + output[2] = (uint8_t)(value >> 8U); + output[3] = (uint8_t)value; +} +/* Write one big-endian uint64. */ +static void store_u64(uint8_t *output, uint64_t value) +{ + store_u32(output, (uint32_t)(value >> 32U)); + store_u32(output + 4U, (uint32_t)value); +} + +/* Read exactly *size* bytes; *clean_eof* distinguishes EOF-before-any-data from a mid-frame close. */ +static int receive_exact(int socket_fd, uint8_t *output, size_t size, bool *clean_eof) +{ + size_t offset = 0; + *clean_eof = false; + while (offset < size) { + const ssize_t received = recv(socket_fd, output + offset, size - offset, 0); + if (received > 0) { + offset += (size_t)received; + continue; + } + if (received == 0) { + *clean_eof = offset == 0; + return -1; + } + if (errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Write exactly *size* bytes; retries on EINTR, fails on short writes. */ +static int send_exact(int socket_fd, const uint8_t *input, size_t size) +{ + size_t offset = 0; + while (offset < size) { + const ssize_t sent = send(socket_fd, input + offset, size - offset, MSG_NOSIGNAL); + if (sent > 0) { + offset += (size_t)sent; + continue; + } + if (sent < 0 && errno == EINTR && !g_stop) { + continue; + } + return -1; + } + return 0; +} +/* Read + validate one request frame into *request*; on error fills *error* and returns -1. */ +static int receive_request(int socket_fd, struct request_frame *request, char *error, size_t error_size) +{ + uint8_t header[FAMEM_PROTOCOL_HEADER_SIZE]; + bool clean_eof = false; + + memset(request, 0, sizeof(*request)); + if (receive_exact(socket_fd, header, sizeof(header), &clean_eof) != 0) { + if (clean_eof) { + return 0; + } + snprintf(error, error_size, "connection closed while reading a frame header"); + return -1; + } + request->operation = load_u16(header + 8U); + request->request_id = load_u32(header + 12U); + request->payload_size = load_u32(header + 20U); + + if (load_u32(header) != FAMEM_PROTOCOL_MAGIC || load_u16(header + 4U) != FAMEM_PROTOCOL_VERSION || + load_u16(header + 6U) != FAMEM_MESSAGE_REQUEST || load_u16(header + 10U) != 0U || + load_u32(header + 16U) != FAMEM_STATUS_OK || request->operation < FAMEM_OP_HELLO || + request->operation > FAMEM_OP_STATS || request->request_id == 0U || + request->payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + snprintf(error, error_size, "invalid protocol header"); + return -2; + } + if (request->operation > FAMEM_OP_STATS) { + snprintf(error, error_size, "unsupported operation %u", request->operation); + return -2; + } + if (request->payload_size != 0U && + receive_exact(socket_fd, request->payload, request->payload_size, &clean_eof) != 0) { + snprintf(error, error_size, "connection closed while reading a frame payload"); + return -1; + } + return 1; +} +/* Frame and send one response (payload optional). */ +static int send_response(int socket_fd, uint16_t operation, uint32_t request_id, uint32_t status, + const uint8_t *payload, uint32_t payload_size) { + uint8_t frame[FAMEM_PROTOCOL_HEADER_SIZE + FAMEM_PROTOCOL_MAX_PAYLOAD]; + if (payload_size > FAMEM_PROTOCOL_MAX_PAYLOAD) { + errno = EMSGSIZE; + return -1; + } + store_u32(frame, FAMEM_PROTOCOL_MAGIC); + store_u16(frame + 4U, FAMEM_PROTOCOL_VERSION); + store_u16(frame + 6U, FAMEM_MESSAGE_RESPONSE); + store_u16(frame + 8U, operation); + store_u16(frame + 10U, 0U); + store_u32(frame + 12U, request_id); + store_u32(frame + 16U, status); + store_u32(frame + 20U, payload_size); + if (payload_size != 0U) { + memcpy(frame + FAMEM_PROTOCOL_HEADER_SIZE, payload, payload_size); + } + return send_exact(socket_fd, frame, FAMEM_PROTOCOL_HEADER_SIZE + payload_size); +} +/* send_response wrapper encoding a UTF-8 error message payload. */ +static int send_error_response(int socket_fd, const struct request_frame *request, uint32_t status, + const char *message) { + const size_t message_size = strnlen(message, FAMEM_PROTOCOL_MAX_PAYLOAD); + return send_response(socket_fd, request->operation, request->request_id, status, (const uint8_t*)message, + (uint32_t)message_size); +} +/* Send an error response and log the rejection at WARNING. */ +static int reject_client(int socket_fd, const struct request_frame *request, uint32_t status, const char *message) +{ + (void)send_error_response(socket_fd, request, status, message); + return -1; +} + +/* True when every byte is a lowercase hex digit (session/UUID validation). */ +static bool is_lower_hex(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (!((value[index] >= (uint8_t)'0' && value[index] <= (uint8_t)'9') || + (value[index] >= (uint8_t)'a' && value[index] <= (uint8_t)'f'))) { + return false; + } + } + return true; +} + +/* True when the buffer is all zero bytes (UUID-unsupported detection). */ +static bool is_all_zero(const uint8_t *value, size_t size) +{ + size_t index; + for (index = 0; index < size; ++index) { + if (value[index] != 0) { + return false; + } + } + return true; +} + +/* Parse /proc/<pid>/stat for the start time (field 22) and parent PID (field 4). */ +static bool read_process_stat(pid_t pid, uint64_t *start_time, pid_t *parent_pid) +{ + char path[64]; + char buffer[4096]; + char *closing_parenthesis; + char *save_pointer = NULL; + char *token; + unsigned int field = 4U; + FILE *process_stat; + char *end = NULL; + unsigned long long parsed; + + if (pid <= 0 || start_time == NULL || snprintf(path, sizeof(path), "/proc/%ld/stat", (long)pid) < 0) { + return false; + } + process_stat = fopen(path, "r"); + if (process_stat == NULL) { + return false; + } + if (fgets(buffer, sizeof(buffer), process_stat) == NULL) { + (void)fclose(process_stat); + return false; + } + (void)fclose(process_stat); + closing_parenthesis = strrchr(buffer, ')'); + if (closing_parenthesis == NULL) { + return false; + } + token = strtok_r(closing_parenthesis + 1, " \t\r\n", &save_pointer); + token = token == NULL ? NULL : strtok_r(NULL, " \t\r\n", &save_pointer); + if (token == NULL) { + return false; + } + if (parent_pid != NULL) { + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0' || parsed > INT_MAX) { + return false; + } + *parent_pid = (pid_t)parsed; + } + while (token != NULL && field < 22U) { + token = strtok_r(NULL, " \t\r\n", &save_pointer); + ++field; + } + if (token == NULL || field != 22U) { + return false; + } + errno = 0; + parsed = strtoull(token, &end, 10); + if (errno != 0 || end == token || *end != '\0') { + return false; + } + *start_time = (uint64_t)parsed; + return true; +} + +/* Open a pidfd for *pid* (0 on kernels without pidfd_open). */ +static int open_process_pidfd(pid_t pid) +{ +#ifdef SYS_pidfd_open + const long descriptor = syscall(SYS_pidfd_open, pid, 0U); + if (descriptor >= 0 && descriptor <= INT_MAX) { + return (int)descriptor; + } +#else + (void)pid; +#endif + return -1; +} +/* Snapshot (pid, pidfd, start_time) for liveness tracking of a lease peer. */ +static void capture_process_identity(struct process_identity *identity, pid_t pid, pid_t *parent_pid) +{ + memset(identity, 0, sizeof(*identity)); + identity->pid = pid; + identity->pidfd = open_process_pidfd(pid); + identity->has_start_time = read_process_stat(pid, &identity->start_time, parent_pid); +} +/* Liveness check: pidfd poll or pid signal-probe plus start_time comparison (PID reuse guard). */ +static bool process_exists(const struct process_identity *identity) +{ + uint64_t current_start_time = 0U; + if (identity->pid <= 0) { + return false; + } + if (kill(identity->pid, 0) == 0) { + if (identity->has_start_time && read_process_stat(identity->pid, &current_start_time, NULL) && + current_start_time != identity->start_time) { + return false; + } + return true; + } + return errno == EPERM; +} +/* Block until the tracked process exits (pidfd poll or 200ms polling loop). */ +static void wait_for_process_exit(const struct process_identity *identity) { + struct timespec interval = {.tv_sec = 0, .tv_nsec = 200000000L}; + if (identity->pid <= 0) { + return; + } + if (identity->pidfd >= 0) { + struct pollfd descriptor = {.fd = identity->pidfd, .events = POLLIN | POLLHUP | POLLERR, .revents = 0}; + while (poll(&descriptor, 1U, -1) < 0) { + if (errno != EINTR) { + break; + } + } + if (descriptor.revents != 0) { + return; + } + } + while (process_exists(identity)) { + (void)nanosleep(&interval, NULL); + } +} +/* Close one identity's pidfd. */ +static void close_process_identity(struct process_identity *identity) { + if (identity->pid > 0 && identity->pidfd >= 0) { + (void)close(identity->pidfd); + } + memset(identity, 0, sizeof(*identity)); + identity->pidfd = -1; +} +/* Close the Worker and Copier identity pidfds of one lease. */ +static void close_lease_process_identities(struct lease *lease) { + close_process_identity(&lease->worker_process); + close_process_identity(&lease->copier_process); +} + +/* Famem last_error() string, or a placeholder when the library is silent. */ +static const char *native_error(void) { + const char *error = famem_last_error(); + return error == NULL || error[0] == '\0' ? "unknown native allocator error" : error; +} + +/* Release N physical extents via famem_server_free; any failure poisons the server. */ +static int free_extents(struct physical_extent *extents, size_t extent_count) { + int result = 0; + while (extent_count > 0U) { + struct physical_extent *extent = &extents[extent_count - 1U]; + if (extent->physical_handle != 0U) { + if (famem_server_free(extent->physical_handle) != 0) { + log_message(LOG_ERROR, "failed to free physical extent: %s", native_error()); + result = -1; + } else { + extent->physical_handle = 0U; + extent->shareable_handle = 0U; + } + } + --extent_count; + } + return result; +} + +enum { EXTENT_ALLOCATION_ROLLBACK_FAILED = -2 }; +/* Allocate N extents and authorize the lease's TGIDs on each shareable handle. */ +static int allocate_extents(int device, struct physical_extent *extents, size_t extent_count) { + size_t index; + for (index = 0; index < extent_count; ++index) { + if (famem_server_allocate_export(device, extents[index].size, extents[index].page_type, + &extents[index].physical_handle, &extents[index].shareable_handle) != 0) { + log_message(LOG_WARNING, "failed to allocate extent %zu: %s", index, native_error()); + if (free_extents(extents, index) != 0) { + return EXTENT_ALLOCATION_ROLLBACK_FAILED; + } + return -1; + } + } + return 0; +} + +/* (Re)authorize both lease TGIDs on every pool shareable handle (called after WAKE too). */ +static int authorize_pool(struct server_state *server, const struct lease *lease) { + const int32_t bare_tgids[] = {lease->bare_tgid, lease->copier_bare_tgid}; + size_t index; + for (index = 0; index < server->pool_extent_count; ++index) { + if (famem_server_authorize(server->pool_extents[index].shareable_handle, bare_tgids, + sizeof(bare_tgids) / sizeof(bare_tgids[0])) != 0) { + log_message(LOG_ERROR, "failed to authorize extent %zu: %s", index, native_error()); + server->poisoned = true; + log_message(LOG_ERROR, "pool authorization state is uncertain; server poisoned"); + return -1; + } + } + return 0; +} + +/* Choose the extent layout: fill with 1G pages up to availability, remainder 2M pages. */ +static int plan_extents(int device, uint64_t size, struct physical_extent *extents, size_t *extent_count) { + uint64_t small_granularity = 0U; + uint64_t large_granularity = 0U; + uint64_t free_large = 0U; + uint64_t large_size = 0U; + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_2M, &small_granularity) != 0 || + small_granularity != FAMEM_HUGE_2M_BYTES) { + return -1; + } + if (size == 0U || size % small_granularity != 0U) { + return -1; + } + if (famem_get_page_granularity(device, FAMEM_PAGE_HUGE_1G, &large_granularity) == 0 && + large_granularity == FAMEM_HUGE_1G_BYTES && + famem_get_page_free_memory(device, FAMEM_PAGE_HUGE_1G, &free_large) == 0) { + large_size = size < free_large ? size : free_large; + large_size -= large_size % large_granularity; + } else { + log_message(LOG_WARNING, "1 GiB pages are unavailable; using 2 MiB pages only: %s", native_error()); + } + memset(extents, 0, sizeof(*extents) * FAMEM_MAX_EXTENTS); + *extent_count = 0U; + if (large_size != 0U) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_1G; + extents[*extent_count].size = large_size; + ++(*extent_count); + } + if (large_size != size) { + extents[*extent_count].page_type = FAMEM_PAGE_HUGE_2M; + extents[*extent_count].size = size - large_size; + ++(*extent_count); + } + return 0; +} + +/* Allocate the pool extents once (first ACQUIRE) and fix server->pool_size. */ +static int initialize_pool(struct server_state *server, uint64_t size) { + int result; + if (plan_extents(server->device, size, server->pool_extents, &server->pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot plan the physical pool layout: %s", native_error()); + return -1; + } + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + if (result == -1 && server->pool_extents[0].page_type == FAMEM_PAGE_HUGE_1G) { + log_message(LOG_WARNING, "preferred pool layout failed; retrying with 2 MiB pages"); + memset(server->pool_extents, 0, sizeof(server->pool_extents)); + server->pool_extents[0].page_type = FAMEM_PAGE_HUGE_2M; + server->pool_extents[0].size = size; + server->pool_extent_count = 1U; + result = allocate_extents(server->device, server->pool_extents, server->pool_extent_count); + } + if (result != 0) { + log_message(LOG_ERROR, "cannot allocate the physical pool%s", + result == EXTENT_ALLOCATION_ROLLBACK_FAILED ? "; rollback left HBM allocated" : ""); + return -1; + } + server->pool_size = size; + log_message(LOG_INFO, "allocated persistent pool of %llu bytes in %zu extent(s)", (unsigned long long)size, + server->pool_extent_count); + return 0; +} + +/* ACQUIRE/WAKE success response: extent page types, sizes, and shareable handles. */ +static int send_mapping_response(int socket_fd, const struct request_frame *request, const struct server_state *server, + const struct lease *lease) { + uint8_t payload[FAMEM_MAPPING_PREFIX_SIZE + FAMEM_MAX_EXTENTS * FAMEM_MAPPING_EXTENT_SIZE]; + size_t offset = FAMEM_MAPPING_PREFIX_SIZE; + size_t index; + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, (uint32_t)server->pool_extent_count); + for (index = 0; index < server->pool_extent_count; ++index) { + store_u32(payload + offset, (uint32_t)server->pool_extents[index].page_type); + store_u64(payload + offset + 4U, server->pool_extents[index].size); + store_u64(payload + offset + 12U, server->pool_extents[index].shareable_handle); + offset += FAMEM_MAPPING_EXTENT_SIZE; + } + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, (uint32_t)offset); +} + +/* Mark *lease *the single active lease and authorize its TGIDs on the pool. */ +static int activate_lease(struct server_state *server, struct lease *lease) { + if (server->epoch == UINT64_MAX) { + server->poisoned = true; + } + if (server->poisoned) { + log_message(LOG_DEBUG, "activate_lease: server is poisoned"); + return -1; + } + if (!g_skip_tgid_check && !process_exists(&lease->worker_process)) { + log_message(LOG_DEBUG, "activate_lease: worker process (pid=%d) does not exist", lease->worker_process.pid); + return -1; + } + if (authorize_pool(server, lease) != 0) { + log_message(LOG_DEBUG, "activate_lease: authorize_pool failed"); + return -1; + } + lease->epoch = ++server->epoch; + lease->active = true; + server->active_lease = lease; + return 0; +} + +/* ACQUIRE: fix pool size on first use, verify capacity match, activate the lease. */ +static int handle_acquire(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) { + log_message(LOG_INFO, "ACQUIRE: lease->acquired=%d, poisoned=%d, worker_pid=%d, copier_pid=%d", + lease->acquired, server->poisoned, lease->worker_process.pid, lease->copier_process.pid); + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid ACQUIRE session or payload size"); + } + if (lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the lease has already acquired an arena"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (load_u64(request->payload + FAMEM_SESSION_SIZE) != server->pool_size) { + if (server->pool_size == 0U) { + /* Pool not yet allocated; use the client's requested size. */ + uint64_t requested_size = load_u64(request->payload + FAMEM_SESSION_SIZE); + log_message(LOG_INFO, "allocating pool on first ACQUIRE: %llu bytes", (unsigned long long)requested_size); + if (initialize_pool(server, requested_size) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "failed to allocate Famem pool"); + } + } else { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, + "requested arena size does not match the configured pool"); + } + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + lease->acquired = true; + return send_mapping_response(socket_fd, request, server, lease); +} + +/* Reject SLEEP/WAKE/RELEASE with a stale epoch (PROTOCOL error). */ +static int validate_epoch_request(int socket_fd, const struct request_frame *request, const struct lease *lease) { + if (request->payload_size != FAMEM_SESSION_U64_REQUEST_SIZE || !lease->acquired || + load_u64(request->payload + FAMEM_SESSION_SIZE) != lease->epoch) { + (void)send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid lease epoch"); + return -1; + } + return 0; +} + +/* SLEEP: free the physical extents; the lease stays and the client keeps the epoch. */ +static int handle_sleep(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) { + uint8_t payload[FAMEM_EPOCH_RESPONSE_SIZE]; + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (!lease->active || server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is already sleeping"); + } + lease->active = false; + server->active_lease = NULL; + store_u64(payload, lease->epoch); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, sizeof(payload)); +} + +/* WAKE: re-allocate the same extent plan, bump the epoch, re-authorize, reply with mapping. */ +static int handle_wake(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease) { + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (lease->active) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "the arena is not sleeping"); + } + if (server->poisoned) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, "server pool authorization is uncertain"); + } + if (server->active_lease != NULL && server->active_lease != lease) { + return send_error_response(socket_fd, request, FAMEM_STATUS_BUSY, "another Famem lease is active"); + } + if (activate_lease(server, lease) != 0) { + return send_error_response(socket_fd, request, FAMEM_STATUS_INTERNAL, + server->poisoned ? "Famem pool is poisoned; restart the server" + : "failed to authorize the Famem pool"); + } + return send_mapping_response(socket_fd, request, server, lease); +} + +/* RELEASE: free extents if still active and destroy the lease. */ +static int handle_release(int socket_fd, const struct request_frame *request, struct server_state *server, + struct lease *lease, bool *released) { + if (validate_epoch_request(socket_fd, request, lease) != 0) { + return 0; + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + *released = true; + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, NULL, 0U); +} + +/* STATS: read-only pool/lease diagnostics. */ +static int handle_stats(int socket_fd, const struct request_frame *request, struct server_state *server, + const struct lease *lease) { + uint8_t payload[FAMEM_STATS_RESPONSE_SIZE]; + uint64_t huge_1g_bytes = 0U; + uint64_t huge_2m_bytes = 0U; + size_t index; + if (request->payload_size != FAMEM_SESSION_REQUEST_SIZE) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "invalid STATS payload size"); + } + if (!lease->acquired) { + return send_error_response(socket_fd, request, FAMEM_STATUS_PROTOCOL, "session has not acquired an arena"); + } + for (index = 0; index < server->pool_extent_count; ++index) { + if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_1G) { + huge_1g_bytes += server->pool_extents[index].size; + } else if (server->pool_extents[index].page_type == FAMEM_PAGE_HUGE_2M) { + huge_2m_bytes += server->pool_extents[index].size; + } + } + store_u64(payload, server->pool_size); + store_u64(payload + 8U, lease->epoch); + store_u32(payload + 16U, lease->active ? 1U : 0U); + store_u32(payload + 20U, (uint32_t)server->pool_extent_count); + store_u64(payload + 24U, huge_1g_bytes); + store_u64(payload + 32U, huge_2m_bytes); + return send_response(socket_fd, request->operation, request->request_id, FAMEM_STATUS_OK, payload, + sizeof(payload)); +} + +/* True when a connected client already holds this session_id (duplicate HELLO guard). */ +static bool session_exists(const struct server_state *server, const uint8_t session_id[FAMEM_SESSION_SIZE]) { + const struct client_context *client = server->clients; + while (client != NULL) { + if (client->lease.session_id[0] != '\0' && memcmp(client->lease.session_id, session_id, FAMEM_SESSION_SIZE) == 0) { + return true; + } + client = client->next; + } + return false; +} + +/* One client's whole lifecycle: HELLO handshake, request loop, disconnect reclamation. */ +static int handle_client(struct client_context *client) { + struct server_state *server = client->server; + struct lease *lease = &client->lease; + const int socket_fd = client->socket_fd; + struct ucred credentials; + socklen_t credentials_size = sizeof(credentials); + struct timeval hello_timeout = {.tv_sec = 10, .tv_usec = 0}; + struct timeval no_timeout = {.tv_sec = 0, .tv_usec = 0}; + struct request_frame request; + uint32_t worker_bare_tgid; + uint32_t copier_bare_tgid; + pid_t copier_parent; + char error[256]; + int receive_result; + int response_result; + bool released = false; + + if (getsockopt(socket_fd, SOL_SOCKET, SO_PEERCRED, &credentials, &credentials_size) != 0 || + credentials_size != sizeof(credentials)) { + log_message(LOG_WARNING, "cannot read client credentials: %s", strerror(errno)); + return -1; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &hello_timeout, sizeof(hello_timeout)); + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result <= 0) { + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid HELLO frame: %s", error); + } + return -1; + } + if (request.operation != FAMEM_OP_HELLO || request.payload_size != FAMEM_HELLO_REQUEST_SIZE) { + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "the first request must be a valid HELLO"); + } + worker_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE); + copier_bare_tgid = load_u32(request.payload + FAMEM_UUID_SIZE + FAMEM_SESSION_SIZE + 4U); + log_message(LOG_INFO, "HELLO: uuid_cmp=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE), + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid); + if ((!g_skip_tgid_check && memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) != 0) || + !is_lower_hex(request.payload, FAMEM_UUID_SIZE) || + !is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) || + copier_bare_tgid == 0U || copier_bare_tgid > INT32_MAX || + copier_bare_tgid == worker_bare_tgid) { + log_message(LOG_WARNING, "HELLO rejected: uuid_match=%d, uuid_hex=%d, sess_hex=%d, pid=%d, tgid=%u, copier_tgid=%u, skip_tgid=%d", + memcmp(request.payload, server->device_uuid, FAMEM_UUID_SIZE) == 0 ? 1 : 0, + is_lower_hex(request.payload, FAMEM_UUID_SIZE) ? 1 : 0, + is_lower_hex(request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE) ? 1 : 0, + credentials.pid, worker_bare_tgid, copier_bare_tgid, g_skip_tgid_check); + return reject_client(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "invalid Famem HELLO identity"); + } + pthread_mutex_lock(&server->mutex); + if (server->poisoned) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_INTERNAL, + "the Famem server has uncertain authorization state and must restart"); + } + if (session_exists(server, request.payload + FAMEM_UUID_SIZE)) { + pthread_mutex_unlock(&server->mutex); + return reject_client(socket_fd, &request, FAMEM_STATUS_BUSY, "this Famem session is already connected"); + } + + memset(lease, 0, sizeof(*lease)); + memcpy(lease->session_id, request.payload + FAMEM_UUID_SIZE, FAMEM_SESSION_SIZE); + lease->session_id[FAMEM_SESSION_SIZE] = '\0'; + capture_process_identity(&lease->worker_process, credentials.pid, NULL); + capture_process_identity(&lease->copier_process, (pid_t)copier_bare_tgid, &copier_parent); + lease->bare_tgid = (int32_t)worker_bare_tgid; + lease->copier_bare_tgid = (int32_t)copier_bare_tgid; + pthread_mutex_unlock(&server->mutex); + if (send_response(socket_fd, request.operation, request.request_id, FAMEM_STATUS_OK, + (const uint8_t*)server->device_uuid, FAMEM_UUID_SIZE) != 0) { + goto disconnect; + } + (void)setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &no_timeout, sizeof(no_timeout)); + + while (!g_stop && !released) { + receive_result = receive_request(socket_fd, &request, error, sizeof(error)); + if (receive_result == 0) { + break; + } + if (receive_result < 0) { + log_message(LOG_WARNING, "invalid request frame: %s", error); + if (receive_result == -2 && request.request_id != 0U && request.operation >= FAMEM_OP_HELLO && + request.operation <= FAMEM_OP_STATS) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, error); + } + break; + } + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] == '\0' || request.payload_size < FAMEM_SESSION_SIZE || + memcmp(request.payload, lease->session_id, FAMEM_SESSION_SIZE) != 0) { + (void)send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, + "the request session does not belong to this connection"); + pthread_mutex_unlock(&server->mutex); + continue; + } + switch (request.operation) { + case FAMEM_OP_ACQUIRE: + response_result = handle_acquire(socket_fd, &request, server, lease); + break; + case FAMEM_OP_SLEEP: + response_result = handle_sleep(socket_fd, &request, server, lease); + break; + case FAMEM_OP_WAKE: + response_result = handle_wake(socket_fd, &request, server, lease); + break; + case FAMEM_OP_RELEASE: + response_result = handle_release(socket_fd, &request, server, lease, &released); + break; + case FAMEM_OP_STATS: + response_result = handle_stats(socket_fd, &request, server, lease); + break; + case FAMEM_OP_HELLO: + default: + response_result = + send_error_response(socket_fd, &request, FAMEM_STATUS_PROTOCOL, "HELLO is only valid as the first request"); + break; + } + pthread_mutex_unlock(&server->mutex); + if (response_result != 0) { + goto disconnect; + } + } + +disconnect: + pthread_mutex_lock(&server->mutex); + if (lease->session_id[0] != '\0') { + if (lease->active) { + log_message(LOG_WARNING, "client disconnected with mapped HBM; waiting for Worker PID %ld and Copier PID %ld", + (long)lease->worker_process.pid, (long)lease->copier_process.pid); + pthread_mutex_unlock(&server->mutex); + if (!g_skip_tgid_check) { + wait_for_process_exit(&lease->worker_process); + } + wait_for_process_exit(&lease->copier_process); + pthread_mutex_lock(&server->mutex); + } + if (server->active_lease == lease) { + server->active_lease = NULL; + } + lease->active = false; + close_lease_process_identities(lease); + memset(lease, 0, sizeof(*lease)); + } + pthread_mutex_unlock(&server->mutex); + return 0; +} + +/* Per-connection thread entry: run handle_client, then flag the context finished. */ +static void *client_thread_main(void *argument) { + struct client_context *client = (struct client_context*)argument; + (void)handle_client(client); + close(client->socket_fd); + pthread_mutex_lock(&client->server->mutex); + client->socket_fd = -1; + client->thread_finished = true; + pthread_mutex_unlock(&client->server->mutex); + return NULL; +} + +/* Join finished client threads and unlink their contexts from the list. */ +static void reap_finished_clients(struct server_state *server) { + while (true) { + struct client_context **link; + struct client_context *client = NULL; + pthread_mutex_lock(&server->mutex); + link = &server->clients; + while (*link != NULL) { + if ((*link)->thread_finished && (*link)->lease.session_id[0] == '\0') { + client = *link; + *link = client->next; + break; + } + link = &(*link)->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + return; + } + (void)pthread_join(client->thread, NULL); + free(client); + } +} + +/* On shutdown: shutdown() every socket, join threads, reclaim still-active leases. */ +static void shutdown_clients(struct server_state *server) { + struct client_context *client; + pthread_mutex_lock(&server->mutex); + for (client = server->clients; client != NULL; client = client->next) { + if (client->socket_fd >= 0) { + (void)shutdown(client->socket_fd, SHUT_RDWR); + } + } + pthread_mutex_unlock(&server->mutex); + + while (true) { + pthread_mutex_lock(&server->mutex); + client = server->clients; + if (client != NULL) { + server->clients = client->next; + } + pthread_mutex_unlock(&server->mutex); + if (client == NULL) { + break; + } + (void)pthread_join(client->thread, NULL); + close_lease_process_identities(&client->lease); + free(client); + } +} + +/* mkdir -p the socket directory with 0750, validating each created component. */ +static int make_directories(const char *directory) { + char path[PATH_MAX]; + size_t length; + size_t index; + struct stat directory_stat; + length = strlen(directory); + if (length >= sizeof(path)) { + errno = ENAMETOOLONG; + return -1; + } + memcpy(path, directory, length + 1U); + while (length > 1U && path[length - 1U] == '/') { + path[--length] = '\0'; + } + for (index = 1U; index < length; ++index) { + if (path[index] != '/') { + continue; + } + path[index] = '\0'; + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + path[index] = '/'; + } + if (mkdir(path, 0750) != 0 && errno != EEXIST) { + return -1; + } + if (stat(path, &directory_stat) != 0 || !S_ISDIR(directory_stat.st_mode)) { + errno = ENOTDIR; + return -1; + } + if (directory_stat.st_uid != getuid() || (directory_stat.st_mode & S_IWOTH) != 0U) { + errno = EACCES; + return -1; + } + return 0; +} + +/* Unlink a dead socket (ECONNREFUSED probe); a live one aborts startup. */ +static int remove_stale_socket(const char *socket_path) { + struct stat socket_stat; + struct stat current_socket_stat; + struct sockaddr_un address; + int probe_fd; + int connect_error; + if (lstat(socket_path, &socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(socket_stat.st_mode) || socket_stat.st_uid != getuid()) { + errno = EACCES; + return -1; + } + probe_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (probe_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, strlen(socket_path) + 1U); + if (connect(probe_fd, (struct sockaddr*)&address, sizeof(address)) == 0) { + close(probe_fd); + errno = EADDRINUSE; + return -1; + } + connect_error = errno; + close(probe_fd); + if (connect_error == ENOENT) { + return 0; + } + if (connect_error != ECONNREFUSED) { + errno = connect_error; + return -1; + } + if (lstat(socket_path, &current_socket_stat) != 0 || !S_ISSOCK(current_socket_stat.st_mode) || + current_socket_stat.st_dev != socket_stat.st_dev || current_socket_stat.st_ino != socket_stat.st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path) == 0 || errno == ENOENT ? 0 : -1; +} + +/* Unlink our socket at exit only if (dev,ino) still matches what we bound. */ +static int unlink_owned_socket(const char *socket_path, const struct stat *bound_socket_stat) { + struct stat current_socket_stat; + if (lstat(socket_path, &current_socket_stat) != 0) { + return errno == ENOENT ? 0 : -1; + } + if (!S_ISSOCK(current_socket_stat.st_mode) || current_socket_stat.st_dev != bound_socket_stat->st_dev || + current_socket_stat.st_ino != bound_socket_stat->st_ino) { + errno = ESTALE; + return -1; + } + return unlink(socket_path); +} + +/* Listener cleanup path shared between fatal startup errors. */ +static int discard_listener(int listener_fd, const char *socket_path, const struct stat *bound_socket_stat, + int saved_errno) { + close(listener_fd); + if (bound_socket_stat != NULL) { + (void)unlink_owned_socket(socket_path, bound_socket_stat); + } + errno = saved_errno; + return -1; +} + +/* bind()+listen() the server socket, chmod 0600, record (dev,ino) identity. */ +static int create_listener(const char *socket_path, struct stat *bound_socket_stat) { + struct sockaddr_un address; + const size_t path_length = strlen(socket_path); + int listener_fd; + listener_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (listener_fd < 0) { + return -1; + } + memset(&address, 0, sizeof(address)); + address.sun_family = AF_UNIX; + memcpy(address.sun_path, socket_path, path_length + 1U); + if (bind(listener_fd, (struct sockaddr*)&address, + offsetof(struct sockaddr_un, sun_path) + path_length + 1U) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (lstat(socket_path, bound_socket_stat) != 0) { + return discard_listener(listener_fd, socket_path, NULL, errno); + } + if (!S_ISSOCK(bound_socket_stat->st_mode) || bound_socket_stat->st_uid != getuid()) { + return discard_listener(listener_fd, socket_path, NULL, EACCES); + } + if (chmod(socket_path, 0600) != 0 || listen(listener_fd, 8) != 0) { + return discard_listener(listener_fd, socket_path, bound_socket_stat, errno); + } + return listener_fd; +} + +/* Print the --help text. */ +static void print_usage(FILE *output, const char *program) { + fprintf(output, + "Usage: %s --device DEVICE [--size-gib GIB | --size-bytes BYTES] [--socket-dir DIR] [--socket-name NAME] [--skip-tgid-check] [--log-level LEVEL]\n" + "Serve Famem physical HBM for one Ascend NPU.\n", + program); +} + +/* Map --log-level names to the enum. */ +static int parse_log_level(const char *value, enum log_level *output) { + static const char *const names[] = {"DEBUG", "INFO", "WARNING", "ERROR"}; + size_t index; + for (index = 0; index < sizeof(names) / sizeof(names[0]); ++index) { + if (strcmp(value, names[index]) == 0) { + *output = (enum log_level)index; + return 0; + } + } + return -1; +} + +/* Parse --device as a non-negative integer. */ +static int parse_device(const char *value, int *output) { + char *end = NULL; + long device; + errno = 0; + device = strtol(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || device < 0 || device > INT_MAX) { + return -1; + } + *output = (int)device; + return 0; +} + +/* Parse --pool-size in GiB (0 = defer to first ACQUIRE). */ +static int parse_pool_size(const char *value, uint64_t *output) { + char *end = NULL; + unsigned long long size_gib; + errno = 0; + size_gib = strtoull(value, &end, 10); + if (errno != 0 || end == value || *end != '\0' || size_gib == 0U || size_gib > FAMEM_MAX_POOL_GIB) { + return -1; + } + *output = (uint64_t)size_gib * FAMEM_HUGE_1G_BYTES; + return 0; +} + +/* Entry point: parse flags, bind the socket, serve until SIGTERM, reclaim everything. */ +int main(int argc, char **argv) { + static const struct option options[] = { + {"device", required_argument, NULL, 'd'}, + {"size-gib", required_argument, NULL, 'm'}, + {"size-bytes", required_argument, NULL, 'b'}, + {"socket-dir", required_argument, NULL, 's'}, + {"socket-name", required_argument, NULL, 'n'}, + {"skip-tgid-check", no_argument, NULL, 'k'}, + {"log-level", required_argument, NULL, 'l'}, + {"help", no_argument, NULL, 'h'}, + {NULL, 0, NULL, 0}, + }; + const char *socket_dir = NULL; + const char *socket_name = NULL; + char socket_path[PATH_MAX] = {0}; + struct sigaction action; + struct stat bound_socket_stat; + struct server_state server; + int device = -1; + int listener_fd = -1; + int socket_dir_fd = -1; + uint64_t pool_size = 0U; + int option; + int exit_code = EXIT_FAILURE; + while ((option = getopt_long(argc, argv, "d:m:b:s:n:kl:h", options, NULL)) != -1) { + switch (option) { + case 'd': + if (parse_device(optarg, &device) != 0) { + fprintf(stderr, "Invalid --device value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'm': + if (parse_pool_size(optarg, &pool_size) != 0) { + fprintf(stderr, "Invalid --size-gib value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'b': { + char *end = NULL; + errno = 0; + pool_size = strtoull(optarg, &end, 10); + if (errno != 0 || end == optarg || *end != '\0' || pool_size == 0U) { + fprintf(stderr, "Invalid --size-bytes value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + } + case 's': + socket_dir = optarg; + break; + case 'n': + socket_name = optarg; + break; + case 'k': + /* --skip-tgid-check: skip PID/TGID verification (needed in containers + where the PID namespace differs from the host) */ + g_skip_tgid_check = 1; + break; + case 'l': + if (parse_log_level(optarg, &g_log_level) != 0) { + fprintf(stderr, "Invalid --log-level value: %s\n", optarg); + return EXIT_FAILURE; + } + break; + case 'h': + print_usage(stdout, argv[0]); + return EXIT_SUCCESS; + default: + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + } + if (device < 0 || optind != argc) { + print_usage(stderr, argv[0]); + return EXIT_FAILURE; + } + prctl(PR_SET_NAME, "hbm_server", 0, 0, 0); + /* Pool size will be determined by the first ACQUIRE request when not + explicitly set via --size-gib / --size-bytes. This allows a single + server binary to serve models with different per-NPU arena sizes + (e.g. 27 GiB for TP1-7B, 54 GiB for TP4-72B) without changing the + launch command. */ + server.pool_size = pool_size; /* 0 means "defer to first ACQUIRE" */ + if (socket_dir == NULL) { + socket_dir = getenv("VLLM_ASCEND_FAMEM_SOCKET_DIR"); + } + if (socket_dir == NULL || socket_dir[0] == '\0') { + socket_dir = FAMEM_DEFAULT_SOCKET_DIR; + } + memset(&action, 0, sizeof(action)); + action.sa_handler = stop_server; + sigemptyset(&action.sa_mask); + if (sigaction(SIGTERM, &action, NULL) != 0 || sigaction(SIGINT, &action, NULL) != 0) { + log_message(LOG_ERROR, "cannot install signal handlers: %s", strerror(errno)); + return EXIT_FAILURE; + } + signal(SIGPIPE, SIG_IGN); + if (famem_server_initialize(device) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem backend: %s", native_error()); + return EXIT_FAILURE; + } + memset(&server, 0, sizeof(server)); + server.device = device; + if (famem_get_device_uuid(device, server.device_uuid, sizeof(server.device_uuid)) != 0 || + !is_lower_hex((const uint8_t*)server.device_uuid, FAMEM_UUID_SIZE) || + is_all_zero((const uint8_t*)server.device_uuid, FAMEM_UUID_SIZE)) { + /* Some NPU models (e.g. Ascend 910B3) do not provide a valid UUID via + aclrtDeviceGetUuid (returns ACL_ERROR_RT_DEVICE_UNAVAILABLE or an + all-zero UUID). In that case, synthesize a deterministic UUID from + the device index so that the server can still operate. */ + log_message(LOG_WARNING, + "NPU UUID not available (device %d), synthesizing from device index", + device); + option = snprintf(server.device_uuid, sizeof(server.device_uuid), + "d%031d", device); + if (option < 0 || (size_t)option != FAMEM_UUID_SIZE) { + log_message(LOG_ERROR, "cannot synthesize device UUID"); + goto cleanup_backend; + } + } + server.device_uuid[FAMEM_UUID_SIZE] = '\0'; + if (pthread_mutex_init(&server.mutex, NULL) != 0) { + log_message(LOG_ERROR, "cannot initialize Famem server mutex"); + goto cleanup_backend; + } + if (make_directories(socket_dir) != 0) { + log_message(LOG_ERROR, "cannot prepare socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + socket_dir_fd = open(socket_dir, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + if (socket_dir_fd < 0 || flock(socket_dir_fd, LOCK_EX) != 0) { + log_message(LOG_ERROR, "cannot lock Famem socket directory %s: %s", socket_dir, strerror(errno)); + goto cleanup_mutex; + } + /* Build socket path: use --socket-name if provided, otherwise device_uuid */ + if (socket_name != NULL) { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, socket_name); + } else { + option = snprintf(socket_path, sizeof(socket_path), "%s/%s.sock", socket_dir, server.device_uuid); + } + if (option < 0 || (size_t)option >= sizeof(socket_path) || (size_t)option > FAMEM_SOCKET_PATH_MAX) { + errno = ENAMETOOLONG; + log_message(LOG_ERROR, "Famem socket path is too long"); + goto cleanup_pool; + } + if (remove_stale_socket(socket_path) != 0) { + log_message(LOG_ERROR, "cannot prepare Famem socket %s: %s", socket_path, strerror(errno)); + goto cleanup_pool; + } + if (pool_size != 0U && initialize_pool(&server, pool_size) != 0) { + goto cleanup_pool; + } + if (pool_size == 0U) { + log_message(LOG_INFO, "pool size not specified; will be determined by the first ACQUIRE request"); + } + listener_fd = create_listener(socket_path, &bound_socket_stat); + option = errno; + /* When --socket-name overrides the socket filename, create a UUID-named + hard link so that clients using famem_get_device_uuid() can find the socket. + A hard link (not symlink) is required because the client checks S_ISSOCK + with follow_symlinks=False, and only a hard link preserves the socket mode. */ + if (socket_name != NULL) { + char uuid_sock[PATH_MAX]; + option = snprintf(uuid_sock, sizeof(uuid_sock), "%s/%s.sock", socket_dir, server.device_uuid); + if (option > 0 && (size_t)option < sizeof(uuid_sock)) { + (void)unlink(uuid_sock); + if (link(socket_path, uuid_sock) != 0) { + log_message(LOG_WARNING, "cannot create UUID hardlink %s: %s", uuid_sock, strerror(errno)); + } + } + } + close(socket_dir_fd); + socket_dir_fd = -1; + errno = option; + if (listener_fd < 0) { + log_message(LOG_ERROR, "cannot create Famem listener: %s", strerror(errno)); + goto cleanup_pool; + } + log_message(LOG_INFO, "pool ready; listening on %s for NPU UUID %s", socket_path, server.device_uuid); + + while (!g_stop) { + struct client_context *client; + int client_fd = accept(listener_fd, NULL, NULL); + if (client_fd < 0) { + if (errno == EINTR) { + continue; + } + log_message(LOG_ERROR, "accept failed: %s", strerror(errno)); + goto cleanup; + } + client = (struct client_context*)calloc(1U, sizeof(*client)); + if (client == NULL) { + log_message(LOG_ERROR, "cannot allocate Famem client context"); + close(client_fd); + continue; + } + client->socket_fd = client_fd; + client->server = &server; + pthread_mutex_lock(&server.mutex); + client->next = server.clients; + server.clients = client; + pthread_mutex_unlock(&server.mutex); + if (pthread_create(&client->thread, NULL, client_thread_main, client) != 0) { + pthread_mutex_lock(&server.mutex); + server.clients = client->next; + pthread_mutex_unlock(&server.mutex); + log_message(LOG_ERROR, "cannot start Famem client thread"); + close(client_fd); + free(client); + continue; + } + reap_finished_clients(&server); + } + exit_code = EXIT_SUCCESS; + +cleanup: + close(listener_fd); + shutdown_clients(&server); + if (unlink_owned_socket(socket_path, &bound_socket_stat) != 0 && errno != ENOENT) { + log_message(LOG_WARNING, "cannot remove owned socket %s: %s", socket_path, strerror(errno)); + } +cleanup_pool: + if (free_extents(server.pool_extents, server.pool_extent_count) != 0) { + log_message(LOG_ERROR, "cannot release every persistent pool extent"); + exit_code = EXIT_FAILURE; + } +cleanup_mutex: + if (socket_dir_fd >= 0) { + close(socket_dir_fd); + } + (void)pthread_mutex_destroy(&server.mutex); +cleanup_backend: + if (famem_server_finalize() != 0) { + log_message(LOG_ERROR, "cannot finalize Famem backend: %s", native_error()); + exit_code = EXIT_FAILURE; + } + return exit_code; +} -- 2.43.0

2 1

Jump to page: