From 9363844a45b867bc7e5a5992e7089769a436e9d9 Mon Sep 17 00:00:00 2001 From: git-hulk Date: Wed, 2 Sep 2026 18:25:39 +0800 Subject: [PATCH] Expire pending requests instead of keeping them for ever Every request went into sniffer->requests and only came out when a matching response arrived. Requests that were never answered -- a dropped connection, a server that never replied, a capture that started mid-response -- stayed for the life of the process, so a long running capture leaked steadily. It also meant a response arriving minutes later was still reported as a latency of minutes: replaying a capture whose answer comes 99 seconds later printed `99000.000 ms`. - hashtable gains a size counter and hashtable_sweep, which removes every entry a predicate selects, correctly unlinking at the head, the middle or the tail of a bucket chain. - requests_expire drops requests older than 60 seconds of capture time. The sweep runs at most once per captured second, and is driven by packet timestamps rather than the wall clock so that replaying a file behaves the same as a live capture. Timestamps that go backwards evict nothing. - 60 seconds is past the last latency bucket, so anything still waiting is lost rather than useful. MAX_PENDING_REQUESTS is a firm ceiling for the case where requests arrive faster than they age out. - hashtable_add now checks its allocations, and no longer half-inserts an entry whose key could not be duplicated. Assistant By Opus 5 Co-Authored-By: Claude Opus 5 (1M context) --- src/hashtable.c | 40 ++++++++++++- src/hashtable.h | 3 + src/packet.c | 50 +++++++++++++++- src/packet.h | 3 + src/sniffer.h | 3 + tests/e2e/run.sh | 7 +++ tests/fixtures/gen_fixtures.py | 16 +++++ tests/fixtures/stale-request.pcap | Bin 0 -> 504 bytes tests/unit/test_hashtable.c | 93 ++++++++++++++++++++++++++++++ tests/unit/test_packet.c | 64 ++++++++++++++++++++ 10 files changed, 276 insertions(+), 3 deletions(-) create mode 100644 tests/fixtures/stale-request.pcap diff --git a/src/hashtable.c b/src/hashtable.c index 580b7cf..64b8f1f 100644 --- a/src/hashtable.c +++ b/src/hashtable.c @@ -61,6 +61,7 @@ hashtable *hashtable_create(int nbucket) { if (!ht) return NULL; ht->free = NULL; ht->nbucket = nbucket; + ht->size = 0; ht->buckets = calloc(nbucket, sizeof(entry*)); if (!ht->buckets) { free(ht); @@ -102,10 +103,16 @@ void *hashtable_add(hashtable *ht, char *key, void *value) { current = current->next; } entry *e = malloc(sizeof(*e)); - e->value = value; + if (!e) return NULL; e->key = strdup(key); + if (!e->key) { + free(e); + return NULL; + } + e->value = value; e->next = ht->buckets[bucket]; ht->buckets[bucket] = e; + ht->size++; return NULL; } @@ -141,6 +148,7 @@ int hashtable_del(hashtable *ht, char *key) { free(current->key); ht->free ? ht->free(current->value):free(current->value); free(current); + ht->size--; return 1; } prev = current; @@ -149,6 +157,36 @@ int hashtable_del(hashtable *ht, char *key) { return 0; } +int hashtable_sweep(hashtable *ht, int (*expired)(void *value, void *arg), void *arg) { + int i, removed = 0; + entry *current, *next, *prev; + + if (!ht || !expired) return 0; + for (i = 0; i < ht->nbucket; i++) { + prev = NULL; + current = ht->buckets[i]; + while (current) { + next = current->next; + if (expired(current->value, arg)) { + if (prev) { + prev->next = next; + } else { + ht->buckets[i] = next; + } + free(current->key); + ht->free ? ht->free(current->value) : free(current->value); + free(current); + ht->size--; + removed++; + } else { + prev = current; + } + current = next; + } + } + return removed; +} + void **hashtable_values(hashtable *ht, int *cnt) { int i, cap = 0; entry *e; diff --git a/src/hashtable.h b/src/hashtable.h index 4242e15..6b201e3 100644 --- a/src/hashtable.h +++ b/src/hashtable.h @@ -20,6 +20,7 @@ typedef struct entry { typedef struct hashtable { int nbucket; + int size; entry **buckets; void (*free)(void *); } hashtable; @@ -30,4 +31,6 @@ void *hashtable_get(hashtable *ht, char *key); void *hashtable_add(hashtable *ht, char *key, void *value); int hashtable_del(hashtable *ht, char *key); void **hashtable_values(hashtable *ht, int *cnt); +/* Removes every entry the predicate accepts and returns how many went. */ +int hashtable_sweep(hashtable *ht, int (*expired)(void *value, void *arg), void *arg); #endif diff --git a/src/packet.c b/src/packet.c index f228df1..223df48 100644 --- a/src/packet.c +++ b/src/packet.c @@ -117,6 +117,50 @@ int process_udp_packet(const struct timeval tv, return 0; } +/* A response later than this is past the last latency bucket, so a request + * still waiting after it is treated as lost rather than kept for ever. */ +#define REQUEST_TIMEOUT_US 60000000 +/* Capture time between sweeps of the pending table. */ +#define EXPIRE_INTERVAL_US 1000000 +/* Firm ceiling for the case where requests arrive faster than they expire. */ +#define MAX_PENDING_REQUESTS 100000 + +struct request_expiry { + struct timeval now; + int64_t timeout_us; +}; + +static int64_t tv_delta_us(struct timeval later, struct timeval earlier) { + return (int64_t)(later.tv_sec - earlier.tv_sec) * 1000000 + + (later.tv_usec - earlier.tv_usec); +} + +static int request_expired(void *value, void *arg) { + const struct request *req = (const struct request *)value; + const struct request_expiry *expiry = (const struct request_expiry *)arg; + + return tv_delta_us(expiry->now, req->tv) > expiry->timeout_us; +} + +int requests_expire(struct hashtable *requests, struct timeval now, int64_t timeout_us) { + struct request_expiry expiry; + + expiry.now = now; + expiry.timeout_us = timeout_us; + return hashtable_sweep(requests, request_expired, &expiry); +} + +static void expire_stale_requests(struct sniffer *sniffer, struct timeval now) { + if (!sniffer->expire_primed) { + sniffer->expire_primed = 1; + sniffer->last_expire = now; + return; + } + if (tv_delta_us(now, sniffer->last_expire) < EXPIRE_INTERVAL_US) return; + sniffer->last_expire = now; + requests_expire(sniffer->requests, now, REQUEST_TIMEOUT_US); +} + static int packet_direction(struct sniffer *sniffer, struct user_packet *upacket) { char key[32]; int direction = -1; @@ -172,6 +216,8 @@ static void process_request_packet(struct sniffer *sniffer, struct user_packet * upacket->ip_src.s_addr, upacket->port_src, upacket->ip_dst.s_addr, upacket->port_dst); if (hashtable_get(sniffer->requests, key)) return; + /* Drop the request rather than grow the table without limit. */ + if (sniffer->requests->size >= MAX_PENDING_REQUESTS) return; req = malloc(sizeof(*req)); if (!req) return; @@ -205,8 +251,7 @@ static void process_response_packet(struct sniffer *sniffer, struct user_packet upacket->ip_dst.s_addr, upacket->port_dst, upacket->ip_src.s_addr, upacket->port_src); if ((req = hashtable_get(sniffer->requests, key)) != NULL) { - delta = (upacket->tv.tv_sec - req->tv.tv_sec) * 1000000 - + (upacket->tv.tv_usec - req->tv.tv_usec); + delta = tv_delta_us(upacket->tv, req->tv); snprintf(target, sizeof(target), "%u:%d", upacket->ip_src.s_addr, upacket->port_src); sniffer_stats_lock(sniffer); if ((stats = hashtable_get(sniffer->syn_tab, target)) != NULL) { @@ -295,6 +340,7 @@ void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket) { print_user_packet(sniffer, upacket); return; } + expire_stale_requests(sniffer, upacket->tv); if (upacket->payload_size == 0) { if ((upacket->flags & syn_mask) != 0) { src = (upacket->flags & ack_mask) != 0; diff --git a/src/packet.h b/src/packet.h index 172ea9f..1c45c5c 100644 --- a/src/packet.h +++ b/src/packet.h @@ -62,5 +62,8 @@ int process_udp_packet(const struct timeval tv, int wirelen, struct user_packet *packet); +/* Drops pending requests older than timeout_us and returns how many went. */ +int requests_expire(struct hashtable *requests, struct timeval now, int64_t timeout_us); + void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket); #endif diff --git a/src/sniffer.h b/src/sniffer.h index 374018d..c92d800 100644 --- a/src/sniffer.h +++ b/src/sniffer.h @@ -14,6 +14,7 @@ #include #include +#include #include #include "tcpkit.h" #include "stats.h" @@ -34,6 +35,8 @@ struct sniffer { int lock_ready; struct hashtable *syn_tab; struct hashtable *requests; + struct timeval last_expire; + int expire_primed; lua_State *lua_state; struct bpf_program *bpf; }; diff --git a/tests/e2e/run.sh b/tests/e2e/run.sh index 6c811c8..b73d115 100755 --- a/tests/e2e/run.sh +++ b/tests/e2e/run.sh @@ -127,6 +127,13 @@ expect_lines "each request stays on one line" "$out" 3 expect_contains "a truncated resp array is summarised" "$out" '*2..$3..GET' expect_contains "a payload with no crlf is summarised" "$out" "GETNOCRLFATALL" +echo "== e2e: stale requests" +replay stale-request.pcap -p redis +expect_status "a stale request exits cleanly" "$status" 0 +expect_lines "only the answered request is reported" "$out" 1 +expect_contains "the healthy exchange is still reported" "$out" "GET fast" +expect_not_contains "the abandoned request is not held for ever" "$out" "GET slow" + echo "== e2e: capture filter" replay redis-session.pcap -p raw udp expect_lines "a udp filter keeps only the dns packet" "$out" 1 diff --git a/tests/fixtures/gen_fixtures.py b/tests/fixtures/gen_fixtures.py index dbb381a..84e4e5d 100644 --- a/tests/fixtures/gen_fixtures.py +++ b/tests/fixtures/gen_fixtures.py @@ -184,6 +184,21 @@ def malformed_payload(): ] +def stale_request(): + """A request whose response arrives long after any latency bucket, followed + by a healthy exchange, so the sweep can be told apart from a broken path.""" + return [ + full(0.000000, to_server(b"", 1000, 0, SYN)), + full(0.000100, to_client(b"", 5000, 1001, SYN | ACK)), + # Answered 99 seconds later, well past the request timeout. + full(1.000000, to_server(resp("GET", "slow"), 1001, 5001, PSH | ACK)), + full(100.000000, to_client(b"$1\r\nb\r\n", 5001, 1024, PSH | ACK)), + # A normal exchange afterwards still has to be reported. + full(101.000000, to_server(resp("GET", "fast"), 1024, 5008, PSH | ACK)), + full(101.000500, to_client(b"$1\r\nc\r\n", 5008, 1047, PSH | ACK)), + ] + + def stress(connections=4000): """Many short-lived connections, to keep the capture thread busy while the stats endpoint is polled. Not a committed fixture: it is generated on demand @@ -208,3 +223,4 @@ def stress(connections=4000): write_pcap("redis-session.pcap", redis_session()) write_pcap("truncated.pcap", truncated()) write_pcap("malformed-payload.pcap", malformed_payload()) + write_pcap("stale-request.pcap", stale_request()) diff --git a/tests/fixtures/stale-request.pcap b/tests/fixtures/stale-request.pcap new file mode 100644 index 0000000000000000000000000000000000000000..5466754809b05f5738345baa6732621c14040690 GIT binary patch literal 504 zcmca|c+)~A1{MYwxWmf8zzE|2X)`Eh0*Nt!Nk$F^R|WFo8^xczygJ1A}k}(4?0ELjV6S-h^etq?j2|$;MgA8X?hqyGEmkR(Z5J(dM literal 0 HcmV?d00001 diff --git a/tests/unit/test_hashtable.c b/tests/unit/test_hashtable.c index 7970996..a17e377 100644 --- a/tests/unit/test_hashtable.c +++ b/tests/unit/test_hashtable.c @@ -99,6 +99,95 @@ static void test_values_lists_every_entry(void) { hashtable_destroy(ht); } +static int always(void *value, void *arg) { + (void)value; + (void)arg; + return 1; +} + +static int value_is(void *value, void *arg) { + return strcmp((char *)value, (char *)arg) == 0; +} + +static void test_size_tracks_the_entries(void) { + hashtable *ht = hashtable_create(8); + char *rejected; + + TK_EQ_INT(ht->size, 0); + hashtable_add(ht, "a", dup_value("1")); + hashtable_add(ht, "b", dup_value("2")); + TK_EQ_INT(ht->size, 2); + + /* A duplicate key adds no entry and hands back the value already there; + * the rejected one still belongs to the caller. */ + rejected = dup_value("3"); + TK_EQ_STR((char *)hashtable_add(ht, "a", rejected), "1"); + free(rejected); + TK_EQ_INT(ht->size, 2); + + hashtable_del(ht, "a"); + TK_EQ_INT(ht->size, 1); + hashtable_del(ht, "missing"); + TK_EQ_INT(ht->size, 1); + + hashtable_destroy(ht); +} + +static void test_sweep_removes_only_what_matches(void) { + hashtable *ht = hashtable_create(4); + + hashtable_add(ht, "a", dup_value("keep")); + hashtable_add(ht, "b", dup_value("drop")); + hashtable_add(ht, "c", dup_value("drop")); + hashtable_add(ht, "d", dup_value("keep")); + + TK_EQ_INT(hashtable_sweep(ht, value_is, (void *)"drop"), 2); + TK_EQ_INT(ht->size, 2); + TK_EQ_STR((char *)hashtable_get(ht, "a"), "keep"); + TK_EQ_STR((char *)hashtable_get(ht, "d"), "keep"); + TK_CHECK(hashtable_get(ht, "b") == NULL, "swept key must be gone"); + TK_CHECK(hashtable_get(ht, "c") == NULL, "swept key must be gone"); + + hashtable_destroy(ht); +} + +static void test_sweep_handles_a_whole_bucket_chain(void) { + hashtable *ht = hashtable_create(1); + char key[16]; + int i; + + /* One bucket, so every entry is chained: head, middle and tail all go. */ + for (i = 0; i < 16; i++) { + snprintf(key, sizeof(key), "key-%d", i); + hashtable_add(ht, key, dup_value("drop")); + } + hashtable_add(ht, "survivor", dup_value("keep")); + + TK_EQ_INT(hashtable_sweep(ht, value_is, (void *)"drop"), 16); + TK_EQ_INT(ht->size, 1); + TK_EQ_STR((char *)hashtable_get(ht, "survivor"), "keep"); + + TK_EQ_INT(hashtable_sweep(ht, always, NULL), 1); + TK_EQ_INT(ht->size, 0); + TK_CHECK(hashtable_get(ht, "survivor") == NULL, "the table must be empty"); + + hashtable_destroy(ht); +} + +static void test_sweep_uses_the_custom_free(void) { + hashtable *ht = hashtable_create(8); + + freed = 0; + ht->free = counting_free; + hashtable_add(ht, "a", dup_value("1")); + hashtable_add(ht, "b", dup_value("2")); + + TK_EQ_INT(hashtable_sweep(ht, always, NULL), 2); + TK_EQ_INT(freed, 2); + + hashtable_destroy(ht); +} + static void test_destroy_uses_the_custom_free(void) { hashtable *ht = hashtable_create(8); @@ -117,6 +206,10 @@ int main(void) { TK_RUN(test_del_removes_the_entry); TK_RUN(test_collisions_keep_every_entry); TK_RUN(test_values_lists_every_entry); + TK_RUN(test_size_tracks_the_entries); + TK_RUN(test_sweep_removes_only_what_matches); + TK_RUN(test_sweep_handles_a_whole_bucket_chain); + TK_RUN(test_sweep_uses_the_custom_free); TK_RUN(test_destroy_uses_the_custom_free); return tk_report("hashtable"); } diff --git a/tests/unit/test_packet.c b/tests/unit/test_packet.c index 63c5362..5e21640 100644 --- a/tests/unit/test_packet.c +++ b/tests/unit/test_packet.c @@ -318,6 +318,68 @@ static void test_udp_rejects_a_length_under_its_own_header(void) { free(c.bytes); } +static void free_test_request(void *value) { + struct request *req = (struct request *)value; + + free(req->payload); + free(req); +} + +static void add_request(struct hashtable *requests, const char *key, long sec) { + struct request *req = malloc(sizeof(*req)); + + req->tv.tv_sec = sec; + req->tv.tv_usec = 0; + req->seq = 1; + req->payload = malloc(8); + memcpy(req->payload, "GET a", 6); + req->size = 5; + hashtable_add(requests, (char *)key, req); +} + +static void test_expiry_drops_only_the_stale_requests(void) { + struct hashtable *requests = hashtable_create(16); + struct timeval now; + + requests->free = free_test_request; + add_request(requests, "old", 100); + add_request(requests, "borderline", 150); + add_request(requests, "fresh", 195); + + now.tv_sec = 200; + now.tv_usec = 0; + + /* A 60s timeout: 100 is 100s old, 150 is 50s old, 195 is 5s old. */ + TK_EQ_INT(requests_expire(requests, now, 60000000), 1); + TK_EQ_INT(requests->size, 2); + TK_CHECK(hashtable_get(requests, "old") == NULL, "the stale request must go"); + TK_CHECK(hashtable_get(requests, "borderline") != NULL, "50s is not stale yet"); + TK_CHECK(hashtable_get(requests, "fresh") != NULL, "5s is not stale"); + + /* Later still, and everything but the newest has aged out. */ + now.tv_sec = 260; + TK_EQ_INT(requests_expire(requests, now, 60000000), 2); + TK_EQ_INT(requests->size, 0); + + hashtable_destroy(requests); +} + +static void test_expiry_keeps_requests_when_time_goes_backwards(void) { + struct hashtable *requests = hashtable_create(16); + struct timeval now; + + requests->free = free_test_request; + add_request(requests, "later", 300); + + /* Out of order capture timestamps must not evict anything. */ + now.tv_sec = 200; + now.tv_usec = 0; + TK_EQ_INT(requests_expire(requests, now, 60000000), 0); + TK_EQ_INT(requests->size, 1); + + hashtable_destroy(requests); +} + int main(void) { TK_RUN(test_tcp_parses_a_complete_packet); TK_RUN(test_tcp_rejects_a_truncated_header); @@ -329,5 +391,7 @@ int main(void) { TK_RUN(test_udp_rejects_a_truncated_header); TK_RUN(test_udp_rejects_a_length_beyond_the_frame); TK_RUN(test_udp_rejects_a_length_under_its_own_header); + TK_RUN(test_expiry_drops_only_the_stale_requests); + TK_RUN(test_expiry_keeps_requests_when_time_goes_backwards); return tk_report("packet"); }