diff --git a/.gitignore b/.gitignore index 9e1a350..ac92afb 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,4 @@ build data_copied +test_data/ +__pycache__/ diff --git a/src/client/client.c b/src/client/client.c index 2cfc027..a026458 100644 --- a/src/client/client.c +++ b/src/client/client.c @@ -207,6 +207,14 @@ int main(int argc, char *argv[]) { config->compression_level); } } + } else if (strcmp(argv[i], "--source-dir") == 0 && i + 1 < argc) { + free(config->send_directory); + config->send_directory = str_dup(argv[++i]); + } else if (strcmp(argv[i], "--dest-dir") == 0 && i + 1 < argc) { + free(config->receive_root_directory); + config->receive_root_directory = str_dup(argv[++i]); + } else if (strcmp(argv[i], "--save-to-disk") == 0) { + config->save_to_disk = true; } else { handle_arg(argv[i], "-m", &config->use_multithreading, "Enabled Multithreading"); diff --git a/src/client/scanner.c b/src/client/scanner.c index 4a543e3..bcb3464 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -7,10 +7,14 @@ #include #include #include +#include +#include DirectoryScanner *directory_scanner_create(char *root_directory) { DirectoryScanner *scanner = malloc(sizeof(DirectoryScanner)); scanner->directories = queue_create(100, free); + scanner->current_dir = NULL; + scanner->current_path = NULL; queue_enqueue(scanner->directories, str_dup(root_directory)); return scanner; } @@ -18,11 +22,16 @@ DirectoryScanner *directory_scanner_create(char *root_directory) { void directory_scanner_destroy(DirectoryScanner *scanner) { if (scanner == NULL) return; + if (scanner->current_dir) { + closedir(scanner->current_dir); + scanner->current_dir = NULL; + } + free(scanner->current_path); queue_destroy(scanner->directories); free(scanner); } -Chunk *chunk_data_to_chunk(ArrayList *chunk_data) { +static Chunk *chunk_data_to_chunk(ArrayList *chunk_data) { void **chunk_items = array_list_to_array(chunk_data); Chunk *chunk = chunk_create((File **)chunk_items, chunk_data->size); free(chunk_items); @@ -31,40 +40,66 @@ Chunk *chunk_data_to_chunk(ArrayList *chunk_data) { return chunk; } +static int open_next_directory(DirectoryScanner *scanner) { + if (scanner->current_dir) { + closedir(scanner->current_dir); + scanner->current_dir = NULL; + } + free(scanner->current_path); + + if (queue_is_empty(scanner->directories)) + return 0; + + scanner->current_path = (char *)queue_dequeue(scanner->directories); + scanner->current_dir = opendir(scanner->current_path); + if (scanner->current_dir == NULL) { + perror("Could not open directory!"); + exit(EXIT_FAILURE); + } + return 1; +} + Chunk *directory_scanner_next(DirectoryScanner *scanner) { ArrayList *chunk_data = array_list_create(file_destroy); unsigned long long chunk_data_size = 0; - while (!queue_is_empty(scanner->directories)) { - char *path = (char *)queue_dequeue(scanner->directories); - DIR *dir; - struct dirent *entry; - dir = opendir(path); - if (dir == NULL) { - perror("Could not open directory!"); - exit(EXIT_FAILURE); + while (1) { + if (scanner->current_dir == NULL) { + if (!open_next_directory(scanner)) + break; } - while ((entry = readdir(dir)) != NULL) { - if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) { - continue; - } - char *cur_path = path_cat(path, entry->d_name); - struct stat stats; - stat(cur_path, &stats); - if (!S_ISREG(stats.st_mode)) - queue_enqueue(scanner->directories, (void *)cur_path); - else { - File *file = file_create(cur_path, &stats); - array_list_add(chunk_data, file); - chunk_data_size += file->stats.st_size; - if (chunk_data_size > DESIRED_CHUNK_SIZE) - return chunk_data_to_chunk(chunk_data); - free(cur_path); - } + + struct dirent *entry = readdir(scanner->current_dir); + if (entry == NULL) { + closedir(scanner->current_dir); + scanner->current_dir = NULL; + free(scanner->current_path); + scanner->current_path = NULL; + continue; + } + + if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) + continue; + + char *cur_path = path_cat(scanner->current_path, entry->d_name); + struct stat stats; + if (stat(cur_path, &stats) != 0) { + free(cur_path); + continue; + } + + if (!S_ISREG(stats.st_mode)) { + queue_enqueue(scanner->directories, (void *)cur_path); + } else { + File *file = file_create(cur_path, &stats); + array_list_add(chunk_data, file); + chunk_data_size += file->stats.st_size; + if (chunk_data_size > DESIRED_CHUNK_SIZE) + return chunk_data_to_chunk(chunk_data); + free(cur_path); } - closedir(dir); - free(path); } + if (chunk_data->size > 0) return chunk_data_to_chunk(chunk_data); return NULL; diff --git a/src/client/scanner.h b/src/client/scanner.h index c8e0205..43596ad 100644 --- a/src/client/scanner.h +++ b/src/client/scanner.h @@ -3,8 +3,12 @@ #include "chunk.h" #include "queue.h" +#include + typedef struct { Queue *directories; + DIR *current_dir; + char *current_path; } DirectoryScanner; DirectoryScanner *directory_scanner_create(char *root_directory); diff --git a/test.py b/test.py index edb13a3..d6dd0b5 100755 --- a/test.py +++ b/test.py @@ -1,27 +1,31 @@ +import argparse +import filecmp import os +import shutil import subprocess +import sys +import tempfile import time -# --- Configuration --- +TEST_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "test_data") +DEFAULT_SOURCE_DIR = os.path.join(TEST_DIR, "source") +DEFAULT_DEST_DIR = os.path.join(TEST_DIR, "dest") + SERVER_CMD = ["./build/server"] base_client_cmd = ["./build/client"] -# --- Resource Limit Configuration --- -# 💾 Disk throttling settings -DISK_DEVICE = ( - "/dev/nvme0n1p5" # IMPORTANT: Change this to your disk (e.g., /dev/nvme0n1) -) -READ_BPS_MAX = "15M" # Max read speed (M for megabytes) -WRITE_BPS_MAX = "10M" # Max write speed +DISK_DEVICE = "/dev/nvme0n1p5" +READ_BPS_MAX = "15M" +WRITE_BPS_MAX = "10M" -# 🐢 Network throttling settings (Linux tc) NET_LIMIT = "100mbit" NET_DELAY = "100ms" NETWORK_INTERFACE = "lo" -NET_LIMIT_CMD = f"sudo tc qdisc add dev {NETWORK_INTERFACE} root netem rate {NET_LIMIT} delay {NET_DELAY}".split() +NET_LIMIT_CMD = ( + f"sudo tc qdisc add dev {NETWORK_INTERFACE} root netem rate {NET_LIMIT} delay {NET_DELAY}".split() +) NET_RESET_CMD = f"sudo tc qdisc del dev {NETWORK_INTERFACE} root".split() -# --- Build the client command prefix with throttling --- CLIENT_CMD_PREFIX = [ "sudo", "systemd-run", @@ -32,11 +36,10 @@ CLIENT_CMD_PREFIX = [ f"IOWriteBandwidthMax={DISK_DEVICE} {WRITE_BPS_MAX}", ] -# --- Test Cases --- TEST_CASES = [ {"name": "Standard (Single-threaded)", "flags": []}, {"name": "Multithreading (-m)", "flags": ["-m"]}, - {"name": "Compression (-c 0)", "flags": ["-c 0"]}, + {"name": "Compression (-c)", "flags": ["-c"]}, {"name": "Chunk Serialization (-s)", "flags": ["-s"]}, {"name": "Compression + Chunk Serialization (-c -s)", "flags": ["-c", "-s"]}, {"name": "Multithreading + Compression (-m -c)", "flags": ["-m", "-c"]}, @@ -48,179 +51,249 @@ TEST_CASES = [ ] -def run_suite(env_name, apply_limits): +def generate_test_files(source_dir): + if os.path.exists(source_dir): + shutil.rmtree(source_dir) + os.makedirs(source_dir) + + target_total = 50 * 1024 * 1024 + written = 0 + + files = { + "small.txt": b"hello world\n", + "medium.txt": b"the quick brown fox jumps over the lazy dog\n" * 5000, + "binary.bin": bytes(range(256)) * 1000, + "nested/subdir/deep.txt": b"deeply nested file\n", + "nested/another.txt": b"another nested file\n" * 50, + } + for rel_path, content in files.items(): + full_path = os.path.join(source_dir, rel_path) + os.makedirs(os.path.dirname(full_path), exist_ok=True) + with open(full_path, "wb") as f: + f.write(content) + written += len(content) + + i = 0 + while written < target_total: + chunk_size = min(5 * 1024 * 1024, target_total - written) + rel_path = f"bulk/file_{i}.dat" + full_path = os.path.join(source_dir, rel_path) + os.makedirs(os.path.dirname(full_path), exist_ok=True) + with open(full_path, "wb") as f: + f.write(b"0" * chunk_size) + written += chunk_size + i += 1 + + total_mb = written / (1024 * 1024) + print(f" Generated {total_mb:.1f} MB of test data in {source_dir}") + + +def verify_transfer(source_dir, dest_dir): + source_dir = os.path.abspath(source_dir) + dest_dir = os.path.abspath(dest_dir) + + received_prefix = os.path.join(dest_dir, source_dir.lstrip(os.sep)) + if not os.path.exists(received_prefix): + return [], ["no received files found"] + + mismatches = [] + missing = [] + + for root, dirs, files in os.walk(source_dir): + for f in files: + src_path = os.path.join(root, f) + rel = os.path.relpath(src_path, source_dir) + dst_path = os.path.join(received_prefix, rel) + + if not os.path.exists(dst_path): + missing.append(rel) + elif not filecmp.cmp(src_path, dst_path, shallow=False): + mismatches.append(rel) + + return mismatches, missing + + +def run_suite(env_name, apply_limits, source_dir, dest_dir): results = [] print(f"\n{'=' * 60}") - print(f"🚀 Starting Suite: {env_name}") + print(f"Suite: {env_name}") print(f"{'=' * 60}") if apply_limits: - print( - f"Applying Disk I/O Limits: Reads <= {READ_BPS_MAX}, Writes <= {WRITE_BPS_MAX}" - ) - print(f"Applying Network Limits: {NET_LIMIT}, {NET_DELAY} delay") + print(f" Disk I/O: Reads <= {READ_BPS_MAX}, Writes <= {WRITE_BPS_MAX}") + print(f" Network: {NET_LIMIT}, {NET_DELAY} delay") client_prefix = CLIENT_CMD_PREFIX else: - print("Running Baseline (No limits applied)") - client_prefix = [] # Run normally without systemd-run/limits + print(" Baseline (no limits)") + client_prefix = [] try: - # SETUP: Apply or ensure clean network limits if apply_limits: subprocess.run(NET_LIMIT_CMD, check=True) else: - # Silently attempt to clear any leftover rules just to ensure a clean baseline subprocess.run(NET_RESET_CMD, capture_output=True) for case in TEST_CASES: name = case["name"] flags = case["flags"] + print(f"\n --- {name} ---") - print(f"\n--- Running: {name} ---") + if os.path.exists(dest_dir): + shutil.rmtree(dest_dir) server_process = None try: - # 1. Start the server - print(" Starting server...") server_process = subprocess.Popen( SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None ) - time.sleep(0.5) # Allow server to bind to port + time.sleep(0.5) - # 2. Build and run the client - client_cmd = client_prefix + base_client_cmd + flags - print(f" Running client: {' '.join(client_cmd)}") + env = os.environ.copy() + + client_cmd = ( + client_prefix + + base_client_cmd + + ["--source-dir", source_dir, "--dest-dir", dest_dir, "--save-to-disk"] + + flags + ) + print(f" Running: {' '.join(client_cmd)}") start_time = time.monotonic() client_result = subprocess.run( - client_cmd, text=True, capture_output=True + client_cmd, env=env, text=True, capture_output=True ) end_time = time.monotonic() - duration = end_time - start_time + if server_process: + try: + server_process.wait(timeout=5) + except subprocess.TimeoutExpired: + server_process.kill() + server_process.wait() + server_process = None + + mismatches, missing = [], [] if client_result.returncode == 0: - results.append( - { - "environment": env_name, - "name": name, - "status": "Success", - "time": f"{duration:.4f}s", - "error": "", - } - ) + mismatches, missing = verify_transfer(source_dir, dest_dir) + + entry = { + "name": name, + "suite": env_name, + "time": f"{duration:.4f}s" if client_result.returncode == 0 else "N/A", + } + + if client_result.returncode == 0 and not mismatches and not missing: + entry["status"] = "Success" + entry["error"] = "" else: - print(f" ⚠️ Failed (code: {client_result.returncode})") - err_msg = ( - client_result.stderr.strip().split("\n")[0] - if client_result.stderr - else ( - client_result.stdout.strip().split("\n")[0] - if client_result.stdout - else "No output" + entry["status"] = "Failed" + errors = [] + if client_result.returncode != 0: + err = ( + client_result.stderr.strip().split("\n")[0] + if client_result.stderr + else ( + client_result.stdout.strip().split("\n")[0] + if client_result.stdout + else "No output" + ) ) - ) - results.append( - { - "environment": env_name, - "name": name, - "status": "Failed", - "time": "N/A", - "error": f"Exit code {client_result.returncode}: {err_msg[:40]}", - } - ) + errors.append(f"Exit code {client_result.returncode}: {err[:80]}") + if missing: + errors.append(f"Missing ({len(missing)}): {', '.join(missing[:5])}") + if mismatches: + errors.append(f"Mismatch ({len(mismatches)}): {', '.join(mismatches[:3])}") + entry["error"] = " | ".join(errors) + + results.append(entry) except subprocess.TimeoutExpired: - print(" ⚠️ Timeout (exceeded 15s)") results.append( - { - "environment": env_name, - "name": name, - "status": "Timeout", - "time": "N/A", - "error": "Exceeded 15 seconds", - } + {"name": name, "suite": env_name, "status": "Timeout", "time": "N/A", "error": "Exceeded 15s"} ) except Exception as e: - print(f" ❌ Error: {e}") results.append( - { - "environment": env_name, - "name": name, - "status": "Error", - "time": "N/A", - "error": str(e), - } + {"name": name, "suite": env_name, "status": "Error", "time": "N/A", "error": str(e)} ) finally: - # Clean up the server for this test case if server_process: - print(" Stopping server...") try: - server_process.terminate() server_process.wait(timeout=5) except subprocess.TimeoutExpired: server_process.kill() server_process.wait() except subprocess.CalledProcessError as e: - print(f"❌ Error running system limit command: {' '.join(e.cmd)}") - print("Are you running this script with 'sudo' privileges?") - + print(f" Error running limit command: {' '.join(e.cmd)}") finally: - # TEARDOWN: Remove network limits if they were applied if apply_limits: - print("\nCleaning up limits for this suite...") try: subprocess.run(NET_RESET_CMD, check=True, capture_output=True) - print("Network limits removed.") - except Exception as e: - print(f"⚠️ Could not reset network settings: {e}") + except Exception: + pass return results -os.system("cmake -B build -S .") -os.system("cd build && make") +def main(): + parser = argparse.ArgumentParser(description="FastSync integration test / benchmark") + parser.add_argument("--source-dir", default=DEFAULT_SOURCE_DIR, + help="Source directory for test files (default: %(default)s)") + parser.add_argument("--dest-dir", default=DEFAULT_DEST_DIR, + help="Destination directory for received files (default: %(default)s)") + parser.add_argument("--keep-data", action="store_true", + help="Keep test_data directory after run") + parser.add_argument("--no-throttled", action="store_true", + help="Skip throttled suite (requires sudo)") + parser.add_argument("--no-unlimited", action="store_true", + help="Skip unlimited suite") + args = parser.parse_args() -# --- Main Execution --- -all_results = [] + os.system("cmake -B build -S . > /dev/null 2>&1") + ret = os.system("cd build && make -j$(nproc) 2>&1 | tail -3") + if ret != 0: + print("Build failed") + sys.exit(1) -# 1. Run Baseline (No Limits) -all_results.extend(run_suite("Unlimited", apply_limits=False)) + generate_test_files(args.source_dir) + os.makedirs(args.dest_dir, exist_ok=True) -# # 2. Run Throttled (With Limits) -all_results.extend(run_suite("Throttled", apply_limits=True)) + try: + all_results = [] -# --- Print Comparison Table --- -print("\n" + "=" * 105) -print(f"{'fastSync BENCHMARK RESULTS (COMPARISON)':^105}") -print("=" * 105) -print( - f"{'Configuration':<45} | {'Environment':<12} | {'Status':<10} | {'Time':<10} | {'Details/Error':<20}" -) -print("-" * 105) + if not args.no_unlimited: + all_results.extend( + run_suite("Unlimited", False, args.source_dir, args.dest_dir) + ) -# Sort results by test case name first, then environment to easily compare -# This groups the baseline and throttled results for the same test next to each other -# sorted_results = sorted( -# all_results, -# key=lambda x: ( -# TEST_CASES.index( -# next(item for item in TEST_CASES if item["name"] == x["name"]) -# ), -# x["environment"], -# ), -# ) + if not args.no_throttled: + all_results.extend( + run_suite("Throttled", True, args.source_dir, args.dest_dir) + ) -for res in all_results: - status_symbol = ( - "✅" - if res["status"] == "Success" - else ("⏳" if res["status"] == "Timeout" else "❌") - ) - status_str = f"{status_symbol} {res['status']}" - print( - f"{res['name']:<45} | {res['environment']:<12} | {status_str:<10} | {res['time']:<10} | {res['error']:<20}" - ) -print("=" * 105) + print("\n" + "=" * 110) + print(f"{'RESULTS':^110}") + print("=" * 110) + print(f"{'Configuration':<45} | {'Suite':<12} | {'Status':<8} | {'Time':<10} | {'Details'}") + print("-" * 110) + + for res in all_results: + print( + f"{res['name']:<45} | {res['suite']:<12} | {res['status']:<8} | {res['time']:<10} | {res['error']}" + ) + + failed = [r for r in all_results if r["status"] != "Success"] + if failed: + print(f"\n {len(failed)} test(s) FAILED") + sys.exit(1) + else: + print(f"\n ALL {len(all_results)} TESTS PASSED") + + finally: + if not args.keep_data: + shutil.rmtree(TEST_DIR, ignore_errors=True) + + +if __name__ == "__main__": + main()