/* * (C) 2026 by sysmocom s.f.m.c GmbH * Author: Vadim Yanitskiy * * All Rights Reserved * * SPDX-License-Identifier: GPL-2.0+ * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * */ #define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "config.h" #define TEST_START() printf("Running %s\n", __func__) static void *ctx = NULL; static unsigned file_bytes_write_compl = 0; /* Shared infrastructure - same as main test file */ static void file_write_cb(struct osmo_io_fd *iofd, int rc, struct msgb *msg) { printf("%s: write() returned rc=%d\n", osmo_iofd_get_name(iofd), rc); if (rc < 0) printf("%s: error: %s\n", osmo_iofd_get_name(iofd), strerror(-rc)); if (msg) { printf("%s\n", osmo_hexdump(msgb_data(msg), msgb_length(msg))); file_bytes_write_compl += msgb_length(msg); } } /* Test osmo_io behavior under network backpressure conditions. * Simulates a scenario where an application sends data faster than the peer * can consume it, causing the OS socket buffers to fill up and creating * backpressure. Tests that osmo_io correctly handles internal message queueing * when write operations return EAGAIN, and properly drains the queue when * the backpressure clears. Uses multiple I/O buffers to exercise the * dequeue fill-up logic that attempts to merge multiple queued messages. */ static void test_backpressure_queue_handling(void) { struct osmo_io_ops ioops; struct osmo_io_fd *iofd; struct msgb *msg; int fd[2] = { 0, 0 }; int rc; TEST_START(); printf("Testing backpressure scenario with queue management\n"); /* Create a socketpair */ rc = socketpair(AF_UNIX, SOCK_STREAM, 0, fd); OSMO_ASSERT(rc == 0); /* Make both ends non-blocking */ rc = fcntl(fd[0], F_SETFL, O_NONBLOCK); OSMO_ASSERT(rc == 0); rc = fcntl(fd[1], F_SETFL, O_NONBLOCK); OSMO_ASSERT(rc == 0); /* Set small socket buffer to force queueing */ int sndbuf = 1024; rc = setsockopt(fd[1], SOL_SOCKET, SO_SNDBUF, &sndbuf, sizeof(sndbuf)); if (rc < 0) printf("Warning: failed to set SO_SNDBUF: %s\n", strerror(errno)); /* Set up iofd with multiple write buffers to enable fill-up logic */ ioops = (struct osmo_io_ops){ .write_cb = file_write_cb }; iofd = osmo_iofd_setup(ctx, fd[1], "backpressure_test", OSMO_IO_FD_MODE_READ_WRITE, &ioops, NULL); OSMO_ASSERT(iofd); /* Use multiple buffers to enable dequeue fill-up logic */ rc = osmo_iofd_set_io_buffers(iofd, OSMO_IO_OP_WRITE, 8); OSMO_ASSERT(rc == 0); osmo_iofd_register(iofd, fd[1]); printf("Queueing messages to simulate network backpressure\n"); file_bytes_write_compl = 0; char drain_buffer[512]; int messages_queued = 0; for (int i = 0; i < 10; i++) { /* Add a message */ msg = msgb_alloc(1024, "qmsg"); memset(msgb_put(msg, 1024), 0xAA + (i % 16), 1024); rc = osmo_iofd_write_msgb(iofd, msg); if (rc == 0) messages_queued++; else msgb_free(msg); /* Process events with explicit timeout */ for (int j = 0; j < 3; j++) osmo_select_main(1); /* Drain to prevent saturation */ if (i % 3 == 0) read(fd[0], drain_buffer, sizeof(drain_buffer)); /* Stop if we have sufficient queue buildup */ if (osmo_iofd_txqueue_len(iofd) >= 10) break; /* Add small delay to prevent overwhelming */ usleep(5000); /* 5ms */ } /* Add varied message sizes */ /* Add tiny messages that might create partially-filled msghdrs */ for (int i = 0; i < 10; i++) { msg = msgb_alloc(50, "Tiny msg"); memset(msgb_put(msg, 50), 0xBB + i, 50); rc = osmo_iofd_write_msgb(iofd, msg); if (rc < 0) msgb_free(msg); } /* Add medium messages */ for (int i = 0; i < 10; i++) { msg = msgb_alloc(500, "Medium msg"); memset(msgb_put(msg, 500), 0xCC + i, 500); rc = osmo_iofd_write_msgb(iofd, msg); if (rc < 0) msgb_free(msg); } /* Add large messages */ for (int i = 0; i < 5; i++) { msg = msgb_alloc(1500, "qmsg_1500"); memset(msgb_put(msg, 1500), 0xDD + i, 1500); rc = osmo_iofd_write_msgb(iofd, msg); if (rc < 0) msgb_free(msg); } printf("Queueing messages completed\n"); /* Controlled draining to trigger dequeue with fill-up logic */ for (int iteration = 0; iteration < 500; iteration++) { /* Drain TINY amounts to slowly unblock writes */ ssize_t drained = read(fd[0], drain_buffer, sizeof(drain_buffer)); if (drained > 0 && iteration % 20 == 0) printf("Iteration %d: drained %zd bytes\n", iteration, drained); /* Process events to trigger dequeue operations */ for (int j = 0; j < 3; j++) osmo_select_main(1); if (iteration % 25 == 0) { printf("Iter %d: queue_len=%u, bytes_completed=%u\n", iteration, osmo_iofd_txqueue_len(iofd), file_bytes_write_compl); } /* Are we done yet? */ if (osmo_iofd_txqueue_len(iofd) == 0 && file_bytes_write_compl > 0) { printf("Queue emptied at iteration %d\n", iteration); break; } usleep(1000); /* 1ms delay to control rate */ } printf("Final: queue_len=%u, bytes_completed=%u\n", osmo_iofd_txqueue_len(iofd), file_bytes_write_compl); /* Clean up */ osmo_iofd_free(iofd); close(fd[0]); for (int i = 0; i < 128; i++) osmo_select_main(1); } int main(int argc, char **argv) { ctx = talloc_named_const(NULL, 0, __FILE__); osmo_init_logging2(ctx, NULL); test_backpressure_queue_handling(); return EXIT_SUCCESS; }