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 <zhilizhao@tencent.com>

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

libavformat/network.c

@@ -29,6 +29,9 @@
 #endif
 
 #include <fcntl.h>
+#if HAVE_EVENTFD
+#include <sys/eventfd.h>
+#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();

libavformat/network.h

@@ -22,6 +22,7 @@
 #define AVFORMAT_NETWORK_H
 
 #include <errno.h>
+#include <stdatomic.h>
 #include <stdint.h>
 
 #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