diff options
Diffstat (limited to 'decompress')
| -rw-r--r-- | decompress/.gitignore | 2 | ||||
| -rw-r--r-- | decompress/Makefile | 25 | ||||
| -rwxr-xr-x | decompress/decompress | bin | 0 -> 59880 bytes | |||
| -rw-r--r-- | decompress/decompress.c | 705 | ||||
| -rwxr-xr-x | decompress/run.sh | 195 |
5 files changed, 927 insertions, 0 deletions
diff --git a/decompress/.gitignore b/decompress/.gitignore new file mode 100644 index 0000000..175da2f --- /dev/null +++ b/decompress/.gitignore @@ -0,0 +1,2 @@ +bins/ +logs/ diff --git a/decompress/Makefile b/decompress/Makefile new file mode 100644 index 0000000..f3d0d32 --- /dev/null +++ b/decompress/Makefile @@ -0,0 +1,25 @@ +CC ?= gcc +TARGET := decompress +SRC := decompress.c + +PKGS := doca-compress doca-common + +CFLAGS ?= -O2 -g -Wall -Wextra -Wno-deprecated-declarations -DALLOW_EXPERIMENTAL_API +CFLAGS += $(shell pkg-config --cflags $(PKGS) 2>/dev/null) + +LDLIBS += $(shell pkg-config --libs $(PKGS) 2>/dev/null) + +ifeq ($(strip $(LDLIBS)),) +CFLAGS += -I/opt/mellanox/doca/include -I/usr/include/libnl3 +LDLIBS += -L/opt/mellanox/doca/lib/aarch64-linux-gnu \ + -L/opt/mellanox/doca/lib64 \ + -ldoca_compress -ldoca_common +endif + +all: $(TARGET) + +$(TARGET): $(SRC) + $(CC) $(CFLAGS) $< -o $@ $(LDLIBS) + +clean: + rm -f $(TARGET) *.o decompressed.out diff --git a/decompress/decompress b/decompress/decompress Binary files differnew file mode 100755 index 0000000..703f5e7 --- /dev/null +++ b/decompress/decompress diff --git a/decompress/decompress.c b/decompress/decompress.c new file mode 100644 index 0000000..2517361 --- /dev/null +++ b/decompress/decompress.c @@ -0,0 +1,705 @@ +#define _GNU_SOURCE + +#include <dirent.h> +#include <errno.h> +#include <getopt.h> +#include <inttypes.h> +#include <limits.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/stat.h> +#include <time.h> +#include <unistd.h> + +#include <doca_buf.h> +#include <doca_buf_inventory.h> +#include <doca_compress.h> +#include <doca_ctx.h> +#include <doca_dev.h> +#include <doca_error.h> +#include <doca_mmap.h> +#include <doca_types.h> + +#ifndef DOCA_WORKQ_RETRIEVE_FLAGS_NONE +#define DOCA_WORKQ_RETRIEVE_FLAGS_NONE 0 +#endif + +#define DEFAULT_OUTPUT "decompressed.out" +#define NUM_BUFS 2 +#define WORKQ_DEPTH 8 + +struct one_result { + uint64_t compressed_bytes; + uint64_t decompressed_bytes; + double elapsed_us; + double bw_mibps; + uint32_t crc; + uint32_t adler; +}; + +static const char *errstr(doca_error_t err) +{ + return doca_get_error_string(err); +} + +static void usage(const char *prog) +{ + fprintf(stderr, + "Usage:\n" + " Single file:\n" + " %s -p <dpu-pci-bdf> -f <compressed-input> -s <dst-size> [-o output-file] [--no-write]\n" + "\n" + " Directory mode:\n" + " %s -p <dpu-pci-bdf> -d <chunk-directory> -S <dst-size-per-chunk>\n" + "\n" + "Examples:\n" + " sudo ./%s -p 03:00.0 -f 1M.bin.compressed -s 1048576 --no-write\n" + " sudo ./%s -p 03:00.0 -d bins/1G -S 1M\n", + prog, prog, prog, prog); +} + +static uint64_t now_nsec(void) +{ + struct timespec ts; + +#ifdef CLOCK_MONOTONIC_RAW + if (clock_gettime(CLOCK_MONOTONIC_RAW, &ts) != 0) { +#else + if (clock_gettime(CLOCK_MONOTONIC, &ts) != 0) { +#endif + perror("clock_gettime"); + exit(EXIT_FAILURE); + } + + return (uint64_t)ts.tv_sec * 1000000000ULL + (uint64_t)ts.tv_nsec; +} + +static void check_doca(doca_error_t err, const char *what) +{ + if (err != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: %s: %s\n", what, errstr(err)); + exit(EXIT_FAILURE); + } +} + +static size_t parse_size_or_die(const char *s) +{ + char *tmp; + char *end = NULL; + uint64_t v; + uint64_t mul = 1; + size_t len; + + len = strlen(s); + if (len == 0) { + fprintf(stderr, "ERROR: invalid size: %s\n", s); + exit(EXIT_FAILURE); + } + + tmp = strdup(s); + if (tmp == NULL) { + fprintf(stderr, "ERROR: strdup failed\n"); + exit(EXIT_FAILURE); + } + + char last = tmp[len - 1]; + + if (last == 'K' || last == 'k') { + mul = 1024ULL; + tmp[len - 1] = '\0'; + } else if (last == 'M' || last == 'm') { + mul = 1024ULL * 1024ULL; + tmp[len - 1] = '\0'; + } else if (last == 'G' || last == 'g') { + mul = 1024ULL * 1024ULL * 1024ULL; + tmp[len - 1] = '\0'; + } + + errno = 0; + v = strtoull(tmp, &end, 10); + + if (errno != 0 || end == tmp || *end != '\0' || v == 0) { + fprintf(stderr, "ERROR: invalid size: %s\n", s); + free(tmp); + exit(EXIT_FAILURE); + } + + free(tmp); + + if (v > SIZE_MAX / mul) { + fprintf(stderr, "ERROR: size overflow: %s\n", s); + exit(EXIT_FAILURE); + } + + return (size_t)(v * mul); +} + +static uint8_t *read_file_or_die(const char *path, size_t *len) +{ + FILE *f; + struct stat st; + uint8_t *buf; + size_t nread; + + if (stat(path, &st) != 0) { + fprintf(stderr, "ERROR: stat(%s): %s\n", path, strerror(errno)); + exit(EXIT_FAILURE); + } + + if (!S_ISREG(st.st_mode)) { + fprintf(stderr, "ERROR: not a regular file: %s\n", path); + exit(EXIT_FAILURE); + } + + if (st.st_size < 0) { + fprintf(stderr, "ERROR: invalid file size: %s\n", path); + exit(EXIT_FAILURE); + } + + *len = (size_t)st.st_size; + + buf = malloc(*len == 0 ? 1 : *len); + if (buf == NULL) { + fprintf(stderr, "ERROR: malloc input failed\n"); + exit(EXIT_FAILURE); + } + + f = fopen(path, "rb"); + if (f == NULL) { + fprintf(stderr, "ERROR: fopen(%s): %s\n", path, strerror(errno)); + free(buf); + exit(EXIT_FAILURE); + } + + nread = fread(buf, 1, *len, f); + if (nread != *len) { + fprintf(stderr, "ERROR: fread(%s): expected %zu, got %zu\n", + path, *len, nread); + fclose(f); + free(buf); + exit(EXIT_FAILURE); + } + + fclose(f); + return buf; +} + +static void write_file_or_die(const char *path, const void *buf, size_t len) +{ + FILE *f; + size_t nwritten; + + f = fopen(path, "wb"); + if (f == NULL) { + fprintf(stderr, "ERROR: fopen(%s): %s\n", path, strerror(errno)); + exit(EXIT_FAILURE); + } + + nwritten = fwrite(buf, 1, len, f); + if (nwritten != len) { + fprintf(stderr, "ERROR: fwrite(%s): expected %zu, wrote %zu\n", + path, len, nwritten); + fclose(f); + exit(EXIT_FAILURE); + } + + fclose(f); +} + +static int open_first_doca_device(const char *pci_addr, struct doca_dev **dev) +{ + struct doca_devinfo **dev_list = NULL; + uint32_t nb_devs = 0; + doca_error_t result; + + /* + * DOCA 2.0.2027: + * doca_devinfo_list_create() + * doca_devinfo_list_destroy() + * + * Keep -p for CLI compatibility, but open the first available local DOCA device. + */ + (void)pci_addr; + + result = doca_devinfo_list_create(&dev_list, &nb_devs); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_devinfo_list_create: %s\n", errstr(result)); + return -1; + } + + if (nb_devs == 0) { + fprintf(stderr, "ERROR: no DOCA devices found\n"); + doca_devinfo_list_destroy(dev_list); + return -1; + } + + result = doca_dev_open(dev_list[0], dev); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_dev_open: %s\n", errstr(result)); + doca_devinfo_list_destroy(dev_list); + return -1; + } + + doca_devinfo_list_destroy(dev_list); + return 0; +} + +static struct doca_mmap *create_mmap_or_die(struct doca_dev *dev, + void *addr, + size_t len, + uint32_t permissions) +{ + struct doca_mmap *mmap = NULL; + union doca_data user_data; + doca_error_t result; + + memset(&user_data, 0, sizeof(user_data)); + + result = doca_mmap_create(&user_data, &mmap); + check_doca(result, "doca_mmap_create"); + + result = doca_mmap_set_memrange(mmap, addr, len); + check_doca(result, "doca_mmap_set_memrange"); + + result = doca_mmap_set_permissions(mmap, permissions); + check_doca(result, "doca_mmap_set_permissions"); + + result = doca_mmap_dev_add(mmap, dev); + check_doca(result, "doca_mmap_dev_add"); + + result = doca_mmap_start(mmap); + check_doca(result, "doca_mmap_start"); + + return mmap; +} + +static void join_path(const char *dir, const char *file, char *out, size_t out_len) +{ + if (file[0] == '/') + snprintf(out, out_len, "%s", file); + else + snprintf(out, out_len, "%s/%s", dir, file); +} + +static int should_skip_dirent(const struct dirent *ent) +{ + if (strcmp(ent->d_name, ".") == 0 || strcmp(ent->d_name, "..") == 0) + return 1; + + if (strcmp(ent->d_name, "manifest.csv") == 0) + return 1; + + return 0; +} + +static int run_one_decompress(struct doca_ctx *ctx, + struct doca_dev *dev, + struct doca_workq *workq, + const char *input_path, + size_t dst_len, + int write_output, + const char *output_path, + struct one_result *out) +{ + uint8_t *src_mem = NULL; + uint8_t *dst_mem = NULL; + size_t src_len = 0; + size_t final_len = 0; + + struct doca_mmap *src_mmap = NULL; + struct doca_mmap *dst_mmap = NULL; + struct doca_buf_inventory *inventory = NULL; + struct doca_buf *src_buf = NULL; + struct doca_buf *dst_buf = NULL; + + union doca_data inventory_user_data; + struct doca_compress_deflate_job job; + struct doca_event event; + + uint64_t checksum = 0; + uint64_t start_ns; + uint64_t end_ns; + uint64_t elapsed_ns; + double elapsed_us; + double mibps; + + doca_error_t result; + int ret = -1; + uint16_t refcnt = 0; + + memset(out, 0, sizeof(*out)); + memset(&inventory_user_data, 0, sizeof(inventory_user_data)); + memset(&job, 0, sizeof(job)); + memset(&event, 0, sizeof(event)); + + src_mem = read_file_or_die(input_path, &src_len); + + dst_mem = calloc(1, dst_len); + if (dst_mem == NULL) { + fprintf(stderr, "ERROR: calloc destination failed\n"); + goto cleanup; + } + + src_mmap = create_mmap_or_die(dev, src_mem, src_len, DOCA_ACCESS_LOCAL_READ_ONLY); + dst_mmap = create_mmap_or_die(dev, dst_mem, dst_len, DOCA_ACCESS_LOCAL_READ_WRITE); + + result = doca_buf_inventory_create(&inventory_user_data, NUM_BUFS, 0, &inventory); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_buf_inventory_create: %s\n", errstr(result)); + goto cleanup; + } + + result = doca_buf_inventory_start(inventory); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_buf_inventory_start: %s\n", errstr(result)); + goto cleanup; + } + + result = doca_buf_inventory_buf_by_args( + inventory, + src_mmap, + src_mem, + src_len, + src_mem, + src_len, + &src_buf); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_buf_inventory_buf_by_args(src): %s\n", errstr(result)); + goto cleanup; + } + + result = doca_buf_inventory_buf_by_args( + inventory, + dst_mmap, + dst_mem, + dst_len, + dst_mem, + 0, + &dst_buf); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_buf_inventory_buf_by_args(dst): %s\n", errstr(result)); + goto cleanup; + } + + job.base.type = DOCA_DECOMPRESS_DEFLATE_JOB; + job.base.flags = DOCA_JOB_FLAGS_NONE; + job.base.ctx = ctx; + job.src_buff = src_buf; + job.dst_buff = dst_buf; + job.output_chksum = &checksum; + + /* + * Timing starts here. + * This excludes file read, malloc/calloc, mmap, inventory, buffer setup, and job construction. + * It includes only job submit + workq progress until completion. + */ + start_ns = now_nsec(); + + result = doca_workq_submit(workq, &job.base); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_workq_submit: %s\n", errstr(result)); + goto cleanup; + } + + do { + result = doca_workq_progress_retrieve(workq, &event, DOCA_WORKQ_RETRIEVE_FLAGS_NONE); + } while (result == DOCA_ERROR_AGAIN); + + end_ns = now_nsec(); + + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_workq_progress_retrieve: %s\n", errstr(result)); + goto cleanup; + } + + if ((doca_error_t)event.result.u64 != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: decompress job failed: %s\n", + errstr((doca_error_t)event.result.u64)); + goto cleanup; + } + + result = doca_buf_get_data_len(dst_buf, &final_len); + if (result != DOCA_SUCCESS) { + fprintf(stderr, "ERROR: doca_buf_get_data_len(dst): %s\n", errstr(result)); + goto cleanup; + } + + elapsed_ns = end_ns - start_ns; + if (elapsed_ns == 0) + elapsed_ns = 1; + + elapsed_us = (double)elapsed_ns / 1000.0; + + mibps = ((double)final_len / (1024.0 * 1024.0)) / + (elapsed_us / 1000000.0); + + if (write_output && output_path != NULL) + write_file_or_die(output_path, dst_mem, final_len); + + out->compressed_bytes = (uint64_t)src_len; + out->decompressed_bytes = (uint64_t)final_len; + out->elapsed_us = elapsed_us; + out->bw_mibps = mibps; + out->crc = (uint32_t)(checksum & 0xffffffffu); + out->adler = (uint32_t)((checksum >> 32) & 0xffffffffu); + + ret = 0; + +cleanup: + if (src_buf != NULL) + (void)doca_buf_refcount_rm(src_buf, &refcnt); + if (dst_buf != NULL) + (void)doca_buf_refcount_rm(dst_buf, &refcnt); + if (inventory != NULL) + (void)doca_buf_inventory_destroy(inventory); + if (src_mmap != NULL) + (void)doca_mmap_destroy(src_mmap); + if (dst_mmap != NULL) + (void)doca_mmap_destroy(dst_mmap); + + free(src_mem); + free(dst_mem); + + return ret; +} + +static int run_directory(struct doca_ctx *ctx, + struct doca_dev *dev, + struct doca_workq *workq, + const char *dir_path, + size_t dst_len) +{ + struct dirent **namelist = NULL; + int n; + int count = 0; + + uint64_t total_compressed_bytes = 0; + uint64_t total_decompressed_bytes = 0; + double total_elapsed_us = 0.0; + double total_bw_mibps; + + n = scandir(dir_path, &namelist, NULL, alphasort); + if (n < 0) { + fprintf(stderr, "ERROR: scandir(%s): %s\n", dir_path, strerror(errno)); + return -1; + } + + for (int i = 0; i < n; i++) { + char path[PATH_MAX]; + struct stat st; + struct one_result r; + + if (should_skip_dirent(namelist[i])) { + free(namelist[i]); + continue; + } + + join_path(dir_path, namelist[i]->d_name, path, sizeof(path)); + + if (stat(path, &st) != 0) { + fprintf(stderr, "ERROR: stat(%s): %s\n", path, strerror(errno)); + free(namelist[i]); + goto fail; + } + + if (!S_ISREG(st.st_mode)) { + free(namelist[i]); + continue; + } + + if (run_one_decompress(ctx, + dev, + workq, + path, + dst_len, + 0, + NULL, + &r) != 0) { + fprintf(stderr, "ERROR: failed on file %s\n", path); + free(namelist[i]); + goto fail; + } + + count++; + total_compressed_bytes += r.compressed_bytes; + total_decompressed_bytes += r.decompressed_bytes; + total_elapsed_us += r.elapsed_us; + + free(namelist[i]); + } + + free(namelist); + + if (count == 0) { + fprintf(stderr, "ERROR: no regular chunk files found in directory: %s\n", dir_path); + return -1; + } + + total_bw_mibps = ((double)total_decompressed_bytes / (1024.0 * 1024.0)) / + (total_elapsed_us / 1000000.0); + + printf("Decompress complete\n"); + printf(" chunks: %d\n", count); + printf(" compressed_bytes: %" PRIu64 "\n", total_compressed_bytes); + printf(" bytes: %" PRIu64 "\n", total_decompressed_bytes); + printf(" time: %.3f usec\n", total_elapsed_us); + printf(" BW: %.6f MiB/s\n", total_bw_mibps); + + return 0; + +fail: + for (int j = 0; j < n; j++) { + if (namelist[j] != NULL) + free(namelist[j]); + } + free(namelist); + return -1; +} + +int main(int argc, char **argv) +{ + const char *pci = NULL; + const char *input_path = NULL; + const char *output_path = DEFAULT_OUTPUT; + const char *dir_path = NULL; + + int write_output = 1; + int directory_mode = 0; + + size_t dst_len = 0; + + struct doca_dev *dev = NULL; + struct doca_compress *compress = NULL; + struct doca_ctx *ctx = NULL; + struct doca_workq *workq = NULL; + + union doca_data compress_user_data; + doca_error_t result; + + int opt; + int option_index = 0; + int ret = EXIT_FAILURE; + + static struct option long_options[] = { + {"no-write", no_argument, 0, 1000}, + {"help", no_argument, 0, 'h'}, + {0, 0, 0, 0} + }; + + memset(&compress_user_data, 0, sizeof(compress_user_data)); + + while ((opt = getopt_long(argc, argv, "p:f:o:s:d:S:h", long_options, &option_index)) != -1) { + switch (opt) { + case 'p': + pci = optarg; + break; + case 'f': + input_path = optarg; + break; + case 'o': + output_path = optarg; + write_output = 1; + break; + case 's': + dst_len = parse_size_or_die(optarg); + break; + case 'd': + dir_path = optarg; + directory_mode = 1; + write_output = 0; + break; + case 'S': + dst_len = parse_size_or_die(optarg); + break; + case 1000: + write_output = 0; + break; + case 'h': + default: + usage(argv[0]); + return EXIT_FAILURE; + } + } + + if (pci == NULL) { + usage(argv[0]); + return EXIT_FAILURE; + } + + if (!directory_mode && (input_path == NULL || dst_len == 0)) { + usage(argv[0]); + return EXIT_FAILURE; + } + + if (directory_mode && (dir_path == NULL || dst_len == 0)) { + usage(argv[0]); + return EXIT_FAILURE; + } + + if (open_first_doca_device(pci, &dev) != 0) + goto cleanup; + + result = doca_compress_create(&compress); + check_doca(result, "doca_compress_create"); + + ctx = doca_compress_as_ctx(compress); + if (ctx == NULL) { + fprintf(stderr, "ERROR: doca_compress_as_ctx returned NULL\n"); + goto cleanup; + } + + result = doca_ctx_dev_add(ctx, dev); + check_doca(result, "doca_ctx_dev_add"); + + result = doca_ctx_start(ctx); + check_doca(result, "doca_ctx_start"); + + result = doca_workq_create(WORKQ_DEPTH, &workq); + check_doca(result, "doca_workq_create"); + + result = doca_ctx_workq_add(ctx, workq); + check_doca(result, "doca_ctx_workq_add"); + + if (directory_mode) { + if (run_directory(ctx, dev, workq, dir_path, dst_len) != 0) + goto cleanup; + } else { + struct one_result r; + + if (run_one_decompress(ctx, + dev, + workq, + input_path, + dst_len, + write_output, + output_path, + &r) != 0) + goto cleanup; + + printf("Decompress complete\n"); + printf(" compressed_bytes: %" PRIu64 "\n", r.compressed_bytes); + printf(" bytes: %" PRIu64 "\n", r.decompressed_bytes); + printf(" time: %.3f usec\n", r.elapsed_us); + printf(" BW: %.6f MiB/s\n", r.bw_mibps); + printf(" crc: 0x%08" PRIx32 "\n", r.crc); + printf(" adler: 0x%08" PRIx32 "\n", r.adler); + if (write_output) + printf(" output: %s\n", output_path); + } + + ret = EXIT_SUCCESS; + +cleanup: + if (workq != NULL) + (void)doca_workq_destroy(workq); + if (ctx != NULL) + (void)doca_ctx_stop(ctx); + if (compress != NULL) + (void)doca_compress_destroy(compress); + if (dev != NULL) + (void)doca_dev_close(dev); + + return ret; +} diff --git a/decompress/run.sh b/decompress/run.sh new file mode 100755 index 0000000..4183c10 --- /dev/null +++ b/decompress/run.sh @@ -0,0 +1,195 @@ +#!/usr/bin/env bash + +set -euo pipefail + +usage() { + echo "Usage: $0 <tag> <dpu-pci-bdf> <chunk-directory> <chunk-size> [run-label]" + echo + echo "Examples:" + echo " $0 VANILLA_1G 03:00.0 bins/1G 1M VANILLA_1G_run_1" + echo " $0 TRACE_5G 03:00.0 bins/5G 1M TRACE_5G_run_1" +} + +if [[ $# -lt 4 || $# -gt 5 ]]; then + usage + exit 1 +fi + +TAG="$1" +DPU_PCI="$2" +CHUNK_DIR="$3" +CHUNK_SIZE="$4" +RUN_LABEL="${5:-${TAG}}" + +DECOMPRESS_BIN="./decompress" + +LOG_DIR="logs" +OUT_CSV="${LOG_DIR}/cpu_${TAG}.csv" + +mkdir -p "$LOG_DIR" + +TMP_DIR="$(mktemp -d)" +DECOMPRESS_LOG="${TMP_DIR}/decompress.log" +TIME_LOG="${TMP_DIR}/time.log" + +cleanup() { + rm -rf "$TMP_DIR" +} +trap cleanup EXIT INT TERM + +die() { + echo "ERROR: $*" >&2 + exit 1 +} + +read_proc_stat() { + # Outputs: idle total softirq + awk ' + /^cpu / { + user_v=$2 + nice_v=$3 + system_v=$4 + idle_v=$5 + iowait_v=$6 + irq_v=$7 + softirq_v=$8 + steal_v=$9 + + idle_all = idle_v + iowait_v + non_idle = user_v + nice_v + system_v + irq_v + softirq_v + steal_v + total = idle_all + non_idle + + printf "%.0f %.0f %.0f\n", idle_all, total, softirq_v + exit + } + ' /proc/stat +} + +calc_system_cpu_pct() { + local before_idle="$1" + local before_total="$2" + local after_idle="$3" + local after_total="$4" + + awk \ + -v bi="$before_idle" \ + -v bt="$before_total" \ + -v ai="$after_idle" \ + -v at="$after_total" \ + 'BEGIN { + total_delta = at - bt + idle_delta = ai - bi + + if (total_delta <= 0) { + printf "" + } else { + cpu = 100.0 * (total_delta - idle_delta) / total_delta + printf "%.6f", cpu + } + }' +} + +calc_softirq_cpu_pct() { + local before_total="$1" + local before_softirq="$2" + local after_total="$3" + local after_softirq="$4" + + awk \ + -v bt="$before_total" \ + -v bs="$before_softirq" \ + -v at="$after_total" \ + -v as="$after_softirq" \ + 'BEGIN { + total_delta = at - bt + softirq_delta = as - bs + + if (total_delta <= 0) { + printf "" + } else { + pct = 100.0 * softirq_delta / total_delta + printf "%.6f", pct + } + }' +} + +parse_time_cpu_pct() { + local file="$1" + + awk -F: ' + /Percent of CPU this job got/ { + val = $2 + gsub(/^[ \t]+|[ \t%]+$/, "", val) + print val + found = 1 + } + END { + if (!found) + print "" + } + ' "$file" 2>/dev/null | tail -n1 +} + +write_header_if_needed() { + if [[ ! -f "$OUT_CSV" ]]; then + echo "system_cpu_pct,process_cpu_pct,softirq_cpu_pct" > "$OUT_CSV" + fi +} + +[[ -x "$DECOMPRESS_BIN" ]] || die "$DECOMPRESS_BIN not found or not executable" +[[ -d "$CHUNK_DIR" ]] || die "chunk directory not found: $CHUNK_DIR" + +echo "[dpu] tag: $TAG" +echo "[dpu] run label: $RUN_LABEL" +echo "[dpu] chunk dir: $CHUNK_DIR" +echo "[dpu] chunk size: $CHUNK_SIZE" +echo "[dpu] pci: $DPU_PCI" +echo "[dpu] csv: $OUT_CSV" + +echo "[dpu] dropping filesystem caches..." +sudo -n sh -c 'sync; echo 3 > /proc/sys/vm/drop_caches' + +read before_idle before_total before_softirq < <(read_proc_stat) + +set +e +( + /usr/bin/time -v sudo -n "$DECOMPRESS_BIN" \ + -p "$DPU_PCI" \ + -d "$CHUNK_DIR" \ + -S "$CHUNK_SIZE" +) > "$DECOMPRESS_LOG" 2> "$TIME_LOG" +DECOMPRESS_STATUS=$? +set -e + +read after_idle after_total after_softirq < <(read_proc_stat) + +system_cpu_pct="$( + calc_system_cpu_pct "$before_idle" "$before_total" "$after_idle" "$after_total" +)" + +softirq_cpu_pct="$( + calc_softirq_cpu_pct "$before_total" "$before_softirq" "$after_total" "$after_softirq" +)" + +process_cpu_pct="$(parse_time_cpu_pct "$TIME_LOG")" + +write_header_if_needed + +printf "%s,%s,%s\n" \ + "${system_cpu_pct:-}" \ + "${process_cpu_pct:-}" \ + "${softirq_cpu_pct:-}" \ + >> "$OUT_CSV" + +# Keep these prints so the host script can parse chunks/compressed_bytes/bytes/time/BW. +cat "$DECOMPRESS_LOG" +cat "$TIME_LOG" + +if [[ "$DECOMPRESS_STATUS" -ne 0 ]]; then + echo "[dpu] decompressor failed with status $DECOMPRESS_STATUS" + echo "[dpu] appended CSV: $OUT_CSV" + exit "$DECOMPRESS_STATUS" +fi + +echo "[dpu] decompressor completed successfully" +echo "[dpu] appended CSV: $OUT_CSV" |