From 2f74158ddf8c6fc33df94f867cf5b58d72c0e62f Mon Sep 17 00:00:00 2001 From: Zhao Zhili Date: Sat, 10 Oct 2026 00:47:15 +0800 Subject: [PATCH] avformat/network: add a wakeup channel for blocking waits ff_network_wait_fd() can only return after its poll() slice, so a thread blocked in it notices a request from another thread up to POLLING_TIME late. Add a one-shot channel that a second thread can signal to end the wait at once, together with a wait function that polls it next to the socket. The channel is an eventfd where available and a pipe otherwise. On Windows neither can be waited on together with sockets, so it is a loopback datagram socket connected to itself. Signed-off-by: Zhao Zhili --- configure | 2 + libavformat/network.c | 118 ++++++++++++++++++++++++++++++++++++++++++ libavformat/network.h | 63 ++++++++++++++++++++++ 3 files changed, 183 insertions(+) diff --git a/configure b/configure index 362e7b12f7..3bb4d9ef80 100755 --- a/configure +++ b/configure @@ -2492,6 +2492,7 @@ SYSTEM_FUNCS=" closesocket CommandLineToArgvW elf_aux_info + eventfd fcntl getaddrinfo getauxval @@ -6893,6 +6894,7 @@ check_func mmap check_func mprotect # Solaris has nanosleep in -lrt, OpenSolaris no longer needs that check_func_headers time.h nanosleep || check_lib nanosleep time.h nanosleep -lrt +check_func_headers sys/eventfd.h eventfd check_func_headers sys/prctl.h prctl check_func sched_getaffinity check_func setrlimit diff --git a/libavformat/network.c b/libavformat/network.c index 2b4c78cfb1..4d2a2a9654 100644 --- a/libavformat/network.c +++ b/libavformat/network.c @@ -29,6 +29,9 @@ #endif #include +#if HAVE_EVENTFD +#include +#endif #include "network.h" #include "tls.h" #include "url.h" @@ -100,6 +103,121 @@ int ff_network_wait_fd_timeout(int fd, int write, int64_t timeout, AVIOInterrupt } } +int ff_network_wakeup_open(FFNetWakeup *w) +{ +#if HAVE_WINSOCK2_H + /* select() and WSAPoll() only accept sockets, so use a datagram socket + * connected to itself: sending a datagram to it makes it readable. + */ + struct sockaddr_in addr = { .sin_family = AF_INET }; + socklen_t len = sizeof(addr); + int ret, fd = ff_socket(AF_INET, SOCK_DGRAM, 0, NULL); + + if (fd < 0) + return ff_neterrno(); + + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0 || + getsockname(fd, (struct sockaddr *)&addr, &len) < 0 || + connect(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) { + ret = ff_neterrno(); + closesocket(fd); + return ret; + } + + w->rfd = w->wfd = fd; +#elif HAVE_EVENTFD + int fd = eventfd(0, EFD_CLOEXEC); + + if (fd < 0) + return AVERROR(errno); + + w->rfd = w->wfd = fd; +#else + int fds[2]; + + if (pipe(fds) < 0) + return AVERROR(errno); + +#if HAVE_FCNTL + for (int i = 0; i < 2; i++) { + if (fcntl(fds[i], F_SETFD, FD_CLOEXEC) < 0) { + int ret = AVERROR(errno); + closesocket(fds[0]); + closesocket(fds[1]); + return ret; + } + } +#endif + + w->rfd = fds[0]; + w->wfd = fds[1]; +#endif + + atomic_init(&w->signaled, 0); + + return 0; +} + +int ff_network_wakeup_signal(FFNetWakeup *w) +{ + /* An eventfd only accepts 8-byte writes; a pipe or a socket takes any + * payload, so the same value serves every channel type. + */ + const uint64_t one = 1; + int ret; + + /* wfd is blocking; writing at most once keeps the write from blocking. + * The flag publishes no other data, so relaxed ordering is enough. + */ + if (atomic_exchange_explicit(&w->signaled, 1, memory_order_relaxed)) + return 0; + +#if HAVE_WINSOCK2_H + ret = send(w->wfd, (const char *)&one, sizeof(one), 0); +#else + ret = write(w->wfd, &one, sizeof(one)); +#endif + + return ret < 0 ? ff_neterrno() : 0; +} + +void ff_network_wakeup_close(FFNetWakeup *w) +{ + if (w->wfd != w->rfd) + closesocket(w->wfd); + closesocket(w->rfd); + + w->rfd = w->wfd = -1; +} + +int ff_network_wait_fd_wakeup(int fd, int write, int timeout_ms, + const FFNetWakeup *w) +{ + int ev = write ? POLLOUT : POLLIN; + struct pollfd p[2] = { + { .fd = w->rfd, .events = POLLIN }, + { .fd = fd, .events = ev }, + }; + int ret; + + /* Return without polling once signaled. This also ends the wait when + * the write in ff_network_wakeup_signal() failed and rfd never became + * readable. + */ + if (atomic_load_explicit(&w->signaled, memory_order_relaxed)) + return AVERROR_EXIT; + + ret = poll(p, 2, timeout_ms); + if (ret < 0) + return ff_neterrno(); + + if (p[0].revents & POLLIN) + return AVERROR_EXIT; + + return p[1].revents & (ev | POLLERR | POLLHUP) ? 0 : AVERROR(EAGAIN); +} + int ff_network_sleep_interruptible(int64_t timeout, AVIOInterruptCB *int_cb) { int64_t wait_start = av_gettime_relative(); diff --git a/libavformat/network.h b/libavformat/network.h index 7c72c902aa..f48b4d7b2a 100644 --- a/libavformat/network.h +++ b/libavformat/network.h @@ -22,6 +22,7 @@ #define AVFORMAT_NETWORK_H #include +#include #include #include "config.h" @@ -107,6 +108,68 @@ int ff_network_wait_fd_timeout(int fd, int write, int64_t timeout, AVIOInterrupt */ int ff_network_sleep_interruptible(int64_t timeout, AVIOInterruptCB *int_cb); +/** + * A channel for waking a thread that is blocked in + * ff_network_wait_fd_wakeup(), from another thread. + * + * Usage: the waiting thread calls ff_network_wait_fd_wakeup(); another + * thread calls ff_network_wakeup_signal() to end that wait; once the + * waiting thread has been joined, ff_network_wakeup_close() is called. + * + * rfd is pollable together with sockets on every platform. It is an + * eventfd, the read end of a pipe, or a loopback datagram socket on + * Windows, where only sockets can be waited on together with sockets. + * wfd is the end ff_network_wakeup_signal() writes to; it equals rfd + * unless a pipe is in use. On systems other than Windows both fds are + * close-on-exec. + */ +typedef struct FFNetWakeup { + int rfd; + int wfd; + atomic_int signaled; +} FFNetWakeup; + +/** + * Create a wakeup channel. On Windows ff_network_init() must have been + * called first. + * + * @return 0 on success, a negative AVERROR code on failure + */ +int ff_network_wakeup_open(FFNetWakeup *w); + +/** + * Wake the thread blocked in ff_network_wait_fd_wakeup() on this channel, + * or the next one that waits on it: the signal is not consumed, so a + * signal sent before the wait starts is not lost. Only the first call has + * an effect; the function may be called any number of times, from any + * thread. + * + * @return 0 on success, a negative AVERROR code if the first call failed + * to write to the channel + */ +int ff_network_wakeup_signal(FFNetWakeup *w); + +/** + * Close the channel. Call it after the waiting thread has been joined and + * no thread can signal any more. + */ +void ff_network_wakeup_close(FFNetWakeup *w); + +/** + * Like ff_network_wait_fd(), but also returns as soon as the wakeup + * channel is signaled, and with a caller chosen timeout. + * + * @param fd Socket descriptor + * @param write Set 1 to wait for the socket to be writable, 0 for readable + * @param timeout_ms Timeout in milliseconds, or -1 to wait indefinitely + * @param w Wakeup channel to watch, must have been opened + * @return 0 if fd is ready, AVERROR(EAGAIN) if the timeout expired, + * AVERROR_EXIT if w was signaled (also when fd is ready), + * or a negative error code + */ +int ff_network_wait_fd_wakeup(int fd, int write, int timeout_ms, + const FFNetWakeup *w); + #if !HAVE_STRUCT_SOCKADDR_STORAGE struct sockaddr_storage { #if HAVE_STRUCT_SOCKADDR_SA_LEN