diff --git a/fastsync_client b/fastsync_client index 3056e44..da2f6ee 100755 Binary files a/fastsync_client and b/fastsync_client differ diff --git a/fastsync_server b/fastsync_server index 6f1c624..2540a92 100755 Binary files a/fastsync_server and b/fastsync_server differ diff --git a/src/client.c b/src/client.c index c07f3e6..4488f85 100644 --- a/src/client.c +++ b/src/client.c @@ -4,6 +4,7 @@ #include #include #include +#include #include #include @@ -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 [-p ] -s [-n ] [-u]\n" + "Usage: %s -h [-p ] -s [-n ] [-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]); } diff --git a/src/client.o b/src/client.o index aa3bef0..783c8e9 100644 Binary files a/src/client.o and b/src/client.o differ diff --git a/src/common.h b/src/common.h index 07b15fe..29a7e76 100644 --- a/src/common.h +++ b/src/common.h @@ -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) diff --git a/src/server.c b/src/server.c index bee1436..622d3f1 100644 --- a/src/server.c +++ b/src/server.c @@ -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"); diff --git a/src/server.o b/src/server.o index d58effe..bf56207 100644 Binary files a/src/server.o and b/src/server.o differ diff --git a/src/utils.c b/src/utils.c index 0fa78d2..d658bd3 100644 --- a/src/utils.c +++ b/src/utils.c @@ -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); } diff --git a/src/utils.o b/src/utils.o index 10a37b6..27d06c7 100644 Binary files a/src/utils.o and b/src/utils.o differ