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 0000000..5466754 Binary files /dev/null and b/tests/fixtures/stale-request.pcap differ 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"); }