diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index edc7b14..dab7bb4 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/Makefile b/Makefile index f6f9aff..e4d7847 100644 --- a/Makefile +++ b/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 diff --git a/src/Makefile b/src/Makefile index a9a8819..ec291fa 100644 --- a/src/Makefile +++ b/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 diff --git a/src/packet.c b/src/packet.c index 32909f0..f228df1 100644 --- a/src/packet.c +++ b/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)) { diff --git a/src/server.c b/src/server.c index 2791ecc..bd699cd 100644 --- a/src/server.c +++ b/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); diff --git a/src/server.h b/src/server.h index dc05fe2..d0bdd5d 100644 --- a/src/server.h +++ b/src/server.h @@ -13,15 +13,26 @@ #define TCPKIT_SERVER_H #include +#include #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); diff --git a/src/sniffer.c b/src/sniffer.c index 17d9354..600fd68 100644 --- a/src/sniffer.c +++ b/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); diff --git a/src/sniffer.h b/src/sniffer.h index 4118cf1..374018d 100644 --- a/src/sniffer.h +++ b/src/sniffer.h @@ -13,6 +13,7 @@ #define TCPKIT_SNIFFER_H #include +#include #include #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; diff --git a/tests/Makefile b/tests/Makefile index 7c0efa7..c863cc7 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -41,7 +41,14 @@ 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 @@ -49,4 +56,4 @@ fixtures: clean: - rm -rf $(SUITES) *.dSYM -.PHONY: unit e2e fixtures clean +.PHONY: unit e2e test-race fixtures clean diff --git a/tests/e2e/race.py b/tests/e2e/race.py new file mode 100644 index 0000000..81bba4b --- /dev/null +++ b/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()) diff --git a/tests/fixtures/gen_fixtures.py b/tests/fixtures/gen_fixtures.py index 37a765c..dbb381a 100644 --- a/tests/fixtures/gen_fixtures.py +++ b/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())