From 18b789b5a7c73d4eeab47c5e8e878f0175db3de1 Mon Sep 17 00:00:00 2001 From: Daniel Byrne Date: Wed, 21 Oct 2026 11:07:14 -0700 Subject: [PATCH 1/3] stats: per-thread counters, log-scaled histogram, intervals, JSON DTO_COLLECT_STATS kept one global table that every thread updated with atomic adds, so the cost of collecting grew with the thread count: a 4 KiB memcpy on the CPU path cost 55 ns more with stats on at one thread and 2.6 us more at 32 threads. Each thread now counts in its own block, mapped when the thread first records a call and folded into a retired total when the thread exits; output sums the live blocks and the retired total. Stats now cost about 40 ns per call (the two clock reads) at any thread count: extra cost per call, CPU path 1 thr 4 thr 16 thr 32 thr 4 KiB, before +55 +357 +1319 +2612 ns 4 KiB, after +44 +46 +42 +37 ns The histogram's 512 linear 4 KiB buckets, the last of which held everything from 2 MiB up, become log-scaled: under 64 bytes, then four buckets per power of two up to 4 GiB, then one for anything larger. New settings: DTO_STATS_INTERVAL_MS=N a sampler thread writes what happened in each N ms; the records add up to the summary DTO_STATS_FORMAT=json JSON Lines: interval and summary records with calls/bytes/ns per op and path, failures, histogram, and the configuration at exit DTO_STATS_FILE=...%p... %p becomes the process ID DTO_STATS_FILE is now appended to rather than truncated, so a forked or exec'd child's records don't overwrite its parent's; each record carries its pid. After fork the child starts with empty counts, and a %p file is reopened under its own pid. The old dto_stats_log() wrote each line twice when DTO_LOG_FILE was set and reused a consumed va_list; it is gone, along with the TRACE log level that turning on stats forced. Adds tests/test_stats.py (ctest -R stats) with a workload of known calls: totals and buckets, intervals summing to totals, 200 short-lived threads all counted and their blocks unmapped, fork, %p, and text output. --- CMakeLists.txt | 17 + README.md | 114 ++--- TESTING.md | 13 + dto.c | 840 ++++++++++++++++++++++++++++++------ tests/test_stats.py | 141 ++++++ tests/test_stats_workload.c | 132 ++++++ 6 files changed, 1078 insertions(+), 179 deletions(-) create mode 100644 tests/test_stats.py create mode 100644 tests/test_stats_workload.c diff --git a/CMakeLists.txt b/CMakeLists.txt index b7a9139..4419a70 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -221,6 +221,23 @@ if(DTO_BUILD_TESTS) LABELS "functional" TIMEOUT 60 ) + # Stats collection (DTO_COLLECT_STATS): a workload of known calls, checked + # by a Python script. Runs without DSA (everything is counted as CPU). + find_package(Python3 COMPONENTS Interpreter QUIET) + add_executable(dto-test-stats-workload tests/test_stats_workload.c) + target_compile_options(dto-test-stats-workload PRIVATE -fno-builtin) + target_link_libraries(dto-test-stats-workload PRIVATE DTO::dto pthread) + if(Python3_Interpreter_FOUND) + add_test(NAME stats + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/tests/test_stats.py + $) + set_tests_properties(stats PROPERTIES + LABELS "functional" + TIMEOUT 120 + ) + else() + message(STATUS "stats test: disabled (python3 not found)") + endif() # ---- A/B performance regression test ---- # The baseline object is compiled from unguarded upstream sources that diff --git a/README.md b/README.md index 71e1424..cb1a842 100644 --- a/README.md +++ b/README.md @@ -50,9 +50,12 @@ dto-4-dsa.conf: An example json config file for configuring DSAs Following environment variables control the behavior of DTO library: DTO_USESTDC_CALLS=0/1, 1 (uses std c memory functions only), 0 (uses DSA along with std c lib call; in case of DSA page fault - reverts to std c lib call). Default is 0. DTO_COLLECT_STATS=0/1, 1 (enables stats collection - #of operations, avg latency for each API, etc.>, 0 (disables stats collection). - Should be enabled for debugging/profiling only, not for perf evaluation (enabling it slows down the workload). Default is 0. - DTO_STATS_FILE= when DTO_COLLECT_STATS=1, redirect histogram output to the provided regular file instead of standard output. - The file is truncated when DTO loads; failures fall back to standard output and emit an error. + Adds about 40 ns per call (two clock reads), whatever the number of threads. Default is 0. See Statistics below. + DTO_STATS_FILE= when DTO_COLLECT_STATS=1, write stats to the provided regular file instead of standard output. Each %p in + the path is replaced by the process ID. Records are appended, not truncated: processes started from this one + append to the same file unless the path has a %p. Failures fall back to standard output and emit an error. + DTO_STATS_FORMAT=text/json text (default): the tables shown below; json: JSON Lines, one record per line (see Statistics). + DTO_STATS_INTERVAL_MS=N when N > 0, also write what happened in every N ms, from a background thread. Default is 0. DTO_WAIT_METHOD= (specifies the method to use while waiting for DSA to complete operation, default is busypoll) DTO_MIN_BYTES=xxxx (specifies minimum size of API call needed for DSA operation execution, default is 65536 bytes) DTO_CPU_SIZE_FRACTION=0.xx (specifies fraction of job performed by CPU, in parallel to DSA). Default is 0.00 @@ -310,63 +313,72 @@ to the DSA device. /cachebench -json_test_config --progress_stats_file=dto.log --report_api_latency -4. Sample histogram given below (generated using DTO_COLLECT_STATS=1) +4. Sample stats output (DTO_COLLECT_STATS=1; see [Statistics](#statistics)) i. Numbers under columns set, cpy, mov, and cmp show number of API calls or per-API completion latency for memset, memcpy, memmove, and memcmp respectively. ii. Numbers in bytes column show total bytes processed across all 4 API calls ------------------------------------------------------------------------------------------------------------------------------------------ -DTO Run Time: 22826 ms +``` +DTO Run Time: 8 ms ******** Number of Memory Operations ******** - <*************** stdc calls ***************> <*************** dsa (success) ***************> <*************** dsa (failed) ***************> <***** failure reason *****> -Byte Range -- set cpy mov cmp bytes set cpy mov cmp bytes set cpy mov cmp bytes Retries PFs Others - 0-4095 -- 24616626 12319310 7676731 244777209 4531255821 0 0 0 0 0 0 0 0 0 0 0 0 0 - 4096-8191 -- 3 290 14 0 1842507 0 0 0 0 0 0 0 0 0 0 0 0 0 - 8192-12287 -- 0 2 6 0 45579 64 176 2 0 2545968 0 2 6 0 25276 0 8 0 - 12288-16383 -- 1 1 0 0 20756 1 137 0 0 1965343 1 1 0 0 11172 0 2 0 - 16384-20479 -- 0 1 3 0 41974 0 126 2 0 2328434 0 1 3 0 24337 0 4 0 - 32768-36863 -- 0 0 0 0 0 0 67 2 0 2378165 0 0 0 0 0 0 0 0 - 53248-57343 -- 0 0 0 0 0 0 46 0 0 2545890 0 0 0 0 0 0 0 0 - 98304-102399 -- 0 0 0 0 0 0 13 0 0 1309627 0 0 0 0 0 0 0 0 - 106496-110591 -- 0 1 0 0 31637 0 8 0 0 869880 0 1 0 0 77586 0 1 0 - 118784-122879 -- 0 0 0 0 0 0 9 0 0 1083519 0 0 0 0 0 0 0 0 - 131072-135167 -- 0 0 2 0 175638 0 5 0 0 665475 0 0 2 0 86506 0 2 0 - 143360-147455 -- 0 0 0 0 0 0 1 0 0 143399 0 0 0 0 0 0 0 0 - 147456-151551 -- 0 0 0 0 0 0 1 0 0 148647 0 0 0 0 0 0 0 0 - 159744-163839 -- 0 0 0 0 0 0 6 0 0 970186 0 0 0 0 0 0 0 0 - 212992-217087 -- 0 0 11 0 1580248 0 4 2 0 1292252 0 0 11 0 779384 0 11 0 - 278528-282623 -- 0 0 0 0 0 0 1 0 0 279719 0 0 0 0 0 0 0 0 - 466944-471039 -- 0 0 0 0 0 0 1 0 0 467783 0 0 0 0 0 0 0 0 - 528384-532479 -- 0 0 0 0 0 0 1 0 0 529927 0 0 0 0 0 0 0 0 - 720896-724991 -- 0 0 0 0 0 1 0 0 0 720896 0 0 0 0 0 0 0 0 - 999424-1003519 -- 0 0 0 0 0 0 1 0 0 1002439 0 0 0 0 0 0 0 0 - >=2093056 -- 0 1 0 0 1975911 0 0 0 0 0 0 1 0 0 973209 0 1 0 + <*************** stdc calls ***************> <*************** dsa (success) ***************> <*************** dsa (failed) ***************> <***** failure reason *****> +Byte Range -- set cpy mov cmp bytes set cpy mov cmp bytes set cpy mov cmp bytes Retry PFs Others + 65536-81919 -- 0 0 0 0 0 200 0 0 0 13926400 0 0 0 0 0 0 0 0 + 98304-114687 -- 0 0 200 0 19662200 0 0 0 0 0 0 0 0 0 0 0 0 0 + 131072-163839 -- 0 0 0 0 0 0 200 0 0 26239000 0 0 0 0 0 0 0 0 + 262144-327679 -- 0 0 0 0 0 0 0 0 200 52441600 0 0 0 0 0 0 0 0 ******** Average Memory Operation Latency (us) ******** - <******** stdc calls ********> <******** dsa (success) ********> <******** dsa (failed) ********> -Byte Range -- set cpy mov cmp set cpy mov cmp set cpy mov cmp - 0-4095 -- 0.01 0.02 0.01 0.04 0 0 0 0 0 0 0 0 - 4096-8191 -- 0.07 0.42 0.47 0 0 0 0 0 0 0 0 0 - 8192-12287 -- 0 1.29 0.47 0 0.84 1.19 1.82 0 0 56.18 2.54 0 - 12288-16383 -- 2.90 3.34 0 0 7.49 1.26 0 0 2.07 1.93 0 0 - 16384-20479 -- 0 0.27 3.24 0 0 1.37 1.86 0 0 84.92 2.47 0 - 32768-36863 -- 0 0 0 0 0 1.87 2.91 0 0 0 0 0 - 53248-57343 -- 0 0 0 0 0 2.45 0 0 0 0 0 0 - 98304-102399 -- 0 0 0 0 0 3.76 0 0 0 0 0 0 - 106496-110591 -- 0 229.18 0 0 0 4.19 0 0 0 3.65 0 0 - 118784-122879 -- 0 0 0 0 0 5.18 0 0 0 0 0 0 - 131072-135167 -- 0 0 25.00 0 0 5.92 0 0 0 0 14.39 0 - 143360-147455 -- 0 0 0 0 0 5.11 0 0 0 0 0 0 - 147456-151551 -- 0 0 0 0 0 6.16 0 0 0 0 0 0 - 159744-163839 -- 0 0 0 0 0 7.65 0 0 0 0 0 0 - 212992-217087 -- 0 0 36.10 0 0 8.15 6.59 0 0 0 18.07 0 - 278528-282623 -- 0 0 0 0 0 15.47 0 0 0 0 0 0 - 466944-471039 -- 0 0 0 0 0 15.91 0 0 0 0 0 0 - 528384-532479 -- 0 0 0 0 0 18.65 0 0 0 0 0 0 - 720896-724991 -- 0 0 0 0 17.71 0 0 0 0 0 0 0 - 999424-1003519 -- 0 0 0 0 0 28.42 0 0 0 0 0 0 - >=2093056 -- 0 422.15 0 0 0 0 0 0 0 272.87 0 0 + <******** stdc calls ********> <******** dsa (success) ********> <******** dsa (failed) ********> +Byte Range -- set cpy mov cmp set cpy mov cmp set cpy mov cmp + 65536-81919 -- 0 0 0 0 8.06 0 0 0 0 0 0 0 + 98304-114687 -- 0 0 4.18 0 0 0 0 0 0 0 0 0 + 131072-163839 -- 0 0 0 0 0 17.02 0 0 0 0 0 0 + 262144-327679 -- 0 0 0 0 0 0 0 34.50 0 0 0 0 +``` + +## Statistics + +With `DTO_COLLECT_STATS=1`, DTO times and counts every memset, memcpy, +memmove and memcmp it intercepts, by operation, by path (CPU, DSA, or DSA +failed and finished on the CPU) and by size, and writes a summary when the +process exits (or on Ctrl-C). Each thread counts on its own, so collection +costs about 40 ns per call however many threads there are, and the counts +are summed for output. + +The size histogram has a bucket for calls under 64 bytes, then four buckets +per power of two up to 4 GiB (65536-81919, 81920-98303, ...), and one for +anything larger. A call that DSA completed only partly (for example on a +page fault) is counted once as `dsa (failed)` and once as a CPU call for the +remainder. + +With `DTO_STATS_INTERVAL_MS=N`, a background thread also writes, every N ms, +what happened since the previous record, so activity can be followed over +time. The interval records add up to the summary's totals. + +With `DTO_STATS_FORMAT=json` the output is JSON Lines, one record per line: + +``` +{"type":"interval","pid":183654,"t_ms":14.333,"dt_ms":14.333,"ops":{"memcpy":{"dsa":{"calls":8,"bytes":1049560,"ns":96273}},"memmove":{"cpu":{"calls":8,"bytes":786488,"ns":31601}}},"failures":{"retry":0,"page_fault":0,"other":0}} +{"type":"summary","pid":183654,"program":"prog","run_ms":23.395, + "config":{"stdc_only":0,"wait_method":"busypoll","min_bytes":65536,"cpu_size_fraction":0.00,"auto_adjust":1,"numa_aware":0,"wqs":1}, + "totals":{"memcpy":{"dsa":{"calls":12,"bytes":1574340,"ns":138591}},...}, + "failures":{"retry":0,"page_fault":0,"other":0}, + "histogram":[{"min":131072,"max":163839,"ops":{"memcpy":{"dsa":{"calls":12,"bytes":1574340,"ns":138591}}},"failures":{...}},...]} +``` + +(The summary is a single line; it is wrapped here.) `ops` maps `memset`, +`memcpy`, `memmove` and `memcmp` to the paths they took, `cpu`, `dsa` and +`dsa_failed`, each with the number of calls, bytes and total nanoseconds; +operations and paths with no calls are left out. `t_ms` and `run_ms` count +from DTO's start, and `config` holds the settings at exit, including the +auto-tuned `min_bytes` and `cpu_size_fraction`. + +Every process that loads DTO with stats enabled writes its own records, +tagged with its `pid`. They share `DTO_STATS_FILE` (records are appended +whole) unless the path contains `%p`, which gives each process its own file. ## Usage Notes diff --git a/TESTING.md b/TESTING.md index db6b543..ab8a41f 100644 --- a/TESTING.md +++ b/TESTING.md @@ -95,6 +95,19 @@ ctest -R dtoctl tests/test_dtoctl.sh build/dtoctl build/libdto.so ``` +## Stats Tests + +`stats` runs `dto-test-stats-workload`, which makes a known number of calls +of known sizes, with `DTO_COLLECT_STATS` set, and checks the output: totals +and histogram buckets, interval records adding up to the totals, calls from +200 short-lived threads all counted without their stats memory piling up, +separate records for a forked child, `%p` file names, and the text format. +It needs Python 3 and no DSA hardware. + +```bash +ctest -R stats +``` + ## Performance Tests The performance test (`perf`) detects regressions in libdto by diff --git a/dto.c b/dto.c index dd76ef4..51899d0 100644 --- a/dto.c +++ b/dto.c @@ -4,6 +4,7 @@ * SPDX-License-Identifier: MIT ******************************************************************************/ +#include #include #include #include @@ -187,9 +188,13 @@ static const char * const memop_names[] = { [MEMCMP] = "cmp" }; -// memory stats -#define HIST_BUCKET_SIZE 4096 -#define HIST_NO_BUCKETS 512 +// memory stats: a histogram by call size, with bucket 0 for calls under +// 64 bytes, then four buckets per power of two up to 4 GiB, and a last +// bucket for anything larger +#define HIST_MIN_SHIFT 6 +#define HIST_MAX_SHIFT 32 +#define HIST_SUBS 4 +#define HIST_NO_BUCKETS (2 + (HIST_MAX_SHIFT - HIST_MIN_SHIFT) * HIST_SUBS) enum stat_group { STDC_CALL = 0x0, DSA_CALL_SUCCESS, @@ -268,10 +273,57 @@ static struct timespec dto_start_time; } \ } while (0) \ -static atomic_int op_counter[HIST_NO_BUCKETS][MAX_STAT_GROUP][MAX_MEMOP]; -static atomic_ullong bytes_counter[HIST_NO_BUCKETS][MAX_STAT_GROUP]; -static atomic_ullong lat_counter[HIST_NO_BUCKETS][MAX_STAT_GROUP][MAX_MEMOP]; -static atomic_int fail_counter[HIST_NO_BUCKETS][MAX_FAILURES]; +/* the groups that have per-operation counters */ +#define STAT_GROUPS DSA_FAIL_CODES + +struct dto_op_stats { + uint64_t calls; + uint64_t bytes; + uint64_t ns; +}; + +struct dto_bucket_stats { + struct dto_op_stats op[STAT_GROUPS][MAX_MEMOP]; + uint64_t fails[MAX_FAILURES]; +}; + +struct dto_stats { + struct dto_op_stats total[STAT_GROUPS][MAX_MEMOP]; + uint64_t fails[MAX_FAILURES]; + struct dto_bucket_stats hist[HIST_NO_BUCKETS]; +}; + +/* one per counting thread (see update_stats) */ +struct dto_thread_stats { + struct dto_thread_stats *prev, *next; + struct dto_stats s; +}; + +static struct dto_thread_stats *stats_threads; /* live threads' blocks */ +static struct dto_stats stats_retired; /* exited threads, summed */ +static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_key_t stats_key; +static bool stats_key_created; +static __thread struct dto_thread_stats *thr_stats; +static __thread bool thr_stats_off; /* don't count this thread's calls */ + +/* output: one formatter at a time (sampler, exit, SIGINT) */ +static pthread_mutex_t stats_out_lock = PTHREAD_MUTEX_INITIALIZER; +static struct dto_stats stats_scratch; +enum { STATS_FORMAT_TEXT, STATS_FORMAT_JSON }; +static int stats_format; + +static bool stats_file_per_pid; /* DTO_STATS_FILE has a %p */ + +/* DTO_STATS_INTERVAL_MS */ +static unsigned long stats_interval_ms; +static atomic_uchar sampler_started; +static struct dto_stats sampler_prev; /* totals at the previous record */ +static uint64_t sampler_prev_ns; + +static void print_stats(void); +static void print_stats_from_signal(void); +static void stats_fork_child(void); #endif /* call initialize/cleanup functions when library is loaded/unloaded */ @@ -292,29 +344,7 @@ static enum { static unsigned int log_level = LOG_LEVEL_FATAL; -static void dto_stats_log(const char *fmt, ...) -{ - char buf[512]; - va_list args; - - if (stats_fd == -1 && LOG_LEVEL_TRACE > log_level) - return; - - va_start(args, fmt); - if (stats_fd != -1) { - if (vsnprintf(buf, sizeof(buf), fmt, args) > 0) - write(stats_fd, buf, strlen(buf)); - } else if (log_fd == -1) { - vprintf(fmt, args); - } else { - vsnprintf(buf, sizeof(buf), fmt, args); - write(log_fd, buf, strlen(buf)); - } - va_end(args); -} - #ifdef DTO_STATS_SUPPORT -static void print_stats(void); static struct sigaction previous_sigint_action; static volatile sig_atomic_t sigint_handler_installed; static int previous_sigint_action_valid; @@ -338,8 +368,7 @@ static void restore_sigint_handler(void) static void dto_handle_sigint(int signo) { - if (collect_stats) - print_stats(); + print_stats_from_signal(); restore_sigint_handler(); @@ -418,20 +447,7 @@ static void dto_log(int req_log_level, const char *fmt, ...) static void child (void) { #ifdef DTO_STATS_SUPPORT - int i, j, k; - - /* Reset the counters */ - for (i = 0; i < HIST_NO_BUCKETS; i++) { - for (j = 0; j < MAX_STAT_GROUP; j++) { - for (k = 0; k < MAX_MEMOP; k++) { - op_counter[i][j][k] = 0; - lat_counter[i][j][k] = 0; - } - bytes_counter[i][j] = 0; - } - for (j = 0; j < MAX_FAILURES; j++) - fail_counter[i][j] = 0; - } + stats_fork_child(); #endif dto_initializing = 0; dto_initialized = 0; @@ -672,116 +688,662 @@ static __always_inline int dsa_execute(struct dto_wq *wq, } #ifdef DTO_STATS_SUPPORT +/* + * Statistics (DTO_COLLECT_STATS). + * + * Each thread counts into its own block, so an update is a few plain loads + * and stores with no shared cache line; the blocks are summed when stats are + * printed. A block is mmap'd on the thread's first update (only the pages it + * touches are ever allocated), registered in a list, and on thread exit + * folded into stats_retired and unmapped. The totals sit at the start of the + * block, so the interval sampler reads one page per thread. + */ + +/* Histogram bucket of a call of n bytes. */ +static __always_inline int stats_bucket(size_t n) +{ + int e; + + if (n < (1UL << HIST_MIN_SHIFT)) + return 0; + e = 63 - __builtin_clzl(n); + if (e >= HIST_MAX_SHIFT) + return HIST_NO_BUCKETS - 1; + return 1 + (e - HIST_MIN_SHIFT) * HIST_SUBS + + (int)((n >> (e - 2)) & (HIST_SUBS - 1)); +} + +/* Smallest size counted in bucket b. */ +static uint64_t stats_bucket_min(int b) +{ + int e; + + if (b == 0) + return 0; + if (b == HIST_NO_BUCKETS - 1) + return 1ULL << HIST_MAX_SHIFT; + e = HIST_MIN_SHIFT + (b - 1) / HIST_SUBS; + return (uint64_t)(HIST_SUBS + (b - 1) % HIST_SUBS) << (e - 2); +} + +/* Relaxed load-add-store: the owning thread is the only writer, and + * readers summing blocks only need each counter to be read whole. */ +#define STAT_ADD(field, v) \ + __atomic_store_n(&(field), __atomic_load_n(&(field), __ATOMIC_RELAXED) + (v), \ + __ATOMIC_RELAXED) +#define STAT_GET(field) __atomic_load_n(&(field), __ATOMIC_RELAXED) + +static __always_inline void stats_add_op(struct dto_op_stats *s, uint64_t bytes, + uint64_t ns) +{ + STAT_ADD(s->calls, 1); + STAT_ADD(s->bytes, bytes); + STAT_ADD(s->ns, ns); +} + +/* dst += src for the totals, or for everything with hist */ +static void stats_sum(struct dto_stats *dst, struct dto_stats *src, bool hist) +{ + for (int g = 0; g < STAT_GROUPS; g++) + for (int o = 0; o < MAX_MEMOP; o++) { + dst->total[g][o].calls += STAT_GET(src->total[g][o].calls); + dst->total[g][o].bytes += STAT_GET(src->total[g][o].bytes); + dst->total[g][o].ns += STAT_GET(src->total[g][o].ns); + } + for (int f = 0; f < MAX_FAILURES; f++) + dst->fails[f] += STAT_GET(src->fails[f]); + if (!hist) + return; + for (int b = 0; b < HIST_NO_BUCKETS; b++) { + struct dto_bucket_stats *d = &dst->hist[b], *s = &src->hist[b]; + + for (int g = 0; g < STAT_GROUPS; g++) + for (int o = 0; o < MAX_MEMOP; o++) { + d->op[g][o].calls += STAT_GET(s->op[g][o].calls); + d->op[g][o].bytes += STAT_GET(s->op[g][o].bytes); + d->op[g][o].ns += STAT_GET(s->op[g][o].ns); + } + for (int f = 0; f < MAX_FAILURES; f++) + d->fails[f] += STAT_GET(s->fails[f]); + } +} + +/* Sum of every thread's stats, live and exited, into *out (zeroed first). + * The caller holds stats_lock. */ +static void stats_collect(struct dto_stats *out, bool hist) +{ + size_t len = hist ? sizeof(*out) : offsetof(struct dto_stats, hist); + + orig_memset(out, 0, len); + stats_sum(out, &stats_retired, hist); + for (struct dto_thread_stats *t = stats_threads; t; t = t->next) + stats_sum(out, &t->s, hist); +} + +static void stats_snapshot(struct dto_stats *out, bool hist) +{ + pthread_mutex_lock(&stats_lock); + stats_collect(out, hist); + pthread_mutex_unlock(&stats_lock); +} + +static void stats_start_sampler(void); + +static void stats_thread_exit(void *arg) +{ + struct dto_thread_stats *ts = arg; + + /* memcpy and friends may still run during the rest of thread exit */ + thr_stats_off = true; + thr_stats = NULL; + pthread_mutex_lock(&stats_lock); + stats_sum(&stats_retired, &ts->s, true); + if (ts->prev) + ts->prev->next = ts->next; + else + stats_threads = ts->next; + if (ts->next) + ts->next->prev = ts->prev; + pthread_mutex_unlock(&stats_lock); + munmap(ts, sizeof(*ts)); +} + +/* First update on this thread: give it a block. NULL leaves the thread + * uncounted (thr_stats_off stays set, so it is not retried every call). */ +static struct dto_thread_stats *stats_thread_register(void) +{ + struct dto_thread_stats *ts; + + /* mmap, the registry and pthread_setspecific must not count themselves */ + thr_stats_off = true; + ts = mmap(NULL, sizeof(*ts), PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS, -1, 0); + if (ts == MAP_FAILED) + return NULL; + + pthread_mutex_lock(&stats_lock); + ts->prev = NULL; + ts->next = stats_threads; + if (stats_threads) + stats_threads->prev = ts; + stats_threads = ts; + pthread_mutex_unlock(&stats_lock); + pthread_setspecific(stats_key, ts); + thr_stats = ts; + + if (stats_interval_ms && !atomic_load_explicit(&sampler_started, + memory_order_acquire)) + stats_start_sampler(); + thr_stats_off = false; + return ts; +} + static void update_stats(int op, size_t n, bool overlapping, size_t bytes_completed, uint64_t elapsed_ns, int group, int error_code) { + struct dto_thread_stats *ts; + struct dto_bucket_stats *b; + // dto_memcpymove didn't actually submit the request to DSA, so there is nothing to log. This will be captured by a second call if (op == MEMMOVE && overlapping && dto_overlapping_memmove_action == OVERLAPPING_CPU && group == DSA_CALL_SUCCESS) { return; } + if (unlikely(thr_stats_off)) + return; + ts = thr_stats; + if (unlikely(ts == NULL)) { + ts = stats_thread_register(); + if (ts == NULL) + return; + } - int bucket = (n / HIST_BUCKET_SIZE); + b = &ts->s.hist[stats_bucket(n)]; + stats_add_op(&ts->s.total[group][op], bytes_completed, elapsed_ns); + stats_add_op(&b->op[group][op], bytes_completed, elapsed_ns); + if (group == DSA_CALL_FAILED) { + STAT_ADD(ts->s.fails[error_code], 1); + STAT_ADD(b->fails[error_code], 1); + } +} - if (bucket >= HIST_NO_BUCKETS) /* last bucket includes remaining sizes */ - bucket = HIST_NO_BUCKETS-1; - ++op_counter[bucket][group][op]; - bytes_counter[bucket][group] += bytes_completed; - lat_counter[bucket][group][op] += elapsed_ns; - if (group == DSA_CALL_FAILED) - ++fail_counter[bucket][error_code]; +/* ---- Output ---- + * + * Records are formatted into stats_out and written with as few write()s as + * possible: to DTO_STATS_FILE (opened O_APPEND, so records from several + * processes sharing it stay whole), else DTO_LOG_FILE, else stdout. */ + +static char stats_out[1 << 16]; +static size_t stats_out_len; + +static void stats_flush(void) +{ + int fd = stats_fd != -1 ? stats_fd : log_fd != -1 ? log_fd : STDOUT_FILENO; + size_t off = 0; + + while (off < stats_out_len) { + ssize_t r = write(fd, stats_out + off, stats_out_len - off); + if (r <= 0) + break; + off += r; + } + stats_out_len = 0; } -static void print_stats(void) +__attribute__((format(printf, 1, 2))) +static void stats_printf(const char *fmt, ...) { - struct timespec dto_end_time; + va_list args; + int len; - if (likely(!collect_stats)) + va_start(args, fmt); + len = vsnprintf(stats_out + stats_out_len, sizeof(stats_out) - stats_out_len, + fmt, args); + va_end(args); + if (len < 0) return; + if ((size_t)len >= sizeof(stats_out) - stats_out_len) { + /* no room: write out what is there and format again */ + stats_flush(); + va_start(args, fmt); + len = vsnprintf(stats_out, sizeof(stats_out), fmt, args); + va_end(args); + if (len < 0) + return; + if ((size_t)len >= sizeof(stats_out)) + len = sizeof(stats_out) - 1; + } + stats_out_len += len; +} + +/* DTO_STATS_FILE, with each %p replaced by the process ID. It is opened + * O_APPEND, not truncated: processes started from this one (fork, exec) + * inherit the setting, and with %p each gets a file of its own, without it + * they append whole records to a shared one. */ +static void stats_open_file(const char *name) +{ + char path[PATH_MAX]; + size_t j = 0; + struct stat st; + + for (const char *c = name; *c && j < sizeof(path) - 1; c++) { + if (c[0] == '%' && c[1] == 'p') { + stats_file_per_pid = true; + int n = snprintf(path + j, sizeof(path) - j, "%d", getpid()); + + if (n < 0 || (size_t)n >= sizeof(path) - j) + break; + j += n; + c++; + } else { + path[j++] = *c; + } + } + path[j] = '\0'; + + if (lstat(path, &st) == 0 && (st.st_mode & S_IFMT) != S_IFREG) { + LOG_ERROR("DTO_STATS_FILE '%s' must reference a regular file\n", path); + return; + } + stats_fd = open(path, O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0600); + if (stats_fd == -1) + LOG_ERROR("Failed to open DTO_STATS_FILE '%s': %s\n", path, strerror(errno)); +} + +static const char * const memop_json[] = { + [MEMSET] = "memset", + [MEMCOPY] = "memcpy", + [MEMMOVE] = "memmove", + [MEMCMP] = "memcmp", +}; - clock_gettime(CLOCK_BOOTTIME, &dto_end_time); +static const char * const group_json[] = { + [STDC_CALL] = "cpu", + [DSA_CALL_SUCCESS] = "dsa", + [DSA_CALL_FAILED] = "dsa_failed", +}; - dto_stats_log("DTO Run Time: %ld ms\n", TS_NS(dto_start_time, dto_end_time)/1000000); +static const char * const failure_json[] = { + [RETRY] = "retry", + [PAGE_FAULT] = "page_fault", + [FAIL_OTHERS] = "other", +}; + +static uint64_t stats_now_ns(void) +{ + struct timespec now; + + clock_gettime(CLOCK_BOOTTIME, &now); + return TS_NS(dto_start_time, now); +} + +/* {"memcpy":{"dsa":{"calls":..,"bytes":..,"ns":..},..},..} without zeroes */ +static void stats_json_ops(struct dto_op_stats op[STAT_GROUPS][MAX_MEMOP]) +{ + bool first_op = true; + + stats_printf("{"); + for (int o = 0; o < MAX_MEMOP; o++) { + bool first_group = true; + + for (int g = 0; g < STAT_GROUPS; g++) { + if (!op[g][o].calls) + continue; + if (first_group) + stats_printf("%s\"%s\":{", first_op ? "" : ",", memop_json[o]); + stats_printf("%s\"%s\":{\"calls\":%lu,\"bytes\":%lu,\"ns\":%lu}", + first_group ? "" : ",", group_json[g], op[g][o].calls, + op[g][o].bytes, op[g][o].ns); + first_group = false; + first_op = false; + } + if (!first_group) + stats_printf("}"); + } + stats_printf("}"); +} + +static void stats_json_fails(uint64_t fails[MAX_FAILURES]) +{ + stats_printf("{"); + for (int f = 1; f < MAX_FAILURES; f++) + stats_printf("%s\"%s\":%lu", f == 1 ? "" : ",", failure_json[f], fails[f]); + stats_printf("}"); +} + +/* Text: "memcpy dsa 120 calls 7864320 B 421.3 us, cpu ..." */ +static void stats_text_ops(struct dto_op_stats op[STAT_GROUPS][MAX_MEMOP]) +{ + bool any = false; + + for (int o = 0; o < MAX_MEMOP; o++) + for (int g = 0; g < STAT_GROUPS; g++) { + if (!op[g][o].calls) + continue; + stats_printf("%s%s %s %lu calls %lu B %.1f us", any ? ", " : "", + memop_json[o], group_json[g], op[g][o].calls, + op[g][o].bytes, op[g][o].ns / 1000.0); + any = true; + } + if (!any) + stats_printf("no calls"); +} + +/* One interval record: the change in the totals since the previous one. + * The caller holds stats_out_lock. */ +static void stats_emit_interval(void) +{ + struct dto_stats *now = &stats_scratch; /* totals part only */ + struct dto_op_stats delta[STAT_GROUPS][MAX_MEMOP]; + uint64_t fails[MAX_FAILURES]; + uint64_t t = stats_now_ns(); + + stats_snapshot(now, false); + for (int g = 0; g < STAT_GROUPS; g++) + for (int o = 0; o < MAX_MEMOP; o++) { + delta[g][o].calls = now->total[g][o].calls - sampler_prev.total[g][o].calls; + delta[g][o].bytes = now->total[g][o].bytes - sampler_prev.total[g][o].bytes; + delta[g][o].ns = now->total[g][o].ns - sampler_prev.total[g][o].ns; + } + for (int f = 0; f < MAX_FAILURES; f++) + fails[f] = now->fails[f] - sampler_prev.fails[f]; + orig_memcpy(&sampler_prev, now, offsetof(struct dto_stats, hist)); + + if (stats_format == STATS_FORMAT_JSON) { + stats_printf("{\"type\":\"interval\",\"pid\":%d,\"t_ms\":%.3f,\"dt_ms\":%.3f,\"ops\":", + getpid(), t / 1e6, (t - sampler_prev_ns) / 1e6); + stats_json_ops(delta); + stats_printf(",\"failures\":"); + stats_json_fails(fails); + stats_printf("}\n"); + } else { + stats_printf("DTO stats t=%.3f ms (+%.3f ms): ", t / 1e6, (t - sampler_prev_ns) / 1e6); + stats_text_ops(delta); + if (fails[RETRY] || fails[PAGE_FAULT] || fails[FAIL_OTHERS]) + stats_printf("; failures retry %lu page_fault %lu other %lu", + fails[RETRY], fails[PAGE_FAULT], fails[FAIL_OTHERS]); + stats_printf("\n"); + } + sampler_prev_ns = t; + stats_flush(); +} + +static bool stats_bucket_empty(struct dto_bucket_stats *b) +{ + for (int g = 0; g < STAT_GROUPS; g++) + for (int o = 0; o < MAX_MEMOP; o++) + if (b->op[g][o].calls) + return false; + return true; +} + +static void stats_print_json(struct dto_stats *s, uint64_t run_ns) +{ + char prog[64]; + bool first = true; + size_t j = 0; + + /* the program name as a JSON string: drop quotes, backslashes and + * control characters rather than escape them */ + for (const char *c = __progname; *c && j < sizeof(prog) - 1; c++) + if (*c != '"' && *c != '\\' && (unsigned char)*c >= 0x20) + prog[j++] = *c; + prog[j] = '\0'; + stats_printf("{\"type\":\"summary\",\"pid\":%d,\"program\":\"%s\",\"run_ms\":%.3f,", + getpid(), prog, run_ns / 1e6); + stats_printf("\"config\":{\"stdc_only\":%d,\"wait_method\":\"%s\",\"min_bytes\":%lu," + "\"cpu_size_fraction\":%.2f,\"auto_adjust\":%d,\"numa_aware\":%d,\"wqs\":%d},", + use_std_lib_calls, wait_names[wait_method], dsa_min_size, + cpu_size_fraction / 100.0, auto_adjust_knobs, (int)is_numa_aware, num_wqs); + stats_printf("\"totals\":"); + stats_json_ops(s->total); + stats_printf(",\"failures\":"); + stats_json_fails(s->fails); + stats_printf(",\"histogram\":["); + for (int b = 0; b < HIST_NO_BUCKETS; b++) { + if (stats_bucket_empty(&s->hist[b])) + continue; + stats_printf("%s{\"min\":%lu,", first ? "" : ",", stats_bucket_min(b)); + if (b < HIST_NO_BUCKETS - 1) + stats_printf("\"max\":%lu,", stats_bucket_min(b + 1) - 1); + stats_printf("\"ops\":"); + stats_json_ops(s->hist[b].op); + stats_printf(",\"failures\":"); + stats_json_fails(s->hist[b].fails); + stats_printf("}"); + first = false; + } + stats_printf("]}\n"); +} + +static void stats_print_text(struct dto_stats *s, uint64_t run_ns) +{ + stats_printf("DTO Run Time: %lu ms\n", run_ns / 1000000); - // display stats for (int t = 0; t < 2; ++t) { if (t == 0) - dto_stats_log("\n******** Number of Memory Operations ********\n"); + stats_printf("\n******** Number of Memory Operations ********\n"); else - dto_stats_log("\n******** Average Memory Operation Latency (us) ********\n"); + stats_printf("\n******** Average Memory Operation Latency (us) ********\n"); - dto_stats_log("%17s ", ""); + stats_printf("%23s ", ""); for (int g = 0; g < MAX_STAT_GROUP; ++g) { if (t == 0) { if (g == DSA_FAIL_CODES) - dto_stats_log("<***** %-13s *****> ", stat_group_names[g]); + stats_printf("<***** %-13s *****> ", stat_group_names[g]); else - dto_stats_log("<*************** %-13s ***************> ", stat_group_names[g]); - } else { - if (g != DSA_FAIL_CODES) - dto_stats_log("<******** %-13s ********> ", stat_group_names[g]); - + stats_printf("<*************** %-13s ***************> ", stat_group_names[g]); + } else if (g != DSA_FAIL_CODES) { + stats_printf("<******** %-13s ********> ", stat_group_names[g]); } } - dto_stats_log("\n"); - - dto_stats_log("%-17s -- ", "Byte Range"); - for (int g = 0; g < MAX_STAT_GROUP - 1; ++g) { + stats_printf("\n%-23s -- ", "Byte Range"); + for (int g = 0; g < STAT_GROUPS; ++g) { for (int o = 0; o < MAX_MEMOP; ++o) - dto_stats_log("%-8s ", memop_names[o]); + stats_printf("%-8s ", memop_names[o]); if (t == 0) - dto_stats_log("%-12s ", "bytes"); + stats_printf("%-12s ", "bytes"); } if (t == 0) for (int o = 1; o < MAX_FAILURES; ++o) - dto_stats_log("%-6s ", failure_names[o]); - dto_stats_log("\n"); + stats_printf("%-6s ", failure_names[o]); + stats_printf("\n"); for (int b = 0; b < HIST_NO_BUCKETS; ++b) { - bool empty = true; + struct dto_bucket_stats *h = &s->hist[b]; - for (int g = 0; g < MAX_STAT_GROUP; ++g) { - for (int o = 0; o < MAX_MEMOP; ++o) { - if (op_counter[b][g][o] != 0) { - empty = false; - break; - } - } - if (!empty) - break; - } - if (empty) + if (stats_bucket_empty(h)) continue; - - if (b < (HIST_NO_BUCKETS-1)) - dto_stats_log("% 8d-%-8d -- ", b*4096, ((b+1)*4096)-1); + if (b < HIST_NO_BUCKETS - 1) + stats_printf("%11lu-%-11lu -- ", stats_bucket_min(b), + stats_bucket_min(b + 1) - 1); else - dto_stats_log(" >=%-12d -- ", b*4096); + stats_printf(" >=%-18lu -- ", stats_bucket_min(b)); + + for (int g = 0; g < STAT_GROUPS; ++g) { + uint64_t bytes = 0; - for (int g = 0; g < MAX_STAT_GROUP - 1; ++g) { for (int o = 0; o < MAX_MEMOP; ++o) { - if (t == 0) { - dto_stats_log("%-8d ", op_counter[b][g][o]); - continue; - } - if (op_counter[b][g][o] != 0) { - double avg_us = ((double) lat_counter[b][g][o])/(((double) op_counter[b][g][o]) * 1000.0); + struct dto_op_stats *op = &h->op[g][o]; - dto_stats_log("%-8.2f ", avg_us); - } else { - dto_stats_log("%-8d ", 0); - } + bytes += op->bytes; + if (t == 0) + stats_printf("%-8lu ", op->calls); + else if (op->calls) + stats_printf("%-8.2f ", op->ns / (op->calls * 1000.0)); + else + stats_printf("%-8d ", 0); } if (t == 0) - dto_stats_log("%-12lld ", bytes_counter[b][g]); + stats_printf("%-12lu ", bytes); } if (t == 0) for (int o = 1; o < MAX_FAILURES; ++o) - dto_stats_log("%-6d ", fail_counter[b][o]); - dto_stats_log("\n"); + stats_printf("%-6lu ", h->fails[o]); + stats_printf("\n"); } } } + +/* The end-of-run summary of *s. The caller holds stats_out_lock. */ +static void stats_print_summary(struct dto_stats *s) +{ + if (stats_format == STATS_FORMAT_JSON) + stats_print_json(s, stats_now_ns()); + else + stats_print_text(s, stats_now_ns()); + stats_flush(); +} + +static void print_stats(void) +{ + bool off = thr_stats_off; + + if (likely(!collect_stats)) + return; + thr_stats_off = true; + pthread_mutex_lock(&stats_out_lock); + stats_snapshot(&stats_scratch, true); + stats_print_summary(&stats_scratch); + pthread_mutex_unlock(&stats_out_lock); + thr_stats_off = off; +} + +/* From the SIGINT handler: the interrupted thread may hold a stats lock, so + * print only if both can be taken right away. */ +static void print_stats_from_signal(void) +{ + bool off = thr_stats_off; + + if (!collect_stats) + return; + thr_stats_off = true; + if (pthread_mutex_trylock(&stats_out_lock) == 0) { + if (pthread_mutex_trylock(&stats_lock) == 0) { + stats_collect(&stats_scratch, true); + pthread_mutex_unlock(&stats_lock); + stats_print_summary(&stats_scratch); + } + pthread_mutex_unlock(&stats_out_lock); + } + thr_stats_off = off; +} + +/* ---- Interval sampler (DTO_STATS_INTERVAL_MS) ---- */ + +static pthread_t sampler_thread; +static pthread_mutex_t sampler_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t sampler_cv; +static bool sampler_stop; + +static void *stats_sampler(void *arg) +{ + struct timespec next; + + thr_stats_off = true; /* its own formatting is not the program's */ + clock_gettime(CLOCK_MONOTONIC, &next); + pthread_mutex_lock(&sampler_lock); + while (!sampler_stop) { + next.tv_nsec += (long)(stats_interval_ms % 1000) * 1000000; + next.tv_sec += stats_interval_ms / 1000 + next.tv_nsec / 1000000000; + next.tv_nsec %= 1000000000; + while (!sampler_stop && + pthread_cond_timedwait(&sampler_cv, &sampler_lock, &next) != ETIMEDOUT) + ; + if (sampler_stop) + break; + pthread_mutex_unlock(&sampler_lock); + pthread_mutex_lock(&stats_out_lock); + stats_emit_interval(); + pthread_mutex_unlock(&stats_out_lock); + pthread_mutex_lock(&sampler_lock); + } + pthread_mutex_unlock(&sampler_lock); + return NULL; +} + +/* Started by the first counted call (not from the constructor or the fork + * handler, where creating a thread is not safe), at most once per process. */ +static void stats_start_sampler(void) +{ + unsigned char expected = 0; + pthread_condattr_t attr; + sigset_t all, old; + + if (!atomic_compare_exchange_strong(&sampler_started, &expected, 1)) + return; + pthread_condattr_init(&attr); + pthread_condattr_setclock(&attr, CLOCK_MONOTONIC); + pthread_cond_init(&sampler_cv, &attr); + pthread_condattr_destroy(&attr); + sampler_stop = false; + /* the sampler must not take the program's signals */ + sigfillset(&all); + pthread_sigmask(SIG_SETMASK, &all, &old); + if (pthread_create(&sampler_thread, NULL, stats_sampler, NULL)) + atomic_store(&sampler_started, 0); + pthread_sigmask(SIG_SETMASK, &old, NULL); +} + +static void stats_stop_sampler(void) +{ + if (!atomic_load(&sampler_started)) + return; + pthread_mutex_lock(&sampler_lock); + sampler_stop = true; + pthread_cond_signal(&sampler_cv); + pthread_mutex_unlock(&sampler_lock); + pthread_join(sampler_thread, NULL); + atomic_store(&sampler_started, 0); + /* the part of the last interval since the sampler's final record */ + pthread_mutex_lock(&stats_out_lock); + stats_emit_interval(); + pthread_mutex_unlock(&stats_out_lock); +} + +/* ---- fork ---- + * + * The forking thread holds both stats locks across fork() so the child never + * inherits one held by a thread that does not exist there. */ +static void stats_fork_prepare(void) +{ + pthread_mutex_lock(&stats_out_lock); + pthread_mutex_lock(&stats_lock); +} + +static void stats_fork_parent(void) +{ + pthread_mutex_unlock(&stats_lock); + pthread_mutex_unlock(&stats_out_lock); +} + +/* In the child: drop every block (their threads are gone, and the child + * counts from zero) and forget the parent's sampler. */ +static void stats_fork_child(void) +{ + struct dto_thread_stats *t = stats_threads, *next; + + pthread_mutex_unlock(&stats_lock); + pthread_mutex_unlock(&stats_out_lock); + for (; t; t = next) { + next = t->next; + munmap(t, sizeof(*t)); + } + stats_threads = NULL; + orig_memset(&stats_retired, 0, sizeof(stats_retired)); + orig_memset(&sampler_prev, 0, sizeof(sampler_prev)); + sampler_prev_ns = 0; + pthread_setspecific(stats_key, NULL); + thr_stats = NULL; + thr_stats_off = false; + atomic_store(&sampler_started, 0); + stats_out_len = 0; + /* with %p the child writes a file of its own: init_dto opens it */ + if (stats_file_per_pid && stats_fd != -1) { + close(stats_fd); + stats_fd = -1; + } +} #endif #define DTO_MAX_PARAM_LEN 16 @@ -1454,28 +2016,43 @@ static int init_dto(void) } if (collect_stats) { + env_str = getenv("DTO_STATS_FORMAT"); + if (env_str != NULL && !strcmp(env_str, "json")) + stats_format = STATS_FORMAT_JSON; + else if (env_str != NULL && strcmp(env_str, "text")) + LOG_ERROR("Unknown DTO_STATS_FORMAT '%s', using text\n", env_str); + + env_str = getenv("DTO_STATS_INTERVAL_MS"); + if (env_str != NULL) { + errno = 0; + stats_interval_ms = strtoul(env_str, NULL, 10); + if (errno) + stats_interval_ms = 0; + } + + /* A forked child keeps the parent's (O_APPEND) file, + * unless the name has a %p (see stats_fork_child). */ env_str = getenv("DTO_STATS_FILE"); - if (env_str != NULL) { - struct stat st; - - if (lstat(env_str, &st) == -1 || - (st.st_mode & S_IFMT) == S_IFREG) { - stats_fd = open(env_str, O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0600); - if (stats_fd == -1) - LOG_ERROR("Failed to open DTO_STATS_FILE '%s': %s\n", env_str, strerror(errno)); - } else { - LOG_ERROR("DTO_STATS_FILE '%s' must reference a regular file\n", env_str); - } - } - - clock_gettime(CLOCK_BOOTTIME, &dto_start_time); - /* Change the log level to 'trace' so that the - * stats can be logged - */ - log_level = LOG_LEVEL_TRACE; - dto_stats_log("Stats collection enabled, some latencies may be higher due to collection overhead.\n"); - install_sigint_handler(); - } + if (env_str != NULL && stats_fd == -1) + stats_open_file(env_str); + + if (!stats_key_created && + pthread_key_create(&stats_key, stats_thread_exit) == 0) + stats_key_created = true; + if (!stats_key_created) { + LOG_ERROR("Stats collection disabled: no thread key\n"); + collect_stats = 0; + } + } + + if (collect_stats) { + clock_gettime(CLOCK_BOOTTIME, &dto_start_time); + if (stats_format == STATS_FORMAT_TEXT) { + stats_printf("Stats collection enabled, some latencies may be higher due to collection overhead.\n"); + stats_flush(); + } + install_sigint_handler(); + } #endif /* Register fork handler for the child process */ @@ -1485,7 +2062,11 @@ static int init_dto(void) * take that risk and disable DSA offload for parent * as well. */ +#ifdef DTO_STATS_SUPPORT + if (!pthread_atfork(stats_fork_prepare, stats_fork_parent, child)) +#else if (!pthread_atfork(NULL, NULL, child)) +#endif fork_handler_registered = 1; else { LOG_ERROR("Setting fork() handler failed. " @@ -1634,7 +2215,10 @@ static void cleanup_dto(void) close(wqs[i].wq_fd); } #ifdef DTO_STATS_SUPPORT - print_stats(); + if (collect_stats) { + stats_stop_sampler(); + print_stats(); + } restore_sigint_handler(); #endif if (log_fd != -1) diff --git a/tests/test_stats.py b/tests/test_stats.py new file mode 100644 index 0000000..fff0449 --- /dev/null +++ b/tests/test_stats.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +# +# Checks DTO_COLLECT_STATS output against test_stats_workload's known calls. +# Needs no DSA hardware: without it every call is counted on the CPU path. +# +# Usage: test_stats.py + +import json +import os +import subprocess +import sys +import tempfile + +WORKLOAD = sys.argv[1] +SIZES = {"memcpy": 131072 + 123, "memset": 65536 + 4096, + "memmove": 98304 + 7, "memcmp": 262144 + 64} +SIZE_CHILD = 524288 + 1 +failures = 0 + + +def check(desc, ok, detail=""): + global failures + print(("PASS: " if ok else "FAIL: ") + desc + ("" if ok else " " + detail)) + failures += not ok + + +def run(args, env_extra, stats_path): + env = dict(os.environ, DTO_COLLECT_STATS="1", DTO_STATS_FILE=stats_path, + **env_extra) + out = subprocess.run([WORKLOAD] + args, env=env, check=True, + capture_output=True, text=True).stdout + with open(stats_path) as f: + return out, f.read() + + +def records(text): + return [json.loads(line) for line in text.splitlines() if line.strip()] + + +def op_sum(ops, op, field): + return sum(g[field] for g in ops.get(op, {}).values()) + + +def bucket_for(summary, size): + for b in summary["histogram"]: + if b["min"] <= size and ("max" not in b or size <= b["max"]): + return b + return None + + +tmp = tempfile.mkdtemp() + +# --- totals and histogram, JSON --- +threads, iters = 4, 50 +_, text = run(["basic", str(threads), str(iters)], {"DTO_STATS_FORMAT": "json"}, + os.path.join(tmp, "basic.json")) +recs = records(text) +summaries = [r for r in recs if r["type"] == "summary"] +check("one summary record", len(summaries) == 1, str(len(summaries))) +s = summaries[0] +n = threads * iters +for op, size in SIZES.items(): + b = bucket_for(s, size) + check(f"{op} lands in the bucket for {size} bytes", b is not None and op in b["ops"], + json.dumps(b)) + if b is None or op not in b["ops"]: + continue + calls = op_sum(b["ops"], op, "calls") + failed = b["ops"][op].get("dsa_failed", {}).get("calls", 0) + # a call the device left partly done is counted as dsa_failed plus cpu + check(f"{op} calls ({n})", calls - failed == n, f"calls {calls} failed {failed}") + check(f"{op} bytes ({n * size})", op_sum(b["ops"], op, "bytes") == n * size, + str(op_sum(b["ops"], op, "bytes"))) + check(f"{op} time recorded", op_sum(b["ops"], op, "ns") > 0) +check("summary carries the configuration", "wait_method" in s["config"]) + +# --- interval records add up to the totals --- +_, text = run(["basic", "2", "40", "2"], + {"DTO_STATS_FORMAT": "json", "DTO_STATS_INTERVAL_MS": "10"}, + os.path.join(tmp, "interval.json")) +recs = records(text) +intervals = [r for r in recs if r["type"] == "interval"] +s = [r for r in recs if r["type"] == "summary"][0] +check("interval records emitted", len(intervals) >= 5, str(len(intervals))) +check("interval times increase", + all(b["t_ms"] > a["t_ms"] for a, b in zip(intervals, intervals[1:]))) +for op in SIZES: + for field in ("calls", "bytes"): + total = sum(op_sum(r["ops"], op, field) for r in intervals) + check(f"intervals sum to the {op} {field} total", + total == op_sum(s["totals"], op, field), + f"{total} vs {op_sum(s['totals'], op, field)}") + +# --- short-lived threads: all counted, blocks released --- +out, text = run(["churn", "200", "5"], {"DTO_STATS_FORMAT": "json"}, + os.path.join(tmp, "churn.json")) +s = [r for r in records(text) if r["type"] == "summary"][0] +b = bucket_for(s, SIZES["memcpy"]) +calls = op_sum(b["ops"], "memcpy", "calls") - \ + b["ops"]["memcpy"].get("dsa_failed", {}).get("calls", 0) +check("calls from 200 exited threads counted", calls == 200 * 5, str(calls)) +before, after = (int(x) for x in out.split()[1::2]) +check("exited threads' stats blocks are unmapped", after - before < 20, + f"mappings {before} -> {after}") + +# --- fork: separate records per process, appended to one file --- +out, text = run(["fork", "30"], {"DTO_STATS_FORMAT": "json"}, + os.path.join(tmp, "fork.json")) +child = int(out.split()[1]) +by_pid = {r["pid"]: r for r in records(text) if r["type"] == "summary"} +check("parent and child both report, in one file", len(by_pid) == 2, str(list(by_pid))) +if child in by_pid and len(by_pid) == 2: + parent = [p for p in by_pid if p != child][0] + cb = bucket_for(by_pid[child], SIZE_CHILD) + check("child counts only its own copies", + cb is not None and bucket_for(by_pid[child], SIZES["memcpy"]) is None) + pb = bucket_for(by_pid[parent], SIZES["memcpy"]) + check("parent counts its copies from before and after the fork", + pb is not None and op_sum(pb["ops"], "memcpy", "bytes") == 60 * SIZES["memcpy"]) + +# --- %p: a file per process --- +pattern = os.path.join(tmp, "pp.%p.json") +env = dict(os.environ, DTO_COLLECT_STATS="1", DTO_STATS_FILE=pattern, + DTO_STATS_FORMAT="json") +subprocess.run([WORKLOAD, "fork", "5"], env=env, check=True, capture_output=True) +files = [f for f in os.listdir(tmp) if f.startswith("pp.")] +check("%p gives each process its own file", len(files) == 2, str(files)) + +# --- text format --- +_, text = run(["basic", "1", "10"], {"DTO_STATS_FORMAT": "text"}, + os.path.join(tmp, "basic.txt")) +check("text summary has the operation table", "Number of Memory Operations" in text + and "Byte Range" in text) +check("text summary lists the memcpy bucket", " 131072-163839" in text) + +print() +if failures: + print(f"{failures} stats test(s) failed") + sys.exit(1) +print("All stats tests passed") diff --git a/tests/test_stats_workload.c b/tests/test_stats_workload.c new file mode 100644 index 0000000..d6f257a --- /dev/null +++ b/tests/test_stats_workload.c @@ -0,0 +1,132 @@ +/******************************************************************************* + * Copyright (C) 2026 Intel Corporation + * + * SPDX-License-Identifier: MIT + * + * test_stats_workload.c - known memory operations for tests/test_stats.py, + * which runs this with DTO_COLLECT_STATS set and checks what DTO reports. + * + * basic [sleep_ms] each thread, per iteration: memcpy, + * memset, memmove, memcmp of + * SIZE_{CPY,SET,MOV,CMP} bytes + * churn the same, from short-lived threads + * started one after another; prints + * the mapping count before and after + * fork memcpy SIZE_CPY in the parent and + * SIZE_CHILD in a forked child + * + * Built with -fno-builtin so every call reaches DTO. + ******************************************************************************/ + +#include +#include +#include +#include +#include +#include +#include +#include + +/* distinctive sizes, each in its own histogram bucket */ +#define SIZE_CPY (131072 + 123) +#define SIZE_SET (65536 + 4096) +#define SIZE_MOV (98304 + 7) +#define SIZE_CMP (262144 + 64) +#define SIZE_CHILD (524288 + 1) +#define BUF (1 << 20) + +static int iters, sleep_ms; + +static void *work(void *arg) +{ + char *a = mmap(NULL, BUF, PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0); + char *b = mmap(NULL, BUF, PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0); + struct timespec pause = { 0, sleep_ms * 1000000L }; + + for (int i = 0; i < iters; i++) { + memcpy(b, a, SIZE_CPY); + memset(a, i, SIZE_SET); + memmove(b + 64, b, SIZE_MOV); + if (memcmp(a, a + 0, SIZE_CMP)) + abort(); + if (sleep_ms) + nanosleep(&pause, NULL); + } + munmap(a, BUF); + munmap(b, BUF); + return NULL; +} + +static int count_maps(void) +{ + char line[512]; + int n = 0; + FILE *f = fopen("/proc/self/maps", "r"); + + while (f && fgets(line, sizeof(line), f)) + n++; + if (f) + fclose(f); + return n; +} + +static void copies(size_t size, int n) +{ + char *a = mmap(NULL, BUF, PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0); + char *b = mmap(NULL, BUF, PROT_READ | PROT_WRITE, + MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0); + + for (int i = 0; i < n; i++) + memcpy(b, a, size); + munmap(a, BUF); + munmap(b, BUF); +} + +int main(int argc, char **argv) +{ + pthread_t t[256]; + + if (argc < 3) + return 2; + if (!strcmp(argv[1], "basic")) { + int nt = atoi(argv[2]); + + iters = atoi(argv[3]); + sleep_ms = argc > 4 ? atoi(argv[4]) : 0; + for (int i = 0; i < nt; i++) + pthread_create(&t[i], NULL, work, NULL); + for (int i = 0; i < nt; i++) + pthread_join(t[i], NULL); + } else if (!strcmp(argv[1], "churn")) { + int nt = atoi(argv[2]), before, after; + + iters = atoi(argv[3]); + /* the first thread's block and allocations, so they don't count */ + pthread_create(&t[0], NULL, work, NULL); + pthread_join(t[0], NULL); + before = count_maps(); + for (int i = 1; i < nt; i++) { + pthread_create(&t[0], NULL, work, NULL); + pthread_join(t[0], NULL); + } + after = count_maps(); + printf("maps_before %d maps_after %d\n", before, after); + } else if (!strcmp(argv[1], "fork")) { + int n = atoi(argv[2]), st; + pid_t pid; + + copies(SIZE_CPY, n); + pid = fork(); + if (pid == 0) { + copies(SIZE_CHILD, n); + exit(0); + } + waitpid(pid, &st, 0); + copies(SIZE_CPY, n); + printf("child %d\n", pid); + } + return 0; +} From fddac82cc72aefc9768b23e197bbbd9651486775 Mon Sep 17 00:00:00 2001 From: Daniel Byrne Date: Wed, 21 Oct 2026 11:29:57 -0700 Subject: [PATCH 2/3] stats: link libdto with pthread The stats rework uses thread keys, a sampler thread and its condition variable. Before glibc 2.34 those functions are in libpthread, not libc, so preloading libdto into a program that doesn't link libpthread itself (sh, true, most single-threaded tools) failed at load time: true: symbol lookup error: libdto.so: undefined symbol: pthread_key_create Link libdto with pthread in CMake and the Makefile (an empty stub on newer glibc), and check in the stats test that stats work in a program without libpthread. Seen on RHEL 8 (glibc 2.28); Ubuntu 24.04 (2.39) is not affected. --- CMakeLists.txt | 7 +++++-- Makefile | 2 +- tests/test_stats.py | 9 ++++++++- 3 files changed, 14 insertions(+), 4 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 4419a70..a383cfe 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -65,7 +65,10 @@ target_compile_definitions(dto PRIVATE DTO_STATS_SUPPORT) # Link libraries set(DTO_STATIC_LIBS "") -set(DTO_DYNAMIC_LIBS dl) +# pthread: stats use thread keys and a sampler thread; before glibc 2.34 these +# are not in libc, so a program that doesn't link libpthread itself would fail +# to load libdto. +set(DTO_DYNAMIC_LIBS dl pthread) # Handle accel-config support if(DTO_ACCEL_CONFIG_SUPPORT) @@ -230,7 +233,7 @@ if(DTO_BUILD_TESTS) if(Python3_Interpreter_FOUND) add_test(NAME stats COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/tests/test_stats.py - $) + $ $) set_tests_properties(stats PROPERTIES LABELS "functional" TIMEOUT 120 diff --git a/Makefile b/Makefile index 25b649a..9c809be 100644 --- a/Makefile +++ b/Makefile @@ -16,7 +16,7 @@ MANDIR ?= /usr/share/man DTO_VERSION := $(shell grep -o 'DTO VERSION [0-9.]*' CMakeLists.txt | cut -d' ' -f3) libdto: dto.c - gcc -shared -fPIC -Wl,-soname,libdto.so dto.c $(DML_LIB_CXX) -DDTO_STATS_SUPPORT -DDTO_ACCEL_CONFIG_SUPPORT -DDTO_NUMA_SUPPORT -o libdto.so.1.0 -laccel-config -ldl -lnuma -mwaitpkg + gcc -shared -fPIC -Wl,-soname,libdto.so dto.c $(DML_LIB_CXX) -DDTO_STATS_SUPPORT -DDTO_ACCEL_CONFIG_SUPPORT -DDTO_NUMA_SUPPORT -o libdto.so.1.0 -laccel-config -ldl -lnuma -lpthread -mwaitpkg libdto_nostats: dto.c gcc -shared -fPIC -Wl,-soname,libdto.so dto.c $(DML_LIB_CXX) -DDTO_ACCEL_CONFIG_SUPPORT -DDTO_NUMA_SUPPORT -o libdto.so.1.0 -laccel-config -ldl -lnuma -mwaitpkg diff --git a/tests/test_stats.py b/tests/test_stats.py index fff0449..0043159 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -4,7 +4,7 @@ # Checks DTO_COLLECT_STATS output against test_stats_workload's known calls. # Needs no DSA hardware: without it every call is counted on the CPU path. # -# Usage: test_stats.py +# Usage: test_stats.py import json import os @@ -13,6 +13,7 @@ import tempfile WORKLOAD = sys.argv[1] +LIBDTO = os.path.abspath(sys.argv[2]) SIZES = {"memcpy": 131072 + 123, "memset": 65536 + 4096, "memmove": 98304 + 7, "memcmp": 262144 + 64} SIZE_CHILD = 524288 + 1 @@ -127,6 +128,12 @@ def bucket_for(summary, size): files = [f for f in os.listdir(tmp) if f.startswith("pp.")] check("%p gives each process its own file", len(files) == 2, str(files)) +# --- a program that doesn't link libpthread (before glibc 2.34, it isn't libc) --- +env = dict(os.environ, LD_PRELOAD=LIBDTO, DTO_COLLECT_STATS="1", + DTO_STATS_INTERVAL_MS="1", DTO_STATS_FILE=os.path.join(tmp, "true.txt")) +p = subprocess.run(["true"], env=env, capture_output=True, text=True) +check("stats work in a program without libpthread", p.returncode == 0, p.stderr.strip()) + # --- text format --- _, text = run(["basic", "1", "10"], {"DTO_STATS_FORMAT": "text"}, os.path.join(tmp, "basic.txt")) From c1ffec8944c9f412455cda240b49fc7f9d470e36 Mon Sep 17 00:00:00 2001 From: Daniel Byrne Date: Wed, 28 Oct 2026 09:24:47 -0700 Subject: [PATCH 3/3] stats: fix fork and exit safety of the stats machinery - stats_fork_child() called pthread_setspecific(stats_key, NULL) even when stats were off and the key had never been created, i.e. on key 0, which belongs to whichever library created it first. Only touch the key when it exists. - sampler_lock was not held across fork(), so a child could inherit it locked by the parent's sampler thread and hang in its first stats_start_sampler(). Take it in the prepare handler with the other two and release it in parent and child. - stats_stop_sampler() reset sampler_started, so a thread whose first counted call landed after cleanup_dto() restarted the sampler during process exit, writing to closed descriptors. Set stats_shutdown in cleanup_dto() and check it before starting the sampler. - stats_flush() gave up on EINTR and EAGAIN and discarded the rest of the record. Retry on EINTR and poll() on EAGAIN. - The 64 KiB output buffer was smaller than a worst-case JSON summary, so a record could go out in two write()s and interleave with another process's in a shared O_APPEND file. Make it 256 KiB. - DTO_STATS_INTERVAL_MS accepted negative, non-numeric and absurd values silently; reject them with an error like DTO_STATS_FORMAT does. --- dto.c | 36 +++++++++++++++++++++++++++++++----- 1 file changed, 31 insertions(+), 5 deletions(-) diff --git a/dto.c b/dto.c index 51899d0..8c7f4dd 100644 --- a/dto.c +++ b/dto.c @@ -10,6 +10,7 @@ #include #include #include +#include #include #include #include @@ -318,6 +319,7 @@ static bool stats_file_per_pid; /* DTO_STATS_FILE has a %p */ /* DTO_STATS_INTERVAL_MS */ static unsigned long stats_interval_ms; static atomic_uchar sampler_started; +static bool stats_shutdown; /* cleanup_dto ran: no new sampler */ static struct dto_stats sampler_prev; /* totals at the previous record */ static uint64_t sampler_prev_ns; @@ -831,7 +833,8 @@ static struct dto_thread_stats *stats_thread_register(void) pthread_setspecific(stats_key, ts); thr_stats = ts; - if (stats_interval_ms && !atomic_load_explicit(&sampler_started, + if (stats_interval_ms && !stats_shutdown && + !atomic_load_explicit(&sampler_started, memory_order_acquire)) stats_start_sampler(); thr_stats_off = false; @@ -872,7 +875,10 @@ static void update_stats(int op, size_t n, bool overlapping, size_t bytes_comple * possible: to DTO_STATS_FILE (opened O_APPEND, so records from several * processes sharing it stay whole), else DTO_LOG_FILE, else stdout. */ -static char stats_out[1 << 16]; +/* Holds a whole record, so a summary (up to ~1 KB for each of the 106 + * histogram buckets in JSON) goes out in one write() and records from + * several processes sharing an O_APPEND file stay whole. */ +static char stats_out[1 << 18]; static size_t stats_out_len; static void stats_flush(void) @@ -883,6 +889,14 @@ static void stats_flush(void) while (off < stats_out_len) { ssize_t r = write(fd, stats_out + off, stats_out_len - off); + if (r < 0 && errno == EINTR) + continue; + if (r < 0 && errno == EAGAIN) { + struct pollfd pfd = { .fd = fd, .events = POLLOUT }; + + if (poll(&pfd, 1, -1) > 0 || errno == EINTR) + continue; + } if (r <= 0) break; off += r; @@ -1309,10 +1323,12 @@ static void stats_fork_prepare(void) { pthread_mutex_lock(&stats_out_lock); pthread_mutex_lock(&stats_lock); + pthread_mutex_lock(&sampler_lock); } static void stats_fork_parent(void) { + pthread_mutex_unlock(&sampler_lock); pthread_mutex_unlock(&stats_lock); pthread_mutex_unlock(&stats_out_lock); } @@ -1323,6 +1339,7 @@ static void stats_fork_child(void) { struct dto_thread_stats *t = stats_threads, *next; + pthread_mutex_unlock(&sampler_lock); pthread_mutex_unlock(&stats_lock); pthread_mutex_unlock(&stats_out_lock); for (; t; t = next) { @@ -1333,9 +1350,11 @@ static void stats_fork_child(void) orig_memset(&stats_retired, 0, sizeof(stats_retired)); orig_memset(&sampler_prev, 0, sizeof(sampler_prev)); sampler_prev_ns = 0; - pthread_setspecific(stats_key, NULL); + if (stats_key_created) + pthread_setspecific(stats_key, NULL); thr_stats = NULL; thr_stats_off = false; + sampler_stop = false; atomic_store(&sampler_started, 0); stats_out_len = 0; /* with %p the child writes a file of its own: init_dto opens it */ @@ -2024,10 +2043,16 @@ static int init_dto(void) env_str = getenv("DTO_STATS_INTERVAL_MS"); if (env_str != NULL) { + char *end; + errno = 0; - stats_interval_ms = strtoul(env_str, NULL, 10); - if (errno) + stats_interval_ms = strtoul(env_str, &end, 10); + /* a day: beyond that it is a typo, and the timespec math stays exact */ + if (errno || end == env_str || *end || *env_str == '-' || + stats_interval_ms > 86400000UL) { + LOG_ERROR("Invalid DTO_STATS_INTERVAL_MS '%s', no interval records\n", env_str); stats_interval_ms = 0; + } } /* A forked child keeps the parent's (O_APPEND) file, @@ -2216,6 +2241,7 @@ static void cleanup_dto(void) } #ifdef DTO_STATS_SUPPORT if (collect_stats) { + stats_shutdown = true; stats_stop_sampler(); print_stats(); }