Rsync-compatible flags: -a, -n, -p, --exclude, --delete #8
@@ -1,36 +1,40 @@
|
|||||||
# FastSync
|
# FastSync
|
||||||
|
|
||||||
A high-performance file synchronization system with a custom TCP-based protocol, optional metadata preservation, compression, multithreading, and zero-copy `sendfile()` support.
|
A high-performance file synchronization system with SSH and TCP transport, streaming zstd compression, multithreaded transfer, metadata preservation, and rsync-compatible CLI flags.
|
||||||
|
|
||||||
## Technical Overview
|
## Technical Overview
|
||||||
|
|
||||||
1. Custom TCP-based client-server protocol with status codes
|
1. **Dual transport**: custom TCP client-server or SSH subprocess (rsync-style `user@host:/path`)
|
||||||
2. Chunked file transfer (files grouped into ~10 MB chunks)
|
2. **Chunked file transfer**: files grouped into configurable-size chunks (default ~10 MB)
|
||||||
3. Optional zstd compression (levels 1–22)
|
3. **Streaming zstd compression** (levels 1–22) using `ZSTD_compressStream2`
|
||||||
4. Multithreading for parallel file processing (producer-consumer with thread-safe queues)
|
4. **Multithreading**: producer-consumer pipeline with thread-safe queues (scanner → loader → sender)
|
||||||
5. Optional file metadata preservation (`mode`, `uid`, `gid`, `mtime`) — restored on disk
|
5. **Metadata preservation**: `mode`, `uid`, `gid`, `mtime` restored on disk when enabled
|
||||||
6. In-memory and disk-based storage options
|
6. **`sendfile()` zero-copy** on TCP (~2× faster on loopback)
|
||||||
7. `sendfile()` zero-copy path (~2× faster on localhost)
|
7. **SSH ControlMaster** for connection reuse across repeated invocations
|
||||||
|
8. **`--delete`**: receiver removes files not present in sender manifest
|
||||||
|
9. **`--exclude`**: glob-pattern filename filtering (`*`, `?`, no `/` crossing)
|
||||||
|
|
||||||
## System Architecture
|
## System Architecture
|
||||||
|
|
||||||
### Client
|
### Client
|
||||||
- Recursively scans source directories (BFS)
|
- Recursively scans source directories (BFS), supports exclude patterns
|
||||||
- Groups files into chunks (default ~10 MB total)
|
- Groups files into chunks (configurable size)
|
||||||
- Optionally compresses with zstd
|
- Streaming zstd compression with configurable level
|
||||||
- Optionally serializes chunks into a compact binary format
|
- Chunk serialization (compact binary format) or per-file transfer
|
||||||
- Optionally attaches per-file metadata (mode, ownership, timestamps)
|
- Manifests all sent paths when `--delete` is active
|
||||||
- Sends via custom protocol or `sendfile()` zero-copy path
|
- Sends via TCP `sendfile()` or SSH pipe
|
||||||
|
- Optional progress display with throughput
|
||||||
|
|
||||||
### Server
|
### Server
|
||||||
- Listens on port 8080
|
- TCP mode: listens on port 8080; SSH mode: runs via `--stdio`
|
||||||
- Receives and reassembles files
|
- Receives and reassembles files
|
||||||
- Decompresses, deserializes, restores metadata on disk
|
- Decompresses (streaming zstd), deserializes, restores metadata
|
||||||
|
- Processes `STATUS_MANIFEST` for `--delete`: walks destination tree, removes extras
|
||||||
- Thread pool for parallel processing
|
- Thread pool for parallel processing
|
||||||
|
|
||||||
## Protocol Details
|
## Protocol Details
|
||||||
|
|
||||||
Status codes:
|
### Status Codes
|
||||||
| Code | Meaning |
|
| Code | Meaning |
|
||||||
|------|---------|
|
|------|---------|
|
||||||
| `STATUS_OK` | Operation successful |
|
| `STATUS_OK` | Operation successful |
|
||||||
@@ -38,56 +42,76 @@ Status codes:
|
|||||||
| `STATUS_FINISHED` | Transfer complete |
|
| `STATUS_FINISHED` | Transfer complete |
|
||||||
| `STATUS_NEXT` | Ready for next file (per-file mode) |
|
| `STATUS_NEXT` | Ready for next file (per-file mode) |
|
||||||
| `STATUS_CHUNK` | Following data is a serialized chunk |
|
| `STATUS_CHUNK` | Following data is a serialized chunk |
|
||||||
|
| `STATUS_MANIFEST` | Following data is a file manifest (for `--delete`) |
|
||||||
|
|
||||||
### Wire Format — Metadata
|
### Wire Format — Metadata
|
||||||
|
|
||||||
When `use_metadata` is enabled (`-M`), each file entry carries a 4-byte `present` flag followed by five fields (`mode`, `uid`, `gid`, `mtime_sec`, `mtime_nsec`). When disabled globally, no metadata bytes are sent — zero wire overhead.
|
When `use_metadata` is enabled (`-M`), each file entry carries a 4-byte `present` flag followed by five fields (`mode`, `uid`, `gid`, `mtime_sec`, `mtime_nsec`). When disabled globally, no metadata bytes are sent — zero wire overhead.
|
||||||
|
|
||||||
## Configuration
|
### Transfer Flow
|
||||||
|
```
|
||||||
|
Config → (STATUS_NEXT | STATUS_CHUNK)* → [STATUS_MANIFEST] → STATUS_FINISHED → STATUS_OK
|
||||||
|
```
|
||||||
|
|
||||||
|
## Command-Line Arguments
|
||||||
|
|
||||||
### Command-Line Arguments
|
|
||||||
| Argument | Description |
|
| Argument | Description |
|
||||||
|----------|-------------|
|
|----------|-------------|
|
||||||
| `-m` | Multithreading mode |
|
| Positional | `<source> <dest>` — automatic SSH detection if dest contains `:` |
|
||||||
| `-c [level]` | Compression with optional level (1–22, default 5) |
|
| `-c [level]` | Compression with optional level (1–22, default 5) |
|
||||||
|
| `-z [level]` | Alias for `-c` |
|
||||||
|
| `-a, --archive` | Archive mode: enables `-c -m -M` (no `-s`) |
|
||||||
|
| `-m` | Multithreading mode |
|
||||||
| `-s` | Chunk serialization (batch all files per chunk) |
|
| `-s` | Chunk serialization (batch all files per chunk) |
|
||||||
| `-f` | Sendfile zero-copy. Incompatible with `-c` / `-s`. |
|
| `-f` | Sendfile zero-copy. Incompatible with `-c` / `-s`. TCP only. |
|
||||||
| `-M, --preserve` | Preserve file metadata (mode, uid, gid, mtime) |
|
| `-M, --preserve` | Preserve file metadata (mode, uid, gid, mtime) |
|
||||||
|
| `-n, --dry-run` | Scan and print what would be transferred |
|
||||||
|
| `-p <port>` | SSH port (default: 22) |
|
||||||
|
| `--progress` | Show real-time transfer speed |
|
||||||
|
| `--delete` | Delete files on receiver not present in source |
|
||||||
|
| `--exclude <pattern>` | Exclude files matching glob pattern (repeatable) |
|
||||||
|
| `--chunk-size <n>` | Chunk size in bytes (default: 10485760) |
|
||||||
| `--source-dir <path>` | Source directory (overrides `FASTSYNC_SOURCE_DIR`) |
|
| `--source-dir <path>` | Source directory (overrides `FASTSYNC_SOURCE_DIR`) |
|
||||||
| `--dest-dir <path>` | Server destination directory (overrides `FASTSYNC_DEST_DIR`) |
|
| `--dest-dir <path>` | Server destination directory (overrides `FASTSYNC_DEST_DIR`) |
|
||||||
| `--save-to-disk` | Write received files to disk |
|
| `--save-to-disk` | Write received files to disk |
|
||||||
|
| `--server-host <ip>` | Server IP address (default: `127.0.0.1`) |
|
||||||
|
| `--server-port <n>` | Server port (default: `8080`) |
|
||||||
|
| `-v, --verbose` | Enable debug logging |
|
||||||
|
|
||||||
|
## Environment Variables
|
||||||
|
|
||||||
### Environment Variables
|
|
||||||
| Variable | Default | Description |
|
| Variable | Default | Description |
|
||||||
|----------|---------|-------------|
|
|----------|---------|-------------|
|
||||||
| `FASTSYNC_SOURCE_DIR` | User documents | Source directory fallback |
|
| `FASTSYNC_SOURCE_DIR` | — | Source directory fallback |
|
||||||
| `FASTSYNC_DEST_DIR` | `./data_copied` | Destination directory fallback |
|
| `FASTSYNC_DEST_DIR` | — | Destination directory fallback |
|
||||||
| `FASTSYNC_SERVER_IP` | `127.0.0.1` | Server address |
|
|
||||||
| `FASTSYNC_SERVER_PORT` | `8080` | Server port |
|
|
||||||
| `FASTSYNC_SAVE_TO_DISK` | `false` | Disk persistence fallback |
|
| `FASTSYNC_SAVE_TO_DISK` | `false` | Disk persistence fallback |
|
||||||
|
|
||||||
## Implementation Details
|
## Implementation Details
|
||||||
|
|
||||||
### Data Structures
|
### Data Structures
|
||||||
1. **Chunk** — collection of files (~10 MB total)
|
1. **Chunk** — collection of files (~10 MB total by default)
|
||||||
2. **File** — path, content (`Data`), optional `FileMetadata` pointer
|
2. **File** — path, content (`Data`), optional `FileMetadata` pointer
|
||||||
3. **FileMetadata** — `mode`, `uid`, `gid`, `mtime_sec`, `mtime_nsec`
|
3. **FileMetadata** — `mode`, `uid`, `gid`, `mtime_sec`, `mtime_nsec`
|
||||||
4. **Config** — runtime parameters
|
4. **Config** — runtime parameters (transported over wire)
|
||||||
5. **Queue** — thread-safe queue with condition variables
|
5. **Queue** — thread-safe bounded queue with condition variables
|
||||||
|
6. **DirectoryScanner** — recursive BFS traversal with exclude pattern support
|
||||||
|
|
||||||
### Key Algorithms
|
### Key Algorithms
|
||||||
1. **File scanning** — recursive BFS directory traversal
|
1. **File scanning** — BFS directory traversal; each entry matched against exclude patterns
|
||||||
2. **Chunking** — files grouped by size limit
|
2. **Chunking** — files accumulated until `chunk_size` threshold, then flushed
|
||||||
3. **Compression** — zstd with configurable level
|
3. **Compression** — streaming zstd via `ZSTD_compressStream2` / `ZSTD_decompressStream`
|
||||||
4. **Network protocol** — custom TCP with status codes and optional metadata packing
|
4. **Network protocol** — status-code-driven exchange with metadata packing
|
||||||
5. **Metadata restoration** — `chmod()`, `chown()`, `utimensat()` on the receiving side
|
5. **Metadata restoration** — `chmod()`, `chown()`, `utimensat()` on the receiving side
|
||||||
|
6. **`--delete`** — sender tracks all sent paths; receiver walks destination tree and removes unlisted files/directories
|
||||||
|
7. **SSH transport** — `socketpair()` + `fork()` + `execvp("ssh", ...)` with `ControlMaster` and port support
|
||||||
|
|
||||||
## Build Requirements
|
## Build Requirements
|
||||||
|
|
||||||
- C11 compiler
|
- C11 compiler
|
||||||
- CMake 4.1+
|
- CMake 4.1+
|
||||||
- zstd library
|
- zstd library (≥ 1.4.0 for streaming API)
|
||||||
- pthreads
|
- pthreads
|
||||||
|
- SSH client (for SSH transport)
|
||||||
|
|
||||||
## Building
|
## Building
|
||||||
|
|
||||||
@@ -97,55 +121,67 @@ cmake -B build -S . && cmake --build build -j$(nproc)
|
|||||||
|
|
||||||
## Running
|
## Running
|
||||||
|
|
||||||
### Server
|
### Server (TCP mode)
|
||||||
```bash
|
```bash
|
||||||
./build/server
|
./build/server
|
||||||
```
|
```
|
||||||
|
|
||||||
### Client
|
### Client — SSH (rsync-style)
|
||||||
|
```bash
|
||||||
|
./build/client /path/to/send user@host:/path/to/receive
|
||||||
|
```
|
||||||
|
|
||||||
|
### Client — TCP
|
||||||
```bash
|
```bash
|
||||||
# Basic
|
|
||||||
./build/client --source-dir /path/to/send --dest-dir /path/to/receive --save-to-disk
|
./build/client --source-dir /path/to/send --dest-dir /path/to/receive --save-to-disk
|
||||||
|
```
|
||||||
|
|
||||||
# With metadata preservation
|
### Common Options
|
||||||
./build/client -M --source-dir ... --dest-dir ...
|
```bash
|
||||||
|
# Archive mode (compression + multithreading + metadata)
|
||||||
|
./build/client -a /path/to/send user@host:/path
|
||||||
|
|
||||||
# Multithreaded + compression
|
# Dry run
|
||||||
./build/client -m -c 10
|
./build/client -n /path/to/send /path/to/receive
|
||||||
|
|
||||||
# Sendfile (zero-copy)
|
# With progress and custom chunk size
|
||||||
./build/client -f
|
./build/client --progress --chunk-size 2097152 /src user@host:/dst
|
||||||
|
|
||||||
|
# Exclude temporary files + delete extras on receiver
|
||||||
|
./build/client --exclude "*.tmp" --exclude "*.o" --delete /src user@host:/dst
|
||||||
|
|
||||||
# All features
|
# All features
|
||||||
./build/client -m -c -s -M
|
./build/client -a --progress --chunk-size 5242880 --exclude "*.log" --delete /src /dst
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### Server via SSH
|
||||||
|
Place the `fastsync-server` binary in the remote `$PATH`. The client runs `ssh user@host fastsync-server --stdio` automatically when an SSH-style destination is given.
|
||||||
|
|
||||||
## Testing
|
## Testing
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
# Unit tests
|
# Unit tests (7 suites)
|
||||||
./build/tests
|
./build/tests
|
||||||
|
|
||||||
# Integration benchmark (~50 MB data, 13 configurations + rsync comparison)
|
# Integration + benchmark suite
|
||||||
python3 test.py
|
python3 test.py
|
||||||
|
|
||||||
# Profiles: --wan (100 Mbit, 50 ms, 1% loss), --unlimited (no throttling)
|
|
||||||
python3 test.py --wan
|
|
||||||
```
|
```
|
||||||
|
|
||||||
The benchmark prints throughput metrics for the best configuration and speedup vs rsync.
|
The benchmark prints throughput metrics, best configuration, and speedup vs rsync.
|
||||||
|
|
||||||
## Performance Considerations
|
## Performance Considerations
|
||||||
|
|
||||||
1. Chunk size (~10 MB) balances memory and transfer efficiency
|
1. Chunk size (~10 MB default) balances memory and transfer efficiency
|
||||||
2. Compression level trades CPU for bandwidth
|
2. Compression level trades CPU for bandwidth
|
||||||
3. `sendfile()` bypasses userspace — ~2× faster on localhost for large files
|
3. `sendfile()` bypasses userspace — ~2× faster on localhost for large files
|
||||||
4. Multithreading scales with core count
|
4. Multithreading scales with core count
|
||||||
5. Metadata transfer adds negligible overhead when disabled, ~24 bytes per file when enabled
|
5. Metadata transfer adds negligible overhead (~24 bytes per file when enabled)
|
||||||
|
6. SSH socketpair buffer set to 1 MB for improved pipe throughput
|
||||||
|
7. SSH ControlMaster reuses connections across repeated invocations
|
||||||
|
|
||||||
## Benchmark Results
|
## Benchmark Results
|
||||||
|
|
||||||
50 MB of mixed file sizes over `localhost` with disk I/O throttled (reads ≤ 15 MB/s, writes ≤ 10 MB/s) and network emulation via `tc netem`. Each test was run 3×; the median is reported below.
|
25 MB of mixed file sizes over `localhost` with disk I/O throttled (reads ≤ 15 MB/s, writes ≤ 10 MB/s) and network emulation via `tc netem`. Each test was run 3×; the median is reported below.
|
||||||
|
|
||||||
### LAN (1000 Mbit, 20 ms ±1 ms, 0.1% loss)
|
### LAN (1000 Mbit, 20 ms ±1 ms, 0.1% loss)
|
||||||
|
|
||||||
@@ -167,4 +203,4 @@ The benchmark prints throughput metrics for the best configuration and speedup v
|
|||||||
| rsync (archive) | 17.44 s | — | — |
|
| rsync (archive) | 17.44 s | — | — |
|
||||||
| rsync (archive + compress) | 1.47 s | — | — |
|
| rsync (archive + compress) | 1.47 s | — | — |
|
||||||
|
|
||||||
Compression reduces the data on the wire enough that the transfer becomes latency-bound rather than bandwidth-bound. On WAN, the best configuration runs 10.8× faster than the theoretical limit for uncompressed data, since zstd shrinks the 50 MB payload to a fraction of its original size over the wire.
|
Compression reduces the data on the wire enough that the transfer becomes latency-bound rather than bandwidth-bound. On WAN, the best configuration runs 10.8× faster than the theoretical limit for uncompressed data, since zstd shrinks the 25 MB payload to a fraction of its original size over the wire.
|
||||||
|
|||||||
@@ -22,7 +22,12 @@ static void print_usage(void) {
|
|||||||
printf("Options:\n");
|
printf("Options:\n");
|
||||||
printf(" -c [level] Enable compression (level 1-22, default 5)\n");
|
printf(" -c [level] Enable compression (level 1-22, default 5)\n");
|
||||||
printf(" -z [level] Alias for -c\n");
|
printf(" -z [level] Alias for -c\n");
|
||||||
|
printf(" -a, --archive Archive mode (-c -m -M)\n");
|
||||||
|
printf(" -n, --dry-run Show what would be transferred\n");
|
||||||
|
printf(" -p <port> SSH port (default: 22)\n");
|
||||||
printf(" --progress Show transfer progress\n");
|
printf(" --progress Show transfer progress\n");
|
||||||
|
printf(" --delete Delete files on receiver not in source\n");
|
||||||
|
printf(" --exclude <pattern> Exclude files matching pattern\n");
|
||||||
printf(" -m Enable multithreading\n");
|
printf(" -m Enable multithreading\n");
|
||||||
printf(" -s Enable chunk serialization\n");
|
printf(" -s Enable chunk serialization\n");
|
||||||
printf(" -f Enable sendfile (TCP only, not with -c or -s)\n");
|
printf(" -f Enable sendfile (TCP only, not with -c or -s)\n");
|
||||||
@@ -58,6 +63,21 @@ int main(int argc, char *argv[]) {
|
|||||||
if (strcmp(argv[i], "--help") == 0) {
|
if (strcmp(argv[i], "--help") == 0) {
|
||||||
print_usage();
|
print_usage();
|
||||||
return 0;
|
return 0;
|
||||||
|
} else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--archive") == 0) {
|
||||||
|
config->use_compression = true;
|
||||||
|
config->use_multithreading = true;
|
||||||
|
config->use_metadata = true;
|
||||||
|
log_message(LOG_LEVEL_INFO, "Enabled archive mode (-c -m -M)");
|
||||||
|
} else if (strcmp(argv[i], "-n") == 0 || strcmp(argv[i], "--dry-run") == 0) {
|
||||||
|
config->dry_run = true;
|
||||||
|
} else if (strcmp(argv[i], "-p") == 0 && i + 1 < argc) {
|
||||||
|
config->ssh_port = atoi(argv[++i]);
|
||||||
|
} else if (strcmp(argv[i], "--delete") == 0) {
|
||||||
|
config->use_delete = true;
|
||||||
|
} else if (strcmp(argv[i], "--exclude") == 0 && i + 1 < argc) {
|
||||||
|
int idx = config->exclude_count++;
|
||||||
|
config->exclude_patterns = realloc(config->exclude_patterns, config->exclude_count * sizeof(char *));
|
||||||
|
config->exclude_patterns[idx] = str_dup(argv[++i]);
|
||||||
} else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "-z") == 0) {
|
} else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "-z") == 0) {
|
||||||
config->use_compression = true;
|
config->use_compression = true;
|
||||||
log_message(LOG_LEVEL_INFO, "Enabled Compression");
|
log_message(LOG_LEVEL_INFO, "Enabled Compression");
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
#include "client_send.h"
|
#include "client_send.h"
|
||||||
|
#include "array_list.h"
|
||||||
#include "chunk.h"
|
#include "chunk.h"
|
||||||
#include "config.h"
|
#include "config.h"
|
||||||
#include "data.h"
|
#include "data.h"
|
||||||
@@ -52,7 +53,7 @@ static int send_chunks_multithreaded(void *pipeline_context) {
|
|||||||
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
|
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
client = client_connect_ssh(context->config->ssh_destination);
|
client = client_connect_ssh(context->config->ssh_destination, context->config->ssh_port);
|
||||||
} else {
|
} else {
|
||||||
client = client_create();
|
client = client_create();
|
||||||
client_connect(client, server_host, server_port);
|
client_connect(client, server_host, server_port);
|
||||||
@@ -65,6 +66,13 @@ static int send_chunks_multithreaded(void *pipeline_context) {
|
|||||||
&context->condition_not_empty_loader,
|
&context->condition_not_empty_loader,
|
||||||
&context->condition_not_full_loader, &context->loader_done);
|
&context->condition_not_full_loader, &context->loader_done);
|
||||||
if (current_chunk == NULL) {
|
if (current_chunk == NULL) {
|
||||||
|
if (context->config->use_delete) {
|
||||||
|
send_status(client->file_descriptor, STATUS_MANIFEST);
|
||||||
|
send_int(client->file_descriptor, context->manifest->size);
|
||||||
|
for (int i = 0; i < context->manifest->size; i++)
|
||||||
|
send_str(client->file_descriptor,
|
||||||
|
(char *)context->manifest->items[i]);
|
||||||
|
}
|
||||||
send_status(client->file_descriptor, STATUS_FINISHED);
|
send_status(client->file_descriptor, STATUS_FINISHED);
|
||||||
int ok = receive_status(client->file_descriptor) == STATUS_OK;
|
int ok = receive_status(client->file_descriptor) == STATUS_OK;
|
||||||
client_disconnect(client);
|
client_disconnect(client);
|
||||||
@@ -82,16 +90,28 @@ static int send_chunks_multithreaded(void *pipeline_context) {
|
|||||||
static int scan_directory_multithreaded(void *pipeline_context) {
|
static int scan_directory_multithreaded(void *pipeline_context) {
|
||||||
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
|
PipelineContextSender *context = (PipelineContextSender *)pipeline_context;
|
||||||
mtx_lock(&context->mutex_scanner);
|
mtx_lock(&context->mutex_scanner);
|
||||||
DirectoryScanner *scanner =
|
DirectoryScanner *scanner = directory_scanner_create(
|
||||||
directory_scanner_create(context->config->send_directory, context->config->use_metadata, context->config->chunk_size);
|
context->config->send_directory, context->config->use_metadata,
|
||||||
|
context->config->chunk_size, context->config->exclude_patterns,
|
||||||
|
context->config->exclude_count);
|
||||||
mtx_unlock(&context->mutex_scanner);
|
mtx_unlock(&context->mutex_scanner);
|
||||||
|
|
||||||
Chunk *current_chunk;
|
Chunk *current_chunk;
|
||||||
while ((current_chunk = directory_scanner_next(scanner)) != NULL)
|
while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
|
||||||
|
if (context->config->use_delete) {
|
||||||
|
mtx_lock(&context->mutex_scanner);
|
||||||
|
for (int i = 0; i < current_chunk->element_count; i++) {
|
||||||
|
const char *p = current_chunk->items[i]->path;
|
||||||
|
if (*p == '/') p++;
|
||||||
|
array_list_add(context->manifest, str_dup(p));
|
||||||
|
}
|
||||||
|
mtx_unlock(&context->mutex_scanner);
|
||||||
|
}
|
||||||
queue_enqueue_multithreaded(context->queue_scanner, current_chunk,
|
queue_enqueue_multithreaded(context->queue_scanner, current_chunk,
|
||||||
&context->mutex_scanner,
|
&context->mutex_scanner,
|
||||||
&context->condition_not_empty_scanner,
|
&context->condition_not_empty_scanner,
|
||||||
&context->condition_not_full_scanner);
|
&context->condition_not_full_scanner);
|
||||||
|
}
|
||||||
mtx_lock(&context->mutex_scanner);
|
mtx_lock(&context->mutex_scanner);
|
||||||
context->scanner_done = true;
|
context->scanner_done = true;
|
||||||
cnd_signal(&context->condition_not_empty_scanner);
|
cnd_signal(&context->condition_not_empty_scanner);
|
||||||
@@ -127,27 +147,59 @@ static int load_files_multithreaded(void *pipeline_context) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
int send_files(Config *config) {
|
int send_files(Config *config) {
|
||||||
|
if (config->dry_run) {
|
||||||
|
DirectoryScanner *scanner = directory_scanner_create(
|
||||||
|
config->send_directory, config->use_metadata, config->chunk_size,
|
||||||
|
config->exclude_patterns, config->exclude_count);
|
||||||
|
Chunk *chunk;
|
||||||
|
int file_count = 0;
|
||||||
|
unsigned long long total_bytes = 0;
|
||||||
|
printf("Dry run: files to be transferred\n");
|
||||||
|
while ((chunk = directory_scanner_next(scanner)) != NULL) {
|
||||||
|
for (int i = 0; i < chunk->element_count; i++) {
|
||||||
|
printf(" %s (%zu bytes)\n", chunk->items[i]->path,
|
||||||
|
chunk->items[i]->data->size);
|
||||||
|
total_bytes += chunk->items[i]->data->size;
|
||||||
|
file_count++;
|
||||||
|
}
|
||||||
|
chunk_destroy(chunk);
|
||||||
|
}
|
||||||
|
directory_scanner_destroy(scanner);
|
||||||
|
printf("Total: %d files, %.1f MB\n", file_count,
|
||||||
|
total_bytes / 1048576.0);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
Client *client;
|
Client *client;
|
||||||
if (config->transport == TRANSPORT_SSH) {
|
if (config->transport == TRANSPORT_SSH) {
|
||||||
if (config->use_sendfile) {
|
if (config->use_sendfile) {
|
||||||
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
|
fprintf(stderr, "Error: -f/--sendfile is not supported with SSH transport\n");
|
||||||
return 1;
|
return 1;
|
||||||
}
|
}
|
||||||
client = client_connect_ssh(config->ssh_destination);
|
client = client_connect_ssh(config->ssh_destination, config->ssh_port);
|
||||||
} else {
|
} else {
|
||||||
client = client_create();
|
client = client_create();
|
||||||
client_connect(client, server_host, server_port);
|
client_connect(client, server_host, server_port);
|
||||||
}
|
}
|
||||||
config_send(client->file_descriptor, config);
|
config_send(client->file_descriptor, config);
|
||||||
DirectoryScanner *scanner = directory_scanner_create(config->send_directory, config->use_metadata, config->chunk_size);
|
DirectoryScanner *scanner = directory_scanner_create(
|
||||||
|
config->send_directory, config->use_metadata, config->chunk_size,
|
||||||
|
config->exclude_patterns, config->exclude_count);
|
||||||
Chunk *current_chunk;
|
Chunk *current_chunk;
|
||||||
unsigned long long total_bytes = 0;
|
unsigned long long total_bytes = 0;
|
||||||
time_t last_progress = 0;
|
time_t last_progress = 0;
|
||||||
time_t start = time(NULL);
|
time_t start = time(NULL);
|
||||||
|
ArrayList *manifest = config->use_delete ? array_list_create(free) : NULL;
|
||||||
while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
|
while ((current_chunk = directory_scanner_next(scanner)) != NULL) {
|
||||||
unsigned long long chunk_bytes = 0;
|
unsigned long long chunk_bytes = 0;
|
||||||
for (int i = 0; i < current_chunk->element_count; i++)
|
for (int i = 0; i < current_chunk->element_count; i++) {
|
||||||
chunk_bytes += current_chunk->items[i]->data->size;
|
chunk_bytes += current_chunk->items[i]->data->size;
|
||||||
|
if (manifest) {
|
||||||
|
const char *p = current_chunk->items[i]->path;
|
||||||
|
if (*p == '/') p++;
|
||||||
|
array_list_add(manifest, str_dup(p));
|
||||||
|
}
|
||||||
|
}
|
||||||
if (!config->use_sendfile) {
|
if (!config->use_sendfile) {
|
||||||
for (int i = 0; i < current_chunk->element_count; i++)
|
for (int i = 0; i < current_chunk->element_count; i++)
|
||||||
file_load_data(current_chunk->items[i]);
|
file_load_data(current_chunk->items[i]);
|
||||||
@@ -166,6 +218,13 @@ int send_files(Config *config) {
|
|||||||
}
|
}
|
||||||
chunk_destroy(current_chunk);
|
chunk_destroy(current_chunk);
|
||||||
}
|
}
|
||||||
|
if (config->use_delete) {
|
||||||
|
send_status(client->file_descriptor, STATUS_MANIFEST);
|
||||||
|
send_int(client->file_descriptor, manifest->size);
|
||||||
|
for (int i = 0; i < manifest->size; i++)
|
||||||
|
send_str(client->file_descriptor, (char *)manifest->items[i]);
|
||||||
|
array_list_delete(manifest);
|
||||||
|
}
|
||||||
send_status(client->file_descriptor, STATUS_FINISHED);
|
send_status(client->file_descriptor, STATUS_FINISHED);
|
||||||
int ok = receive_status(client->file_descriptor) == STATUS_OK;
|
int ok = receive_status(client->file_descriptor) == STATUS_OK;
|
||||||
if (config->show_progress) {
|
if (config->show_progress) {
|
||||||
@@ -180,9 +239,34 @@ int send_files(Config *config) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
int send_files_multithreaded(Config *config) {
|
int send_files_multithreaded(Config *config) {
|
||||||
|
if (config->dry_run) {
|
||||||
|
DirectoryScanner *scanner = directory_scanner_create(
|
||||||
|
config->send_directory, config->use_metadata, config->chunk_size,
|
||||||
|
config->exclude_patterns, config->exclude_count);
|
||||||
|
Chunk *chunk;
|
||||||
|
int file_count = 0;
|
||||||
|
unsigned long long total_bytes = 0;
|
||||||
|
printf("Dry run: files to be transferred\n");
|
||||||
|
while ((chunk = directory_scanner_next(scanner)) != NULL) {
|
||||||
|
for (int i = 0; i < chunk->element_count; i++) {
|
||||||
|
printf(" %s (%zu bytes)\n", chunk->items[i]->path,
|
||||||
|
chunk->items[i]->data->size);
|
||||||
|
total_bytes += chunk->items[i]->data->size;
|
||||||
|
file_count++;
|
||||||
|
}
|
||||||
|
chunk_destroy(chunk);
|
||||||
|
}
|
||||||
|
directory_scanner_destroy(scanner);
|
||||||
|
printf("Total: %d files, %.1f MB\n", file_count,
|
||||||
|
total_bytes / 1048576.0);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
PipelineContextSender *context =
|
PipelineContextSender *context =
|
||||||
pipeline_context_sender_create(config, queue_create(100, chunk_destroy),
|
pipeline_context_sender_create(config, queue_create(100, chunk_destroy),
|
||||||
queue_create(100, chunk_destroy));
|
queue_create(100, chunk_destroy));
|
||||||
|
if (config->use_delete)
|
||||||
|
context->manifest = array_list_create(free);
|
||||||
|
|
||||||
thrd_t scanner, loader, sender;
|
thrd_t scanner, loader, sender;
|
||||||
if (thrd_create(&scanner, scan_directory_multithreaded, context) !=
|
if (thrd_create(&scanner, scan_directory_multithreaded, context) !=
|
||||||
|
|||||||
+14
-1
@@ -11,13 +11,15 @@
|
|||||||
#include <sys/stat.h>
|
#include <sys/stat.h>
|
||||||
#include <unistd.h>
|
#include <unistd.h>
|
||||||
|
|
||||||
DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metadata, unsigned long long chunk_size) {
|
DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metadata, unsigned long long chunk_size, char **exclude_patterns, int exclude_count) {
|
||||||
DirectoryScanner *scanner = malloc(sizeof(DirectoryScanner));
|
DirectoryScanner *scanner = malloc(sizeof(DirectoryScanner));
|
||||||
scanner->directories = queue_create(100, free);
|
scanner->directories = queue_create(100, free);
|
||||||
scanner->current_dir = NULL;
|
scanner->current_dir = NULL;
|
||||||
scanner->current_path = NULL;
|
scanner->current_path = NULL;
|
||||||
scanner->use_metadata = use_metadata;
|
scanner->use_metadata = use_metadata;
|
||||||
scanner->chunk_size = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE;
|
scanner->chunk_size = chunk_size > 0 ? chunk_size : DESIRED_CHUNK_SIZE;
|
||||||
|
scanner->exclude_patterns = exclude_patterns;
|
||||||
|
scanner->exclude_count = exclude_count;
|
||||||
queue_enqueue(scanner->directories, str_dup(root_directory));
|
queue_enqueue(scanner->directories, str_dup(root_directory));
|
||||||
return scanner;
|
return scanner;
|
||||||
}
|
}
|
||||||
@@ -94,6 +96,17 @@ Chunk *directory_scanner_next(DirectoryScanner *scanner) {
|
|||||||
if (S_ISDIR(stats.st_mode)) {
|
if (S_ISDIR(stats.st_mode)) {
|
||||||
queue_enqueue(scanner->directories, (void *)cur_path);
|
queue_enqueue(scanner->directories, (void *)cur_path);
|
||||||
} else {
|
} else {
|
||||||
|
bool excluded = false;
|
||||||
|
for (int i = 0; i < scanner->exclude_count; i++) {
|
||||||
|
if (glob_match(scanner->exclude_patterns[i], entry->d_name)) {
|
||||||
|
excluded = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (excluded) {
|
||||||
|
free(cur_path);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
File *file = file_create(cur_path);
|
File *file = file_create(cur_path);
|
||||||
file->data->size = stats.st_size;
|
file->data->size = stats.st_size;
|
||||||
if (scanner->use_metadata)
|
if (scanner->use_metadata)
|
||||||
|
|||||||
@@ -12,9 +12,11 @@ typedef struct {
|
|||||||
char *current_path;
|
char *current_path;
|
||||||
bool use_metadata;
|
bool use_metadata;
|
||||||
unsigned long long chunk_size;
|
unsigned long long chunk_size;
|
||||||
|
char **exclude_patterns;
|
||||||
|
int exclude_count;
|
||||||
} DirectoryScanner;
|
} DirectoryScanner;
|
||||||
|
|
||||||
DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metadata, unsigned long long chunk_size);
|
DirectoryScanner *directory_scanner_create(char *root_directory, bool use_metadata, unsigned long long chunk_size, char **exclude_patterns, int exclude_count);
|
||||||
Chunk *directory_scanner_next(DirectoryScanner *scanner);
|
Chunk *directory_scanner_next(DirectoryScanner *scanner);
|
||||||
void directory_scanner_destroy(DirectoryScanner *scanner);
|
void directory_scanner_destroy(DirectoryScanner *scanner);
|
||||||
|
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
#include "array_list.h"
|
||||||
#include "chunk.h"
|
#include "chunk.h"
|
||||||
#include "compression.h"
|
#include "compression.h"
|
||||||
#include "config.h"
|
#include "config.h"
|
||||||
@@ -55,6 +56,16 @@ int receive_files(Config *config, int file_descriptor) {
|
|||||||
}
|
}
|
||||||
status = receive_status(file_descriptor);
|
status = receive_status(file_descriptor);
|
||||||
}
|
}
|
||||||
|
if (status == STATUS_MANIFEST) {
|
||||||
|
int count = receive_int(file_descriptor);
|
||||||
|
ArrayList *manifest = array_list_create(free);
|
||||||
|
for (int i = 0; i < count; i++)
|
||||||
|
array_list_add(manifest, receive_str(file_descriptor));
|
||||||
|
fprintf(stderr, "Deleting files not in manifest...\n");
|
||||||
|
delete_extras(config->receive_root_directory, manifest);
|
||||||
|
array_list_delete(manifest);
|
||||||
|
status = receive_status(file_descriptor);
|
||||||
|
}
|
||||||
if (status != STATUS_FINISHED) {
|
if (status != STATUS_FINISHED) {
|
||||||
log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
|
log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
|
||||||
send_status(file_descriptor, STATUS_ERROR);
|
send_status(file_descriptor, STATUS_ERROR);
|
||||||
|
|||||||
@@ -23,11 +23,16 @@ Config *config_create(char *version, char *send_directory,
|
|||||||
config->use_compression = use_compression;
|
config->use_compression = use_compression;
|
||||||
config->use_metadata = use_metadata;
|
config->use_metadata = use_metadata;
|
||||||
config->show_progress = false;
|
config->show_progress = false;
|
||||||
|
config->dry_run = false;
|
||||||
|
config->use_delete = false;
|
||||||
config->compression_level = compression_level;
|
config->compression_level = compression_level;
|
||||||
config->use_sendfile = use_sendfile;
|
config->use_sendfile = use_sendfile;
|
||||||
config->chunk_size = chunk_size > 0 ? chunk_size : DEFAULT_CHUNK_SIZE;
|
config->chunk_size = chunk_size > 0 ? chunk_size : DEFAULT_CHUNK_SIZE;
|
||||||
|
config->ssh_port = 22;
|
||||||
config->transport = TRANSPORT_TCP;
|
config->transport = TRANSPORT_TCP;
|
||||||
config->ssh_destination = NULL;
|
config->ssh_destination = NULL;
|
||||||
|
config->exclude_patterns = NULL;
|
||||||
|
config->exclude_count = 0;
|
||||||
return config;
|
return config;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -57,6 +62,9 @@ void config_delete(Config *config) {
|
|||||||
free(config->send_directory);
|
free(config->send_directory);
|
||||||
free(config->receive_root_directory);
|
free(config->receive_root_directory);
|
||||||
free(config->ssh_destination);
|
free(config->ssh_destination);
|
||||||
|
for (int i = 0; i < config->exclude_count; i++)
|
||||||
|
free(config->exclude_patterns[i]);
|
||||||
|
free(config->exclude_patterns);
|
||||||
free(config);
|
free(config);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -72,6 +80,7 @@ void config_send(int file_descriptor, Config *config) {
|
|||||||
send_int(file_descriptor, config->compression_level);
|
send_int(file_descriptor, config->compression_level);
|
||||||
send_int(file_descriptor, (int)config->chunk_size);
|
send_int(file_descriptor, (int)config->chunk_size);
|
||||||
send_int(file_descriptor, config->use_sendfile);
|
send_int(file_descriptor, config->use_sendfile);
|
||||||
|
send_int(file_descriptor, config->use_delete);
|
||||||
if (receive_status(file_descriptor) != STATUS_OK) {
|
if (receive_status(file_descriptor) != STATUS_OK) {
|
||||||
perror("Error transmitting config!");
|
perror("Error transmitting config!");
|
||||||
exit(EXIT_FAILURE);
|
exit(EXIT_FAILURE);
|
||||||
@@ -91,8 +100,14 @@ Config *config_receive(int file_descriptor) {
|
|||||||
config->compression_level = receive_int(file_descriptor);
|
config->compression_level = receive_int(file_descriptor);
|
||||||
config->chunk_size = (unsigned long long)receive_int(file_descriptor);
|
config->chunk_size = (unsigned long long)receive_int(file_descriptor);
|
||||||
config->use_sendfile = receive_int(file_descriptor);
|
config->use_sendfile = receive_int(file_descriptor);
|
||||||
|
config->use_delete = receive_int(file_descriptor);
|
||||||
|
config->show_progress = false;
|
||||||
|
config->dry_run = false;
|
||||||
|
config->ssh_port = 22;
|
||||||
config->transport = TRANSPORT_TCP;
|
config->transport = TRANSPORT_TCP;
|
||||||
config->ssh_destination = NULL;
|
config->ssh_destination = NULL;
|
||||||
|
config->exclude_patterns = NULL;
|
||||||
|
config->exclude_count = 0;
|
||||||
send_status(file_descriptor, STATUS_OK);
|
send_status(file_descriptor, STATUS_OK);
|
||||||
return config;
|
return config;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,10 +19,15 @@ typedef struct Config {
|
|||||||
bool use_sendfile;
|
bool use_sendfile;
|
||||||
bool use_metadata;
|
bool use_metadata;
|
||||||
bool show_progress;
|
bool show_progress;
|
||||||
|
bool dry_run;
|
||||||
|
bool use_delete;
|
||||||
int compression_level;
|
int compression_level;
|
||||||
unsigned long long chunk_size;
|
unsigned long long chunk_size;
|
||||||
|
int ssh_port;
|
||||||
TransportType transport;
|
TransportType transport;
|
||||||
char *ssh_destination;
|
char *ssh_destination;
|
||||||
|
char **exclude_patterns;
|
||||||
|
int exclude_count;
|
||||||
} Config;
|
} Config;
|
||||||
|
|
||||||
#define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
|
#define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
#include "multiprocessing.h"
|
#include "multiprocessing.h"
|
||||||
|
#include "array_list.h"
|
||||||
#include "chunk.h"
|
#include "chunk.h"
|
||||||
#include "compression.h"
|
#include "compression.h"
|
||||||
#include "config.h"
|
#include "config.h"
|
||||||
@@ -36,6 +37,9 @@ PipelineContextSender *pipeline_context_sender_create(Config *config,
|
|||||||
}
|
}
|
||||||
|
|
||||||
void pipeline_context_sender_destroy(PipelineContextSender *context) {
|
void pipeline_context_sender_destroy(PipelineContextSender *context) {
|
||||||
|
if (context->manifest) {
|
||||||
|
array_list_delete(context->manifest);
|
||||||
|
}
|
||||||
config_delete(context->config);
|
config_delete(context->config);
|
||||||
queue_destroy(context->queue_scanner);
|
queue_destroy(context->queue_scanner);
|
||||||
queue_destroy(context->queue_loader);
|
queue_destroy(context->queue_loader);
|
||||||
@@ -119,6 +123,17 @@ int receive_thread(void *pipeline_context) {
|
|||||||
}
|
}
|
||||||
status = receive_status(file_descriptor);
|
status = receive_status(file_descriptor);
|
||||||
}
|
}
|
||||||
|
if (status == STATUS_MANIFEST) {
|
||||||
|
int count = receive_int(file_descriptor);
|
||||||
|
ArrayList *manifest = array_list_create(free);
|
||||||
|
for (int i = 0; i < count; i++)
|
||||||
|
array_list_add(manifest, receive_str(file_descriptor));
|
||||||
|
delete_extras(context->config->receive_root_directory, manifest);
|
||||||
|
for (int i = 0; i < manifest->size; i++)
|
||||||
|
free(manifest->items[i]);
|
||||||
|
array_list_delete(manifest);
|
||||||
|
status = receive_status(file_descriptor);
|
||||||
|
}
|
||||||
mtx_lock(&context->mutex);
|
mtx_lock(&context->mutex);
|
||||||
context->receiver_done = true;
|
context->receiver_done = true;
|
||||||
cnd_signal(&context->condition_not_empty);
|
cnd_signal(&context->condition_not_empty);
|
||||||
|
|||||||
@@ -3,6 +3,7 @@
|
|||||||
|
|
||||||
#include <threads.h>
|
#include <threads.h>
|
||||||
|
|
||||||
|
#include "array_list.h"
|
||||||
#include "config.h"
|
#include "config.h"
|
||||||
#include "file.h"
|
#include "file.h"
|
||||||
#include "queue.h"
|
#include "queue.h"
|
||||||
@@ -19,6 +20,7 @@ typedef struct {
|
|||||||
cnd_t condition_not_full_loader;
|
cnd_t condition_not_full_loader;
|
||||||
cnd_t condition_not_empty_loader;
|
cnd_t condition_not_empty_loader;
|
||||||
bool loader_done;
|
bool loader_done;
|
||||||
|
ArrayList *manifest;
|
||||||
} PipelineContextSender;
|
} PipelineContextSender;
|
||||||
|
|
||||||
typedef struct PipelineContextReceiver {
|
typedef struct PipelineContextReceiver {
|
||||||
|
|||||||
@@ -5,7 +5,7 @@
|
|||||||
#include <stddef.h>
|
#include <stddef.h>
|
||||||
|
|
||||||
typedef int Status;
|
typedef int Status;
|
||||||
enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK };
|
enum NET_STATUS { STATUS_OK, STATUS_ERROR, STATUS_FINISHED, STATUS_NEXT, STATUS_CHUNK, STATUS_MANIFEST };
|
||||||
|
|
||||||
void io_set_fds(int read_fd, int write_fd);
|
void io_set_fds(int read_fd, int write_fd);
|
||||||
void send_n_data(int file_descriptor, void *data, size_t data_size);
|
void send_n_data(int file_descriptor, void *data, size_t data_size);
|
||||||
|
|||||||
@@ -42,7 +42,7 @@ static int parse_remote_dest(const char *dest, RemoteDest *r) {
|
|||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
Client *client_connect_ssh(char *destination) {
|
Client *client_connect_ssh(char *destination, int port) {
|
||||||
RemoteDest r;
|
RemoteDest r;
|
||||||
if (parse_remote_dest(destination, &r) != 0) {
|
if (parse_remote_dest(destination, &r) != 0) {
|
||||||
fprintf(stderr, "Invalid remote destination: %s\n", destination);
|
fprintf(stderr, "Invalid remote destination: %s\n", destination);
|
||||||
@@ -90,10 +90,26 @@ Client *client_connect_ssh(char *destination) {
|
|||||||
else
|
else
|
||||||
snprintf(ssh_user, sizeof(ssh_user), "%s", r.host);
|
snprintf(ssh_user, sizeof(ssh_user), "%s", r.host);
|
||||||
|
|
||||||
execlp("ssh", "ssh", "-o", "Compression=no", "-o",
|
char *ssh_argv[16];
|
||||||
"ControlMaster=auto", "-o",
|
int ac = 0;
|
||||||
"ControlPath=~/.cache/fastsync-%r@%h:%p", ssh_user,
|
char port_str[16];
|
||||||
"fastsync-server", "--stdio", (char *)NULL);
|
ssh_argv[ac++] = "ssh";
|
||||||
|
ssh_argv[ac++] = "-o";
|
||||||
|
ssh_argv[ac++] = "Compression=no";
|
||||||
|
ssh_argv[ac++] = "-o";
|
||||||
|
ssh_argv[ac++] = "ControlMaster=auto";
|
||||||
|
ssh_argv[ac++] = "-o";
|
||||||
|
ssh_argv[ac++] = "ControlPath=~/.cache/fastsync-%r@%h:%p";
|
||||||
|
if (port > 0 && port != 22) {
|
||||||
|
ssh_argv[ac++] = "-p";
|
||||||
|
snprintf(port_str, sizeof(port_str), "%d", port);
|
||||||
|
ssh_argv[ac++] = port_str;
|
||||||
|
}
|
||||||
|
ssh_argv[ac++] = ssh_user;
|
||||||
|
ssh_argv[ac++] = "fastsync-server";
|
||||||
|
ssh_argv[ac++] = "--stdio";
|
||||||
|
ssh_argv[ac] = NULL;
|
||||||
|
execvp("ssh", ssh_argv);
|
||||||
perror("exec of ssh failed");
|
perror("exec of ssh failed");
|
||||||
ssize_t wret = write(exec_pipe[1], "x", 1);
|
ssize_t wret = write(exec_pipe[1], "x", 1);
|
||||||
(void)wret;
|
(void)wret;
|
||||||
|
|||||||
@@ -3,6 +3,6 @@
|
|||||||
|
|
||||||
#include "transport_tcp.h"
|
#include "transport_tcp.h"
|
||||||
|
|
||||||
Client *client_connect_ssh(char *destination);
|
Client *client_connect_ssh(char *destination, int port);
|
||||||
|
|
||||||
#endif
|
#endif
|
||||||
|
|||||||
@@ -1,9 +1,12 @@
|
|||||||
#include "utils.h"
|
#include "utils.h"
|
||||||
|
#include "array_list.h"
|
||||||
#include "libgen.h"
|
#include "libgen.h"
|
||||||
|
#include <dirent.h>
|
||||||
#include <stdio.h>
|
#include <stdio.h>
|
||||||
#include <stdlib.h>
|
#include <stdlib.h>
|
||||||
#include <string.h>
|
#include <string.h>
|
||||||
#include <sys/stat.h>
|
#include <sys/stat.h>
|
||||||
|
#include <unistd.h>
|
||||||
|
|
||||||
void mkdir_r(char *path) {
|
void mkdir_r(char *path) {
|
||||||
char *path_duplicate = malloc(strlen(path) + 1);
|
char *path_duplicate = malloc(strlen(path) + 1);
|
||||||
@@ -44,6 +47,75 @@ char *str_dup(const char *string) {
|
|||||||
return new_string;
|
return new_string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
bool glob_match(const char *pattern, const char *str) {
|
||||||
|
while (*pattern) {
|
||||||
|
if (*pattern == '*') {
|
||||||
|
pattern++;
|
||||||
|
while (*str && *str != '/') {
|
||||||
|
if (glob_match(pattern, str))
|
||||||
|
return true;
|
||||||
|
str++;
|
||||||
|
}
|
||||||
|
return glob_match(pattern, str);
|
||||||
|
} else if (*pattern == '?') {
|
||||||
|
if (!*str || *str == '/')
|
||||||
|
return false;
|
||||||
|
pattern++;
|
||||||
|
str++;
|
||||||
|
} else {
|
||||||
|
if (*pattern != *str)
|
||||||
|
return false;
|
||||||
|
pattern++;
|
||||||
|
str++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return *str == '\0';
|
||||||
|
}
|
||||||
|
|
||||||
|
static void delete_extras_walk(const char *abs_path, const char *rel_path,
|
||||||
|
ArrayList *manifest) {
|
||||||
|
DIR *dir = opendir(abs_path);
|
||||||
|
if (!dir)
|
||||||
|
return;
|
||||||
|
struct dirent *entry;
|
||||||
|
while ((entry = readdir(dir)) != NULL) {
|
||||||
|
if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
|
||||||
|
continue;
|
||||||
|
char *child_abs = path_cat((char *)abs_path, entry->d_name);
|
||||||
|
char *child_rel = path_cat((char *)rel_path, entry->d_name);
|
||||||
|
struct stat st;
|
||||||
|
if (stat(child_abs, &st) != 0) {
|
||||||
|
free(child_abs);
|
||||||
|
free(child_rel);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if (S_ISDIR(st.st_mode)) {
|
||||||
|
delete_extras_walk(child_abs, child_rel, manifest);
|
||||||
|
} else {
|
||||||
|
// Check if relative path is in manifest
|
||||||
|
bool found = false;
|
||||||
|
for (int i = 0; i < manifest->size; i++) {
|
||||||
|
if (strcmp((char *)manifest->items[i], child_rel) == 0) {
|
||||||
|
found = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!found) {
|
||||||
|
unlink(child_abs);
|
||||||
|
fprintf(stderr, " Deleted: %s\n", child_rel);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
free(child_abs);
|
||||||
|
free(child_rel);
|
||||||
|
}
|
||||||
|
closedir(dir);
|
||||||
|
rmdir(abs_path);
|
||||||
|
}
|
||||||
|
|
||||||
|
void delete_extras(const char *dest_root, ArrayList *manifest) {
|
||||||
|
delete_extras_walk(dest_root, "", manifest);
|
||||||
|
}
|
||||||
|
|
||||||
char *path_cat(char *path1, char *path2) {
|
char *path_cat(char *path1, char *path2) {
|
||||||
if (path1 == NULL || *path1 == '\0')
|
if (path1 == NULL || *path1 == '\0')
|
||||||
return str_dup(path2);
|
return str_dup(path2);
|
||||||
|
|||||||
@@ -1,8 +1,13 @@
|
|||||||
#ifndef UTILS_H
|
#ifndef UTILS_H
|
||||||
#define UTILS_H
|
#define UTILS_H
|
||||||
|
|
||||||
|
#include "array_list.h"
|
||||||
|
#include <stdbool.h>
|
||||||
|
|
||||||
void mkdir_r(char *path);
|
void mkdir_r(char *path);
|
||||||
char *str_dup(const char *string);
|
char *str_dup(const char *string);
|
||||||
char *path_cat(char *path1, char *path2);
|
char *path_cat(char *path1, char *path2);
|
||||||
|
bool glob_match(const char *pattern, const char *str);
|
||||||
|
void delete_extras(const char *dest_root, ArrayList *manifest);
|
||||||
|
|
||||||
#endif
|
#endif
|
||||||
|
|||||||
@@ -195,7 +195,7 @@ def start_rsync_daemon(source_dir):
|
|||||||
return port, conf, daemon
|
return port, conf, daemon
|
||||||
|
|
||||||
|
|
||||||
def run_single_test(cmd, name, source_dir, dest_dir, *, source_prefix=None, no_server=False):
|
def run_single_test(cmd, name, source_dir, dest_dir, *, source_prefix=None, no_server=False, expected_missing=None):
|
||||||
if os.path.exists(dest_dir):
|
if os.path.exists(dest_dir):
|
||||||
shutil.rmtree(dest_dir)
|
shutil.rmtree(dest_dir)
|
||||||
if no_server:
|
if no_server:
|
||||||
@@ -216,6 +216,8 @@ def run_single_test(cmd, name, source_dir, dest_dir, *, source_prefix=None, no_s
|
|||||||
received = os.path.join(dest_dir, source_prefix if source_prefix is not None
|
received = os.path.join(dest_dir, source_prefix if source_prefix is not None
|
||||||
else os.path.abspath(source_dir).lstrip(os.sep))
|
else os.path.abspath(source_dir).lstrip(os.sep))
|
||||||
mismatches, missing = verify_transfer(source_dir, received)
|
mismatches, missing = verify_transfer(source_dir, received)
|
||||||
|
if expected_missing:
|
||||||
|
missing = [m for m in missing if m not in expected_missing]
|
||||||
|
|
||||||
first_line = lambda s: (s or "").strip().split("\n")[0]
|
first_line = lambda s: (s or "").strip().split("\n")[0]
|
||||||
entry = {
|
entry = {
|
||||||
@@ -318,6 +320,152 @@ def run_profile(profile_name, source_dir, dest_dir):
|
|||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
# Feature-specific tests for rsync-compatible flags
|
||||||
|
print("\n " + "─" * 56 + "\n Feature Tests\n " + "─" * 56)
|
||||||
|
|
||||||
|
# Dry run (-n) — no server needed
|
||||||
|
print("\n --- Dry run (-n) ---")
|
||||||
|
flags = BASE_CLIENT_FLAGS + ["-n"]
|
||||||
|
cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags
|
||||||
|
print(f" Running: {' '.join(cmd)}")
|
||||||
|
try:
|
||||||
|
start = time.monotonic()
|
||||||
|
result = subprocess.run(cmd, text=True, capture_output=True)
|
||||||
|
duration = time.monotonic() - start
|
||||||
|
r = {"name": "Dry run (-n)", "suite": profile_name}
|
||||||
|
if result.returncode == 0 and "Dry run:" in result.stdout:
|
||||||
|
r["status"] = "Success"
|
||||||
|
r["time"] = f"{duration:.4f}s"
|
||||||
|
r["error"] = ""
|
||||||
|
else:
|
||||||
|
r["status"] = "Failed"
|
||||||
|
r["time"] = "N/A"
|
||||||
|
r["error"] = f"Exit {result.returncode}: {(result.stderr or result.stdout)[:100]}"
|
||||||
|
results.append(r)
|
||||||
|
except Exception as e:
|
||||||
|
results.append({"name": "Dry run (-n)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
|
||||||
|
|
||||||
|
# Archive mode (-a)
|
||||||
|
feature_flags = BASE_CLIENT_FLAGS + ["-a"]
|
||||||
|
cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
|
||||||
|
print(f"\n --- Archive mode (-a) ---\n Running: {' '.join(cmd)}")
|
||||||
|
try:
|
||||||
|
r = run_single_test(cmd, "Archive mode (-a)", source_dir, dest_dir)
|
||||||
|
r["suite"] = profile_name
|
||||||
|
results.append(r)
|
||||||
|
except Exception as e:
|
||||||
|
results.append({"name": "Archive mode (-a)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
|
||||||
|
|
||||||
|
# Exclude (--exclude small.txt)
|
||||||
|
feature_flags = BASE_CLIENT_FLAGS + ["--exclude", "small.txt"]
|
||||||
|
cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
|
||||||
|
print(f"\n --- Exclude (--exclude small.txt) ---\n Running: {' '.join(cmd)}")
|
||||||
|
try:
|
||||||
|
r = run_single_test(cmd, "Exclude (--exclude small.txt)", source_dir, dest_dir,
|
||||||
|
expected_missing=["small.txt"])
|
||||||
|
r["suite"] = profile_name
|
||||||
|
results.append(r)
|
||||||
|
except Exception as e:
|
||||||
|
results.append({"name": "Exclude (--exclude small.txt)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
|
||||||
|
|
||||||
|
# Progress (--progress)
|
||||||
|
feature_flags = BASE_CLIENT_FLAGS + ["--progress"]
|
||||||
|
cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
|
||||||
|
print(f"\n --- Progress (--progress) ---\n Running: {' '.join(cmd)}")
|
||||||
|
try:
|
||||||
|
r = run_single_test(cmd, "Progress (--progress)", source_dir, dest_dir)
|
||||||
|
r["suite"] = profile_name
|
||||||
|
results.append(r)
|
||||||
|
except Exception as e:
|
||||||
|
results.append({"name": "Progress (--progress)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
|
||||||
|
|
||||||
|
# Chunk size (--chunk-size 5242880)
|
||||||
|
feature_flags = BASE_CLIENT_FLAGS + ["--chunk-size", "5242880"]
|
||||||
|
cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + feature_flags
|
||||||
|
print(f"\n --- Chunk size (--chunk-size 5242880) ---\n Running: {' '.join(cmd)}")
|
||||||
|
try:
|
||||||
|
r = run_single_test(cmd, "Chunk size (--chunk-size 5242880)", source_dir, dest_dir)
|
||||||
|
r["suite"] = profile_name
|
||||||
|
results.append(r)
|
||||||
|
except Exception as e:
|
||||||
|
results.append({"name": "Chunk size (--chunk-size 5242880)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
|
||||||
|
|
||||||
|
# Delete (--delete) — pre-populate dest, add extra files, then sync with --delete
|
||||||
|
# Note: server handles one client per launch, so we restart between syncs
|
||||||
|
print(f"\n --- Delete (--delete) ---")
|
||||||
|
try:
|
||||||
|
flags = BASE_CLIENT_FLAGS + ["-M"]
|
||||||
|
# First sync (no delete) to populate dest
|
||||||
|
s1 = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
|
||||||
|
time.sleep(0.5)
|
||||||
|
first_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags
|
||||||
|
r1 = subprocess.run(first_cmd, text=True, capture_output=True)
|
||||||
|
wait_proc(s1)
|
||||||
|
if r1.returncode != 0:
|
||||||
|
raise RuntimeError(f"First sync failed: {r1.stderr[:100]}")
|
||||||
|
# Add extra files to received dir
|
||||||
|
received = os.path.join(dest_dir, os.path.abspath(source_dir).lstrip(os.sep))
|
||||||
|
extra_path = os.path.join(received, "extra_file.txt")
|
||||||
|
with open(extra_path, "w") as f:
|
||||||
|
f.write("should be deleted")
|
||||||
|
extra_dir = os.path.join(received, "extra_dir")
|
||||||
|
os.makedirs(extra_dir, exist_ok=True)
|
||||||
|
with open(os.path.join(extra_dir, "nested.txt"), "w") as f:
|
||||||
|
f.write("nested extra")
|
||||||
|
# Second sync with --delete (fresh server)
|
||||||
|
s2 = subprocess.Popen(SERVER_CMD, stdout=subprocess.DEVNULL, stderr=None)
|
||||||
|
time.sleep(0.5)
|
||||||
|
second_cmd = client_prefix + BASE_CLIENT_CMD + ["--source-dir", source_dir, "--dest-dir", dest_dir] + flags + ["--delete"]
|
||||||
|
start = time.monotonic()
|
||||||
|
r2 = subprocess.run(second_cmd, text=True, capture_output=True)
|
||||||
|
duration = time.monotonic() - start
|
||||||
|
wait_proc(s2)
|
||||||
|
r = {"name": "Delete (--delete)", "suite": profile_name}
|
||||||
|
if r2.returncode == 0 and not os.path.exists(extra_path) and not os.path.exists(extra_dir):
|
||||||
|
mismatches, missing = verify_transfer(source_dir, received)
|
||||||
|
if not mismatches and not missing:
|
||||||
|
r["status"] = "Success"
|
||||||
|
r["time"] = f"{duration:.4f}s"
|
||||||
|
r["error"] = ""
|
||||||
|
else:
|
||||||
|
r["status"] = "Failed"
|
||||||
|
r["time"] = "N/A"
|
||||||
|
r["error"] = f"post-delete verify: mismatches={len(mismatches)}, missing={len(missing)}"
|
||||||
|
else:
|
||||||
|
r["status"] = "Failed"
|
||||||
|
r["time"] = "N/A"
|
||||||
|
errs = []
|
||||||
|
if r2.returncode != 0:
|
||||||
|
errs.append(f"Exit {r2.returncode}: {(r2.stderr or r2.stdout)[:60]}")
|
||||||
|
if os.path.exists(extra_path):
|
||||||
|
errs.append("extra_file.txt remains")
|
||||||
|
if os.path.exists(extra_dir):
|
||||||
|
errs.append("extra_dir remains")
|
||||||
|
r["error"] = " | ".join(errs)
|
||||||
|
results.append(r)
|
||||||
|
except Exception as e:
|
||||||
|
results.append({"name": "Delete (--delete)", "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
|
||||||
|
|
||||||
|
# SSH feature tests
|
||||||
|
if SSH_AVAILABLE:
|
||||||
|
ssh_dest = f"localhost:{dest_dir}_ssh"
|
||||||
|
ssh_feature_cases = [
|
||||||
|
{"name": "SSH Archive (-a)", "flags": ["-a"]},
|
||||||
|
{"name": "SSH Exclude (--exclude small.txt)", "flags": ["--exclude", "small.txt"], "expected_missing": ["small.txt"]},
|
||||||
|
]
|
||||||
|
for case in ssh_feature_cases:
|
||||||
|
flags = BASE_CLIENT_FLAGS + case["flags"]
|
||||||
|
cmd = BASE_CLIENT_CMD + [source_dir, ssh_dest] + flags
|
||||||
|
print(f"\n --- {case['name']} ---\n Running: {' '.join(cmd)}")
|
||||||
|
try:
|
||||||
|
r = run_single_test(cmd, case["name"], source_dir, f"{dest_dir}_ssh",
|
||||||
|
no_server=True,
|
||||||
|
expected_missing=case.get("expected_missing"))
|
||||||
|
r["suite"] = profile_name
|
||||||
|
results.append(r)
|
||||||
|
except Exception as e:
|
||||||
|
results.append({"name": case["name"], "suite": profile_name, "status": "Error", "time": "N/A", "error": str(e)})
|
||||||
|
|
||||||
except (subprocess.CalledProcessError, RuntimeError) as e:
|
except (subprocess.CalledProcessError, RuntimeError) as e:
|
||||||
print(f" Error: {e}")
|
print(f" Error: {e}")
|
||||||
results = []
|
results = []
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ static void test_scanner_single_file() {
|
|||||||
mkdir(dir, 0755);
|
mkdir(dir, 0755);
|
||||||
create_test_file(file1, content1);
|
create_test_file(file1, content1);
|
||||||
|
|
||||||
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0);
|
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0);
|
||||||
EXPECT_NOT_NULL(scanner);
|
EXPECT_NOT_NULL(scanner);
|
||||||
|
|
||||||
Chunk *chunk = directory_scanner_next(scanner);
|
Chunk *chunk = directory_scanner_next(scanner);
|
||||||
@@ -46,7 +46,7 @@ static void test_scanner_multiple_files() {
|
|||||||
create_test_file(file1, content1);
|
create_test_file(file1, content1);
|
||||||
create_test_file(file2, content2);
|
create_test_file(file2, content2);
|
||||||
|
|
||||||
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0);
|
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0);
|
||||||
EXPECT_NOT_NULL(scanner);
|
EXPECT_NOT_NULL(scanner);
|
||||||
|
|
||||||
Chunk *chunk = directory_scanner_next(scanner);
|
Chunk *chunk = directory_scanner_next(scanner);
|
||||||
@@ -83,7 +83,7 @@ static void test_scanner_subdirectory() {
|
|||||||
create_test_file(root_file, content);
|
create_test_file(root_file, content);
|
||||||
create_test_file(sub_file, content);
|
create_test_file(sub_file, content);
|
||||||
|
|
||||||
DirectoryScanner *scanner = directory_scanner_create((char *)root, false, 0);
|
DirectoryScanner *scanner = directory_scanner_create((char *)root, false, 0, NULL, 0);
|
||||||
EXPECT_NOT_NULL(scanner);
|
EXPECT_NOT_NULL(scanner);
|
||||||
|
|
||||||
int total_files = 0;
|
int total_files = 0;
|
||||||
@@ -106,7 +106,7 @@ static void test_scanner_empty_directory() {
|
|||||||
|
|
||||||
mkdir(dir, 0755);
|
mkdir(dir, 0755);
|
||||||
|
|
||||||
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0);
|
DirectoryScanner *scanner = directory_scanner_create((char *)dir, false, 0, NULL, 0);
|
||||||
EXPECT_NOT_NULL(scanner);
|
EXPECT_NOT_NULL(scanner);
|
||||||
|
|
||||||
Chunk *chunk = directory_scanner_next(scanner);
|
Chunk *chunk = directory_scanner_next(scanner);
|
||||||
|
|||||||
Reference in New Issue
Block a user