Synchronise the endpoint stats between the capture and stats threads

.github/workflows/ci.yml

@@ -18,6 +18,10 @@ jobs:
       - name: Install libpcap
         if: runner.os == 'Linux'
         run: sudo apt-get update && sudo apt-get install -y libpcap-dev
+      # The sanitizers the test suite uses need this on newer Ubuntu kernels.
+      - name: Widen mmap randomisation for the sanitizers
+        if: runner.os == 'Linux'
+        run: sudo sysctl -w vm.mmap_rnd_bits=28 || true
       - name: Build
         run: make
       - name: Test
@@ -33,6 +37,18 @@ jobs:
       - name: Build
         run: make USE_SYSTEM_PCAP=0
 
+  race:
+    name: thread sanitizer
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install libpcap
+        run: sudo apt-get update && sudo apt-get install -y libpcap-dev
+      - name: Widen mmap randomisation for the sanitizers
+        run: sudo sysctl -w vm.mmap_rnd_bits=28 || true
+      - name: Replay a busy capture while polling the stats endpoint
+        run: make test-race
+
   fixtures:
     name: fixtures are reproducible
     runs-on: ubuntu-latest

Makefile

@@ -6,6 +6,9 @@ all:
 test: all
 	cd tests && $(MAKE) test
 
+test-race:
+	cd tests && $(MAKE) test-race
+
 fixtures:
 	cd tests && $(MAKE) fixtures
 
@@ -19,4 +22,4 @@ clean:
 .DEFAULT:
 	cd src && $(MAKE) $@
 
-.PHONY: default all test fixtures install clean
+.PHONY: default all test test-race fixtures install clean

src/Makefile

@@ -10,7 +10,6 @@ CFLAGS = -g $(OPTIMIZATION) -Wall
 # SANITIZE=address builds the binary and the test suite under ASan/UBSan.
 ifdef SANITIZE
 CFLAGS += -fsanitize=$(SANITIZE) -fno-omit-frame-pointer
-LDFLAGS += -fsanitize=$(SANITIZE)
 endif
 
 LUA_DIR = ../deps/lua/src

src/packet.c

@@ -119,19 +119,23 @@ int process_udp_packet(const struct timeval tv,
 
 static int packet_direction(struct sniffer *sniffer, struct user_packet *upacket) {
     char key[32];
+    int direction = -1;
     struct query_stats *stats;
 
+    sniffer_stats_lock(sniffer);
     snprintf(key, sizeof(key), "%u:%d", upacket->ip_src.s_addr, upacket->port_src);
     if ((stats = hashtable_get(sniffer->syn_tab, key)) != NULL) {
         stats_incr(stats, 0, upacket->size);
-        return 0;
-    }
-    snprintf(key, sizeof(key), "%u:%d", upacket->ip_dst.s_addr, upacket->port_dst);
-    if ((stats = hashtable_get(sniffer->syn_tab, key)) != NULL)  {
-        stats_incr(stats, 1, upacket->size);
-        return 1;
+        direction = 0;
+    } else {
+        snprintf(key, sizeof(key), "%u:%d", upacket->ip_dst.s_addr, upacket->port_dst);
+        if ((stats = hashtable_get(sniffer->syn_tab, key)) != NULL)  {
+            stats_incr(stats, 1, upacket->size);
+            direction = 1;
+        }
     }
-    return -1;
+    sniffer_stats_unlock(sniffer);
+    return direction;
 }
 
 static void push_packet_to_lua_state(lua_State *state, struct user_packet *upacket) {
@@ -204,9 +208,11 @@ static void process_response_packet(struct sniffer *sniffer, struct user_packet
         delta = (upacket->tv.tv_sec - req->tv.tv_sec) * 1000000
             + (upacket->tv.tv_usec - req->tv.tv_usec);
         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) {
             stats_observer_latency(stats, delta);
         }
+        sniffer_stats_unlock(sniffer);
         if (sniffer->threshold && delta < sniffer->threshold*1000) {
             hashtable_del(sniffer->requests, key);
             return;
@@ -297,12 +303,16 @@ void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket) {
             } else {
                 snprintf(key, sizeof(key), "%u:%d", upacket->ip_dst.s_addr, upacket->port_dst);
             }
+            sniffer_stats_lock(sniffer);
             if (!hashtable_get(sniffer->syn_tab, key)) {
                 stats = calloc(1, sizeof(*stats));
-                stats->ip = src ? upacket->ip_src : upacket->ip_dst;
-                stats->port = src ? upacket->port_src : upacket->port_dst;
-                hashtable_add(sniffer->syn_tab, key, stats);
+                if (stats) {
+                    stats->ip = src ? upacket->ip_src : upacket->ip_dst;
+                    stats->port = src ? upacket->port_src : upacket->port_dst;
+                    hashtable_add(sniffer->syn_tab, key, stats);
+                }
             }
+            sniffer_stats_unlock(sniffer);
         }
     } else {
         switch(packet_direction(sniffer, upacket)) {

src/server.c

@@ -95,8 +95,8 @@ int server_run(struct server *srv, char *err) {
 }
 
 void server_terminate(struct server *srv) {
-    if (!srv || srv->stopped) return;
-    srv->stopped = 1;
+    if (!srv || TK_LOAD(&srv->stopped)) return;
+    TK_STORE(&srv->stopped, 1);
     sniffer_terminate(srv->sniffer);
     if (srv->dumper) dumper_terminate(srv->dumper);
 }
@@ -195,13 +195,17 @@ static char *server_stats_to_json(struct server *srv) {
     char buf[64], *stats_json_str;
 
     object = cJSON_CreateObject();
+    /* Held across the whole walk: the capture thread may otherwise add an
+     * endpoint or bump a counter half way through. */
+    sniffer_stats_lock(srv->sniffer);
     values = hashtable_values(srv->sniffer->syn_tab, &cnt);
     for (i = 0; i < cnt; i++) {
-        stats = (struct query_stats *)values[i];        
+        stats = (struct query_stats *)values[i];
         if (!stats) continue;
-        snprintf(buf, 64, "%s:%d", inet_ntoa(stats->ip), stats->port);
+        snprintf(buf, sizeof(buf), "%s:%d", inet_ntoa(stats->ip), stats->port);
         cJSON_AddItemToObject(object, buf, create_stats_object(stats));
     }
+    sniffer_stats_unlock(srv->sniffer);
     stats_json_str = cJSON_Print(object);
     cJSON_Delete(object);
     free(values);
@@ -223,7 +227,7 @@ static void *server_stats_loop(void *arg) {
     }
     fds[0].fd = listen_fd;
     fds[0].events = POLLIN;
-    while(!srv->stopped) {
+    while(!TK_LOAD(&srv->stopped)) {
         rc = poll(fds, 1, 100);
         if (rc <= 0) continue;
         new_fd = accept(listen_fd, NULL, NULL);

src/server.h

@@ -13,15 +13,26 @@
 #define TCPKIT_SERVER_H
 
 #include <pthread.h>
+#include <signal.h>
 #include "tcpkit.h"
 
+#if defined(__GNUC__)
+#define TK_LOAD(p)      __atomic_load_n((p), __ATOMIC_ACQUIRE)
+#define TK_STORE(p, v)  __atomic_store_n((p), (v), __ATOMIC_RELEASE)
+#else
+#define TK_LOAD(p)      (*(p))
+#define TK_STORE(p, v)  (*(p) = (v))
+#endif
+
 struct server {
     struct options *opts;
     struct sniffer *sniffer;
     struct dumper* dumper;
     pthread_t dumper_tid;
     pthread_t stats_tid;
-    int stopped;
+    /* Set by server_terminate, which also runs from the signal handler, and
+     * read by the stats thread. */
+    sig_atomic_t stopped;
 };
 
 struct server *server_create(struct options *opts, char *err); 

src/sniffer.c

@@ -82,6 +82,11 @@ struct sniffer *sniffer_create(struct options *opts, char *err) {
         snprintf(err, MAX_ERR_BUFF_SIZE, "out of memory");
         goto error;
     }
+    if (pthread_mutex_init(&sniffer->stats_lock, NULL) != 0) {
+        snprintf(err, MAX_ERR_BUFF_SIZE, "failed to init the stats lock");
+        goto error;
+    }
+    sniffer->lock_ready = 1;
     sniffer->syn_tab->free = free_stats;
     sniffer->requests->free = free_request;
 
@@ -120,6 +125,7 @@ struct sniffer *sniffer_create(struct options *opts, char *err) {
 
 void sniffer_destroy(struct sniffer *sniffer) {
     if (!sniffer) return;
+    if (sniffer->lock_ready) pthread_mutex_destroy(&sniffer->stats_lock);
     if (sniffer->pcap) pcap_close(sniffer->pcap);
     free(sniffer->dev);
     free(sniffer->filter);

src/sniffer.h

@@ -13,6 +13,7 @@
 #define TCPKIT_SNIFFER_H
 
 #include <pcap.h>
+#include <pthread.h>
 #include <lua.h>
 #include "tcpkit.h"
 #include "stats.h"
@@ -26,12 +27,25 @@ struct sniffer {
     int threshold;
     int ascii;
 
+    /* syn_tab and the query_stats it owns are read by the stats thread while
+     * the capture thread updates them, so both sides hold stats_lock.
+     * requests belongs to the capture thread alone and needs no lock. */
+    pthread_mutex_t stats_lock;
+    int lock_ready;
     struct hashtable *syn_tab;
     struct hashtable *requests;
     lua_State *lua_state;
     struct bpf_program *bpf;
 };
 
+static inline void sniffer_stats_lock(struct sniffer *sniffer) {
+    pthread_mutex_lock(&sniffer->stats_lock);
+}
+
+static inline void sniffer_stats_unlock(struct sniffer *sniffer) {
+    pthread_mutex_unlock(&sniffer->stats_lock);
+}
+
 struct request {
     struct timeval tv;
     int seq;

tests/Makefile

@@ -41,12 +41,19 @@ PACKET_SRCS = $(SRC_DIR)/packet.c $(SRC_DIR)/protocol.c $(SRC_DIR)/hashtable.c \
 	      $(SRC_DIR)/stats.c $(SRC_DIR)/cJSON.c $(SRC_DIR)/log.c $(SRC_DIR)/lua.c
 
 test_packet: unit/test_packet.c $(PACKET_SRCS) $(LUA_DIR)/liblua.a
-	$(CC) $(CFLAGS) -o $@ $^ -lm -ldl
+	$(CC) $(CFLAGS) -o $@ $^ -lm -ldl -lpthread
+
+# Opt-in: rebuilds the binary under ThreadSanitizer, so it is kept out of
+# `make test` and clobbers the ordinary build.
+test-race:
+	$(MAKE) -C $(SRC_DIR) clean
+	$(MAKE) -C $(SRC_DIR) SANITIZE=thread
+	python3 e2e/race.py
 
 fixtures:
 	cd fixtures && python3 gen_fixtures.py
 
 clean:
 	- rm -rf $(SUITES) *.dSYM
 
-.PHONY: unit e2e fixtures clean
+.PHONY: unit e2e test-race fixtures clean

tests/e2e/race.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""Replays a busy capture while hammering the stats endpoint.
+
+Meant to be run against a binary built with -fsanitize=thread: the stats
+thread walks the endpoint table that the capture thread is still filling in,
+so any missing synchronisation shows up as a ThreadSanitizer report.
+"""
+
+import os
+import re
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TESTS = os.path.dirname(HERE)
+TCPKIT = os.environ.get("TCPKIT", os.path.join(TESTS, os.pardir, "src", "tcpkit"))
+PORT = int(os.environ.get("STATS_PORT", "34100"))
+TIMEOUT = float(os.environ.get("RACE_TIMEOUT", "60"))
+
+
+def main():
+    if not os.path.exists(TCPKIT):
+        print("race: %s not built" % TCPKIT)
+        return 1
+
+    with tempfile.TemporaryDirectory() as tmp:
+        capture = os.path.join(tmp, "stress.pcap")
+        subprocess.run([sys.executable, "gen_fixtures.py", "--stress", capture],
+                       cwd=os.path.join(TESTS, "fixtures"), check=True,
+                       stdout=subprocess.DEVNULL)
+
+        env = dict(os.environ)
+        env["TSAN_OPTIONS"] = "halt_on_error=0 " + env.get("TSAN_OPTIONS", "")
+        proc = subprocess.Popen(
+            [TCPKIT, "-r", capture, "-p", "redis", "-P", str(PORT)],
+            stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, env=env)
+
+        polls = 0
+        deadline = time.time() + TIMEOUT
+        while proc.poll() is None and time.time() < deadline:
+            try:
+                conn = socket.create_connection(("127.0.0.1", PORT), timeout=0.5)
+                conn.recv(1 << 20)
+                conn.close()
+                polls += 1
+            except OSError:
+                time.sleep(0.01)
+
+        try:
+            err = proc.communicate(timeout=30)[1]
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            print("race: the capture did not finish in time")
+            return 1
+
+    races = re.findall(r"SUMMARY: ThreadSanitizer: data race (.+)", err)
+    print("race: %d stats requests served during the replay" % polls)
+    if polls == 0:
+        print("race: the stats endpoint was never reached, nothing was exercised")
+        return 1
+    if races:
+        print("race: %d data races reported" % len(races))
+        for where in sorted(set(races)):
+            print("  " + where)
+        return 1
+    if proc.returncode != 0:
+        print("race: the capture exited with %d" % proc.returncode)
+        sys.stderr.write(err)
+        return 1
+    print("race: no data races reported")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/fixtures/gen_fixtures.py

@@ -6,6 +6,7 @@
 """
 
 import struct
+import sys
 
 PCAP_MAGIC = 0xA1B2C3D4
 DLT_EN10MB = 1
@@ -183,7 +184,27 @@ def malformed_payload():
     ]
 
 
+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
+    by the race check because of its size."""
+    global CLIENT_PORT
+    packets = []
+    ts = 0.0
+    for i in range(connections):
+        CLIENT_PORT = 20000 + (i % 40000)
+        packets.append(full(ts, to_server(b"", 1000, 0, SYN)))
+        packets.append(full(ts + 0.0001, to_client(b"", 5000, 1001, SYN | ACK)))
+        packets.append(full(ts + 0.001, to_server(resp("GET", "a"), 1001, 5001, PSH | ACK)))
+        packets.append(full(ts + 0.002, to_client(b"$1\r\nb\r\n", 5001, 1021, PSH | ACK)))
+        ts += 0.01
+    return packets
+
+
 if __name__ == "__main__":
-    write_pcap("redis-session.pcap", redis_session())
-    write_pcap("truncated.pcap", truncated())
-    write_pcap("malformed-payload.pcap", malformed_payload())
+    if len(sys.argv) > 2 and sys.argv[1] == "--stress":
+        write_pcap(sys.argv[2], stress())
+    else:
+        write_pcap("redis-session.pcap", redis_session())
+        write_pcap("truncated.pcap", truncated())
+        write_pcap("malformed-payload.pcap", malformed_payload())