Use nonblocking srt_connect(), to allow faster abort on quit.

[nageru] / nageru / video_encoder.cpp

diff --git a/nageru/video_encoder.cpp b/nageru/video_encoder.cpp
index 4a84ddce1efe786a0f401b2d4e700f358c5482d5..338428ba16e703235002b00ad28d49dfc3a61dcf 100644 (file)
--- a/nageru/video_encoder.cpp
+++ b/nageru/video_encoder.cpp
@@ -114,6 +114,7 @@ VideoEncoder::VideoEncoder(ResourcePool *resource_pool, QSurface *surface, const
 
 VideoEncoder::~VideoEncoder()
 {
+       should_quit = true;
        quicksync_encoder->shutdown();
        x264_encoder.reset(nullptr);
        x264_disk_encoder.reset(nullptr);
@@ -375,6 +376,8 @@ int VideoEncoder::connect_to_srt()
                return -1;
        }
 
+       unique_ptr<addrinfo, decltype(freeaddrinfo) *> ai_cleanup(ai, &freeaddrinfo);
+
        for (const addrinfo *cur = ai; cur != nullptr; cur = cur->ai_next) {
                // Seemingly, srt_create_socket() isn't universal; once we try to connect,
                // it gets locked to either IPv4 or IPv6. So we need to create a new one
@@ -385,18 +388,73 @@ int VideoEncoder::connect_to_srt()
                        return sock;
                }
                ++metric_srt_num_connection_attempts;
+
+               // We do a non-blocking connect, so that we can check should_quit
+               // every now and then.
+               int blocking = 0;
+               if (srt_setsockopt(sock, 0, SRTO_RCVSYN, &blocking, sizeof(blocking)) < 0) {
+                       fprintf(stderr, "srt_setsockopt(SRTO_SNDSYN=0): %s\n", srt_getlasterror_str());
+                       srt_close(sock);
+                       continue;
+               }
                if (srt_connect(sock, cur->ai_addr, cur->ai_addrlen) < 0) {
                        fprintf(stderr, "srt_connect(%s): %s\n", print_addrinfo(cur).c_str(), srt_getlasterror_str());
                        srt_close(sock);
                        continue;
                }
-               fprintf(stderr, "Connected to destination SRT endpoint at %s.\n", print_addrinfo(cur).c_str());
-               freeaddrinfo(ai);
-               return sock;
+               int eid = srt_epoll_create();
+               if (eid < 0) {
+                       fprintf(stderr, "srt_epoll_create(): %s\n", srt_getlasterror_str());
+                       srt_close(sock);
+                       continue;
+               }
+               int modes = SRT_EPOLL_ERR | SRT_EPOLL_OUT;
+               if (srt_epoll_add_usock(eid, sock, &modes) < 0) {
+                       fprintf(stderr, "srt_epoll_usock(): %s\n", srt_getlasterror_str());
+                       srt_close(sock);
+                       srt_epoll_release(eid);
+                       continue;
+               }
+               bool ok;
+               while (!should_quit.load()) {
+                       SRTSOCKET errfds[1], writefds[1];
+                       int num_errfds = 1, num_writefds = 1;
+                       int poll_time_ms = 100;
+                       int ret = srt_epoll_wait(eid, errfds, &num_errfds, writefds, &num_writefds, poll_time_ms, 0, 0, 0, 0);
+                       if (ret < 0) {
+                               if (srt_getlasterror(nullptr) == SRT_ETIMEOUT) {
+                                       continue;
+                               } else {
+                                       fprintf(stderr, "srt_epoll_wait(): %s\n", srt_getlasterror_str());
+                                       srt_close(sock);
+                                       srt_epoll_release(eid);
+                                       return -1;
+                               }
+                       } else if (ret > 0) {
+                               // The SRT epoll framework is pretty odd, but seemingly,
+                               // this is the way. Getting the same error code as srt_connect()
+                               // would normally return seems to be impossible, though.
+                               ok = (num_errfds == 0);
+                               break;
+                               fprintf(stderr, "num_errfds=%d num_writefds=%d last_err=%s\n", num_errfds, num_writefds, srt_getlasterror_str());
+                               break;
+                       }
+               }
+               srt_epoll_release(eid);
+               if (should_quit.load()) {
+                       srt_close(sock);
+                       return -1;
+               }
+               if (ok) {
+                       fprintf(stderr, "Connected to destination SRT endpoint at %s.\n", print_addrinfo(cur).c_str());
+                       return sock;
+               } else {
+                       fprintf(stderr, "srt_connect(%s): %s\n", print_addrinfo(cur).c_str(), srt_getlasterror_str());
+                       srt_close(sock);
+               }
        }
 
        // Out of candidates, so give up.
-       freeaddrinfo(ai);
        return -1;
 }
 
@@ -405,7 +463,7 @@ int VideoEncoder::write_srt_packet(uint8_t *buf, int buf_size)
        if (want_srt_metric_update.exchange(false) && srt_sock != -1) {
                srt_metrics.update_srt_stats(srt_sock);
        }
-       while (buf_size > 0) {
+       while (buf_size > 0 && !should_quit.load()) {
                if (srt_sock == -1) {
                        srt_sock = connect_to_srt();
                        if (srt_sock == -1) {