[M] Change in libosmocore[master]: osmo_io: add osmo_iofd_flush() to transmit the pending Tx queue

28 Jul 2026

fixeria has uploaded this change for review. ( https://gerrit.osmocom.org/c/libosmocore/+/43102?usp=email )
Change subject: osmo_io: add osmo_iofd_flush() to transmit the pending Tx queue
......................................................................
osmo_io: add osmo_iofd_flush() to transmit the pending Tx queue
osmo_iofd_write_msgb() and friends only enqueue the given message;
the actual transmission happens asynchronously from within
osmo_select_main().  An application that needs to say 'goodbye'
right before calling osmo_iofd_free() (think of a connection
teardown datagram, like TRXC "CMD POWEROFF") currently has no way
to do so reliably: osmo_iofd_free() discards the Tx queue, so a
message enqueued in the same event loop iteration never reaches
the socket.
Add osmo_iofd_flush(), which synchronously transmits the pending
Tx queue, stopping on -EAGAIN or other errors.  Only the poll
backend implements it for now;  the io_uring backend returns
-ENOTSUP, since in-flight SQE submissions make a synchronous
flush non-trivial there.
Change-Id: I4e09e91e472a4441977a2c199c07c2cb8005632f
---
M include/osmocom/core/osmo_io.h
M src/core/libosmocore.map
M src/core/osmo_io.c
M src/core/osmo_io_internal.h
M src/core/osmo_io_poll.c
M tests/osmo_io/osmo_io_test.c
M tests/osmo_io/osmo_io_test.ok
7 files changed, 144 insertions(+), 0 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/libosmocore refs/changes/02/43102/1

diff --git a/include/osmocom/core/osmo_io.h b/include/osmocom/core/osmo_io.h
index c92492c..a526a7c 100644
--- a/include/osmocom/core/osmo_io.h
+++ b/include/osmocom/core/osmo_io.h
@@ -220,6 +220,7 @@
 int osmo_iofd_unregister(struct osmo_io_fd *iofd);
 unsigned int osmo_iofd_txqueue_len(struct osmo_io_fd *iofd);
 void osmo_iofd_txqueue_clear(struct osmo_io_fd *iofd);
+int osmo_iofd_flush(struct osmo_io_fd *iofd);
 int osmo_iofd_close(struct osmo_io_fd *iofd);
 void osmo_iofd_free(struct osmo_io_fd *iofd);
diff --git a/src/core/libosmocore.map b/src/core/libosmocore.map
index 0cb696b..90b1ef2 100644
--- a/src/core/libosmocore.map
+++ b/src/core/libosmocore.map
@@ -268,6 +268,7 @@
 osmo_io_get_backend;
 osmo_io_backend_names;
 osmo_iofd_close;
+osmo_iofd_flush;
 osmo_iofd_free;
 osmo_iofd_get_data;
 osmo_iofd_get_ioops;
diff --git a/src/core/osmo_io.c b/src/core/osmo_io.c
index f14f03b..b0acc04 100644
--- a/src/core/osmo_io.c
+++ b/src/core/osmo_io.c
@@ -1114,6 +1114,39 @@
    }
 }
+/*! Flush the transmit queue of the given osmo_io_fd.
+ *
+ *  This function attempts to synchronously transmit all messages currently
+ *  pending in the transmit queue, stopping at the first message that cannot
+ *  be written without blocking.  This is useful during shutdown, to make
+ *  sure that pending messages (e.g. a final 'goodbye' datagram) actually
+ *  leave the socket before the file descriptor gets closed: unlike
+ *  osmo_iofd_write_msgb(), which only enqueues, actual transmission
+ *  normally happens once the event loop (osmo_select_main()) iterates.
+ *
+ *  The write/sendto/sendmsg call-back is invoked for every transmitted
+ *  message, just like on transmission from the event loop.
+ *
+ *  \param[in] iofd the file descriptor
+ *  \returns 0 if the transmit queue was fully drained; -EAGAIN if a message
+ *	     could not be transmitted without blocking; -ENOTSUP if the
+ *	     osmo_io backend does not implement flushing; other negative
+ *	     values on send errors */
+int osmo_iofd_flush(struct osmo_io_fd *iofd)
+{
+	int rc;
+
+	if (iofd->fd < 0 || IOFD_FLAG_ISSET(iofd, IOFD_FLAG_CLOSED))
+		return -EBADF;
+	if (osmo_iofd_ops.flush == NULL)
+		return -ENOTSUP;
+
+	rc = osmo_iofd_ops.flush(iofd);
+	if (rc == 0) /* nothing left to transmit */
+		osmo_iofd_ops.write_disable(iofd);
+	return rc;
+}
+
 /*! Free the given osmo_io_fd.
  *
  *  The iofd will be automatically closed before via osmo_iofd_close() [which in turn will unregister
diff --git a/src/core/osmo_io_internal.h b/src/core/osmo_io_internal.h
index 81a70f3..5f4614f 100644
--- a/src/core/osmo_io_internal.h
+++ b/src/core/osmo_io_internal.h
@@ -35,6 +35,7 @@
    void (*read_enable)(struct osmo_io_fd *iofd);
    void (*read_disable)(struct osmo_io_fd *iofd);
    void (*notify_connected)(struct osmo_io_fd *iofd);
+	int (*flush)(struct osmo_io_fd *iofd);
 };
#define IOFD_FLAG_CLOSED (1<<0)
diff --git a/src/core/osmo_io_poll.c b/src/core/osmo_io_poll.c
index b861675..f07c501 100644
--- a/src/core/osmo_io_poll.c
+++ b/src/core/osmo_io_poll.c
@@ -258,6 +258,42 @@
    osmo_fd_write_enable(&iofd->u.poll.ofd);
 }
+static int iofd_poll_flush(struct osmo_io_fd *iofd)
+{
+	struct iofd_msghdr *msghdr;
+	int rc;
+
+	while ((msghdr = iofd_txqueue_dequeue(iofd))) {
+		/* a zero-length write is progress, unlike a partial write of 0
+		 * bytes (see the interpretation in iofd_handle_send_completion()) */
+		bool zero_len = msgb_length(msghdr->msg[0]) == 0;
+
+		switch (iofd->mode) {
+		case OSMO_IO_FD_MODE_READ_WRITE:
+			rc = writev(iofd->fd, msghdr->iov, msghdr->io_len);
+			break;
+		case OSMO_IO_FD_MODE_RECVFROM_SENDTO:
+		case OSMO_IO_FD_MODE_RECVMSG_SENDMSG:
+			rc = sendmsg(iofd->fd, &msghdr->hdr, msghdr->flags);
+			break;
+		default:
+			OSMO_ASSERT(0);
+		}
+
+		rc = (rc < 0 && errno > 0) ? -errno : rc;
+		/* completes transmitted msgbs; re-enqueues the msghdr at the
+		 * front of the queue on -EAGAIN and on partial writes */
+		iofd_handle_send_completion(iofd, rc, msghdr);
+
+		if (rc < 0)
+			return rc;
+		if (rc == 0 && !zero_len)
+			return -EAGAIN; /* no progress was made, give up */
+	}
+
+	return 0;
+}
+
 const struct iofd_backend_ops iofd_poll_ops = {
    .register_fd = iofd_poll_register,
    .unregister_fd = iofd_poll_unregister,
@@ -267,6 +303,7 @@
    .read_enable = iofd_poll_read_enable,
    .read_disable = iofd_poll_read_disable,
    .notify_connected = iofd_poll_notify_connected,
+	.flush = iofd_poll_flush,
 };
#endif /* ifndef EMBEDDED */
diff --git a/tests/osmo_io/osmo_io_test.c b/tests/osmo_io/osmo_io_test.c
index ca7fa67..8d245fb 100644
--- a/tests/osmo_io/osmo_io_test.c
+++ b/tests/osmo_io/osmo_io_test.c
@@ -251,6 +251,69 @@
    	osmo_select_main(1);
 }
+static unsigned int flush_num_compl = 0;
+
+static void flush_sendto_cb(struct osmo_io_fd *iofd, int rc, struct msgb *msg,
+			    const struct osmo_sockaddr *daddr)
+{
+	printf("%s: sendto() returned rc=%d\n", osmo_iofd_get_name(iofd), rc);
+	flush_num_compl++;
+}
+
+struct osmo_io_ops ioops_flush = {
+	.sendto_cb = flush_sendto_cb,
+	.recvfrom_cb = recvfrom_cb,
+};
+
+static void test_flush(void)
+{
+	int fds[2] = {0, 0}, rc;
+	struct osmo_io_fd *iofd;
+	struct msgb *msg;
+	uint8_t buf[64];
+
+	TEST_START();
+
+	rc = socketpair(AF_UNIX, SOCK_DGRAM, 0, fds);
+	OSMO_ASSERT(rc == 0);
+
+	iofd = osmo_iofd_setup(ctx, fds[0], "ep1", OSMO_IO_FD_MODE_RECVFROM_SENDTO, &ioops_flush, NULL);
+	osmo_iofd_register(iofd, fds[0]);
+
+	/* flushing an empty Tx queue is a no-op */
+	rc = osmo_iofd_flush(iofd);
+	OSMO_ASSERT(rc == 0 || rc == -ENOTSUP);
+
+	for (unsigned int i = 0; i < 3; i++) {
+		msg = msgb_alloc(1024, "Test data");
+		memcpy(msgb_put(msg, sizeof(TESTDATA)), TESTDATA, sizeof(TESTDATA));
+		osmo_iofd_sendto_msgb(iofd, msg, 0, NULL);
+	}
+
+	rc = osmo_iofd_flush(iofd);
+	if (rc == -ENOTSUP) {
+		/* backend without flush support (io_uring): drain via the event
+		 * loop instead, so that the test output remains identical */
+		while (flush_num_compl < 3)
+			osmo_select_main(1);
+	} else {
+		OSMO_ASSERT(rc == 0);
+	}
+	OSMO_ASSERT(osmo_iofd_txqueue_len(iofd) == 0);
+
+	/* all three datagrams must have hit the socket by now */
+	for (unsigned int i = 0; i < 3; i++) {
+		rc = recv(fds[1], buf, sizeof(buf), MSG_DONTWAIT);
+		printf("ep2: recv() returned rc=%d\n", rc);
+	}
+
+	osmo_iofd_free(iofd);
+	close(fds[1]);
+
+	for (int i = 0; i < 128; i++)
+		osmo_select_main(1);
+}
+
 int segmentation_cb(struct osmo_io_fd *iofd, struct msgb *msg)
 {
    printf("%s: segmentation_cb() returning %d\n", osmo_iofd_get_name(iofd), 4);
@@ -668,6 +731,7 @@
    test_file();
    test_connected();
    test_unconnected();
+	test_flush();
    test_segmentation();
    test_segmentation_uint16_max(10000, UINT16_MAX, 0);
    test_segmentation_uint16_max(10000, UINT16_MAX - 320, 320);
diff --git a/tests/osmo_io/osmo_io_test.ok b/tests/osmo_io/osmo_io_test.ok
index d368db4..5a9193c 100644
--- a/tests/osmo_io/osmo_io_test.ok
+++ b/tests/osmo_io/osmo_io_test.ok
@@ -16,6 +16,13 @@
 ep1: sendto() returned rc=16
 ep2: recvfrom() msg with len=16
 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 10 
+Running test_flush
+ep1: sendto() returned rc=16
+ep1: sendto() returned rc=16
+ep1: sendto() returned rc=16
+ep2: recv() returned rc=16
+ep2: recv() returned rc=16
+ep2: recv() returned rc=16
 Running test_segmentation
 Enable write
 seg_iofd: write() returned rc=12
-- 
To view, visit https://gerrit.osmocom.org/c/libosmocore/+/43102?usp=email
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings?usp=email

Gerrit-MessageType: newchange
Gerrit-Project: libosmocore
Gerrit-Branch: master
Gerrit-Change-Id: I4e09e91e472a4441977a2c199c07c2cb8005632f
Gerrit-Change-Number: 43102
Gerrit-PatchSet: 1
Gerrit-Owner: fixeria vyanitskiy@sysmocom.de



    

2026

2025

2024

2023

2022

[M] Change in libosmocore[master]: osmo_io: add osmo_iofd_flush() to transmit the pending Tx queue