This is an automated email from the git hooks/post-receive script.
git pushed a commit to branch tymux
in repository terminology.
View the commit online.
commit 0424d174fbded204e022e60acce48a941643c962
Author: [email protected] <[email protected]>
AuthorDate: Sun Mar 22 13:50:17 2026 -0600
feat: async backlog buffer prefetching for tymux scrollback
Replace blocking synchronous buffer fetches (which failed when
NOTIFY messages interleaved with BACKLOG_BUF responses) with fully
asynchronous prefetching. termsrv_msg_recv now uses recvmsg to
extract SCM_RIGHTS fds, enabling BACKLOG_BUF handling in the normal
fd handler. On attach, placeholder entries are created for all
older buffers and a cascading prefetch chain fetches them one by
one. Proximity-based prefetch triggers when scrolling within 20%
of a buffer boundary. This fixes the 256-line scrollback limit
and eliminates UI freezes during buffer fetches.
Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
---
src/bin/termd_tymux.c | 2 +-
src/bin/termsrv.c | 159 ++++++++----------
src/bin/termsrv.h | 15 +-
src/bin/termtymux.c | 426 ++++++++++++++++++++++++++++++-----------------
src/bin/termtymux.h | 14 +-
src/tests/test_termsrv.c | 4 +-
6 files changed, 355 insertions(+), 265 deletions(-)
diff --git a/src/bin/termd_tymux.c b/src/bin/termd_tymux.c
index d83c4e88..0bd0314a 100644
--- a/src/bin/termd_tymux.c
+++ b/src/bin/termd_tymux.c
@@ -425,7 +425,7 @@ _cb_client_data(void *data, Ecore_Fd_Handler *fd_handler EINA_UNUSED)
uint32_t size;
int type;
- type = termsrv_msg_recv(client->fd, &payload, &size);
+ type = termsrv_msg_recv(client->fd, &payload, &size, NULL);
if (type < 0)
{
free(payload);
diff --git a/src/bin/termsrv.c b/src/bin/termsrv.c
index 6f299690..9180b35d 100644
--- a/src/bin/termsrv.c
+++ b/src/bin/termsrv.c
@@ -75,48 +75,6 @@ termsrv_msg_send(int fd, TermSrvMsgType type,
return (n == expected) ? 0 : -1;
}
-/*
- * Read exactly @need bytes from @fd into @buf, tolerating short reads
- * that can occur on non-blocking sockets. Spins with EAGAIN up to a
- * tight retry limit — on AF_UNIX the peer always sends header+payload
- * atomically via sendmsg/iovec, so the bytes arrive together; any
- * EAGAIN after a partial read is an extremely transient condition.
- *
- * Returns 0 on success, -1 on EOF or unrecoverable error.
- */
-static int
-_recv_exact(int fd, void *buf, size_t need)
-{
- size_t got = 0;
- int retries = 0;
-
- while (got < need)
- {
- ssize_t r = recv(fd, (char *)buf + got, need - got, 0);
- if (r > 0)
- {
- got += (size_t)r;
- retries = 0;
- }
- else if (r == 0)
- {
- return -1; /* EOF */
- }
- else
- {
- if (errno == EINTR) continue;
- if ((errno == EAGAIN || errno == EWOULDBLOCK) && retries < 100)
- {
- /* Payload bytes haven't arrived yet — yield briefly */
- retries++;
- usleep(1000); /* 1 ms */
- continue;
- }
- return -1;
- }
- }
- return 0;
-}
/*
* Receive one tymux message from @fd.
@@ -128,15 +86,26 @@ _recv_exact(int fd, void *buf, size_t need)
* (EAGAIN/EWOULDBLOCK on the header read) — caller should
* return ECORE_CALLBACK_RENEW and wait for the next wakeup.
* Returns -1 on EOF or protocol error.
+ *
+ * If fd_out is non-NULL and the message carries an SCM_RIGHTS fd,
+ * *fd_out is set to the received fd; otherwise it is set to -1.
+ * Pass NULL for fd_out when the caller does not expect an ancillary fd
+ * (e.g. the daemon side); in that case any received fd is closed immediately
+ * to prevent leaks.
*/
int
-termsrv_msg_recv(int fd, void **payload_out, uint32_t *size_out)
+termsrv_msg_recv(int fd, void **payload_out, uint32_t *size_out, int *fd_out)
{
TermSrvMsgHdr hdr;
*payload_out = NULL;
*size_out = 0;
+ if (fd_out)
+ *fd_out = -1;
- ssize_t r = recv(fd, &hdr, sizeof(hdr), MSG_DONTWAIT);
+ /* Peek-check: use MSG_DONTWAIT so we can return 0 (no data yet) when the
+ * socket is non-blocking and empty, without consuming the data.
+ * We only read the header here; the payload is read with recvmsg below. */
+ ssize_t r = recv(fd, &hdr, sizeof(hdr), MSG_DONTWAIT | MSG_PEEK);
if (r == 0) return -1; /* EOF */
if (r < 0)
{
@@ -146,10 +115,9 @@ termsrv_msg_recv(int fd, void **payload_out, uint32_t *size_out)
}
if (r != (ssize_t)sizeof(hdr))
{
- /* Partial header: the rest must be in the kernel buffer already
- * (same sendmsg atomicity guarantee). Retrieve the remainder. */
- if (_recv_exact(fd, (char *)&hdr + r, sizeof(hdr) - (size_t)r) < 0)
- return -1;
+ /* Partial header in peek: the rest hasn't arrived yet.
+ * Return 0 to retry on the next fd-ready event. */
+ return 0;
}
/* Reject unknown message types before allocating — valid range is
@@ -158,15 +126,64 @@ termsrv_msg_recv(int fd, void **payload_out, uint32_t *size_out)
if (hdr.type == 0 || hdr.type > 11) return -1;
if (hdr.size > TERMSRV_MAX_MSG_PAYLOAD) return -1;
+ /* Now do the real read. Use recvmsg so that we can extract an optional
+ * SCM_RIGHTS fd (present in BACKLOG_BUF responses). The control buffer
+ * is sized for exactly one fd; MSG_WAITALL ensures we get the full
+ * header + payload in one call (AF_UNIX sends atomically). */
+ void *buf = NULL;
+ size_t total = sizeof(hdr) + (size_t)hdr.size;
+ char *databuf = malloc(total);
+ if (!databuf) return -1;
+
+ char cmsgbuf[CMSG_SPACE(sizeof(int))];
+ memset(cmsgbuf, 0, sizeof(cmsgbuf));
+
+ struct iovec iov = { .iov_base = databuf, .iov_len = total };
+ struct msghdr msg = {
+ .msg_iov = &iov,
+ .msg_iovlen = 1,
+ .msg_control = cmsgbuf,
+ .msg_controllen = sizeof(cmsgbuf),
+ };
+
+ ssize_t n = recvmsg(fd, &msg, MSG_WAITALL);
+ if (n != (ssize_t)total)
+ {
+ free(databuf);
+ return (n == 0) ? -1 : -1;
+ }
+
+ /* Re-read the header from the data buffer (consumed by recvmsg) */
+ memcpy(&hdr, databuf, sizeof(hdr));
+
+ /* Extract ancillary fd if present */
+ struct cmsghdr *cm = CMSG_FIRSTHDR(&msg);
+ if (cm && cm->cmsg_level == SOL_SOCKET &&
+ cm->cmsg_type == SCM_RIGHTS &&
+ cm->cmsg_len >= CMSG_LEN(sizeof(int)))
+ {
+ int recv_fd;
+ memcpy(&recv_fd, CMSG_DATA(cm), sizeof(int));
+ if (fd_out)
+ *fd_out = recv_fd;
+ else
+ close(recv_fd); /* caller did not want it — prevent leak */
+ }
+
if (hdr.size > 0)
{
- void *buf = malloc(hdr.size);
- if (!buf) return -1;
- if (_recv_exact(fd, buf, hdr.size) < 0)
- { free(buf); return -1; }
+ buf = malloc(hdr.size);
+ if (!buf)
+ {
+ free(databuf);
+ if (fd_out && *fd_out >= 0) { close(*fd_out); *fd_out = -1; }
+ return -1;
+ }
+ memcpy(buf, databuf + sizeof(hdr), hdr.size);
*payload_out = buf;
}
+ free(databuf);
*size_out = hdr.size;
return (int)hdr.type;
}
@@ -297,43 +314,5 @@ termsrv_send_backlog_buf(int sock_fd, int buf_fd,
return (n == (ssize_t)(sizeof(hdr) + sizeof(pl))) ? 0 : -1;
}
-int
-termsrv_recv_backlog_buf(int sock_fd, int *buf_fd_out,
- uint32_t *buf_id_out, uint32_t *cols_out,
- uint32_t *num_lines_out, uint32_t *max_lines_out)
-{
- TermSrvMsgHdr hdr;
- TermSrvMsgBacklogBuf pl;
- char cmsgbuf[CMSG_SPACE(sizeof(int))];
- memset(cmsgbuf, 0, sizeof(cmsgbuf));
- struct iovec iov[2] = {
- { &hdr, sizeof(hdr) },
- { &pl, sizeof(pl) }
- };
- struct msghdr msg = {
- .msg_iov = iov,
- .msg_iovlen = 2,
- .msg_control = cmsgbuf,
- .msg_controllen = sizeof(cmsgbuf)
- };
-
- *buf_fd_out = -1;
- ssize_t n = recvmsg(sock_fd, &msg, MSG_WAITALL);
- if (n < (ssize_t)(sizeof(hdr) + sizeof(pl))) return -1;
- if (msg.msg_flags & MSG_CTRUNC) return -1;
- if (hdr.type != TSRV_MSG_BACKLOG_BUF) return -1;
- if (hdr.size < sizeof(TermSrvMsgBacklogBuf)) return -1;
-
- struct cmsghdr *cm = CMSG_FIRSTHDR(&msg);
- if (!cm || cm->cmsg_type != SCM_RIGHTS ||
- cm->cmsg_len < CMSG_LEN(sizeof(int))) return -1;
- memcpy(buf_fd_out, CMSG_DATA(cm), sizeof(int));
-
- *buf_id_out = pl.buf_id;
- *cols_out = pl.cols;
- *num_lines_out = pl.num_lines;
- *max_lines_out = pl.max_lines;
- return 0;
-}
#endif /* HAVE_TYMUX */
diff --git a/src/bin/termsrv.h b/src/bin/termsrv.h
index 035db6c7..41050d3c 100644
--- a/src/bin/termsrv.h
+++ b/src/bin/termsrv.h
@@ -188,8 +188,13 @@ int termsrv_msg_send(int fd, TermSrvMsgType type,
* Returns 0 if the fd is non-blocking and no data is available yet
* (EAGAIN/EWOULDBLOCK) — caller should retry on the next fd-ready event.
* Returns -1 on EOF or protocol error.
- * *payload_out is NULL and *size_out is 0 when there is no payload. */
-int termsrv_msg_recv(int fd, void **payload_out, uint32_t *size_out);
+ * *payload_out is NULL and *size_out is 0 when there is no payload.
+ *
+ * If fd_out is non-NULL and the message carries an SCM_RIGHTS fd,
+ * *fd_out receives it; otherwise *fd_out is set to -1.
+ * Pass NULL for fd_out if the caller is not expecting an fd (e.g. daemon). */
+int termsrv_msg_recv(int fd, void **payload_out, uint32_t *size_out,
+ int *fd_out);
/* Send ATTACHED with shm memfd (and optional backlog memfd) via SCM_RIGHTS.
* Pass backlog_fd >= 0 to include the head backlog buffer fd.
@@ -215,12 +220,6 @@ int termsrv_send_backlog_buf(int sock_fd, int buf_fd,
uint32_t buf_id, uint32_t cols,
uint32_t num_lines, uint32_t max_lines);
-/* Receive BACKLOG_BUF and extract the buffer memfd.
- * Blocking receive — caller must ensure no concurrent reads on sock_fd.
- * Returns 0/-1. */
-int termsrv_recv_backlog_buf(int sock_fd, int *buf_fd_out,
- uint32_t *buf_id_out, uint32_t *cols_out,
- uint32_t *num_lines_out, uint32_t *max_lines_out);
#endif /* HAVE_TYMUX */
#endif /* TERMINOLOGY_TERMSRV_H_ */
diff --git a/src/bin/termtymux.c b/src/bin/termtymux.c
index 72203836..aa0a3668 100644
--- a/src/bin/termtymux.c
+++ b/src/bin/termtymux.c
@@ -19,6 +19,11 @@
#include <stdatomic.h>
#include <time.h>
+/* ── forward declarations ───────────────────────────────────────────── */
+
+static void _backlog_update_backsize(TermTymux *ts);
+static void _backlog_request_async(TermTymux *ts, uint32_t buf_id);
+
/* ── backlog buffer helpers ─────────────────────────────────────────── */
/*
@@ -103,78 +108,132 @@ _backlog_bufs_evict_old(TermTymux *ts, uint32_t new_oldest_id)
}
/*
- * Request and mmap a backlog buffer from the daemon.
- *
- * If slot >= 0, the fetched buffer fills the existing NULL entry at
- * ts->backlog_bufs[slot] (called from _backlog_get_cb for a known-position
- * unmapped buffer). If slot < 0, the buffer is appended at the tail
- * (called from the NOTIFY handler for new head buffers).
- *
- * Returns EINA_TRUE on success.
- *
- * NOTE: this temporarily clears O_NONBLOCK on the socket (via
- * termtymux_request_backlog_buf) and must only be called from the main
- * loop (not from an Ecore fd-handler that is itself running inside a
- * non-blocking recv).
+ * Send an async BACKLOG_BUF_REQ to the daemon for buf_id.
+ * No-op if a request is already in flight (one at a time) or the socket
+ * is not open. On success ts->pending_fetch_id is set to buf_id.
*/
-static Eina_Bool
-_backlog_buf_fetch(TermTymux *ts, uint32_t buf_id, int slot)
+static void
+_backlog_request_async(TermTymux *ts, uint32_t buf_id)
{
- int fd = -1;
- uint32_t cols = 0, num_lines = 0, max_lines = 0;
- size_t map_size;
- void *ptr;
- TermBacklogBufHeader *hdr;
+ TermSrvMsgBacklogBufReq req;
- if (termtymux_request_backlog_buf(ts, buf_id,
- &fd, &cols,
- &num_lines, &max_lines) < 0)
- return EINA_FALSE;
+ if (ts->pending_fetch_id >= 0)
+ return; /* one request in flight at a time */
+ if (ts->sock_fd < 0)
+ return;
- if (cols == 0 || cols > TERMSRV_MAX_COLS)
+ req.buf_id = buf_id;
+ if (termsrv_msg_send(ts->sock_fd, TSRV_MSG_BACKLOG_BUF_REQ,
+ &req, sizeof(req)) < 0)
{
- ERR("termtymux: backlog buf %u has invalid cols %u", buf_id, cols);
- close(fd);
- return EINA_FALSE;
+ ERR("termtymux: send BACKLOG_BUF_REQ buf_id=%u: %s",
+ buf_id, strerror(errno));
+ return;
+ }
+ ts->pending_fetch_id = (int)buf_id;
+}
+
+/*
+ * Handle an incoming BACKLOG_BUF response: mmap the fd, validate the
+ * buffer header, install it in the tracking array, then chain-prefetch
+ * the next older buffer if one is still unmapped.
+ *
+ * Called from _cb_tymux_data when TSRV_MSG_BACKLOG_BUF is received.
+ * Takes ownership of recv_fd (closes it after mmap or on any error path).
+ */
+static void
+_backlog_handle_buf_response(TermTymux *ts,
+ int recv_fd,
+ const TermSrvMsgBacklogBuf *bp)
+{
+ size_t map_size;
+ void *ptr;
+ TermBacklogBufHeader *hdr;
+ int slot = -1;
+ int j;
+
+ if (bp->cols == 0 || bp->cols > TERMSRV_MAX_COLS)
+ {
+ WRN("termtymux: BACKLOG_BUF buf_id=%u invalid cols %u",
+ bp->buf_id, bp->cols);
+ close(recv_fd);
+ return;
}
- map_size = TERMSRV_BACKLOG_TOTAL_SIZE(cols);
- ptr = mmap(NULL, map_size, PROT_READ, MAP_SHARED, fd, 0);
- close(fd);
+ map_size = TERMSRV_BACKLOG_TOTAL_SIZE(bp->cols);
+ ptr = mmap(NULL, map_size, PROT_READ, MAP_SHARED, recv_fd, 0);
+ close(recv_fd); /* fd no longer needed after mmap */
if (ptr == MAP_FAILED)
{
- ERR("termtymux: mmap backlog buf %u: %s", buf_id, strerror(errno));
- return EINA_FALSE;
+ ERR("termtymux: mmap BACKLOG_BUF buf_id=%u: %s",
+ bp->buf_id, strerror(errno));
+ return;
}
hdr = ptr;
if (hdr->magic != TERMSRV_BACKLOG_MAGIC ||
hdr->version != TERMSRV_BACKLOG_VERSION)
{
- WRN("termtymux: backlog buf %u magic/version mismatch", buf_id);
+ WRN("termtymux: BACKLOG_BUF buf_id=%u magic/version mismatch",
+ bp->buf_id);
munmap(ptr, map_size);
- return EINA_FALSE;
+ return;
+ }
+
+ /* Find the slot in the tracking array that matches this buf_id */
+ for (j = 0; j < ts->backlog_buf_count; j++)
+ {
+ if (ts->backlog_bufs[j].buf_id == bp->buf_id)
+ {
+ slot = j;
+ break;
+ }
}
if (slot >= 0)
{
- /* Fill the pre-existing NULL slot directly — no array growth needed */
- TermBacklogBuf *entry = &ts->backlog_bufs[slot];
- entry->hdr = hdr;
- entry->size = map_size;
- /* entry->buf_id is already set */
+ if (ts->backlog_bufs[slot].hdr != NULL)
+ {
+ /* Already mapped — daemon sent a duplicate, discard */
+ WRN("termtymux: BACKLOG_BUF buf_id=%u already mapped, ignoring",
+ bp->buf_id);
+ munmap(ptr, map_size);
+ return;
+ }
+ ts->backlog_bufs[slot].hdr = hdr;
+ ts->backlog_bufs[slot].size = map_size;
}
else
{
- /* Append new head buffer */
- if (!_backlog_buf_push(ts, hdr, map_size, buf_id))
+ /* buf_id not in array — treat as a new head buffer */
+ if (!_backlog_buf_push(ts, hdr, map_size, bp->buf_id))
{
munmap(ptr, map_size);
- return EINA_FALSE;
+ return;
}
+ slot = ts->backlog_buf_count - 1;
}
- return EINA_TRUE;
+ /* Clear in-flight flag if this is the response we were waiting for */
+ if (ts->pending_fetch_id == (int)bp->buf_id)
+ ts->pending_fetch_id = -1;
+
+ /* Update scrollback size so the renderer sees the new lines */
+ _backlog_update_backsize(ts);
+
+ /* Notify renderer that new backlog data is available */
+ if (ts->cb_change.func)
+ ts->cb_change.func(ts->cb_change.data);
+
+ /* Chain-prefetch: request the next older unmapped buffer */
+ for (j = slot - 1; j >= 0; j--)
+ {
+ if (!ts->backlog_bufs[j].hdr)
+ {
+ _backlog_request_async(ts, ts->backlog_bufs[j].buf_id);
+ break;
+ }
+ }
}
/*
@@ -228,10 +287,11 @@ _backlog_update_backsize(TermTymux *ts)
*
* We walk the buffer chain newest-to-oldest, counting lines until we find
* the buffer that owns offset = -y. For lines that land in a buffer that
- * has not been mmap'd yet we request it on the fly.
+ * has not been mmap'd yet, an async request is sent and NULL is returned;
+ * the response handler triggers a re-render once the data arrives.
*
* Returns a pointer into the mmap'd cells on success; NULL if y is out
- * of range or a fetch fails.
+ * of range or the buffer is not yet fetched.
*/
static Termcell *
_backlog_get_cb(void *data, int y, ssize_t *wret)
@@ -281,11 +341,12 @@ _backlog_get_cb(void *data, int y, ssize_t *wret)
/* Ensure buffer is mmap'd */
if (!b->hdr)
{
- /* Pass slot=i so _backlog_buf_fetch fills this existing entry
- * directly rather than appending a duplicate at the tail. */
- if (!_backlog_buf_fetch(ts, b->buf_id, i))
- return NULL;
- b = &ts->backlog_bufs[i]; /* re-look-up in case array was reallocated */
+ /* Not yet fetched — fire an async request if one is not already
+ * in flight. Return NULL for now; the response handler will
+ * trigger a re-render once the data arrives. */
+ if (ts->pending_fetch_id < 0)
+ _backlog_request_async(ts, b->buf_id);
+ return NULL;
}
hdr = b->hdr;
if (!hdr)
@@ -303,6 +364,18 @@ _backlog_get_cb(void *data, int y, ssize_t *wret)
}
*wret = (ssize_t)hdr->line_widths[line_idx];
+
+ /* Proximity prefetch: if we are accessing a line in the older 20%
+ * of this buffer, speculatively request the next older unmapped buffer
+ * so it is ready before the user scrolls past the boundary. */
+ if (i > 0 && local_offset > (n_lines * 4u / 5u))
+ {
+ TermBacklogBuf *prev = &ts->backlog_bufs[i - 1];
+
+ if (!prev->hdr && ts->pending_fetch_id < 0)
+ _backlog_request_async(ts, prev->buf_id);
+ }
+
return TERMSRV_BACKLOG_LINE(hdr, line_idx);
}
@@ -496,8 +569,9 @@ _cb_tymux_data(void *data, Ecore_Fd_Handler *handler EINA_UNUSED)
void *payload = NULL;
uint32_t size = 0;
int type;
+ int recv_fd = -1;
- type = termsrv_msg_recv(ts->sock_fd, &payload, &size);
+ type = termsrv_msg_recv(ts->sock_fd, &payload, &size, &recv_fd);
if (type == 0)
{
/* EAGAIN — no complete message available yet, wait for next wakeup */
@@ -542,17 +616,21 @@ _cb_tymux_data(void *data, Ecore_Fd_Handler *handler EINA_UNUSED)
if (next_new_id < new_oldest)
next_new_id = new_oldest;
- /* Fetch new head buffer(s) produced since the last NOTIFY.
+ /* Register new head buffer(s) as placeholder entries and
+ * request the newest one asynchronously. The response
+ * handler will chain-fetch older buffers automatically.
* new_head = oldest + count - 1; we want every id from
* next_new_id through new_head. */
if (new_count > 0)
{
uint32_t new_head = new_oldest + new_count - 1;
uint32_t id;
+ uint32_t newest_untracked = 0;
+ Eina_Bool have_untracked = EINA_FALSE;
for (id = next_new_id; id <= new_head; id++)
{
- /* Don't re-fetch an id already tracked */
+ /* Don't duplicate an id already tracked */
Eina_Bool already = EINA_FALSE;
int k;
@@ -565,8 +643,28 @@ _cb_tymux_data(void *data, Ecore_Fd_Handler *handler EINA_UNUSED)
}
}
if (!already)
- _backlog_buf_fetch(ts, id, -1);
+ {
+ /* Add a NULL placeholder — will be filled when
+ * the BACKLOG_BUF response arrives */
+ if (_backlog_bufs_grow(ts))
+ {
+ TermBacklogBuf *slot;
+
+ slot = &ts->backlog_bufs[ts->backlog_buf_count];
+ slot->hdr = NULL;
+ slot->size = 0;
+ slot->buf_id = id;
+ ts->backlog_buf_count++;
+ newest_untracked = id;
+ have_untracked = EINA_TRUE;
+ }
+ }
}
+
+ /* Request the newest untracked buffer; the response
+ * handler will cascade through older unmapped ones */
+ if (have_untracked)
+ _backlog_request_async(ts, newest_untracked);
}
ts->last_backlog_buf_count = new_count;
@@ -606,16 +704,36 @@ _cb_tymux_data(void *data, Ecore_Fd_Handler *handler EINA_UNUSED)
break;
case TSRV_MSG_EXIT:
+ if (recv_fd >= 0) { close(recv_fd); recv_fd = -1; }
if (ts->cb_exit.func)
ts->cb_exit.func(ts->cb_exit.data);
free(payload);
return ECORE_CALLBACK_CANCEL;
+ case TSRV_MSG_BACKLOG_BUF:
+ if (recv_fd >= 0 && payload &&
+ size >= sizeof(TermSrvMsgBacklogBuf))
+ {
+ _backlog_handle_buf_response(ts, recv_fd,
+ (TermSrvMsgBacklogBuf *)payload);
+ recv_fd = -1; /* ownership transferred to response handler */
+ }
+ else
+ {
+ WRN("termtymux: BACKLOG_BUF missing fd or payload");
+ /* recv_fd closed below */
+ }
+ break;
+
default:
WRN("termtymux: unexpected message type %d", type);
break;
}
+ /* Close any ancillary fd that was not consumed by a case handler */
+ if (recv_fd >= 0)
+ close(recv_fd);
+
free(payload);
return ECORE_CALLBACK_RENEW;
}
@@ -865,71 +983,125 @@ termtymux_attach(const char *tymux_name,
ts->oldest_buf_id = oldest_buf_id;
ts->last_backlog_buf_count = backlog_buf_count;
+ ts->pending_fetch_id = -1;
- /* mmap the head backlog buffer received in ATTACHED.
- * Use a two-stage map: first map just the header page to read the actual
- * cols width, then re-map at the correct total size. Mapping at
- * TERMSRV_BACKLOG_TOTAL_SIZE(TERMSRV_MAX_COLS) when the file is smaller
- * (because the session was started with fewer columns) would cause SIGBUS
- * on any access past the file's actual end. */
- if (backlog_fd >= 0 && backlog_buf_count > 0)
+ /* Build the backlog tracking array.
+ *
+ * The daemon told us there are backlog_buf_count buffers with ids
+ * [oldest_buf_id .. head_id]. It sent us the head buffer's memfd in
+ * the ATTACHED message. All older buffers are registered as NULL
+ * placeholders here so that _backlog_get_cb knows they exist and can
+ * trigger async fetches; they will be filled as BACKLOG_BUF responses
+ * arrive.
+ *
+ * Array order: oldest (index 0) → newest (index backlog_buf_count-1).
+ *
+ * Use a two-stage mmap for the head: first map just the header page to
+ * read the actual cols width, then re-map at the correct total size.
+ * Mapping at TERMSRV_BACKLOG_TOTAL_SIZE(TERMSRV_MAX_COLS) when the file
+ * is smaller would cause SIGBUS on access past the file's actual end. */
+ if (backlog_buf_count > 0)
{
- TermBacklogBufHeader *hdr_tmp;
- void *hdr_page;
- uint32_t head_cols = 0;
uint32_t head_id = oldest_buf_id + backlog_buf_count - 1;
+ uint32_t id;
+ TermBacklogBufHeader *head_hdr = NULL;
+ size_t head_size = 0;
- hdr_page = mmap(NULL, TERMSRV_BACKLOG_HEADER_SIZE,
- PROT_READ, MAP_SHARED, backlog_fd, 0);
- if (hdr_page != MAP_FAILED)
+ /* Step 1: mmap the head buffer if its fd was provided */
+ if (backlog_fd >= 0)
{
- hdr_tmp = hdr_page;
- if (hdr_tmp->magic == TERMSRV_BACKLOG_MAGIC &&
- hdr_tmp->version == TERMSRV_BACKLOG_VERSION &&
- hdr_tmp->cols > 0 &&
- hdr_tmp->cols <= TERMSRV_MAX_COLS)
- head_cols = hdr_tmp->cols;
- munmap(hdr_page, TERMSRV_BACKLOG_HEADER_SIZE);
- }
- else
- WRN("termtymux: mmap head backlog buf header: %s", strerror(errno));
+ TermBacklogBufHeader *hdr_tmp;
+ void *hdr_page;
+ uint32_t head_cols = 0;
- if (head_cols > 0)
- {
- size_t map_size = TERMSRV_BACKLOG_TOTAL_SIZE(head_cols);
- void *ptr;
-
- ptr = mmap(NULL, map_size, PROT_READ, MAP_SHARED, backlog_fd, 0);
- close(backlog_fd);
- backlog_fd = -1;
-
- if (ptr != MAP_FAILED)
+ hdr_page = mmap(NULL, TERMSRV_BACKLOG_HEADER_SIZE,
+ PROT_READ, MAP_SHARED, backlog_fd, 0);
+ if (hdr_page != MAP_FAILED)
{
- TermBacklogBufHeader *hdr = ptr;
+ hdr_tmp = hdr_page;
+ if (hdr_tmp->magic == TERMSRV_BACKLOG_MAGIC &&
+ hdr_tmp->version == TERMSRV_BACKLOG_VERSION &&
+ hdr_tmp->cols > 0 &&
+ hdr_tmp->cols <= TERMSRV_MAX_COLS)
+ head_cols = hdr_tmp->cols;
+ munmap(hdr_page, TERMSRV_BACKLOG_HEADER_SIZE);
+ }
+ else
+ WRN("termtymux: mmap head backlog buf header: %s",
+ strerror(errno));
- if (hdr->magic == TERMSRV_BACKLOG_MAGIC &&
- hdr->version == TERMSRV_BACKLOG_VERSION)
+ if (head_cols > 0)
+ {
+ void *ptr;
+
+ head_size = TERMSRV_BACKLOG_TOTAL_SIZE(head_cols);
+ ptr = mmap(NULL, head_size, PROT_READ, MAP_SHARED,
+ backlog_fd, 0);
+ if (ptr != MAP_FAILED)
{
- if (!_backlog_buf_push(ts, hdr, map_size, head_id))
+ TermBacklogBufHeader *h = ptr;
+
+ if (h->magic == TERMSRV_BACKLOG_MAGIC &&
+ h->version == TERMSRV_BACKLOG_VERSION)
+ head_hdr = h;
+ else
{
- WRN("termtymux: failed to push head backlog buf");
- munmap(ptr, map_size);
+ WRN("termtymux: head backlog buf magic/version mismatch");
+ munmap(ptr, head_size);
+ head_size = 0;
}
}
else
- {
- WRN("termtymux: head backlog buf magic/version mismatch");
- munmap(ptr, map_size);
- }
+ WRN("termtymux: mmap head backlog buf: %s", strerror(errno));
}
- else
- WRN("termtymux: mmap head backlog buf: %s", strerror(errno));
- }
- else
- {
+
close(backlog_fd);
backlog_fd = -1;
}
+
+ /* Step 2: add NULL placeholder entries for all older buffers */
+ for (id = oldest_buf_id; id < head_id; id++)
+ {
+ if (!_backlog_bufs_grow(ts))
+ break;
+ {
+ TermBacklogBuf *slot = &ts->backlog_bufs[ts->backlog_buf_count];
+
+ slot->hdr = NULL;
+ slot->size = 0;
+ slot->buf_id = id;
+ ts->backlog_buf_count++;
+ }
+ }
+
+ /* Step 3: push the head buffer (with real mapping, or NULL if mmap
+ * failed — the async path will fetch it later) */
+ if (!_backlog_buf_push(ts, head_hdr, head_size, head_id))
+ {
+ WRN("termtymux: failed to push head backlog buf");
+ if (head_hdr)
+ {
+ munmap((void *)head_hdr, head_size);
+ head_hdr = NULL;
+ head_size = 0;
+ }
+ /* Still add a NULL placeholder for the head so backlog_get
+ * can kick an async fetch */
+ if (_backlog_bufs_grow(ts))
+ {
+ TermBacklogBuf *slot = &ts->backlog_bufs[ts->backlog_buf_count];
+
+ slot->hdr = NULL;
+ slot->size = 0;
+ slot->buf_id = head_id;
+ ts->backlog_buf_count++;
+ }
+ }
+
+ /* Step 4: if there are older unmapped buffers, request the newest
+ * one; the response handler will chain through the rest */
+ if (backlog_buf_count > 1)
+ _backlog_request_async(ts, head_id - 1);
}
else if (backlog_fd >= 0)
{
@@ -1077,53 +1249,5 @@ termtymux_resize(TermTymux *ts, int cols, int rows)
ERR("termtymux: send RESIZE: %s", strerror(errno));
}
-int
-termtymux_request_backlog_buf(TermTymux *ts, uint32_t buf_id,
- int *fd_out, uint32_t *cols_out,
- uint32_t *num_lines_out,
- uint32_t *max_lines_out)
-{
- TermSrvMsgBacklogBufReq req;
- int flags;
- int ret;
-
- EINA_SAFETY_ON_NULL_RETURN_VAL(ts, -1);
- EINA_SAFETY_ON_NULL_RETURN_VAL(fd_out, -1);
- EINA_SAFETY_ON_NULL_RETURN_VAL(cols_out, -1);
- EINA_SAFETY_ON_NULL_RETURN_VAL(num_lines_out, -1);
- EINA_SAFETY_ON_NULL_RETURN_VAL(max_lines_out, -1);
-
- if (ts->sock_fd < 0)
- return -1;
-
- req.buf_id = buf_id;
- if (termsrv_msg_send(ts->sock_fd, TSRV_MSG_BACKLOG_BUF_REQ,
- &req, sizeof(req)) < 0)
- {
- ERR("termtymux: send BACKLOG_BUF_REQ: %s", strerror(errno));
- return -1;
- }
-
- /* Temporarily clear O_NONBLOCK so termsrv_recv_backlog_buf can use
- * MSG_WAITALL to receive header + payload + cmsg atomically.
- * The socket was set non-blocking after the attach handshake; restore
- * it unconditionally after the recv so the Ecore fd handler is not
- * affected regardless of the recv outcome. */
- flags = fcntl(ts->sock_fd, F_GETFL, 0);
- if (flags >= 0)
- fcntl(ts->sock_fd, F_SETFL, flags & ~O_NONBLOCK);
-
- ret = termsrv_recv_backlog_buf(ts->sock_fd, fd_out,
- &buf_id, cols_out,
- num_lines_out, max_lines_out);
- if (ret < 0)
- ERR("termtymux: recv BACKLOG_BUF failed");
-
- /* Restore non-blocking mode regardless of recv outcome */
- if (flags >= 0)
- fcntl(ts->sock_fd, F_SETFL, flags);
-
- return ret;
-}
#endif /* HAVE_TYMUX */
diff --git a/src/bin/termtymux.h b/src/bin/termtymux.h
index 04f4c904..fbcb8152 100644
--- a/src/bin/termtymux.h
+++ b/src/bin/termtymux.h
@@ -36,6 +36,7 @@ struct _TermTymux {
int backlog_buf_alloc; /* allocated slots */
uint32_t oldest_buf_id; /* first valid buffer id from daemon */
uint32_t last_backlog_buf_count; /* daemon's buf_count at last NOTIFY */
+ int pending_fetch_id; /* buf_id of in-flight BACKLOG_BUF_REQ, or -1 */
/* Callbacks — set these after termtymux_attach() returns,
* before the Ecore main loop fires. */
@@ -63,19 +64,6 @@ void termtymux_free(TermTymux *ts);
/* Send a resize event to the daemon. */
void termtymux_resize(TermTymux *ts, int cols, int rows);
-/*
- * Request a specific backlog buffer from the daemon and receive it
- * synchronously. The socket is briefly set to blocking mode for the
- * recv, then restored to O_NONBLOCK.
- *
- * On success *fd_out is a new memfd (caller must close it when done);
- * *cols_out, *num_lines_out and *max_lines_out describe the buffer.
- * Returns 0 on success, -1 on failure.
- */
-int termtymux_request_backlog_buf(TermTymux *ts, uint32_t buf_id,
- int *fd_out, uint32_t *cols_out,
- uint32_t *num_lines_out,
- uint32_t *max_lines_out);
#endif /* HAVE_TYMUX */
#endif /* TERMINOLOGY_TERMTYMUX_H_ */
diff --git a/src/tests/test_termsrv.c b/src/tests/test_termsrv.c
index 9d8639e6..dde4a527 100644
--- a/src/tests/test_termsrv.c
+++ b/src/tests/test_termsrv.c
@@ -56,7 +56,7 @@ test_msg_round_trip(void)
void *payload = NULL;
uint32_t size = 0;
- int type = termsrv_msg_recv(sv[1], &payload, &size);
+ int type = termsrv_msg_recv(sv[1], &payload, &size, NULL);
assert(type == TSRV_MSG_RESIZE);
assert(size == sizeof(TermSrvMsgResize));
TermSrvMsgResize *r = payload;
@@ -65,7 +65,7 @@ test_msg_round_trip(void)
/* zero-payload message */
assert(termsrv_msg_send(sv[0], TSRV_MSG_BELL, NULL, 0) == 0);
- type = termsrv_msg_recv(sv[1], &payload, &size);
+ type = termsrv_msg_recv(sv[1], &payload, &size, NULL);
assert(type == TSRV_MSG_BELL && size == 0 && payload == NULL);
close(sv[0]); close(sv[1]);
--
To stop receiving notification emails like this one, please contact
the administrator of this repository.
