From 1b32a27e06e0eda7d5aebf36dcd18a2998f8092e Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Mon, 28 Sep 2026 20:18:41 -0500 Subject: [PATCH 01/13] Adding a small host test harness --- .gitignore | 1 + test/Makefile | 11 +++++++++++ test/check.h | 25 +++++++++++++++++++++++++ 3 files changed, 37 insertions(+) create mode 100644 test/Makefile create mode 100644 test/check.h diff --git a/.gitignore b/.gitignore index b66e7014..725163d2 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ /toolchain compile_flags.txt divinus +/test/*_test diff --git a/test/Makefile b/test/Makefile new file mode 100644 index 00000000..7f18a080 --- /dev/null +++ b/test/Makefile @@ -0,0 +1,11 @@ +CFLAGS = -std=gnu99 -Wall -Wno-parentheses -I../src \ + -DCLOCK_MONOTONIC_COARSE=CLOCK_MONOTONIC -Dmmap64=mmap +TESTS = + +all: $(TESTS) + @for t in $(TESTS); do ./$$t || exit 1; done + +clean: + rm -f $(TESTS) + +.PHONY: all clean diff --git a/test/check.h b/test/check.h new file mode 100644 index 00000000..cddfb641 --- /dev/null +++ b/test/check.h @@ -0,0 +1,25 @@ +#pragma once + +#include +#include +#include + +static int checkFailures = 0; + +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + fprintf(stderr, "%s:%d: CHECK failed: %s\n", __FILE__, __LINE__, #cond); \ + checkFailures++; \ + } \ + } while (0) + +#define CHECK_DONE() \ + do { \ + if (checkFailures) { \ + fprintf(stderr, "%d check(s) failed\n", checkFailures); \ + return EXIT_FAILURE; \ + } \ + printf("%s: ok\n", __FILE__); \ + return EXIT_SUCCESS; \ + } while (0) From 5ecce7b4f1dee50ab71bb738f92eb3bb21a7bce4 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:27:33 -0500 Subject: [PATCH 02/13] Detaching an RTSP connection once when a request is cut short --- src/rtsp/rtsp.c | 13 +++--- src/rtsp/rtsp.h | 3 +- test/Makefile | 6 ++- test/rtsp_con_test.c | 104 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 116 insertions(+), 10 deletions(-) create mode 100644 test/rtsp_con_test.c diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index 4a61d68d..a33c779e 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -468,7 +468,9 @@ static int __message_proc_sock(struct list_t *e, void *p) __PARSE_ERROR(con); } - if (con->parser_state == __PARSER_S_ERROR) { + if (con->con_state == __CON_S_DISCONNECTED) { + ASSERT(bufpool_detach(con->pool, con) == SUCCESS, ERR("connection detach failed\n")); + } else if (con->parser_state == __PARSER_S_ERROR) { __method_error(con, h); } else { if (con->method != __METHOD_NONE && h->isAuthOn && !isAuthValid) @@ -483,13 +485,8 @@ static int __message_proc_sock(struct list_t *e, void *p) case __METHOD_RECORDING: __method_record(con, h); break; case __METHOD_TEARDOWN: __method_teardown(con, h); break; case __METHOD_NONE: - /* state DISCONNECTED connections should be garbage collected immediately. - but sending thread might watches the connection right now. - so the connection might live at here */ - if (con->con_state != __CON_S_DISCONNECTED) { - ERR("unexpected empty request, forcing disconnect\n"); - con->con_state = __CON_S_DISCONNECTED; - } + ERR("unexpected empty request, forcing disconnect\n"); + con->con_state = __CON_S_DISCONNECTED; ASSERT(bufpool_detach(con->pool, con) == SUCCESS, ERR("connection detach failed\n")); break; default: ERR("unexpected method state\n"); return FAILURE; diff --git a/src/rtsp/rtsp.h b/src/rtsp/rtsp.h index dbd2e120..754fa246 100644 --- a/src/rtsp/rtsp.h +++ b/src/rtsp/rtsp.h @@ -194,8 +194,9 @@ static inline int __read_line(struct connection_item_t *p, char *buf) ERR("message end before delimiter\n"); } + /* the caller detaches: doing it here may reset the connection + * under its feet */ p->con_state = __CON_S_DISCONNECTED; - ASSERT(bufpool_detach(p->pool, p) == SUCCESS, ERR("connection detach failed\n")); return FALSE; } diff --git a/test/Makefile b/test/Makefile index 7f18a080..aef6642a 100644 --- a/test/Makefile +++ b/test/Makefile @@ -1,6 +1,6 @@ CFLAGS = -std=gnu99 -Wall -Wno-parentheses -I../src \ -DCLOCK_MONOTONIC_COARSE=CLOCK_MONOTONIC -Dmmap64=mmap -TESTS = +TESTS = rtsp_con_test all: $(TESTS) @for t in $(TESTS); do ./$$t || exit 1; done @@ -8,4 +8,8 @@ all: $(TESTS) clean: rm -f $(TESTS) +# rtsp.c is included by the test to reach its static functions +rtsp_con_test: rtsp_con_test.c ../src/rtsp/mime.c ../src/hal/tools.c ../src/rtsp/rtsp.c $(wildcard ../src/rtsp/*.h) + $(CC) $(CFLAGS) $(filter-out ../src/rtsp/rtsp.c %.h,$^) -lpthread -o $@ + .PHONY: all clean diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c new file mode 100644 index 00000000..16113282 --- /dev/null +++ b/test/rtsp_con_test.c @@ -0,0 +1,104 @@ +#include +#include +#include +#include +#include +#include + +#ifndef SO_SNDBUFFORCE +#define SO_SNDBUFFORCE SO_SNDBUF +#endif + +#include "check.h" +#undef CHECK +#include "../src/rtsp/rtsp.c" +#undef CHECK +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + fprintf(stderr, "%s:%d: CHECK failed: %s\n", __FILE__, __LINE__, #cond); \ + checkFailures++; \ + } \ + } while (0) + +void request_idr(void) {} + +static rtsp_handle make_server(int max_con) +{ + rtsp_handle h = calloc(1, sizeof(*h)); + pthread_mutex_init(&h->mutex, NULL); + h->audioPt = 255; + h->max_con = max_con; + h->con_pool = __connectionpool_create(max_con); + h->transfer_pool = __transpool_create(max_con); + return h; +} + +/* Returns the client end of a socketpair accepted by the server */ +static int connect_client(rtsp_handle h, struct connection_item_t **con) +{ + int sv[2]; + struct sockaddr_in addr = {}; + + if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv)) return -1; + fcntl(sv[0], F_SETFL, fcntl(sv[0], F_GETFL) | O_NONBLOCK); + if (__connection_list_add(h->con_pool, &h->con_list, sv[0], addr) != SUCCESS) return -1; + for (int i = 0; i < h->max_con; i++) + if (__connection_pool[i].client_fd == sv[0]) *con = &__connection_pool[i]; + return sv[1]; +} + +static void serve_once(rtsp_handle h) +{ + struct sock_select_t socks = {}; + struct timeval tv = { .tv_sec = 1 }; + + socks.h_rtsp = h; + FD_ZERO(&socks.rfds); + list_map_inline(&h->con_list, __set_select_sock, &socks); + select(__find_fd_max(&h->con_list) + 1, &socks.rfds, NULL, NULL, &tv); + list_map_inline(&h->con_list, __message_proc_sock, &socks); + list_sweep(&h->con_list, __connection_is_dead); +} + +static int pool_ref(rtsp_handle h, struct connection_item_t *con) +{ + return h->con_pool->elems[con - __connection_pool].ref_count; +} + +static void send_str(int fd, const char *s) +{ + if (write(fd, s, strlen(s)) != (ssize_t)strlen(s)) perror("write"); +} + +/* A request cut short by the peer must drop the connection exactly once */ +static void test_truncated_request(void) +{ + const char *requests[] = { + "TEARDOWN rtsp://cam/ RTSP/1.0\r\nCSeq: 2\r\n", + "GARBAGE rtsp://cam/ RTSP/1.0\r\nCSeq: 2\r\n", + }; + + for (int i = 0; i < 2; i++) { + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + int peer = connect_client(h, &con); + CHECK(peer >= 0 && con); + + send_str(peer, requests[i]); + shutdown(peer, SHUT_WR); + serve_once(h); + + CHECK(con->con_state == __CON_S_DISCONNECTED); + CHECK(pool_ref(h, con) == 0); + CHECK(h->con_list.list == NULL); + CHECK(list_length(&h->con_pool->free_list) == 1); + close(peer); + } +} + +int main(void) +{ + test_truncated_request(); + CHECK_DONE(); +} From 1e575b00d81c9a914efe12afcda23dff70ac5d69 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:28:23 -0500 Subject: [PATCH 03/13] Sending RTSP responses whole on the non-blocking client socket --- src/rtsp/rtp.c | 17 ++---------- src/rtsp/rtsp.c | 18 ++++++++---- src/rtsp/rtsp.h | 22 ++++++++++++++- test/rtsp_con_test.c | 65 ++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 100 insertions(+), 22 deletions(-) diff --git a/src/rtsp/rtp.c b/src/rtsp/rtp.c index 080a83d7..4091ca13 100644 --- a/src/rtsp/rtp.c +++ b/src/rtsp/rtp.c @@ -5,7 +5,6 @@ #include #include #include -#include #include "rtsp_server.h" #include "common.h" @@ -38,23 +37,11 @@ struct __transfer_set_t { static unsigned int __frame_ts_video, __frame_ts_audio; -static inline void __wait_out(int fd) -{ - struct pollfd p = { .fd = fd, .events = POLLOUT }; - poll(&p, 1, 100); -} - static int __tcp_flush(struct connection_item_t *con) { - unsigned int sent = 0; - while (sent < con->tx_len) { - int r = send(con->client_fd, con->tx_buf + sent, con->tx_len - sent, 0); - if (r > 0) sent += r; - else if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { __wait_out(con->client_fd); } - else { con->tx_len = 0; return FAILURE; } - } + int ret = __tcp_send_all(con->client_fd, con->tx_buf, con->tx_len); con->tx_len = 0; - return SUCCESS; + return ret; } static int __tcp_flush_each(struct list_t *e, void *v) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index a33c779e..96b65407 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -133,18 +133,26 @@ static inline bufpool_handle __transpool_create(int num) /****************************************************************************** * RESPONSE IMPLEMENTATIONS ******************************************************************************/ +/* Formatted locally and sent like the interleaved packets: stdio on a + * non-blocking socket drops what it cannot write, leaving a truncated + * response between RTP frames */ static int __rtsp_write(struct connection_item_t *p, const char *fmt, ...) { + char msg[__RTSP_TCP_BUF_SIZE + 256]; va_list args; - int ret; + int len, ret; - pthread_mutex_lock(&p->write_mutex); va_start(args, fmt); - ret = vfprintf(p->fp_tcp_write, fmt, args); + len = vsnprintf(msg, sizeof(msg), fmt, args); va_end(args); - fflush(p->fp_tcp_write); + ASSERT(len >= 0 && len < (int)sizeof(msg), return FAILURE); + + pthread_mutex_lock(&p->write_mutex); + ret = __tcp_send_all(p->client_fd, (unsigned char *)msg, len); pthread_mutex_unlock(&p->write_mutex); + if (ret != SUCCESS) ERR("send:%s\n", strerror(errno)); + return ret; } @@ -511,7 +519,6 @@ static int __connection_reset(void *v) } FCLOSE(p->fp_tcp_read); - FCLOSE(p->fp_tcp_write); CLOSE(p->client_fd); p->client_fd = 0; @@ -558,7 +565,6 @@ __connection_list_add(bufpool_handle con_pool, struct list_head_t *head, int fd, p->client_fd=fd; ASSERT((p->fp_tcp_read = fdopen(fd, "r")), goto error); - ASSERT((p->fp_tcp_write = fdopen(fd, "w")), goto error); p->tx_len = 0; if (!p->tx_buf) p->tx_buf = malloc(RTSP_TX_BATCH); diff --git a/src/rtsp/rtsp.h b/src/rtsp/rtsp.h index 754fa246..558e38b5 100644 --- a/src/rtsp/rtsp.h +++ b/src/rtsp/rtsp.h @@ -4,6 +4,8 @@ #include #include #include +#include +#include #include "rtsp_server.h" @@ -98,7 +100,6 @@ typedef struct { struct connection_item_t { struct sockaddr_in addr; FILE *fp_tcp_read; - FILE *fp_tcp_write; int client_fd; int track_id; int cseq; @@ -206,6 +207,25 @@ static inline int __read_line(struct connection_item_t *p, char *buf) return !(SCMP(__TERM, buf)); } +static inline void __wait_out(int fd) +{ + struct pollfd p = { .fd = fd, .events = POLLOUT }; + poll(&p, 1, 100); +} + +/* The client socket is non-blocking: retry until everything is queued */ +static inline int __tcp_send_all(int fd, const unsigned char *buf, unsigned int len) +{ + unsigned int sent = 0; + while (sent < len) { + int r = send(fd, buf + sent, len - sent, 0); + if (r > 0) sent += r; + else if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { __wait_out(fd); } + else return FAILURE; + } + return SUCCESS; +} + static inline unsigned long long __get_random_byte(unsigned *ctx) { return (unsigned long long)(((int)(random())*(*ctx)) % 256); diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index 16113282..d30032bf 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -2,6 +2,8 @@ #include #include #include +#include +#include #include #include @@ -97,8 +99,71 @@ static void test_truncated_request(void) } } +struct drain_t { + int fd; + char tail[256]; + int found; +}; + +/* Waits for the server to hit EAGAIN, then reads everything it sent */ +static void *drain_peer(void *v) +{ + struct drain_t *d = v; + char chunk[4096]; + size_t tail_len = 0; + + usleep(100 * 1000); + for (;;) { + struct pollfd pfd = { .fd = d->fd, .events = POLLIN }; + if (poll(&pfd, 1, 1000) <= 0) break; + ssize_t r = read(d->fd, chunk, sizeof(chunk)); + if (r <= 0) break; + for (ssize_t i = 0; i < r; i++) { + if (tail_len == sizeof(d->tail) - 1) { + memmove(d->tail, d->tail + 1, tail_len - 1); + tail_len--; + } + d->tail[tail_len++] = chunk[i]; + } + d->tail[tail_len] = 0; + if (strstr(d->tail, "CSeq: 3\r\nPublic: OPTIONS, DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE\r\n\r\n")) { + d->found = 1; + break; + } + } + return NULL; +} + +/* A response must go out whole even when the socket buffer is full */ +static void test_response_on_full_socket(void) +{ + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + struct drain_t drain = {}; + pthread_t thread; + char junk[1024]; + int size = 4096; + + drain.fd = connect_client(h, &con); + CHECK(drain.fd >= 0 && con); + setsockopt(con->client_fd, SOL_SOCKET, SO_SNDBUF, &size, sizeof(size)); + memset(junk, 'x', sizeof(junk)); + while (write(con->client_fd, junk, sizeof(junk)) > 0); + + send_str(drain.fd, "OPTIONS rtsp://cam/ RTSP/1.0\r\nCSeq: 3\r\n\r\n"); + pthread_create(&thread, NULL, drain_peer, &drain); + serve_once(h); + pthread_join(thread, NULL); + + CHECK(drain.found); + CHECK(con->con_state == __CON_S_INIT); + close(drain.fd); +} + int main(void) { + signal(SIGPIPE, SIG_IGN); test_truncated_request(); + test_response_on_full_socket(); CHECK_DONE(); } From 6bfd5a42bbbbaac58b723cde6d676178373024f3 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:28:57 -0500 Subject: [PATCH 04/13] Closing the RTSP client socket exactly once --- src/rtsp/rtsp.c | 10 +++++++--- test/rtsp_con_test.c | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index 96b65407..0c24596a 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -518,10 +518,14 @@ static int __connection_reset(void *v) DBG("force connection to close\n"); } - FCLOSE(p->fp_tcp_read); + /* the stream owns client_fd: close it once, as another connection may + * already have been given the same number */ + if (p->fp_tcp_read) { + FCLOSE(p->fp_tcp_read); + p->client_fd = 0; + } CLOSE(p->client_fd); - p->client_fd = 0; p->con_state = __CON_S_DISCONNECTED; for (int i = 0; i < sizeof(p->trans) / sizeof(*p->trans); i++) { @@ -572,7 +576,7 @@ __connection_list_add(bufpool_handle con_pool, struct list_head_t *head, int fd, return list_add(head, &(p->list_entry)); error: - __connection_reset(&p->list_entry); + ASSERT(bufpool_detach(con_pool, p) == SUCCESS, ERR("connection detach failed\n")); return FAILURE; } diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index d30032bf..ae417e38 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -11,6 +11,15 @@ #define SO_SNDBUFFORCE SO_SNDBUF #endif +/* Counts explicit close() calls on the client socket made by rtsp.c */ +static int watched_fd = -1, watched_closes; +static int counted_close(int fd) +{ + if (fd == watched_fd) watched_closes++; + return (close)(fd); +} +#define close(fd) counted_close(fd) + #include "check.h" #undef CHECK #include "../src/rtsp/rtsp.c" @@ -160,10 +169,34 @@ static void test_response_on_full_socket(void) close(drain.fd); } +/* fclose() on the read stream already closes the client fd: closing it again + * would hit whichever socket the kernel has given that number meanwhile */ +static void test_client_fd_closed_once(void) +{ + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + int peer = connect_client(h, &con); + int fd = con->client_fd; + + CHECK(peer >= 0 && con); + watched_fd = fd; + watched_closes = 0; + shutdown(peer, SHUT_WR); + serve_once(h); + + CHECK(con->con_state == __CON_S_DISCONNECTED); + CHECK(fcntl(fd, F_GETFD) == -1); + CHECK(watched_closes == 0); + CHECK(con->client_fd == 0 && con->fp_tcp_read == NULL); + watched_fd = -1; + close(peer); +} + int main(void) { signal(SIGPIPE, SIG_IGN); test_truncated_request(); test_response_on_full_socket(); + test_client_fd_closed_once(); CHECK_DONE(); } From 1a4a7369f0c060720652e120cd55d547e178f5b6 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:29:41 -0500 Subject: [PATCH 05/13] Clearing transports between RTSP clients, pacing RTCP on interleaved tracks --- src/rtsp/rtsp.c | 10 +++++++--- test/rtsp_con_test.c | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index 0c24596a..c01faa73 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -334,11 +334,13 @@ static void __method_play(struct connection_item_t *p, rtsp_handle h) "\r\n" , p->cseq); for (int i = 0; i < sizeof(p->trans) / sizeof(*p->trans); i++) { - if (!p->trans[i].server_port_rtp) continue; + if (!p->trans[i].server_port_rtp && !p->trans[i].is_tcp) continue; p->track_id = i; - ASSERT(__bind_rtcp(p) == SUCCESS, return); - ASSERT(__bind_rtp(p) == SUCCESS, return); + if (!p->trans[i].is_tcp) { + ASSERT(__bind_rtcp(p) == SUCCESS, return); + ASSERT(__bind_rtp(p) == SUCCESS, return); + } p->trans[p->track_id].rtp_timestamp = (millis() * 90) & UINT32_MAX; p->trans[p->track_id].rtp_seq = rand_r(&h->ctx); p->trans[p->track_id].rtcp_octet = 0; @@ -544,9 +546,11 @@ static int __connection_reset(void *v) p->cseq = 0; ctx = p->trans[0].rtp_timestamp; + memset(p->trans, 0, sizeof(p->trans)); /* randomize session id to avoid conflict */ p->session_id = __get_random_llu(&ctx); + p->ssrc = (unsigned int)__get_random_llu(&ctx); return SUCCESS; } diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index ae417e38..136bf1f6 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -192,11 +192,46 @@ static void test_client_fd_closed_once(void) close(peer); } +/* Interleaved tracks get the same RTCP pacing as UDP ones, and a reused + * connection slot starts with no transport left from the previous client */ +static void test_tcp_play_and_reuse(void) +{ + static const transport_t clean; + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + char reply[1024]; + int peer = connect_client(h, &con); + + CHECK(peer >= 0 && con); + send_str(peer, "SETUP rtsp://cam/track=0 RTSP/1.0\r\nCSeq: 1\r\n" + "Transport: RTP/AVP/TCP;unicast;interleaved=0-1\r\n\r\n"); + serve_once(h); + send_str(peer, "PLAY rtsp://cam/ RTSP/1.0\r\nCSeq: 2\r\n\r\n"); + serve_once(h); + CHECK(read(peer, reply, sizeof(reply)) > 0); + + CHECK(con->con_state == __CON_S_PLAYING); + CHECK(con->trans[0].is_tcp); + CHECK(con->trans[0].rtcp_tick_org == 150); + CHECK(con->trans[0].rtcp_tick == 150); + + con->ssrc = 0x12345678; + shutdown(peer, SHUT_WR); + serve_once(h); + close(peer); + + CHECK(con->con_state == __CON_S_DISCONNECTED); + CHECK(!memcmp(&con->trans[0], &clean, sizeof(clean))); + CHECK(!memcmp(&con->trans[1], &clean, sizeof(clean))); + CHECK(con->ssrc != 0x12345678); +} + int main(void) { signal(SIGPIPE, SIG_IGN); test_truncated_request(); test_response_on_full_socket(); test_client_fd_closed_once(); + test_tcp_play_and_reuse(); CHECK_DONE(); } From c3633e870b3329ce7123959165cf07636ee46e09 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:30:20 -0500 Subject: [PATCH 06/13] Waiting for the rest of a split interleaved packet from the RTSP client --- src/rtsp/rtsp.c | 39 +++++++++++++++++++++++++++++++-------- test/rtsp_con_test.c | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 8 deletions(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index c01faa73..08de91c4 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -373,6 +373,25 @@ static int __method_teardown(struct connection_item_t *p, rtsp_handle h) * METHOD IMPLEMENTATIONS ******************************************************************************/ +/* Interleaved frames from the client (RTCP receiver reports) may arrive split + * and the socket is non-blocking: wait briefly for the rest instead of leaving + * it to be parsed as a request */ +static int __read_interleaved(struct connection_item_t *con, void *dst, size_t len) +{ + size_t got = 0; + int waits = 0; + + while ((got += fread((char *)dst + got, 1, len - got, con->fp_tcp_read)) < len) { + struct pollfd pfd = { .fd = con->client_fd, .events = POLLIN }; + if (!ferror(con->fp_tcp_read) || errno != EAGAIN || ++waits > 10) + return FAILURE; + clearerr(con->fp_tcp_read); + poll(&pfd, 1, 20); + } + + return SUCCESS; +} + static int __message_proc_sock(struct list_t *e, void *p) { struct connection_item_t *con; @@ -394,15 +413,19 @@ static int __message_proc_sock(struct list_t *e, void *p) int first_char = fgetc(con->fp_tcp_read); if (first_char == '$') { unsigned char head[3]; - if (fread(head, 1, 3, con->fp_tcp_read) == 3) { - int len = (head[1] << 8) | head[2]; - while (len > 0) { - int r = fread(buf, 1, min(len, sizeof(buf)), con->fp_tcp_read); - if (r <= 0) break; - len -= r; - } - DBG("discarded interleaved packet (%d bytes)\n", (head[1] << 8) | head[2]); + int len = -1; + if (__read_interleaved(con, head, 3) == SUCCESS) { + len = (head[1] << 8) | head[2]; + while (len > 0 && __read_interleaved(con, buf, min(len, sizeof(buf))) == SUCCESS) + len -= min(len, sizeof(buf)); + } + if (len != 0) { + ERR("interleaved packet cut short, dropping connection\n"); + con->con_state = __CON_S_DISCONNECTED; + ASSERT(bufpool_detach(con->pool, con) == SUCCESS, ERR("connection detach failed\n")); + return SUCCESS; } + DBG("discarded interleaved packet (%d bytes)\n", (head[1] << 8) | head[2]); return SUCCESS; } else if (first_char != EOF) { ungetc(first_char, con->fp_tcp_read); diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index 136bf1f6..7500619f 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -226,6 +226,38 @@ static void test_tcp_play_and_reuse(void) CHECK(con->ssrc != 0x12345678); } +static void *send_rest_later(void *v) +{ + usleep(30 * 1000); + send_str(*(int *)v, "defgh"); + return NULL; +} + +/* An interleaved packet from the client split across segments is skipped + * whole, and the request after it still gets its answer */ +static void test_split_interleaved_packet(void) +{ + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + pthread_t thread; + char reply[1024] = {}; + int peer = connect_client(h, &con); + + CHECK(peer >= 0 && con); + CHECK(write(peer, "$\x01\x00\x08" "abc", 7) == 7); + pthread_create(&thread, NULL, send_rest_later, &peer); + serve_once(h); + pthread_join(thread, NULL); + + send_str(peer, "OPTIONS rtsp://cam/ RTSP/1.0\r\nCSeq: 4\r\n\r\n"); + serve_once(h); + + CHECK(con->con_state == __CON_S_INIT); + CHECK(read(peer, reply, sizeof(reply) - 1) > 0); + CHECK(strstr(reply, "CSeq: 4\r\n")); + close(peer); +} + int main(void) { signal(SIGPIPE, SIG_IGN); @@ -233,5 +265,6 @@ int main(void) test_response_on_full_socket(); test_client_fd_closed_once(); test_tcp_play_and_reuse(); + test_split_interleaved_packet(); CHECK_DONE(); } From 4a8e7556f459e5937deb579fc8ab3ffafe990c18 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:49:28 -0500 Subject: [PATCH 07/13] Turning away RTSP clients beyond the connection pool without stopping the server --- src/rtsp/rtsp.c | 106 +++++++++++++++++++++++++++++++------------ test/rtsp_con_test.c | 98 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 29 deletions(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index 08de91c4..a21fa755 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -765,46 +765,94 @@ static inline int __bind_rtcp(struct connection_item_t *con ) return FAILURE; } +/* Refused clients come in bursts: one line per burst, not per client */ +static void __accept_warn(const char *reason) +{ + static unsigned int last_ms, dropped; + static char warned; + unsigned int now = millis(); + + if (warned && now - last_ms < 10000) { + dropped++; + return; + } + ERR("refusing RTSP client: %s (%u more refused since last notice)\n", reason, dropped); + warned = 1; + last_ms = now; + dropped = 0; +} + static inline int __accept_proc_sock(rtsp_handle h, int server_fd, struct sock_select_t *p_socks) { + /* kept open to be given up when out of descriptors, so the pending + * client can still be accepted and closed instead of spinning select */ + static int spare_fd = -1; unsigned int len; - int fd; + int fd, full; struct sockaddr_in from_addr; - if (FD_ISSET(server_fd, &p_socks->rfds) != 0) { - /* accept new connection */ - len = sizeof(from_addr); - - fd = accept(server_fd, (struct sockaddr *)&from_addr, - &len); + if (FD_ISSET(server_fd, &p_socks->rfds) == 0) + return SUCCESS; - /* we have selected this fd, but still EAGAIN may occur */ - if (fd < 0){ - ASSERT(errno == EAGAIN, ({ - ERR("accept:%s\n",strerror(errno)); - return FAILURE;})); - return SUCCESS; + if (spare_fd < 0) + spare_fd = open("/dev/null", O_RDONLY | O_CLOEXEC); + + len = sizeof(from_addr); + fd = accept(server_fd, (struct sockaddr *)&from_addr, &len); + + if (fd < 0) { + int err = errno; + switch (err) { + case EBADF: case EINVAL: case ENOTSOCK: case EFAULT: + ERR("accept:%s\n", strerror(err)); + return FAILURE; + case EMFILE: case ENFILE: + if (spare_fd >= 0) { + close(spare_fd); + fd = accept(server_fd, NULL, NULL); + if (fd >= 0) close(fd); + spare_fd = open("/dev/null", O_RDONLY | O_CLOEXEC); + } + __accept_warn(strerror(err)); + return SUCCESS; + case EAGAIN: +#if EWOULDBLOCK != EAGAIN + case EWOULDBLOCK: +#endif + case EINTR: + return SUCCESS; + default: + __accept_warn(strerror(err)); + return SUCCESS; } + } - /* set server fd to non-blocking */ - fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK); - - /* A keyframe is a few hundred KB and the kernel's default TCP send - * buffer (64 KB on small boards) cannot hold it: the interleaved sender - * then spins on partial writes for as long as the link takes to drain, - * hundreds of ms of CPU per keyframe. Ask for a buffer that fits one - * (SO_SNDBUFFORCE bypasses wmem_max, which needs root). */ - { - int sz = 512 * 1024; - if (setsockopt(fd, SOL_SOCKET, SO_SNDBUFFORCE, &sz, sizeof(sz))) - setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &sz, sizeof(sz)); - } + pthread_mutex_lock(&h->con_pool->mutex); + full = h->con_pool->free_list.list == NULL; + pthread_mutex_unlock(&h->con_pool->mutex); + if (full) { + close(fd); + __accept_warn("all connection slots in use"); + return SUCCESS; + } + + fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK); - /* update connection-list exclusively */ - ASSERT(__connection_list_add(h->con_pool, &h->con_list, fd, from_addr) == SUCCESS, - return FAILURE); + /* A keyframe is a few hundred KB and the kernel's default TCP send + * buffer (64 KB on small boards) cannot hold it: the interleaved sender + * then spins on partial writes for as long as the link takes to drain, + * hundreds of ms of CPU per keyframe. Ask for a buffer that fits one + * (SO_SNDBUFFORCE bypasses wmem_max, which needs root). */ + { + int sz = 512 * 1024; + if (setsockopt(fd, SOL_SOCKET, SO_SNDBUFFORCE, &sz, sizeof(sz))) + setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &sz, sizeof(sz)); } + /* on failure the slot reset has already closed fd */ + if (__connection_list_add(h->con_pool, &h->con_list, fd, from_addr) != SUCCESS) + __accept_warn("cannot set up the connection"); + return SUCCESS; } diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index 7500619f..8821b978 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -6,6 +6,8 @@ #include #include #include +#include +#include #ifndef SO_SNDBUFFORCE #define SO_SNDBUFFORCE SO_SNDBUF @@ -258,6 +260,101 @@ static void test_split_interleaved_packet(void) close(peer); } +static int tcp_connect(int port) +{ + struct sockaddr_in addr = { .sin_family = AF_INET, .sin_port = htons(port) }; + int fd = socket(AF_INET, SOCK_STREAM, 0); + + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if (fd < 0 || connect(fd, (struct sockaddr *)&addr, sizeof(addr))) return -1; + return fd; +} + +static int accept_once(rtsp_handle h, int server_fd) +{ + struct sock_select_t socks = {}; + struct timeval tv = { .tv_sec = 1 }; + + FD_ZERO(&socks.rfds); + FD_SET(server_fd, &socks.rfds); + select(server_fd + 1, &socks.rfds, NULL, NULL, &tv); + return __accept_proc_sock(h, server_fd, &socks); +} + +/* 1 when the server closed its end, 0 when the connection stays open */ +static int peer_closed(int fd) +{ + struct pollfd pfd = { .fd = fd, .events = POLLIN }; + char c; + + return poll(&pfd, 1, 500) == 1 && read(fd, &c, 1) == 0; +} + +/* Clients beyond the pool, or beyond the fd limit, are turned away while + * the server keeps serving; a freed slot is taken by the next client */ +static void test_connection_overflow(void) +{ + rtsp_handle h = make_server(RTSP_MAXIMUM_CONNECTIONS); + int server_fd = __bind_tcp(0); + struct sockaddr_in addr; + socklen_t len = sizeof(addr); + int peers[RTSP_MAXIMUM_CONNECTIONS], extra, port; + struct rlimit old_lim, lim; + char reply[1024] = {}; + + CHECK(server_fd > 0); + getsockname(server_fd, (struct sockaddr *)&addr, &len); + port = ntohs(addr.sin_port); + + for (int i = 0; i < RTSP_MAXIMUM_CONNECTIONS; i++) { + peers[i] = tcp_connect(port); + CHECK(peers[i] >= 0); + CHECK(accept_once(h, server_fd) == SUCCESS); + } + CHECK(list_length(&h->con_list) == RTSP_MAXIMUM_CONNECTIONS); + CHECK(h->con_pool->free_list.list == NULL); + + extra = tcp_connect(port); + CHECK(extra >= 0); + CHECK(accept_once(h, server_fd) == SUCCESS); + CHECK(peer_closed(extra)); + CHECK(list_length(&h->con_list) == RTSP_MAXIMUM_CONNECTIONS); + close(extra); + + close(peers[0]); + serve_once(h); + CHECK(list_length(&h->con_list) == RTSP_MAXIMUM_CONNECTIONS - 1); + + /* with no descriptor left the pending client is still accepted and dropped */ + getrlimit(RLIMIT_NOFILE, &old_lim); + extra = tcp_connect(port); + CHECK(extra >= 0); + lim = old_lim; + lim.rlim_cur = dup(0); + close(lim.rlim_cur); + setrlimit(RLIMIT_NOFILE, &lim); + CHECK(accept_once(h, server_fd) == SUCCESS); + setrlimit(RLIMIT_NOFILE, &old_lim); + CHECK(peer_closed(extra)); + CHECK(list_length(&h->con_list) == RTSP_MAXIMUM_CONNECTIONS - 1); + close(extra); + + peers[0] = tcp_connect(port); + CHECK(peers[0] >= 0); + CHECK(accept_once(h, server_fd) == SUCCESS); + CHECK(list_length(&h->con_list) == RTSP_MAXIMUM_CONNECTIONS); + CHECK(!peer_closed(peers[0])); + send_str(peers[0], "OPTIONS rtsp://cam/ RTSP/1.0\r\nCSeq: 5\r\n\r\n"); + serve_once(h); + CHECK(read(peers[0], reply, sizeof(reply) - 1) > 0); + CHECK(strstr(reply, "RTSP/1.0 200 OK\r\nCSeq: 5\r\n")); + + for (int i = 0; i < RTSP_MAXIMUM_CONNECTIONS; i++) close(peers[i]); + for (int i = 0; i < 10 && h->con_list.list; i++) serve_once(h); + CHECK(h->con_list.list == NULL); + close(server_fd); +} + int main(void) { signal(SIGPIPE, SIG_IGN); @@ -266,5 +363,6 @@ int main(void) test_client_fd_closed_once(); test_tcp_play_and_reuse(); test_split_interleaved_packet(); + test_connection_overflow(); CHECK_DONE(); } From 996764873c6ca549f6bc7b94b8e0edcdbd09e7aa Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:49:36 -0500 Subject: [PATCH 08/13] Retrying the RTSP select when a signal interrupts it --- src/rtsp/rtsp.c | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index a21fa755..602a03b1 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -896,7 +896,9 @@ static void *rtspThrFxn(void *v) ASSERT(list_map_inline(&rh->con_list, (__set_select_sock), &socks) == SUCCESS, goto error); - ASSERT((ret_select = select(socks.nfds, &(socks.rfds), NULL, NULL, &(socks.timeout))) >= 0, ({ + ret_select = select(socks.nfds, &(socks.rfds), NULL, NULL, &(socks.timeout)); + if (ret_select < 0 && errno == EINTR) continue; + ASSERT(ret_select >= 0, ({ ERR("select:%s\n", strerror(errno)); goto error;})); From e3f58652601730e9bcb02d393bca61dfcf8963ea Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 10:50:04 -0500 Subject: [PATCH 09/13] Answering GET_PARAMETER keepalives instead of dropping the RTSP client --- src/rtsp/rtsp.c | 14 +++++++++++++- src/rtsp/rtsp.h | 1 + test/rtsp_con_test.c | 23 ++++++++++++++++++++++- 3 files changed, 36 insertions(+), 2 deletions(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index 602a03b1..7568246d 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -38,6 +38,7 @@ extern void request_idr(); #define __STR_SESSION "SESSION" #define __STR_PAUSE "PAUSE" #define __STR_RECORDING "RECORDING" +#define __STR_GET_PARAMETER "GET_PARAMETER" #define __STR_RANGE "RANGE" #define __SPACE " " @@ -65,6 +66,7 @@ static void __method_setup(struct connection_item_t *p, rtsp_handle h); static void __method_play(struct connection_item_t *p, rtsp_handle h); static void __method_pause(struct connection_item_t *p, rtsp_handle h); static void __method_record(struct connection_item_t *p, rtsp_handle h); +static void __method_get_parameter(struct connection_item_t *p, rtsp_handle h); static void __method_error(struct connection_item_t *p, rtsp_handle h); static void *rtspThrFxn(void *v); @@ -168,7 +170,7 @@ static void __method_options(struct connection_item_t *p, rtsp_handle h) { __rtsp_write(p, "RTSP/1.0 200 OK\r\n" "CSeq: %d\r\n" - "Public: OPTIONS, DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE\r\n" + "Public: OPTIONS, DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE, GET_PARAMETER\r\n" "\r\n", p->cseq); } @@ -319,6 +321,14 @@ static void __method_record(struct connection_item_t *p, rtsp_handle h) "RTSP/1.0 " __RESPONCE_STR_METHODNOTALLOWED "\r\n"); } +/* Clients send it empty as a keepalive */ +static void __method_get_parameter(struct connection_item_t *p, rtsp_handle h) +{ + __rtsp_write(p, "RTSP/1.0 200 OK\r\n" + "CSeq: %d\r\n" + "\r\n", p->cseq); +} + static void __method_error(struct connection_item_t *p, rtsp_handle h) { __rtsp_write(p, @@ -452,6 +462,7 @@ static int __message_proc_sock(struct list_t *e, void *p) } else if (SCMP(__STR_RECORDING, buf)) { con->method = __METHOD_RECORDING; } else if (SCMP(__STR_PAUSE, buf)) { con->method = __METHOD_PAUSE; } else if (SCMP(__STR_TEARDOWN, buf)) { con->method = __METHOD_TEARDOWN; + } else if (SCMP(__STR_GET_PARAMETER, buf)) { con->method = __METHOD_GET_PARAMETER; } header++; } @@ -517,6 +528,7 @@ static int __message_proc_sock(struct list_t *e, void *p) case __METHOD_PAUSE: __method_pause(con, h); break; case __METHOD_RECORDING: __method_record(con, h); break; case __METHOD_TEARDOWN: __method_teardown(con, h); break; + case __METHOD_GET_PARAMETER: __method_get_parameter(con, h); break; case __METHOD_NONE: ERR("unexpected empty request, forcing disconnect\n"); con->con_state = __CON_S_DISCONNECTED; diff --git a/src/rtsp/rtsp.h b/src/rtsp/rtsp.h index 558e38b5..8e449bf9 100644 --- a/src/rtsp/rtsp.h +++ b/src/rtsp/rtsp.h @@ -60,6 +60,7 @@ enum __method_e { __METHOD_TEARDOWN, __METHOD_PAUSE, __METHOD_RECORDING, + __METHOD_GET_PARAMETER, __METHOD_AUTH, __METHOD_NONE, __METHOD_COUNT diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index 8821b978..cf1f029f 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -137,7 +137,7 @@ static void *drain_peer(void *v) d->tail[tail_len++] = chunk[i]; } d->tail[tail_len] = 0; - if (strstr(d->tail, "CSeq: 3\r\nPublic: OPTIONS, DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE\r\n\r\n")) { + if (strstr(d->tail, "CSeq: 3\r\nPublic: OPTIONS, DESCRIBE, SETUP, TEARDOWN, PLAY, PAUSE, GET_PARAMETER\r\n\r\n")) { d->found = 1; break; } @@ -355,6 +355,26 @@ static void test_connection_overflow(void) close(server_fd); } +/* An empty GET_PARAMETER is a keepalive: answer it and keep the client */ +static void test_get_parameter_keepalive(void) +{ + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + char reply[256] = {}; + int peer = connect_client(h, &con); + + CHECK(peer >= 0 && con); + send_str(peer, "GET_PARAMETER rtsp://cam/ RTSP/1.0\r\nCSeq: 6\r\n" + "Session: 1234abcd\r\n\r\n"); + serve_once(h); + + CHECK(con->con_state == __CON_S_INIT); + CHECK(h->con_list.list != NULL); + CHECK(read(peer, reply, sizeof(reply) - 1) > 0); + CHECK(!strcmp(reply, "RTSP/1.0 200 OK\r\nCSeq: 6\r\n\r\n")); + close(peer); +} + int main(void) { signal(SIGPIPE, SIG_IGN); @@ -364,5 +384,6 @@ int main(void) test_tcp_play_and_reuse(); test_split_interleaved_packet(); test_connection_overflow(); + test_get_parameter_keepalive(); CHECK_DONE(); } From 4f5885d321878e67c2ca1a9be140943b23c55331 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 16:41:22 -0500 Subject: [PATCH 10/13] Sharing one wait budget across a split interleaved packet --- src/rtsp/rtsp.c | 13 ++++++------- test/rtsp_con_test.c | 30 ++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index 7568246d..445f475c 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -385,15 +385,14 @@ static int __method_teardown(struct connection_item_t *p, rtsp_handle h) /* Interleaved frames from the client (RTCP receiver reports) may arrive split * and the socket is non-blocking: wait briefly for the rest instead of leaving - * it to be parsed as a request */ -static int __read_interleaved(struct connection_item_t *con, void *dst, size_t len) + * it to be parsed as a request. The wait budget is shared by the whole packet */ +static int __read_interleaved(struct connection_item_t *con, void *dst, size_t len, int *waits) { size_t got = 0; - int waits = 0; while ((got += fread((char *)dst + got, 1, len - got, con->fp_tcp_read)) < len) { struct pollfd pfd = { .fd = con->client_fd, .events = POLLIN }; - if (!ferror(con->fp_tcp_read) || errno != EAGAIN || ++waits > 10) + if (!ferror(con->fp_tcp_read) || errno != EAGAIN || ++*waits > 10) return FAILURE; clearerr(con->fp_tcp_read); poll(&pfd, 1, 20); @@ -423,10 +422,10 @@ static int __message_proc_sock(struct list_t *e, void *p) int first_char = fgetc(con->fp_tcp_read); if (first_char == '$') { unsigned char head[3]; - int len = -1; - if (__read_interleaved(con, head, 3) == SUCCESS) { + int len = -1, waits = 0; + if (__read_interleaved(con, head, 3, &waits) == SUCCESS) { len = (head[1] << 8) | head[2]; - while (len > 0 && __read_interleaved(con, buf, min(len, sizeof(buf))) == SUCCESS) + while (len > 0 && __read_interleaved(con, buf, min(len, sizeof(buf)), &waits) == SUCCESS) len -= min(len, sizeof(buf)); } if (len != 0) { diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index cf1f029f..dfb9f08b 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -260,6 +260,35 @@ static void test_split_interleaved_packet(void) close(peer); } +static void *trickle_packet(void *v) +{ + usleep(150 * 1000); + if (write(*(int *)v, "\x01\x00\x08", 3) != 3) perror("write"); + usleep(150 * 1000); + /* fails once the server has given up on the packet */ + send(*(int *)v, "abcdefgh", 8, 0); + return NULL; +} + +/* A packet trickled in piece by piece gets one wait budget, not one per read */ +static void test_trickled_interleaved_packet(void) +{ + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + pthread_t thread; + int peer = connect_client(h, &con); + + CHECK(peer >= 0 && con); + CHECK(write(peer, "$", 1) == 1); + pthread_create(&thread, NULL, trickle_packet, &peer); + serve_once(h); + pthread_join(thread, NULL); + + CHECK(con->con_state == __CON_S_DISCONNECTED); + CHECK(h->con_list.list == NULL); + close(peer); +} + static int tcp_connect(int port) { struct sockaddr_in addr = { .sin_family = AF_INET, .sin_port = htons(port) }; @@ -383,6 +412,7 @@ int main(void) test_client_fd_closed_once(); test_tcp_play_and_reuse(); test_split_interleaved_packet(); + test_trickled_interleaved_packet(); test_connection_overflow(); test_get_parameter_keepalive(); CHECK_DONE(); From 278f5b85abf5a63a4258719c069f3286f430ea0e Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 16:42:02 -0500 Subject: [PATCH 11/13] Skipping RTSP request bodies so they are not parsed as the next request --- src/rtsp/rtsp.c | 30 ++++++++++++++++++++++++------ test/rtsp_con_test.c | 26 ++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 6 deletions(-) diff --git a/src/rtsp/rtsp.c b/src/rtsp/rtsp.c index 445f475c..e51e02bf 100644 --- a/src/rtsp/rtsp.c +++ b/src/rtsp/rtsp.c @@ -36,6 +36,7 @@ extern void request_idr(); #define __STR_INTERLEAVED "interleaved" #define __STR_RTP_AVP_TCP "RTP/AVP/TCP" #define __STR_SESSION "SESSION" +#define __STR_CONTENT_LENGTH "CONTENT-LENGTH" #define __STR_PAUSE "PAUSE" #define __STR_RECORDING "RECORDING" #define __STR_GET_PARAMETER "GET_PARAMETER" @@ -383,10 +384,11 @@ static int __method_teardown(struct connection_item_t *p, rtsp_handle h) * METHOD IMPLEMENTATIONS ******************************************************************************/ -/* Interleaved frames from the client (RTCP receiver reports) may arrive split - * and the socket is non-blocking: wait briefly for the rest instead of leaving - * it to be parsed as a request. The wait budget is shared by the whole packet */ -static int __read_interleaved(struct connection_item_t *con, void *dst, size_t len, int *waits) +/* Interleaved frames from the client (RTCP receiver reports) and request + * bodies may arrive split and the socket is non-blocking: wait briefly for the + * rest instead of leaving it to be parsed as a request. The wait budget is + * shared by the whole packet or body */ +static int __read_client_data(struct connection_item_t *con, void *dst, size_t len, int *waits) { size_t got = 0; @@ -423,9 +425,9 @@ static int __message_proc_sock(struct list_t *e, void *p) if (first_char == '$') { unsigned char head[3]; int len = -1, waits = 0; - if (__read_interleaved(con, head, 3, &waits) == SUCCESS) { + if (__read_client_data(con, head, 3, &waits) == SUCCESS) { len = (head[1] << 8) | head[2]; - while (len > 0 && __read_interleaved(con, buf, min(len, sizeof(buf)), &waits) == SUCCESS) + while (len > 0 && __read_client_data(con, buf, min(len, sizeof(buf)), &waits) == SUCCESS) len -= min(len, sizeof(buf)); } if (len != 0) { @@ -448,6 +450,7 @@ static int __message_proc_sock(struct list_t *e, void *p) con->method = __METHOD_NONE; char header = 0, isAuthValid = 0, *tok, *last; + int body_len = 0; unsigned long long session_id; /* parse line by line. hereafter parser is switched according to the finite state machine */ while (__read_line(con, buf)) { @@ -484,6 +487,10 @@ static int __message_proc_sock(struct list_t *e, void *p) ASSERT(sscanf(tok, "%llx", &session_id) > 0, goto error); con->given_session_id = session_id; con->parser_state = __PARSER_S_SESSION; + } else if (SCMP(__STR_CONTENT_LENGTH, buf)) { + /* without a length the next request cannot be found */ + ASSERT(sscanf(buf + strlen(__STR_CONTENT_LENGTH), " : %d", &body_len) == 1 && + body_len >= 0 && body_len <= 0xFFFF, con->con_state = __CON_S_DISCONNECTED); } else if (SCMP(__STR_TRANSPORT, buf)) { if (strstr(buf, __STR_RTP_AVP_TCP)) { con->trans[con->track_id].is_tcp = 1; @@ -511,6 +518,17 @@ static int __message_proc_sock(struct list_t *e, void *p) __PARSE_ERROR(con); } + /* no method takes a body: skip it so it is not read as the next request */ + if (body_len > 0 && con->con_state != __CON_S_DISCONNECTED) { + int waits = 0; + while (body_len > 0 && __read_client_data(con, buf, min(body_len, sizeof(buf)), &waits) == SUCCESS) + body_len -= min(body_len, sizeof(buf)); + if (body_len != 0) { + ERR("request body cut short, dropping connection\n"); + con->con_state = __CON_S_DISCONNECTED; + } + } + if (con->con_state == __CON_S_DISCONNECTED) { ASSERT(bufpool_detach(con->pool, con) == SUCCESS, ERR("connection detach failed\n")); } else if (con->parser_state == __PARSER_S_ERROR) { diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index dfb9f08b..03a3acd1 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -404,6 +404,31 @@ static void test_get_parameter_keepalive(void) close(peer); } +/* A body declared by Content-Length is consumed with its request instead of + * being read as the start of the next one */ +static void test_request_body_skipped(void) +{ + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + char reply[256] = {}; + int peer = connect_client(h, &con); + + CHECK(peer >= 0 && con); + send_str(peer, "GET_PARAMETER rtsp://cam/ RTSP/1.0\r\nCSeq: 7\r\n" + "Content-Type: text/parameters\r\nContent-Length: 10\r\n\r\nposition\r\n"); + serve_once(h); + CHECK(read(peer, reply, sizeof(reply) - 1) > 0); + CHECK(!strcmp(reply, "RTSP/1.0 200 OK\r\nCSeq: 7\r\n\r\n")); + + memset(reply, 0, sizeof(reply)); + send_str(peer, "OPTIONS rtsp://cam/ RTSP/1.0\r\nCSeq: 8\r\n\r\n"); + serve_once(h); + CHECK(con->con_state == __CON_S_INIT); + CHECK(read(peer, reply, sizeof(reply) - 1) > 0); + CHECK(strstr(reply, "RTSP/1.0 200 OK\r\nCSeq: 8\r\n")); + close(peer); +} + int main(void) { signal(SIGPIPE, SIG_IGN); @@ -415,5 +440,6 @@ int main(void) test_trickled_interleaved_packet(); test_connection_overflow(); test_get_parameter_keepalive(); + test_request_body_skipped(); CHECK_DONE(); } From 4c323461ec54961a196bd43d8cc1d905da147c77 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Wed, 30 Sep 2026 16:42:40 -0500 Subject: [PATCH 12/13] Shutting down an RTSP client that stops reading instead of waiting on it forever --- src/rtsp/rtsp.h | 19 ++++++++++++++----- test/rtsp_con_test.c | 26 ++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/src/rtsp/rtsp.h b/src/rtsp/rtsp.h index 8e449bf9..c536e261 100644 --- a/src/rtsp/rtsp.h +++ b/src/rtsp/rtsp.h @@ -208,20 +208,29 @@ static inline int __read_line(struct connection_item_t *p, char *buf) return !(SCMP(__TERM, buf)); } -static inline void __wait_out(int fd) +static inline int __wait_out(int fd) { struct pollfd p = { .fd = fd, .events = POLLOUT }; - poll(&p, 1, 100); + return poll(&p, 1, 100); } -/* The client socket is non-blocking: retry until everything is queued */ +/* The client socket is non-blocking: retry until everything is queued. A peer + * that takes nothing for 5 s is shut down, as a partial write has already + * broken its stream, and the RTSP thread then sees EOF and drops it */ static inline int __tcp_send_all(int fd, const unsigned char *buf, unsigned int len) { unsigned int sent = 0; + int stalls = 0; while (sent < len) { int r = send(fd, buf + sent, len - sent, 0); - if (r > 0) sent += r; - else if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { __wait_out(fd); } + if (r > 0) { sent += r; stalls = 0; } + else if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + if (__wait_out(fd) == 0 && ++stalls >= 50) { + shutdown(fd, SHUT_RDWR); + errno = ETIMEDOUT; + return FAILURE; + } + } else return FAILURE; } return SUCCESS; diff --git a/test/rtsp_con_test.c b/test/rtsp_con_test.c index 03a3acd1..3224ea76 100644 --- a/test/rtsp_con_test.c +++ b/test/rtsp_con_test.c @@ -171,6 +171,31 @@ static void test_response_on_full_socket(void) close(drain.fd); } +/* A client that stops reading is given up on instead of holding the RTSP + * thread, and every stream with it, forever */ +static void test_response_to_stalled_client(void) +{ + rtsp_handle h = make_server(1); + struct connection_item_t *con = NULL; + char junk[1024]; + int size = 4096; + int peer = connect_client(h, &con); + + CHECK(peer >= 0 && con); + setsockopt(con->client_fd, SOL_SOCKET, SO_SNDBUF, &size, sizeof(size)); + memset(junk, 'x', sizeof(junk)); + while (write(con->client_fd, junk, sizeof(junk)) > 0); + + send_str(peer, "OPTIONS rtsp://cam/ RTSP/1.0\r\nCSeq: 9\r\n\r\n"); + serve_once(h); + CHECK(h->con_list.list != NULL); + serve_once(h); + + CHECK(con->con_state == __CON_S_DISCONNECTED); + CHECK(h->con_list.list == NULL); + close(peer); +} + /* fclose() on the read stream already closes the client fd: closing it again * would hit whichever socket the kernel has given that number meanwhile */ static void test_client_fd_closed_once(void) @@ -434,6 +459,7 @@ int main(void) signal(SIGPIPE, SIG_IGN); test_truncated_request(); test_response_on_full_socket(); + test_response_to_stalled_client(); test_client_fd_closed_once(); test_tcp_play_and_reuse(); test_split_interleaved_packet(); From 9cecf31d35ac8d5ca92fa3032570aacd2206ea80 Mon Sep 17 00:00:00 2001 From: Jaro Marval Date: Thu, 1 Oct 2026 22:16:45 -0500 Subject: [PATCH 13/13] Checking the NAL size before reading its header, as an empty pack has no data --- src/rtsp/rtp.c | 5 +++-- test/Makefile | 6 +++++- test/rtp_nal_test.c | 29 +++++++++++++++++++++++++++++ 3 files changed, 37 insertions(+), 3 deletions(-) create mode 100644 test/rtp_nal_test.c diff --git a/src/rtsp/rtp.c b/src/rtsp/rtp.c index 4091ca13..43f46fc9 100644 --- a/src/rtsp/rtp.c +++ b/src/rtsp/rtp.c @@ -66,6 +66,9 @@ static int __tcp_flush_each(struct list_t *e, void *v) static inline int __transfer_nal_h26x(struct list_head_t *trans_list, unsigned char *nalptr, size_t nalsize, char isH265) { + /* Checked before the header is read: an empty pack comes with no data */ + if (nalsize < 4) return SUCCESS; + struct nal_rtp_t rtp; unsigned int nri = isH265 ? (nalptr[0] & 0x81) : (nalptr[0] & 0x60); unsigned int pt = isH265 ? (nalptr[0] >> 1 & 0x3F) : (nalptr[0] & 0x1F); @@ -81,8 +84,6 @@ static inline int __transfer_nal_h26x(struct list_head_t *trans_list, unsigned c p_header->cc = 0; p_header->pt = 96 & 0x7F; - if (nalsize < 4) return SUCCESS; - if (nalsize <= __RTP_MAXPAYLOADSIZE) { /* single packet */ /* SPS, PPS, SEI is not marked */ diff --git a/test/Makefile b/test/Makefile index aef6642a..f1c0be96 100644 --- a/test/Makefile +++ b/test/Makefile @@ -1,6 +1,6 @@ CFLAGS = -std=gnu99 -Wall -Wno-parentheses -I../src \ -DCLOCK_MONOTONIC_COARSE=CLOCK_MONOTONIC -Dmmap64=mmap -TESTS = rtsp_con_test +TESTS = rtsp_con_test rtp_nal_test all: $(TESTS) @for t in $(TESTS); do ./$$t || exit 1; done @@ -12,4 +12,8 @@ clean: rtsp_con_test: rtsp_con_test.c ../src/rtsp/mime.c ../src/hal/tools.c ../src/rtsp/rtsp.c $(wildcard ../src/rtsp/*.h) $(CC) $(CFLAGS) $(filter-out ../src/rtsp/rtsp.c %.h,$^) -lpthread -o $@ +# rtp.c likewise, linked against the rest of the RTSP server +rtp_nal_test: rtp_nal_test.c ../src/rtsp/mime.c ../src/hal/tools.c ../src/rtsp/rtsp.c ../src/rtsp/rtp.c $(wildcard ../src/rtsp/*.h) + $(CC) $(CFLAGS) $(filter-out ../src/rtsp/rtp.c %.h,$^) -lpthread -o $@ + .PHONY: all clean diff --git a/test/rtp_nal_test.c b/test/rtp_nal_test.c new file mode 100644 index 00000000..753291fd --- /dev/null +++ b/test/rtp_nal_test.c @@ -0,0 +1,29 @@ +#include "check.h" +#undef CHECK +#include "../src/rtsp/rtp.c" +#undef CHECK +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + fprintf(stderr, "%s:%d: CHECK failed: %s\n", __FILE__, __LINE__, #cond); \ + checkFailures++; \ + } \ + } while (0) + +void request_idr(void) {} + +/* The encoder can hand out a pack with no data and no length: nothing to send, + * and its header must not be read */ +static void test_empty_nal(void) +{ + struct list_head_t trans_list = {}; + + CHECK(__transfer_nal_h26x(&trans_list, NULL, 0, 0) == SUCCESS); + CHECK(__transfer_nal_h26x(&trans_list, NULL, 0, 1) == SUCCESS); +} + +int main(void) +{ + test_empty_nal(); + CHECK_DONE(); +}