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 <[email protected]>
