Skip to content

upipe_xfer: use ueventfd to suspend remote thread #1227

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nto wants to merge 2 commits into
base: master
Choose a base branch
from
Open

upipe_xfer: use ueventfd to suspend remote thread #1227

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ee9b142
ueventfd: add helper to change blocking flag
nto Aug 7, 2026
385884c
upipe_xfer: use ueventfd to suspend remote thread
nto Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 25 additions & 5 deletions include/upipe/ueventfd.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,30 @@ static inline bool ueventfd_write(struct ueventfd *fd)
}
}

/** @This makes the ueventfd blocking or non-blocking.
*
* @param fd pointer to a ueventfd
* @param blocking whether to set or unset the flag
* @return false in case of failure
*/
static inline bool ueventfd_set_blocking(struct ueventfd *fd, bool blocking)
{
int ret = fcntl(fd->event_fd, F_GETFL);
if (unlikely(ret < 0))
return false;

if (blocking)
ret &= ~O_NONBLOCK;
else
ret |= O_NONBLOCK;

ret = fcntl(fd->event_fd, F_SETFL, ret);
if (unlikely(ret < 0))
return false;

return true;
}

/** @This initializes a ueventfd.
*
* @param fd pointer to a ueventfd
Expand All @@ -218,11 +242,7 @@ static inline bool ueventfd_init(struct ueventfd *fd, bool readable)
if (unlikely(ret < 0))
return false;

ret = fcntl(fd->event_fd, F_GETFL);
if (unlikely(ret < 0))
return false;
ret = fcntl(fd->event_fd, F_SETFL | O_NONBLOCK);
if (unlikely(ret < 0))
if (!ueventfd_set_blocking(fd, false))
return false;

if (unlikely(fd->event_fd == -1)) { // eventfd() fails, fallback to pipe()
Expand Down
41 changes: 39 additions & 2 deletions lib/upipe-modules/upipe_transfer.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,10 @@ struct upipe_xfer_mgr {

/** mutual exclusion primitives to access the remote event loop */
struct umutex *mutex;
/** eventfd used to wait until a remote event loop has suspended */
struct ueventfd suspend_fd;
/** eventfd used to resume a remote event loop */
struct ueventfd resume_fd;
/** watcher */
struct upump *upump;
/** remote upump_mgr */
Expand All @@ -68,6 +72,8 @@ enum upipe_xfer_msg_type {
UPIPE_XFER_SET_URI,
/** set output of a pipe */
UPIPE_XFER_SET_OUTPUT,
/** ask the remote upump_mgr to suspend until resume_fd is signalled */
UPIPE_XFER_YIELD,
/** release pipe */
UPIPE_XFER_RELEASE,
/** detach from remote upump_mgr */
Expand Down Expand Up @@ -497,6 +503,8 @@ static void upipe_xfer_mgr_free(struct upipe_mgr *mgr)
upump_stop(xfer_mgr->upump);
upump_free(xfer_mgr->upump);
upump_mgr_release(xfer_mgr->upump_mgr);
ueventfd_clean(&xfer_mgr->suspend_fd);
ueventfd_clean(&xfer_mgr->resume_fd);
uqueue_clean(&xfer_mgr->uqueue);
umutex_release(xfer_mgr->mutex);
upipe_xfer_mgr_vacuum(mgr);
Expand Down Expand Up @@ -526,6 +534,10 @@ static void upipe_xfer_mgr_worker(struct upump *upump)
upipe_set_output(msg->upipe_remote, msg->arg.pipe);
upipe_release(msg->arg.pipe);
break;
case UPIPE_XFER_YIELD:
ueventfd_write(&xfer_mgr->suspend_fd);
ueventfd_read(&xfer_mgr->resume_fd);
break;
case UPIPE_XFER_RELEASE:
upipe_release(msg->upipe_remote);
break;
Expand Down Expand Up @@ -632,8 +644,14 @@ static int _upipe_xfer_mgr_attach(struct upipe_mgr *mgr,
static inline int _upipe_xfer_mgr_freeze(struct upipe_mgr *mgr)
{
struct upipe_xfer_mgr *xfer_mgr = upipe_xfer_mgr_from_upipe_mgr(mgr);
union upipe_xfer_arg arg = { .pipe = NULL };
int ret = upipe_xfer_mgr_send(mgr, UPIPE_XFER_YIELD, NULL, arg);
if (unlikely(!ubase_check(ret)))
return ret;
if (unlikely(!ueventfd_read(&xfer_mgr->suspend_fd)))
return UBASE_ERR_EXTERNAL;
upipe_mgr_use(mgr);
return umutex_lock(xfer_mgr->mutex);
return UBASE_ERR_NONE;
}

/** @This thaws the remote event loop previously frozen by @ref
Expand All @@ -645,7 +663,9 @@ static inline int _upipe_xfer_mgr_freeze(struct upipe_mgr *mgr)
static inline int _upipe_xfer_mgr_thaw(struct upipe_mgr *mgr)
{
struct upipe_xfer_mgr *xfer_mgr = upipe_xfer_mgr_from_upipe_mgr(mgr);
int err = umutex_unlock(xfer_mgr->mutex);
int err = UBASE_ERR_NONE;
if (unlikely(!ueventfd_write(&xfer_mgr->resume_fd)))
err = UBASE_ERR_EXTERNAL;
upipe_mgr_release(mgr);
return err;
}
Expand Down Expand Up @@ -713,6 +733,23 @@ struct upipe_mgr *upipe_xfer_mgr_alloc(uint8_t queue_length,
ulifo_init(&xfer_mgr->msg_pool, msg_pool_depth,
xfer_mgr->extra + uqueue_sizeof(queue_length));

if (!ueventfd_init(&xfer_mgr->suspend_fd, false)) {
free(xfer_mgr);
return NULL;
}
if (!ueventfd_init(&xfer_mgr->resume_fd, false)) {
ueventfd_clean(&xfer_mgr->suspend_fd);
free(xfer_mgr);
return NULL;
}
if (!ueventfd_set_blocking(&xfer_mgr->suspend_fd, true) ||
!ueventfd_set_blocking(&xfer_mgr->resume_fd, true)) {
ueventfd_clean(&xfer_mgr->suspend_fd);
ueventfd_clean(&xfer_mgr->resume_fd);
free(xfer_mgr);
return NULL;
}

struct upipe_mgr *mgr = upipe_xfer_mgr_to_upipe_mgr(xfer_mgr);
urefcount_init(upipe_xfer_mgr_to_urefcount(xfer_mgr),
upipe_xfer_mgr_detach);
Expand Down
Loading