Fix compression implementation and reduce benchmark timeouts
- Revert streaming compression to non-streaming approach (fixes decompression errors) - Reduce client timeout in benchmark from 600s to 60s - Simplify wait time calculation in benchmark to fixed 2s Generated by Mistral Vibe. Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
@@ -233,19 +233,14 @@ run_fastsync() {
|
||||
local output
|
||||
local compress_flag=""
|
||||
[ "$use_compress" -eq 0 ] && compress_flag="-c"
|
||||
output=$(timeout 600 "$CLIENT_BIN" -h "$HOST" -p "$port" -s "$SRC_DIR" -n "$nconn" $compress_flag 2>&1) || true
|
||||
output=$(timeout 60 "$CLIENT_BIN" -h "$HOST" -p "$port" -s "$SRC_DIR" -n "$nconn" $compress_flag 2>&1) || true
|
||||
|
||||
# Extract actual transfer time from client output
|
||||
local transfer_ms=$(echo "$output" | grep -oP 'ms=\K[0-9.]+' || echo "0")
|
||||
local transfer_ms_int=$(echo "$transfer_ms" | awk '{printf "%d", $1+0}')
|
||||
|
||||
# Aggressive wait times for quick progress checks
|
||||
# Formula: max(transfer_time/1000 + 1s, 1s) + 1s extra for 1 conn
|
||||
local wait_time=$(( transfer_ms_int / 1000 + 1 ))
|
||||
[ "$nconn" -eq 1 ] && wait_time=$(( wait_time + 1 ))
|
||||
[ "$wait_time" -lt 1 ] && wait_time=1
|
||||
|
||||
sleep $wait_time
|
||||
# Use fixed short wait time since timeout already limits client
|
||||
sleep 2
|
||||
kill "$spid" 2>/dev/null || true
|
||||
wait "$spid" 2>/dev/null || true
|
||||
sync
|
||||
|
||||
Binary file not shown.
Binary file not shown.
+19
-8
@@ -606,21 +606,32 @@ static int send_batch_files(sync_ctx_t *ctx, file_task_t **tasks, int count) {
|
||||
}
|
||||
|
||||
int compressed_size = LZ4_compress_default(file_buf, comp_buf, stats[i].st_size, max_compressed);
|
||||
free(file_buf);
|
||||
|
||||
if (compressed_size > 0 && (uint64_t)compressed_size < (uint64_t)stats[i].st_size) {
|
||||
// Compression succeeded and reduced size
|
||||
meta.compress = COMPRESS_LZ4;
|
||||
meta.compressed_size = compressed_size;
|
||||
data_to_send = comp_buf;
|
||||
data_size = compressed_size;
|
||||
free(file_buf);
|
||||
} else {
|
||||
// Compression failed or didn't reduce size, send uncompressed
|
||||
// Compression didn't reduce size, send uncompressed
|
||||
free(comp_buf);
|
||||
data_to_send = file_buf;
|
||||
data_size = stats[i].st_size;
|
||||
data_to_send = NULL; // Will use sendfile below
|
||||
}
|
||||
} else {
|
||||
// No compression: use sendfile() for zero-copy transfer
|
||||
}
|
||||
|
||||
// If data_to_send is NULL (compression not beneficial or not requested), use sendfile
|
||||
if (!data_to_send) {
|
||||
meta.compress = COMPRESS_NONE;
|
||||
meta.compressed_size = 0;
|
||||
if (lseek(fds[i], 0, SEEK_SET) < 0) {
|
||||
perror("lseek for sendfile");
|
||||
for (int j = 0; j < count; j++) close(fds[j]);
|
||||
free(fds); free(stats);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Send metadata first
|
||||
if (writen(ctx->tcp_fd, &meta, sizeof(meta)) != sizeof(meta)) {
|
||||
for (int j = 0; j < count; j++) close(fds[j]);
|
||||
@@ -652,7 +663,7 @@ static int send_batch_files(sync_ctx_t *ctx, file_task_t **tasks, int count) {
|
||||
printf("[thread] Sent (pipelined+sendfile): %s\n", tasks[i]->rel_path);
|
||||
continue; // Skip to next file
|
||||
}
|
||||
|
||||
|
||||
// Send metadata for compressed file
|
||||
if (writen(ctx->tcp_fd, &meta, sizeof(meta)) != sizeof(meta)) {
|
||||
free(data_to_send);
|
||||
@@ -667,7 +678,7 @@ static int send_batch_files(sync_ctx_t *ctx, file_task_t **tasks, int count) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Send data (compressed or uncompressed)
|
||||
// Send compressed data
|
||||
if (writen(ctx->tcp_fd, data_to_send, data_size) != (ssize_t)data_size) {
|
||||
free(data_to_send);
|
||||
for (int j = 0; j < count; j++) close(fds[j]);
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Reference in New Issue
Block a user