From patchwork Sun Jan 31 18:32:19 2021 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Marton Balint X-Patchwork-Id: 25288 Return-Path: X-Original-To: patchwork@ffaux-bg.ffmpeg.org Delivered-To: patchwork@ffaux-bg.ffmpeg.org Received: from ffbox0-bg.mplayerhq.hu (ffbox0-bg.ffmpeg.org [79.124.17.100]) by ffaux.localdomain (Postfix) with ESMTP id 3A04044A299 for ; Sun, 31 Jan 2021 20:32:37 +0200 (EET) Received: from [127.0.1.1] (localhost [127.0.0.1]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTP id 2160B689D84; Sun, 31 Jan 2021 20:32:37 +0200 (EET) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from iq.passwd.hu (iq.passwd.hu [217.27.212.140]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTP id 3493A689D3A for ; Sun, 31 Jan 2021 20:32:31 +0200 (EET) Received: from localhost (localhost [127.0.0.1]) by iq.passwd.hu (Postfix) with ESMTP id 209AEE4CDF; Sun, 31 Jan 2021 19:32:31 +0100 (CET) X-Virus-Scanned: amavisd-new at passwd.hu Received: from iq.passwd.hu ([127.0.0.1]) by localhost (iq.passwd.hu [127.0.0.1]) (amavisd-new, port 10024) with ESMTP id ypgl0eZ_juCI; Sun, 31 Jan 2021 19:32:29 +0100 (CET) Received: from bluegene.passwd.hu (localhost [127.0.0.1]) by iq.passwd.hu (Postfix) with ESMTP id 00A67E4EF7; Sun, 31 Jan 2021 19:32:28 +0100 (CET) From: Marton Balint To: ffmpeg-devel@ffmpeg.org Date: Sun, 31 Jan 2021 19:32:19 +0100 Message-Id: <20210131183219.27814-3-cus@passwd.hu> X-Mailer: git-send-email 2.26.2 In-Reply-To: <20210131183219.27814-1-cus@passwd.hu> References: <20210131183219.27814-1-cus@passwd.hu> MIME-Version: 1.0 Subject: [FFmpeg-devel] [PATCH 3/3] avformat/libsrt: fix race condition with libsrt_network_wait_fd and epoll X-BeenThere: ffmpeg-devel@ffmpeg.org X-Mailman-Version: 2.1.20 Precedence: list List-Id: FFmpeg development discussions and patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Reply-To: FFmpeg development discussions and patches Cc: Marton Balint Errors-To: ffmpeg-devel-bounces@ffmpeg.org Sender: "ffmpeg-devel" The way SRT's async / epoll-based IO works is that the event status is stored in the epoll containers. That is, if an event occurs on an SRT socket, and that SRT socket isn't part of any epoll container, then that event is lost. If we later add that socket to an epoll container, we still won't receive the event even if it wasn't serviced. Therefore we create the epoll and put the fd into it right after the connection is established. See http://lists.ffmpeg.org/pipermail/ffmpeg-devel/2021-January/275334.html Signed-off-by: Marton Balint --- libavformat/libsrt.c | 63 +++++++++++++++++++++++++++----------------- 1 file changed, 39 insertions(+), 24 deletions(-) diff --git a/libavformat/libsrt.c b/libavformat/libsrt.c index f10b043c36..233e9096fa 100644 --- a/libavformat/libsrt.c +++ b/libavformat/libsrt.c @@ -163,15 +163,25 @@ static int libsrt_socket_nonblock(int socket, int enable) return srt_setsockopt(socket, 0, SRTO_RCVSYN, &blocking, sizeof(blocking)); } -static int libsrt_network_wait_fd(URLContext *h, int eid, int fd, int write) +static int libsrt_epoll_create(URLContext *h, int fd, int write) { - int ret, len = 1, errlen = 1; int modes = SRT_EPOLL_ERR | (write ? SRT_EPOLL_OUT : SRT_EPOLL_IN); + int eid = srt_epoll_create(); + if (eid < 0) + return libsrt_neterrno(h); + if (srt_epoll_add_usock(eid, fd, &modes) < 0) { + srt_epoll_release(eid); + return libsrt_neterrno(h); + } + return eid; +} + +static int libsrt_network_wait_fd(URLContext *h, int eid, int write) +{ + int ret, len = 1, errlen = 1; SRTSOCKET ready[1]; SRTSOCKET error[1]; - if (srt_epoll_add_usock(eid, fd, &modes) < 0) - return libsrt_neterrno(h); if (write) { ret = srt_epoll_wait(eid, error, &errlen, ready, &len, POLLING_TIME, 0, 0, 0, 0); } else { @@ -185,14 +195,12 @@ static int libsrt_network_wait_fd(URLContext *h, int eid, int fd, int write) } else { ret = errlen ? AVERROR(EIO) : 0; } - if (srt_epoll_remove_usock(eid, fd) < 0) - return libsrt_neterrno(h); return ret; } /* TODO de-duplicate code from ff_network_wait_fd_timeout() */ -static int libsrt_network_wait_fd_timeout(URLContext *h, int eid, int fd, int write, int64_t timeout, AVIOInterruptCB *int_cb) +static int libsrt_network_wait_fd_timeout(URLContext *h, int eid, int write, int64_t timeout, AVIOInterruptCB *int_cb) { int ret; int64_t wait_start = 0; @@ -200,7 +208,7 @@ static int libsrt_network_wait_fd_timeout(URLContext *h, int eid, int fd, int wr while (1) { if (ff_check_interrupt(int_cb)) return AVERROR_EXIT; - ret = libsrt_network_wait_fd(h, eid, fd, write); + ret = libsrt_network_wait_fd(h, eid, write); if (ret != AVERROR(EAGAIN)) return ret; if (timeout > 0) { @@ -225,7 +233,7 @@ static int libsrt_listen(int eid, int fd, const struct sockaddr *addr, socklen_t if (srt_listen(fd, 1)) return libsrt_neterrno(h); - ret = libsrt_network_wait_fd_timeout(h, eid, fd, 1, timeout, &h->interrupt_callback); + ret = libsrt_network_wait_fd_timeout(h, eid, 1, timeout, &h->interrupt_callback); if (ret < 0) return ret; @@ -245,7 +253,7 @@ static int libsrt_listen_connect(int eid, int fd, const struct sockaddr *addr, s if (srt_connect(fd, addr, addrlen) < 0) return libsrt_neterrno(h); - ret = libsrt_network_wait_fd_timeout(h, eid, fd, 1, timeout, &h->interrupt_callback); + ret = libsrt_network_wait_fd_timeout(h, eid, 1, timeout, &h->interrupt_callback); if (ret < 0) { if (will_try_next) { av_log(h, AV_LOG_WARNING, @@ -359,7 +367,7 @@ static int libsrt_set_options_pre(URLContext *h, int fd) static int libsrt_setup(URLContext *h, const char *uri, int flags) { struct addrinfo hints = { 0 }, *ai, *cur_ai; - int port, fd = -1; + int port, fd; SRTContext *s = h->priv_data; const char *p; char buf[256]; @@ -367,7 +375,7 @@ static int libsrt_setup(URLContext *h, const char *uri, int flags) char hostname[1024],proto[1024],path[1024]; char portstr[10]; int64_t open_timeout = 0; - int eid; + int eid, write_eid; av_url_split(proto, sizeof(proto), NULL, 0, hostname, sizeof(hostname), &port, path, sizeof(path), uri); @@ -404,11 +412,6 @@ static int libsrt_setup(URLContext *h, const char *uri, int flags) cur_ai = ai; - eid = srt_epoll_create(); - if (eid < 0) - return libsrt_neterrno(h); - s->eid = eid; - restart: fd = srt_socket(cur_ai->ai_family, cur_ai->ai_socktype, 0); @@ -432,9 +435,14 @@ static int libsrt_setup(URLContext *h, const char *uri, int flags) if (libsrt_socket_nonblock(fd, 1) < 0) av_log(h, AV_LOG_DEBUG, "libsrt_socket_nonblock failed\n"); + ret = write_eid = libsrt_epoll_create(h, fd, 1); + if (ret < 0) + goto fail1; if (s->mode == SRT_MODE_LISTENER) { // multi-client - if ((ret = libsrt_listen(s->eid, fd, cur_ai->ai_addr, cur_ai->ai_addrlen, h, s->listen_timeout)) < 0) + ret = libsrt_listen(write_eid, fd, cur_ai->ai_addr, cur_ai->ai_addrlen, h, s->listen_timeout); + srt_epoll_release(write_eid); + if (ret < 0) goto fail1; srt_close(fd); fd = ret; @@ -442,12 +450,15 @@ static int libsrt_setup(URLContext *h, const char *uri, int flags) if (s->mode == SRT_MODE_RENDEZVOUS) { if (srt_bind(fd, cur_ai->ai_addr, cur_ai->ai_addrlen)) { ret = libsrt_neterrno(h); + srt_epoll_release(write_eid); goto fail1; } } - if ((ret = libsrt_listen_connect(s->eid, fd, cur_ai->ai_addr, cur_ai->ai_addrlen, - open_timeout, h, !!cur_ai->ai_next)) < 0) { + ret = libsrt_listen_connect(write_eid, fd, cur_ai->ai_addr, cur_ai->ai_addrlen, + open_timeout, h, !!cur_ai->ai_next); + srt_epoll_release(write_eid); + if (ret < 0) { if (ret == AVERROR_EXIT) goto fail1; else @@ -468,8 +479,13 @@ static int libsrt_setup(URLContext *h, const char *uri, int flags) h->max_packet_size = packet_size; } + ret = eid = libsrt_epoll_create(h, fd, flags & AVIO_FLAG_WRITE); + if (eid < 0) + goto fail1; + h->is_streamed = 1; s->fd = fd; + s->eid = eid; freeaddrinfo(ai); return 0; @@ -487,7 +503,6 @@ static int libsrt_setup(URLContext *h, const char *uri, int flags) if (fd >= 0) srt_close(fd); freeaddrinfo(ai); - srt_epoll_release(s->eid); return ret; } @@ -644,7 +659,7 @@ static int libsrt_read(URLContext *h, uint8_t *buf, int size) int ret; if (!(h->flags & AVIO_FLAG_NONBLOCK)) { - ret = libsrt_network_wait_fd_timeout(h, s->eid, s->fd, 0, h->rw_timeout, &h->interrupt_callback); + ret = libsrt_network_wait_fd_timeout(h, s->eid, 0, h->rw_timeout, &h->interrupt_callback); if (ret) return ret; } @@ -663,7 +678,7 @@ static int libsrt_write(URLContext *h, const uint8_t *buf, int size) int ret; if (!(h->flags & AVIO_FLAG_NONBLOCK)) { - ret = libsrt_network_wait_fd_timeout(h, s->eid, s->fd, 1, h->rw_timeout, &h->interrupt_callback); + ret = libsrt_network_wait_fd_timeout(h, s->eid, 1, h->rw_timeout, &h->interrupt_callback); if (ret) return ret; } @@ -680,8 +695,8 @@ static int libsrt_close(URLContext *h) { SRTContext *s = h->priv_data; - srt_close(s->fd); srt_epoll_release(s->eid); + srt_close(s->fd); srt_cleanup();