From 1493f1806dd8a27163f0ef6e1d3713a96cbfd7d8 Mon Sep 17 00:00:00 2001 From: TapTap Date: Wed, 16 Sep 2026 23:00:39 +0200 Subject: [PATCH] feat(parity): rsync-style per-file --progress and differential tests Replace the aggregate stderr progress with rsync 3.4.1's per-file progress block (name, 32 KiB first frame, final frame with (xfr#N, to-chk=X/Y)). Add differential tests against real rsync for --out-format %C/%b, the --progress frames, selected --stats lines and -n --delete lines. --- src/client/client_send.c | 182 +++++++++++++----------- tests/integration/test_features.py | 8 +- tests/integration/test_output_parity.py | 165 ++++++++++++++++++++- 3 files changed, 265 insertions(+), 90 deletions(-) diff --git a/src/client/client_send.c b/src/client/client_send.c index 090ce56..e676ee7 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -65,9 +65,6 @@ static void log_server_rejection(const char* context) { } } -/* Forward declaration for progress-reporting thread used in multithreaded send. */ -static int progress_thread_fn(void* arg); - static const char* display_bytes(unsigned long long bytes, bool human_readable, char* buffer, size_t buffer_size) { if (human_readable && format_human_size_decimal(bytes, buffer, buffer_size)) @@ -140,6 +137,93 @@ static void report_transfer_stats(const Config* config, int total_files, fflush(stdout); } +/* ---- rsync-style per-file --progress ------------------------------------ + * rsync prints, for each transferred regular file, the file name followed by a + * two-frame progress line: the first at the initial 32 KiB read window (always + * 0.00 kB/s / 0:00:00 on a sub-second transfer) and a final 100% frame carrying + * `(xfr#N, to-chk=X/Y)`. Rates are wall-clock dependent, so only the final + * rate is measured here; the layout matches rsync 3.4.1's progress.c. */ +#define RSYNC_PROGRESS_IO_WINDOW (32ULL * 1024ULL) + +static const char* delete_display_path(const Config* config, const char* path); + +static bool g_progress_active; +static unsigned long long g_progress_xferred; +static unsigned long long g_progress_seen; +static struct timespec g_progress_file_start; + +static void progress_first_frame(unsigned long long size, char* out, size_t out_size) { + char ofs_buf[32]; + unsigned long long ofs = size < RSYNC_PROGRESS_IO_WINDOW ? size : RSYNC_PROGRESS_IO_WINDOW; + if (!format_big_num(ofs, false, ofs_buf, sizeof(ofs_buf))) + snprintf(ofs_buf, sizeof(ofs_buf), "%llu", ofs); + int pct = size == 0 ? 100 : (ofs == size ? 100 : (int)(100.0 * (double)ofs / (double)size)); + snprintf(out, out_size, "\r%15s %3d%% %7.2f%s %s%s", ofs_buf, pct, 0.0, "kB/s", " 0:00:00", + " "); +} + +static void progress_final_frame(unsigned long long size, char* out, size_t out_size) { + char ofs_buf[32]; + char rembuf[32]; + unsigned long long last_ofs = size < RSYNC_PROGRESS_IO_WINDOW ? size : RSYNC_PROGRESS_IO_WINDOW; + if (!format_big_num(size, false, ofs_buf, sizeof(ofs_buf))) + snprintf(ofs_buf, sizeof(ofs_buf), "%llu", size); + struct timespec now; + clock_gettime(CLOCK_MONOTONIC, &now); + long long diff_ms = (long long)(now.tv_sec - g_progress_file_start.tv_sec) * 1000 + + (now.tv_nsec - g_progress_file_start.tv_nsec) / 1000000; + if (diff_ms <= 0) + diff_ms = 1; + double rate = size > last_ofs + ? (double)(size - last_ofs) * 1000.0 / (double)diff_ms / 1024.0 + : 0.0; + const char* units = "kB/s"; + if (rate > 1024.0 * 1024.0) { + rate /= 1024.0 * 1024.0; + units = "GB/s"; + } else if (rate > 1024.0) { + rate /= 1024.0; + units = "MB/s"; + } + unsigned long long remain = (unsigned long long)(diff_ms / 1000); + snprintf(rembuf, sizeof(rembuf), "%4u:%02u:%02u", (unsigned)(remain / 3600), + (unsigned)((remain / 60) % 60), (unsigned)(remain % 60)); + unsigned long long to_chk = g_progress_seen > g_progress_xferred ? g_progress_seen - g_progress_xferred : 0; + snprintf(out, out_size, "\r%15s %3d%% %7.2f%s %s (xfr#%llu, to-chk=%llu/%llu)\n", ofs_buf, 100, + rate, units, rembuf, g_progress_xferred, to_chk, g_progress_seen); +} + +static void client_progress_begin(const Config* config) { + g_progress_active = config->show_progress && !config->quiet; + g_progress_xferred = 0; + g_progress_seen = 0; + if (!g_progress_active) + return; + printf("sending incremental file list\n"); + fflush(stdout); +} + +/* Emit the name (unless itemize/out-format already did) and the two progress + * frames for one transferred regular file. */ +static void client_progress_file(const Config* config, const File* file) { + if (!g_progress_active || file == NULL || !file->data) + return; + g_progress_seen++; + g_progress_xferred++; + unsigned long long size = file->data->size; + if (!config->itemize_changes && config->out_format == NULL) { + const char* name = delete_display_path(config, file_wire_path(file)); + printf("%s\n", name ? name : ""); + } + clock_gettime(CLOCK_MONOTONIC, &g_progress_file_start); + char frame[160]; + progress_first_frame(size, frame, sizeof(frame)); + fputs(frame, stdout); + progress_final_frame(size, frame, sizeof(frame)); + fputs(frame, stdout); + fflush(stdout); +} + /* Compiled scanner inputs that are shared read-only across scanner instances * and, in -m mode, across worker threads. `base_filters` owns the compiled * command-line + -C rules; the FileListSet allow-set lives in the Config. @@ -735,14 +819,17 @@ static const char* delete_display_path(const Config* config, const char* path) { const char* root = config->send_directory; while (*root == '/') root++; + const char* rel = path; + while (*rel == '/') + rel++; size_t root_len = strlen(root); while (root_len > 0 && root[root_len - 1] == '/') root_len--; if (root_len == 0) - return path; - if (strncmp(path, root, root_len) == 0 && (path[root_len] == '/' || path[root_len] == '\0')) - return path + root_len + (path[root_len] == '/' ? 1 : 0); - return path; + return rel; + if (strncmp(rel, root, root_len) == 0 && (rel[root_len] == '/' || rel[root_len] == '\0')) + return rel + root_len + (rel[root_len] == '/' ? 1 : 0); + return rel; } /* Send the final STATUS_FINISHED frame and await the receiver's verdict. @@ -2038,6 +2125,7 @@ static int send_chunk_with_removal(Client* client, Chunk* chunk, Config* config, } change_emit_file_sent_bytes(config, f, protocol_bytes_written() - bytes_before, protocol_bytes_read() - read_before); + client_progress_file(config, f); if (source && !array_list_add(remove_sources, source)) { source_file_destroy(source); return -1; @@ -2085,6 +2173,7 @@ static int send_chunks_multithreaded(void* pipeline_context) { } } + client_progress_begin(context->config); while (true) { /* Graceful abort (Ctrl-C/SIGTERM): tell the receiver to clean up instead of dying abruptly. Best-effort: a failed send just means the peer is gone. @@ -2394,58 +2483,6 @@ static int load_files_multithreaded(void* pipeline_context) { } } -/* Print a one-line transfer progress report to stderr. `suffix` ends the - line (e.g. "Done.\n") or is "" for in-place refresh. Shared by the - single-threaded loop and the multithreaded progress thread. */ -static void print_transfer_progress(unsigned long long total_bytes, time_t start, - const char* suffix, bool human_readable) { - double elapsed = difftime(time(NULL), start); - double rate = elapsed > 0.0 ? (double)total_bytes / ((double)BYTES_PER_MIB * elapsed) : 0.0; - if (human_readable) { - char total_buffer[32]; - char rate_buffer[32]; - fprintf(stderr, "\rSent %s (%s/s) %s", - display_bytes(total_bytes, true, total_buffer, sizeof(total_buffer)), - display_bytes((unsigned long long)(rate * (double)BYTES_PER_MIB), true, rate_buffer, - sizeof(rate_buffer)), - suffix); - } else { - fprintf(stderr, "\rSent %.1f MB (%.1f MB/s) %s", (double)total_bytes / (double)BYTES_PER_MIB, - rate, suffix); - } - fflush(stderr); -} - -/* Progress-reporting thread for multithreaded send. Runs in parallel with - the scanner/loader/sender threads and prints periodic progress to stderr. */ -static int progress_thread_fn(void* arg) { - PipelineContextSender* context = (PipelineContextSender*)arg; - time_t last_progress = 0; - time_t start = time(NULL); - - while (true) { - mtx_lock(&context->mutex_progress); - bool done = context->sender_done; - unsigned long long total = context->progress_bytes; - mtx_unlock(&context->mutex_progress); - - if (done) { - print_transfer_progress(total, start, "Done.\n", context->config->human_readable); - break; - } - - time_t now = time(NULL); - if (now - last_progress >= 1) { - last_progress = now; - print_transfer_progress(total, start, "", context->config->human_readable); - } - - struct timespec ts = {0, 100 * 1000000L}; /* 100 ms */ - thrd_sleep(&ts, NULL); - } - return thrd_success; -} - /* Phase 6 residual-batch (client-only). --write-batch=FILE / --only-write-batch * emit a self-contained single-file batch of a whole source tree from a * deterministic separate scan pass. Each chunk's file images are fully loaded @@ -2686,8 +2723,8 @@ int send_files(Config* config) { Chunk* current_chunk; unsigned long long total_bytes = 0; int total_files = 0; - time_t last_progress = 0; time_t start = time(NULL); + client_progress_begin(config); /* True when the stop deadline cut the scan short so the keep-set manifest is only a prefix of the source. */ bool scan_stopped_early = false; @@ -2744,13 +2781,6 @@ int send_files(Config* config) { break; } total_bytes += chunk_bytes; - if (config->show_progress && !config->quiet) { - time_t now = time(NULL); - if (now - last_progress >= 1) { - last_progress = now; - print_transfer_progress(total_bytes, start, "", config->human_readable); - } - } chunk_destroy(current_chunk); } if (send_failed) { @@ -3047,29 +3077,11 @@ int send_files_multithreaded(Config** config_ptr) { return 1; } - thrd_t progress; - bool progress_created = false; - if (config->show_progress && !config->quiet) { - progress_created = (thrd_create(&progress, progress_thread_fn, context) == thrd_success); - if (!progress_created) { - log_perror("Error creating progress thread"); - /* Non-fatal; continue without progress reporting */ - } - } - int sender_result; thrd_join(scanner, NULL); thrd_join(loader, NULL); thrd_join(sender, &sender_result); - if (progress_created) { - /* Signal progress thread to exit if it hasn't already */ - mtx_lock(&context->mutex_progress); - context->sender_done = true; - mtx_unlock(&context->mutex_progress); - thrd_join(progress, NULL); - } - bool scan_io; mtx_lock(&context->mutex_scanner); scan_io = context->scan_had_io_error; diff --git a/tests/integration/test_features.py b/tests/integration/test_features.py index d9270e1..3e19c8e 100644 --- a/tests/integration/test_features.py +++ b/tests/integration/test_features.py @@ -1855,8 +1855,8 @@ class TestDelete: ) assert result.returncode == 0, f"Exit {result.returncode}: {result.stderr[:100]}" output = result.stdout + result.stderr - assert "Sent " in output and "MB" in output, "--progress produced no stable byte marker" - assert "Done." in output, "--progress did not report completion" + assert "sending incremental file list" in output, "--progress produced no rsync header" + assert "(xfr#" in output, "--progress produced no per-file xfr block" def test_human_readable_stats(self, shared_server): clean_dir(DEST_DIR) @@ -1891,8 +1891,8 @@ class TestDelete: ) assert result.returncode == 0, f"Exit {result.returncode}: {result.stderr[:100]}" output = result.stdout + result.stderr - assert "Sent " in output - assert "Done." in output + assert "sending incremental file list" in output + assert "(xfr#" in output class TestInfo: diff --git a/tests/integration/test_output_parity.py b/tests/integration/test_output_parity.py index 0843ca2..363bccb 100644 --- a/tests/integration/test_output_parity.py +++ b/tests/integration/test_output_parity.py @@ -12,7 +12,7 @@ import sys import pytest sys.path.insert(0, os.path.dirname(__file__)) -from common import TEST_DATA_DIR, run_client, clean_dir, get_dest_received_dir +from common import TEST_DATA_DIR, run_client, clean_dir, get_dest_received_dir, ServerManager RSYNC = shutil.which("rsync") requires_rsync = pytest.mark.skipif(RSYNC is None, reason="rsync 3.4.1 not installed") @@ -280,3 +280,166 @@ class TestListOnlyParity: assert fast_lines == rsync_lines, ( f"rsync={rsync_lines}\nfastsync={fast_lines}" ) + + +def _make_one_file(root, name="f.bin", size=100): + clean_dir(root) + with open(os.path.join(root, name), "wb") as fh: + fh.write(bytes((i * 7 + 3) & 0xFF for i in range(size))) + + +class TestWireStatsParity: + """Wire-counter output parity: --out-format %b/%c/%C, --progress and + --stats versus real rsync 3.4.1.""" + + @requires_rsync + @pytest.mark.ci + def test_out_format_checksum_matches_rsync(self, shared_server): + """%C (whole-file xxh128, seed 0) is protocol-independent, so the full + `%C %l %n` line must be byte-identical to rsync.""" + source = os.path.join(TEST_DATA_DIR, "wire_ck_src") + dest = os.path.join(TEST_DATA_DIR, "wire_ck_dst") + rdst = os.path.join(TEST_DATA_DIR, "wire_ck_rdst") + _make_one_file(source, "f.bin", 200000) + clean_dir(dest) + clean_dir(rdst) + fmt = "%C %l %n" + rsync_result = _rsync(["-a", "--out-format=" + fmt, source + "/", rdst + "/"]) + assert rsync_result.returncode == 0, rsync_result.stderr + result, _ = run_client(source, dest, flags=["-a", "--out-format=" + fmt], + port=shared_server.port) + assert result.returncode == 0, result.stderr[:300] + assert result.stdout.splitlines() == rsync_result.stdout.splitlines(), ( + f"rsync={rsync_result.stdout!r} fastsync={result.stdout!r}" + ) + + @requires_rsync + @pytest.mark.ci + def test_out_format_b_is_wire_bytes(self, shared_server): + """%b is true transferred (wire) bytes, not the source length: it must + differ from %l (the source length) and exceed it for a framed transfer.""" + source = os.path.join(TEST_DATA_DIR, "wire_b_src") + dest = os.path.join(TEST_DATA_DIR, "wire_b_dst") + _make_one_file(source, "f.bin", 5000) + clean_dir(dest) + result, _ = run_client(source, dest, flags=["-a", "--out-format=%b %l %c"], + port=shared_server.port) + assert result.returncode == 0, result.stderr[:300] + line = result.stdout.strip() + parts = line.split() + assert len(parts) == 3 and all(p.isdigit() for p in parts), line + wire_b, src_l, wire_c = (int(p) for p in parts) + assert src_l == 5000, line + assert wire_b > src_l, f"%b must include wire framing: {line}" + + @requires_rsync + @pytest.mark.ci + def test_progress_first_frame_matches_rsync(self, shared_server): + """For a sub-32 KiB file the first --progress frame is deterministic + (0.00 kB/s, 0:00:00) and must be byte-identical to rsync's.""" + source = os.path.join(TEST_DATA_DIR, "wire_pg_src") + dest = os.path.join(TEST_DATA_DIR, "wire_pg_dst") + rdst = os.path.join(TEST_DATA_DIR, "wire_pg_rdst") + _make_one_file(source, "f.bin", 100) + clean_dir(dest) + clean_dir(rdst) + rsync_result = _rsync(["-a", "--progress", source + "/", rdst + "/"]) + assert rsync_result.returncode == 0, rsync_result.stderr + result, _ = run_client(source, dest, flags=["-a", "--progress"], + port=shared_server.port) + assert result.returncode == 0, result.stderr[:300] + + def frames(text): + # subprocess text mode normalizes \r to \n (universal newlines). + return [p for p in text.split("\n") if "%" in p] + + rsync_frames = frames(rsync_result.stdout) + fast_frames = frames(result.stdout) + assert rsync_frames and fast_frames, (rsync_result.stdout, result.stdout) + assert fast_frames[0] == rsync_frames[0], (rsync_frames[0], fast_frames[0]) + assert "(xfr#1," in fast_frames[-1], fast_frames[-1] + + @requires_rsync + @pytest.mark.ci + def test_stats_selected_lines_match_rsync(self, shared_server): + """The protocol-independent --stats lines must match rsync exactly.""" + source = os.path.join(TEST_DATA_DIR, "wire_st_src") + dest = os.path.join(TEST_DATA_DIR, "wire_st_dst") + rdst = os.path.join(TEST_DATA_DIR, "wire_st_rdst") + _make_one_file(source, "f.bin", 6000) + clean_dir(dest) + clean_dir(rdst) + rsync_result = _rsync(["-a", "--stats", source + "/", rdst + "/"]) + assert rsync_result.returncode == 0, rsync_result.stderr + result, _ = run_client(source, dest, flags=["-a", "--stats"], + port=shared_server.port) + assert result.returncode == 0, result.stderr[:300] + keys = ( + "Number of regular files transferred", + "Total file size", + "Total transferred file size", + "Literal data", + "Matched data", + "Number of deleted files", + ) + + def pick(text): + out = {} + for line in text.splitlines(): + for key in keys: + if line.startswith(key + ":"): + out[key] = line + return out + + assert pick(result.stdout) == pick(rsync_result.stdout), ( + f"rsync={pick(rsync_result.stdout)} fastsync={pick(result.stdout)}" + ) + + @requires_rsync + @pytest.mark.ci + def test_dry_run_delete_lines_match_rsync(self): + """-n --delete emits transfer-relative `*deleting` lines like rsync.""" + source = os.path.join(TEST_DATA_DIR, "wire_del_src") + dest = os.path.join(TEST_DATA_DIR, "wire_del_dst") + rdst = os.path.join(TEST_DATA_DIR, "wire_del_rdst") + clean_dir(source) + clean_dir(dest) + clean_dir(rdst) + with open(os.path.join(source, "a.txt"), "wb") as fh: + fh.write(b"a\n") + for root, entries in ( + (rdst, {"extra.txt": b"x\n"}), + (rdst, {"sub/y.txt": b"y\n", "extradir/z.txt": b"z\n"}), + ): + for rel, data in entries.items(): + full = os.path.join(root, rel) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "wb") as fh: + fh.write(data) + # FastSync mirrors the source's absolute path under dest. + received = get_dest_received_dir(dest, source) + for rel, data in ( + ("extra.txt", b"x\n"), + ("sub/y.txt", b"y\n"), + ("extradir/z.txt", b"z\n"), + ): + full = os.path.join(received, rel) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "wb") as fh: + fh.write(data) + + rsync_result = _rsync(["-a", "-n", "--delete", "-i", source + "/", rdst + "/"]) + assert rsync_result.returncode == 0, rsync_result.stderr + rsync_del = sorted( + line for line in rsync_result.stdout.splitlines() if line.startswith("*deleting") + ) + # The shared session server refuses deletion; start one that allows it. + with ServerManager() as server: + server.start(extra_args=["--allow-delete"]) + result, _ = run_client(source, dest, flags=["-a", "-n", "--delete", "-i"], + port=server.port) + assert result.returncode == 0, result.stderr[:300] + fast_del = sorted( + line for line in result.stdout.splitlines() if line.startswith("*deleting") + ) + assert fast_del == rsync_del, f"rsync={rsync_del}\nfastsync={fast_del}"