diff --git a/CMakeLists.txt b/CMakeLists.txt index b7a9139..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) @@ -221,6 +224,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/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/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..8c7f4dd 100644 --- a/dto.c +++ b/dto.c @@ -4,11 +4,13 @@ * SPDX-License-Identifier: MIT ******************************************************************************/ +#include #include #include #include #include #include +#include #include #include #include @@ -187,9 +189,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 +274,58 @@ 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 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; + +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 +346,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 +370,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 +449,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,114 +690,677 @@ 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 && !stats_shutdown && + !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. */ + +/* 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) +{ + 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 && 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; + } + 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", +}; + +static const char * const group_json[] = { + [STDC_CALL] = "cpu", + [DSA_CALL_SUCCESS] = "dsa", + [DSA_CALL_FAILED] = "dsa_failed", +}; + +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("}"); +} - clock_gettime(CLOCK_BOOTTIME, &dto_end_time); +/* 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; - dto_stats_log("DTO Run Time: %ld ms\n", TS_NS(dto_start_time, dto_end_time)/1000000); + 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); + 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); +} + +/* 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(&sampler_lock); + 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; + 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 */ + if (stats_file_per_pid && stats_fd != -1) { + close(stats_fd); + stats_fd = -1; } } #endif @@ -1454,28 +2035,49 @@ 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) { + char *end; + + errno = 0; + 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, + * 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 +2087,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 +2240,11 @@ static void cleanup_dto(void) close(wqs[i].wq_fd); } #ifdef DTO_STATS_SUPPORT - print_stats(); + if (collect_stats) { + stats_shutdown = true; + 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..0043159 --- /dev/null +++ b/tests/test_stats.py @@ -0,0 +1,148 @@ +#!/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] +LIBDTO = os.path.abspath(sys.argv[2]) +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)) + +# --- 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")) +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; +}