TapTap/FastSyncAi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement reliable UDP transfer with retransmission

Browse Source 
- Add seq_num and flags to udp_packet_header_t for reliability
- Add MAGIC_UDP_ACK packet type and udp_ack_t struct for ACKs
- Add UDP constants: WINDOW_SIZE=16, MAX_RETRIES=5, RETRANSMIT_MS=100
- Server: Send cumulative + selective ACKs during UDP transfer
- Client: Implement sliding window with retransmission
- Client: Add send_file_udp() and send_file_udp_single() functions
- Fix server make_path call to not create filename as directory
- UDP mode uses single-file protocol (not batch) for reliability

Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>
This commit is contained in:
taptap
2026-06-24 21:49:50 +02:00
parent 3c98ca5803
commit 948542394e
9 changed files with 576 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastsync_client
BIN
View File
Binary file not shown.
fastsync_server
BIN
View File
Binary file not shown.
src/client.c
+436 -23
View File
@@ -4,6 +4,7 @@
#include <dirent.h>
#include <stdatomic.h>
#include <time.h>
#include <poll.h>
#include <netinet/tcp.h>
#include <stdio.h>
@@ -61,6 +62,7 @@ static uint64_t fast_hash_file(const char *path, uint64_t file_size) {
typedef struct {
int tcp_fd;
int use_udp;
int use_raw;
int udp_fd;
struct sockaddr_in udp_server_addr;
uint64_t udp_session_id;
@@ -120,6 +122,391 @@ static int setup_udp(sync_ctx_t *ctx, const char *host) {
return 0;
}
// ── Helper: get current time in milliseconds ────────────────────────────────
static uint64_t get_current_time_ms(void) {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (uint64_t)ts.tv_sec * 1000 + (uint64_t)ts.tv_nsec / 1000000;
}
// ── Per-block tracking for retransmission ───────────────────────────────────
typedef struct {
uint32_t block_id;
uint64_t block_offset;
uint32_t payload_len;
char *data; // Allocated buffer with payload
uint32_t seq_num;
uint32_t retries;
uint64_t last_send_time_ms; // For timeout tracking
int acked;
} udp_block_t;
// ── Send a single file over UDP with reliability ───────────────────────────
static int send_file_udp(sync_ctx_t *ctx, const char *filepath, uint32_t file_id, uint64_t file_size) {
int fd = open(filepath, O_RDONLY);
if (fd < 0) {
perror("open file for UDP send");
return -1;
}
uint32_t total_blocks = (file_size + UDP_PAYLOAD_MAX - 1) / UDP_PAYLOAD_MAX;
// Allocate and initialize block tracking
udp_block_t *blocks = calloc(total_blocks, sizeof(udp_block_t));
if (!blocks) {
perror("malloc blocks");
close(fd);
return -1;
}
// Read file into blocks
for (uint32_t bid = 0; bid < total_blocks; bid++) {
uint64_t offset = (uint64_t)bid * UDP_PAYLOAD_MAX;
uint32_t to_read = (bid == total_blocks - 1)
? (uint32_t)(file_size - offset)
: UDP_PAYLOAD_MAX;
blocks[bid].block_id = bid;
blocks[bid].block_offset = offset;
blocks[bid].payload_len = to_read;
blocks[bid].seq_num = bid; // seq_num == block_id for simplicity
blocks[bid].retries = 0;
blocks[bid].acked = 0;
blocks[bid].data = malloc(to_read);
if (!blocks[bid].data) {
perror("malloc block data");
for (uint32_t j = 0; j < bid; j++) free(blocks[j].data);
free(blocks);
close(fd);
return -1;
}
if (lseek(fd, offset, SEEK_SET) < 0) {
perror("lseek");
for (uint32_t j = 0; j <= bid; j++) free(blocks[j].data);
free(blocks);
close(fd);
return -1;
}
ssize_t n = read(fd, blocks[bid].data, to_read);
if (n != (ssize_t)to_read) {
perror("read block");
for (uint32_t j = 0; j <= bid; j++) free(blocks[j].data);
free(blocks);
close(fd);
return -1;
}
}
close(fd);
// Send initial window of blocks
uint32_t window_base = 0;
uint32_t window_end = (total_blocks < UDP_WINDOW_SIZE) ? total_blocks : UDP_WINDOW_SIZE;
// Send all blocks in the window
for (uint32_t bid = window_base; bid < window_end; bid++) {
udp_packet_header_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.magic = MAGIC_UDP_DATA;
hdr.session_id = ctx->udp_session_id;
hdr.file_id = file_id;
hdr.block_id = blocks[bid].block_id;
hdr.block_offset = blocks[bid].block_offset;
hdr.payload_len = blocks[bid].payload_len;
hdr.seq_num = blocks[bid].seq_num;
hdr.flags = (bid == total_blocks - 1) ? UDP_FLAG_LAST_BLOCK : UDP_FLAG_NONE;
// Build packet: header + payload
char pkt[sizeof(udp_packet_header_t) + UDP_PAYLOAD_MAX];
memcpy(pkt, &hdr, sizeof(hdr));
memcpy(pkt + sizeof(hdr), blocks[bid].data, blocks[bid].payload_len);
ssize_t sent = sendto(ctx->udp_fd, pkt, sizeof(hdr) + blocks[bid].payload_len, 0,
(struct sockaddr *)&ctx->udp_server_addr,
sizeof(ctx->udp_server_addr));
if (sent < 0) {
perror("sendto UDP data");
// Continue trying
} else {
blocks[bid].last_send_time_ms = get_current_time_ms();
blocks[bid].retries = 1;
}
}
// Now poll for ACKs and retransmit as needed
struct pollfd fds[1];
fds[0].fd = ctx->tcp_fd;
fds[0].events = POLLIN;
int completed = 0;
uint64_t start_time_ms = get_current_time_ms();
while (completed < (int)total_blocks) {
// Check for timeout on un-ACKed blocks in window
uint64_t now_ms = get_current_time_ms();
uint32_t timeout_ms = UDP_RETRANSMIT_MS * (1 + (now_ms - start_time_ms) / 5000); // Exponential backoff
if (timeout_ms > 5000) timeout_ms = 5000; // Cap at 5 seconds
int has_timed_out = 0;
for (uint32_t bid = window_base; bid < window_end && bid < total_blocks; bid++) {
if (!blocks[bid].acked) {
if (now_ms - blocks[bid].last_send_time_ms > timeout_ms) {
if (blocks[bid].retries >= UDP_MAX_RETRIES) {
fprintf(stderr, "[UDP] Block %u exceeded max retries\n", bid);
// Give up on this block
blocks[bid].acked = 1; // Mark as "done" to avoid infinite loop
blocks[bid].retries = UDP_MAX_RETRIES + 1;
completed++;
continue;
}
// Retransmit
udp_packet_header_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.magic = MAGIC_UDP_DATA;
hdr.session_id = ctx->udp_session_id;
hdr.file_id = file_id;
hdr.block_id = blocks[bid].block_id;
hdr.block_offset = blocks[bid].block_offset;
hdr.payload_len = blocks[bid].payload_len;
hdr.seq_num = blocks[bid].seq_num;
hdr.flags = UDP_FLAG_RETRANSMIT |
((bid == total_blocks - 1) ? UDP_FLAG_LAST_BLOCK : UDP_FLAG_NONE);
char pkt[sizeof(udp_packet_header_t) + UDP_PAYLOAD_MAX];
memcpy(pkt, &hdr, sizeof(hdr));
memcpy(pkt + sizeof(hdr), blocks[bid].data, blocks[bid].payload_len);
sendto(ctx->udp_fd, pkt, sizeof(hdr) + blocks[bid].payload_len, 0,
(struct sockaddr *)&ctx->udp_server_addr,
sizeof(ctx->udp_server_addr));
blocks[bid].last_send_time_ms = now_ms;
blocks[bid].retries++;
has_timed_out = 1;
}
}
}
// Poll for ACKs
int poll_timeout_ms = has_timed_out ? 10 : 50; // Short poll if we retransmitted
int poll_ret = poll(fds, 1, poll_timeout_ms);
if (poll_ret > 0) {
// Read ACK from TCP - read entire packet at once
udp_ack_t ack_pkt;
ssize_t n = readn(ctx->tcp_fd, &ack_pkt, sizeof(ack_pkt));
if (n < 0) {
fprintf(stderr, "[UDP] Failed to read ACK packet (error)\n");
break;
}
if (n != sizeof(ack_pkt)) {
// Short read - server might have sent RESP_OK or closed connection
// This can happen if all blocks are done and server sent RESP_OK
if (completed >= (int)total_blocks) {
// Transfer is complete, don't treat as error
break;
}
break;
}
if (ack_pkt.magic == MAGIC_UDP_ACK) {
if (ack_pkt.session_id != ctx->udp_session_id || ack_pkt.file_id != file_id) {
// Not for us, put it back (can't really, just skip)
continue;
}
// Process cumulative ACK
uint32_t cumulative_ack = ack_pkt.ack_block_id;
// Mark all blocks up to cumulative_ack as ACKed
for (uint32_t bid = 0; bid <= cumulative_ack && bid < total_blocks; bid++) {
if (!blocks[bid].acked) {
blocks[bid].acked = 1;
completed++;
}
}
// Process selective ACK bitmap
for (int bmp_idx = 0; bmp_idx < 4; bmp_idx++) {
for (int bit = 0; bit < 32; bit++) {
uint32_t block_idx = bmp_idx * 32 + bit;
if (block_idx < total_blocks) {
if (ack_pkt.ack_bitmap[bmp_idx] & (1U << bit)) {
if (!blocks[block_idx].acked) {
blocks[block_idx].acked = 1;
completed++;
}
}
}
}
}
// Check if transfer is complete
if (completed >= (int)total_blocks) {
break; // All blocks ACKed, transfer complete
}
// Advance window: send new blocks that fit in window
// Sliding window: we can have up to UDP_WINDOW_SIZE unacked blocks
// Move window_base forward to first unacked block
while (window_base < total_blocks && blocks[window_base].acked) {
window_base++;
}
// Send new blocks to fill the window
while (window_end < total_blocks && (window_end - window_base) < UDP_WINDOW_SIZE) {
uint32_t bid = window_end;
udp_packet_header_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.magic = MAGIC_UDP_DATA;
hdr.session_id = ctx->udp_session_id;
hdr.file_id = file_id;
hdr.block_id = blocks[bid].block_id;
hdr.block_offset = blocks[bid].block_offset;
hdr.payload_len = blocks[bid].payload_len;
hdr.seq_num = blocks[bid].seq_num;
hdr.flags = (bid == total_blocks - 1) ? UDP_FLAG_LAST_BLOCK : UDP_FLAG_NONE;
char pkt[sizeof(udp_packet_header_t) + UDP_PAYLOAD_MAX];
memcpy(pkt, &hdr, sizeof(hdr));
memcpy(pkt + sizeof(hdr), blocks[bid].data, blocks[bid].payload_len);
ssize_t sent = sendto(ctx->udp_fd, pkt, sizeof(hdr) + blocks[bid].payload_len, 0,
(struct sockaddr *)&ctx->udp_server_addr,
sizeof(ctx->udp_server_addr));
if (sent < 0) {
perror("sendto UDP data");
} else {
blocks[bid].last_send_time_ms = now_ms;
blocks[bid].retries = 1;
window_end++;
}
}
} else {
// Some other TCP message - put it back
// We can't really put it back, so just break
fprintf(stderr, "[UDP] Unexpected TCP magic: 0x%08x during UDP transfer\n", ack_pkt.magic);
break;
}
} else if (poll_ret < 0 && errno != EINTR) {
perror("poll");
break;
}
// If poll_ret == 0, just continue (timeout)
}
// Clean up
for (uint32_t bid = 0; bid < total_blocks; bid++) {
free(blocks[bid].data);
}
free(blocks);
if (completed == (int)total_blocks) {
printf("[thread] UDP sent: %s (%lu bytes, %u blocks)\n", filepath, (unsigned long)file_size, total_blocks);
atomic_fetch_add(&g_bytes_sent, file_size);
atomic_fetch_add(&g_files_sent, 1);
return 0;
} else {
fprintf(stderr, "[thread] UDP send incomplete: %d/%u blocks for %s\n", completed, total_blocks, filepath);
return -1;
}
}
// ── Send a single file using UDP (metadata over TCP, data over UDP) ──────────
static int send_file_udp_single(sync_ctx_t *ctx, file_task_t *task) {
// Send file metadata over TCP (single-file mode, matching server's read pattern)
// Server expects: magic(4) + name_len(4) + file_size(8) + mode(4) + filename
uint32_t name_len = strlen(task->rel_path);
uint64_t file_size = task->file_size;
uint32_t mode = 0644; // Default permissions for UDP mode
// Send magic
uint32_t magic = MAGIC_META;
if (writen(ctx->tcp_fd, &magic, sizeof(magic)) != sizeof(magic)) {
perror("write magic");
return -1;
}
// Send name_len
if (writen(ctx->tcp_fd, &name_len, sizeof(name_len)) != sizeof(name_len)) {
perror("write name_len");
return -1;
}
// Send file_size
if (writen(ctx->tcp_fd, &file_size, sizeof(file_size)) != sizeof(file_size)) {
perror("write file_size");
return -1;
}
// Send mode
if (writen(ctx->tcp_fd, &mode, sizeof(mode)) != sizeof(mode)) {
perror("write mode");
return -1;
}
// Send filename
if (writen(ctx->tcp_fd, task->rel_path, name_len) != (ssize_t)name_len) {
perror("write filename");
return -1;
}
// Read server response
file_response_t resp;
if (readn(ctx->tcp_fd, &resp, sizeof(resp)) != sizeof(resp)) {
perror("read response");
return -1;
}
if (resp.response == RESP_SKIP) {
// File already exists on server with same checksum
printf("[thread] Skip (checksum match): %s\n", task->rel_path);
atomic_fetch_add(&g_files_sent, 1);
return 0;
}
if (resp.response == RESP_ERROR) {
fprintf(stderr, "[thread] Server error for %s\n", task->rel_path);
return -1;
}
if (resp.response != RESP_USE_UDP) {
fprintf(stderr, "[thread] Expected RESP_USE_UDP, got %u for %s\n", resp.response, task->rel_path);
return -1;
}
uint32_t file_id = resp.file_id;
// Now send file data over UDP with reliability
if (send_file_udp(ctx, task->full_path, file_id, task->file_size) < 0) {
fprintf(stderr, "[thread] UDP data send failed for %s\n", task->rel_path);
return -1;
}
// Read final ACK from server
uint32_t final_ack;
if (readn(ctx->tcp_fd, &final_ack, sizeof(final_ack)) != sizeof(final_ack)) {
perror("read final ack");
return -1;
}
if (final_ack != RESP_OK) {
fprintf(stderr, "[thread] Server returned error ack: %u\n", final_ack);
return -1;
}
return 0;
}
// ─── Batch file sending ────────────────────────────────────────────────────
static int send_batch_files(sync_ctx_t *ctx, file_task_t **tasks, int count) {
@@ -221,6 +608,7 @@ static void *worker_thread(void *arg) {
sync_ctx_t ctx;
memset(&ctx, 0, sizeof(ctx));
ctx.use_udp = wa->use_udp;
ctx.use_raw = wa->use_raw;
ctx.tcp_fd = socket(AF_INET, SOCK_STREAM, 0);
if (ctx.tcp_fd < 0) { perror("socket"); wa->result = -1; return NULL; }
@@ -244,36 +632,52 @@ static void *worker_thread(void *arg) {
if (connect(ctx.tcp_fd, (struct sockaddr *)&sa, sizeof(sa)) < 0) {
perror("connect"); close(ctx.tcp_fd); wa->result = -1; return NULL;
}
printf("[thread %d] TCP connected to %s:%d\n", wa->thread_id, wa->host, wa->port);
// Send mode (SYNC or RAW)
uint32_t mode = wa->use_raw ? MODE_RAW : MODE_SYNC;
if (writen(ctx.tcp_fd, &mode, sizeof(mode)) != sizeof(mode)) {
perror("write mode"); close(ctx.tcp_fd); wa->result = -1; return NULL; }
printf("[thread %d] TCP connected to %s:%d (mode=%s)\n", wa->thread_id, wa->host, wa->port, wa->use_raw ? "raw" : "sync");
if (wa->use_udp && setup_udp(&ctx, wa->host) < 0) {
close(ctx.tcp_fd); wa->result = -1; return NULL;
}
// Drain queue with batching
// Drain queue with batching (or single-file for UDP mode)
file_task_t *batch[BATCH_MAX_FILES];
int batch_count = 0;
file_task_t *task;
while ((task = wq_pop(wa->queue)) != NULL) {
batch[batch_count++] = task;
// If batch is full, send it
if (batch_count >= BATCH_MAX_FILES) {
if (send_batch_files(&ctx, batch, batch_count) < 0) {
fprintf(stderr, "[thread %d] Batch send failed\n", wa->thread_id);
if (ctx.use_udp) {
// UDP mode: send each file individually (single-file protocol)
if (send_file_udp_single(&ctx, task) < 0) {
fprintf(stderr, "[thread %d] UDP send failed for %s\n", wa->thread_id, task->rel_path);
wa->result = -1;
}
// Free all tasks in batch
for (int i = 0; i < batch_count; i++) {
free(batch[i]);
free(task);
} else {
// TCP mode: batch files
batch[batch_count++] = task;
// If batch is full, send it
if (batch_count >= BATCH_MAX_FILES) {
if (send_batch_files(&ctx, batch, batch_count) < 0) {
fprintf(stderr, "[thread %d] Batch send failed\n", wa->thread_id);
wa->result = -1;
}
// Free all tasks in batch
for (int i = 0; i < batch_count; i++) {
free(batch[i]);
}
batch_count = 0;
}
batch_count = 0;
}
}
// Send remaining files in partial batch
if (batch_count > 0) {
// Send remaining files in partial batch (TCP mode only)
if (!ctx.use_udp && batch_count > 0) {
if (send_batch_files(&ctx, batch, batch_count) < 0) {
fprintf(stderr, "[thread %d] Final batch send failed\n", wa->thread_id);
wa->result = -1;
@@ -300,6 +704,7 @@ static void *worker_thread(void *arg) {
typedef struct {
work_queue_t *queue;
char base_path[1024];
int use_raw;
} scanner_arg_t;
// Comparison function for sorting files by size (ascending: small files first)
@@ -312,7 +717,7 @@ static int compare_files_by_size(const void *a, const void *b) {
}
// Collect all files recursively into a list
static int collect_files(const char *base, const char *sub, file_task_t **list, int *count, int max_files) {
static int collect_files(const char *base, const char *sub, file_task_t **list, int *count, int max_files, int use_raw) {
char cur[2048];
if (sub && sub[0])
snprintf(cur, sizeof(cur), "%s/%s", base, sub);
@@ -340,7 +745,7 @@ static int collect_files(const char *base, const char *sub, file_task_t **list,
if (lstat(full, &st) < 0) { perror("lstat"); continue; }
if (S_ISDIR(st.st_mode)) {
if (collect_files(base, rel, list, count, max_files) < 0) {
if (collect_files(base, rel, list, count, max_files, use_raw) < 0) {
closedir(dir);
return -1;
}
@@ -355,7 +760,7 @@ static int collect_files(const char *base, const char *sub, file_task_t **list,
snprintf(list[*count]->full_path, sizeof(list[*count]->full_path), "%s", full);
snprintf(list[*count]->rel_path, sizeof(list[*count]->rel_path), "%s", rel);
list[*count]->file_size = st.st_size;
list[*count]->checksum = fast_hash_file(full, st.st_size);
list[*count]->checksum = use_raw ? 0 : fast_hash_file(full, st.st_size);
list[*count]->next = NULL;
(*count)++;
}
@@ -403,7 +808,7 @@ static void *scanner_thread(void *arg) {
if (!file_list) { perror("malloc"); wq_finish(sa->queue); return NULL; }
int count = 0;
if (collect_files(sa->base_path, "", file_list, &count, file_count) < 0) {
if (collect_files(sa->base_path, "", file_list, &count, file_count, sa->use_raw) < 0) {
for (int i = 0; i < count; i++) free(file_list[i]);
free(file_list);
wq_finish(sa->queue);
@@ -428,9 +833,11 @@ static void *scanner_thread(void *arg) {
static void print_usage(const char *prog) {
fprintf(stderr,
"Usage: %s -h <host> [-p <port>] -s <source> [-n <connections>] [-u]\n"
"Usage: %s -h <host> [-p <port>] -s <source> [-n <connections>] [-u] [-r]\n"
" -n number of parallel TCP connections (default: 4)\n"
" -u enable UDP data channel\n", prog);
" -u enable UDP data channel\n"
" -r raw mode: send ALL files, no checksum/skip logic\n"
" (default: sync mode - only send new/updated files)\n", prog);
}
int main(int argc, char *argv[]) {
@@ -439,14 +846,16 @@ int main(int argc, char *argv[]) {
char *source = NULL;
int nconn = 4;
int use_udp = 0;
int use_raw = 0; // Default: sync mode
int opt;
while ((opt = getopt(argc, argv, "h:p:s:n:u")) != -1) {
while ((opt = getopt(argc, argv, "h:p:s:n:ru")) != -1) {
switch (opt) {
case 'h': host = optarg; break;
case 'p': port = atoi(optarg); break;
case 's': source = optarg; break;
case 'n': nconn = atoi(optarg); break;
case 'r': use_raw = 1; break;
case 'u': use_udp = 1; break;
default: print_usage(argv[0]); exit(EXIT_FAILURE);
}
@@ -459,8 +868,9 @@ int main(int argc, char *argv[]) {
struct stat st;
if (stat(source, &st) < 0) { perror("stat source"); exit(EXIT_FAILURE); }
printf("Starting sync: %s → %s:%d (connections=%d, udp=%s)\n",
source, host, port, nconn, use_udp ? "yes" : "no");
printf("Starting %s: %s → %s:%d (connections=%d, udp=%s, mode=%s)\n",
use_raw ? "raw send" : "sync", source, host, port, nconn,
use_udp ? "yes" : "no", use_raw ? "raw" : "sync");
// ── Create temp dir for single files (to use directory scanning) ─────────
char temp_dir[2048] = "";
@@ -492,10 +902,12 @@ int main(int argc, char *argv[]) {
// ── Init shared work queue ────────────────────────────────────────────
work_queue_t queue;
wq_init(&queue);
queue.use_raw = use_raw; // Set mode for scanner
// ── Start scanner thread ──────────────────────────────────────────────
scanner_arg_t sarg;
sarg.queue = &queue;
sarg.use_raw = use_raw;
snprintf(sarg.base_path, sizeof(sarg.base_path), "%s", final_source);
pthread_t stid;
pthread_create(&stid, NULL, scanner_thread, &sarg);
@@ -516,6 +928,7 @@ int main(int argc, char *argv[]) {
args[i].host = host;
args[i].port = port;
args[i].use_udp = use_udp;
args[i].use_raw = use_raw;
args[i].result = 0;
pthread_create(&tids[i], NULL, worker_thread, &args[i]);
}
src/client.o
BIN
View File
Binary file not shown.
src/common.h
+29
View File
@@ -30,6 +30,7 @@ typedef struct {
file_task_t *tail;
int done; // set to 1 when scanner is finished
int error; // set to 1 on fatal scan error
int use_raw; // raw mode: skip checksum/skip logic
pthread_mutex_t lock;
pthread_cond_t cond;
} work_queue_t;
@@ -48,6 +49,7 @@ typedef struct {
const char *host;
int port;
int use_udp;
int use_raw; // 1 = raw mode (no checksum/skip)
int result; // 0 = ok, -1 = error
} worker_arg_t;
@@ -65,6 +67,10 @@ typedef struct {
#define BATCH_MAX_FILES 64 // Max files per batch
// Mode flags
#define MODE_SYNC 0x53594E43 // 'SYNC' - checksum-based sync (default)
#define MODE_RAW 0x524157 // 'RAW' - send all files, no checksum/skip
// Response Codes
#define RESP_SEND_DATA 1
#define RESP_SKIP 2
@@ -76,6 +82,13 @@ typedef struct {
#define UDP_PAYLOAD_MAX 1400
#define MAGIC_UDP_DATA 0x55445044 // 'UDPD'
#define MAGIC_UDP_KNOCK 0x5544504B // 'UDPK'
#define MAGIC_UDP_ACK 0x55445041 // 'UDPA'
// Reliable UDP settings
#define UDP_WINDOW_SIZE 16 // Sliding window: max unacked packets
#define UDP_MAX_RETRIES 5 // Max retransmit attempts per block
#define UDP_RETRANSMIT_MS 100 // Initial retransmit timeout in ms
#define UDP_ACK_INTERVAL 10 // Send ACK every N packets or on timer
#pragma pack(push, 1)
typedef struct {
@@ -120,7 +133,23 @@ typedef struct {
uint32_t block_id;
uint64_t block_offset;
uint32_t payload_len;
uint32_t seq_num; // Sequence number for reliability
uint32_t flags; // Flags (e.g., RETRANSMIT, LAST_BLOCK)
} udp_packet_header_t;
// UDP ACK packet - sent over TCP for reliability
typedef struct {
uint32_t magic; // MAGIC_UDP_ACK
uint64_t session_id;
uint32_t file_id;
uint32_t ack_block_id; // Highest contiguous block received
uint32_t ack_bitmap[4]; // Bitmap for selective ACK (128 bits)
} udp_ack_t;
// Flags for udp_packet_header_t
#define UDP_FLAG_NONE 0x00
#define UDP_FLAG_RETRANSMIT 0x01 // This is a retransmitted packet
#define UDP_FLAG_LAST_BLOCK 0x02 // This is the last block of the file
#pragma pack(pop)
src/server.c
+110 -18
View File
@@ -129,6 +129,14 @@ int setup_udp_session(int tcp_fd, int *udp_fd_out, struct sockaddr_in *client_ud
return 0;
}
// Track ACK state for sending cumulative + selective ACKs
typedef struct {
uint32_t last_ack_block; // Last contiguous block we've ACKed
uint32_t last_ack_bitmap[4]; // Bitmap of selectively received blocks
uint32_t ack_count; // Packets since last ACK
uint64_t last_ack_time_ms; // Timestamp of last ACK send
} ack_tracker_t;
int receive_file_udp_loop(int tcp_fd, int udp_fd, uint64_t session_id, active_file_t *active) {
struct pollfd fds[2];
fds[0].fd = tcp_fd;
@@ -142,6 +150,12 @@ int receive_file_udp_loop(int tcp_fd, int udp_fd, uint64_t session_id, active_fi
tv.tv_usec = 0;
setsockopt(udp_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
// Initialize ACK tracker
ack_tracker_t ack;
memset(&ack, 0, sizeof(ack));
ack.last_ack_block = (uint32_t)-1; // Invalid, force first ACK
ack.ack_count = 0;
while (1) {
int poll_ret = poll(fds, 2, -1);
if (poll_ret < 0) {
@@ -162,6 +176,7 @@ int receive_file_udp_loop(int tcp_fd, int udp_fd, uint64_t session_id, active_fi
if (n > 0) {
if (n >= (ssize_t)sizeof(udp_packet_header_t)) {
udp_packet_header_t *hdr = (udp_packet_header_t *)pkt;
if (hdr->magic == MAGIC_UDP_DATA && hdr->session_id == session_id && hdr->file_id == active->file_id) {
uint32_t bid = hdr->block_id;
if (bid < active->total_blocks) {
@@ -171,6 +186,60 @@ int receive_file_udp_loop(int tcp_fd, int udp_fd, uint64_t session_id, active_fi
perror("pwrite file block failed");
} else {
active->received_blocks[bid] = 1;
// Update ACK tracker
ack.ack_count++;
// Build selective ACK bitmap
for (int i = 0; i < 4; i++) {
ack.last_ack_bitmap[i] = 0;
}
for (uint32_t i = 0; i < active->total_blocks && i < 128; i++) {
if (active->received_blocks[i]) {
ack.last_ack_bitmap[i / 32] |= (1U << (i % 32));
}
}
// Find highest contiguous block
uint32_t highest_contiguous = 0;
while (highest_contiguous < active->total_blocks && active->received_blocks[highest_contiguous]) {
highest_contiguous++;
}
ack.last_ack_block = highest_contiguous - 1;
// Check if all blocks received first
int all_received = 1;
for (uint32_t i = 0; i < active->total_blocks; i++) {
if (!active->received_blocks[i]) {
all_received = 0;
break;
}
}
// Send ACK periodically (every UDP_ACK_INTERVAL packets or if window advancing)
if (ack.ack_count >= UDP_ACK_INTERVAL || bid == ack.last_ack_block) {
udp_ack_t ack_pkt;
ack_pkt.magic = MAGIC_UDP_ACK;
ack_pkt.session_id = session_id;
ack_pkt.file_id = active->file_id;
ack_pkt.ack_block_id = ack.last_ack_block;
for (int i = 0; i < 4; i++) {
ack_pkt.ack_bitmap[i] = ack.last_ack_bitmap[i];
}
if (writen(tcp_fd, &ack_pkt, sizeof(ack_pkt)) != sizeof(ack_pkt)) {
perror("Failed to send UDP ACK");
return -1;
}
ack.ack_count = 0;
}
if (all_received) {
// All blocks received - just break, don't send final ACK
// The client already knows from cumulative ACKs that all blocks are received
break;
}
}
}
}
@@ -266,6 +335,16 @@ static void *client_thread(void *arg) {
void handle_client(int sock_fd, const char *dest_dir) {
printf("Client connected. Starting transfer session...\n");
// Read mode (first 4 bytes from client)
uint32_t mode = MODE_SYNC; // Default to sync mode
ssize_t n = readn(sock_fd, &mode, sizeof(mode));
if (n != sizeof(mode)) {
fprintf(stderr, "Failed to read mode from client\n");
return;
}
int use_raw = (mode == MODE_RAW);
printf("Mode: %s\n", use_raw ? "RAW" : "SYNC");
int use_udp = 0;
int udp_fd = -1;
struct sockaddr_in client_udp_addr;
@@ -275,7 +354,7 @@ void handle_client(int sock_fd, const char *dest_dir) {
while (1) {
uint32_t magic = 0;
// Peek or read magic
ssize_t n = readn(sock_fd, &magic, sizeof(magic));
n = readn(sock_fd, &magic, sizeof(magic));
if (n <= 0) {
if (n < 0) {
perror("read magic failed");
@@ -286,12 +365,14 @@ void handle_client(int sock_fd, const char *dest_dir) {
}
if (magic == MAGIC_UDP_REQ) {
if (setup_udp_session(sock_fd, &udp_fd, &client_udp_addr, &udp_session_id) < 0) {
fprintf(stderr, "Failed to initialize UDP session. Falling back to TCP.\n");
uint32_t resp_fail = RESP_ERROR;
writen(sock_fd, &resp_fail, sizeof(resp_fail));
} else {
use_udp = 1;
}
continue;
}
@@ -346,17 +427,20 @@ void handle_client(int sock_fd, const char *dest_dir) {
char target_path[2048];
snprintf(target_path, sizeof(target_path), "%s/%s", dest_dir, filename);
// Check if file already exists with same size and checksum (skip if so)
struct stat existing_st;
if (stat(target_path, &existing_st) == 0 &&
(uint64_t)existing_st.st_size == meta.file_size &&
S_ISREG(existing_st.st_mode)) {
// File exists with same size - check checksum
uint64_t existing_checksum = fast_hash_file(target_path, meta.file_size);
if (existing_checksum == meta.checksum) {
printf("Skip (checksum match): %s\n", target_path);
free(filename);
continue;
// In SYNC mode: check if file already exists with same size and checksum (skip if so)
// In RAW mode: always send/receive the file
if (!use_raw) {
struct stat existing_st;
if (stat(target_path, &existing_st) == 0 &&
(uint64_t)existing_st.st_size == meta.file_size &&
S_ISREG(existing_st.st_mode)) {
// File exists with same size - check checksum
uint64_t existing_checksum = fast_hash_file(target_path, meta.file_size);
if (existing_checksum == meta.checksum) {
printf("Skip (checksum match): %s\n", target_path);
free(filename);
continue;
}
}
}
@@ -428,6 +512,7 @@ void handle_client(int sock_fd, const char *dest_dir) {
break;
}
// We received a file metadata packet (read rest of it)
uint32_t name_len = 0;
uint64_t file_size = 0;
@@ -455,17 +540,24 @@ void handle_client(int sock_fd, const char *dest_dir) {
char target_path[2048];
snprintf(target_path, sizeof(target_path), "%s/%s", dest_dir, filename);
// Ensure directories exist
if (make_path(target_path, 0755) < 0) {
perror("Failed to create parent directory path");
file_response_t resp_err = {RESP_ERROR, 0, {0, 0}};
writen(sock_fd, &resp_err, sizeof(resp_err));
continue;
// Ensure parent directories exist
char *slash = strrchr(target_path, '/');
if (slash) {
*slash = '\0';
if (make_path(target_path, 0755) < 0) {
perror("Failed to create parent directory path");
*slash = '/';
file_response_t resp_err = {RESP_ERROR, 0, {0, 0}};
writen(sock_fd, &resp_err, sizeof(resp_err));
continue;
}
*slash = '/';
}
uint32_t file_id = ++file_id_counter;
if (use_udp) {
file_response_t resp_udp = {RESP_USE_UDP, file_id, {0, 0}};
if (writen(sock_fd, &resp_udp, sizeof(resp_udp)) != sizeof(resp_udp)) {
perror("Failed to send file response header");
src/server.o
BIN
View File
Binary file not shown.
src/utils.c
+1
View File
@@ -71,6 +71,7 @@ void wq_init(work_queue_t *q) {
q->tail = NULL;
q->done = 0;
q->error = 0;
q->use_raw = 0;
pthread_mutex_init(&q->lock, NULL);
pthread_cond_init(&q->cond, NULL);
}
src/utils.o
BIN
View File
Binary file not shown.