perf(send,scanner): byte-bound sender queues; drop redundant stat

src/client/client_send.c

@@ -37,6 +37,13 @@
 
 #define STREAM_THRESHOLD (64ULL * 1024 * 1024)
 
+/* Aggregate loaded payload bytes the sender may buffer across the loader queue
+   and the chunk in flight.  Sending one chunk adds up to ~2 * MAX_CHUNK_SIZE of
+   transient serialize/compress buffers on top of the queued payloads, so this
+   ceiling keeps total pipeline memory within MAX_CONNECTION_MEMORY (mirrors the
+   receiver's RECEIVER_QUEUE_MAX_BYTES). */
+#define SENDER_QUEUE_MAX_BYTES (MAX_CONNECTION_MEMORY - 2 * MAX_CHUNK_SIZE)
+
 /* Forward declaration for progress-reporting thread used in multithreaded send. */
 static int progress_thread_fn(void* arg);
 
@@ -1493,6 +1500,10 @@ static int send_chunks_multithreaded(void* pipeline_context) {
       protocol_session_unbind();
       return thrd_error;
     }
+    /* Payload bytes this chunk was charged for on the loader's byte budget.
+       Computed before destruction and released after the memory is actually
+       freed, so a loader blocked on the budget wakes only once room exists. */
+    size_t queued_payload = pipeline_context_sender_chunk_bytes(current_chunk);
     unsigned long long chunk_bytes = 0;
     int chunk_files = 0;
     for (int i = 0; i < current_chunk->element_count; i++) {
@@ -1507,6 +1518,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
     context->progress_bytes = context->total_bytes;
     mtx_unlock(&context->mutex_progress);
     chunk_destroy(current_chunk);
+    pipeline_context_sender_note_bytes_released(context, queued_payload);
   }
 
   /* Completion tail: reached on natural exhaustion or an early stop deadline.
@@ -1728,11 +1740,7 @@ static int load_files_multithreaded(void* pipeline_context) {
         }
       }
     }
-    if (!queue_enqueue_multithreaded_cancel(context->queue_loader, chunk, &context->mutex_loader,
-                                            &context->condition_not_empty_loader,
-                                            &context->condition_not_full_loader,
-                                            &context->cancelled)) {
-      chunk_destroy(chunk);
+    if (!pipeline_context_sender_enqueue_chunk(context, chunk)) {
       pipeline_cancel(context);
       protocol_session_unbind();
       return thrd_error;
@@ -2197,8 +2205,13 @@ int send_files_multithreaded(Config** config_ptr) {
   unsigned long long available_memory =
       pages > 0 && page_size > 0 ? (unsigned long long)pages * (unsigned long long)page_size
                                  : 512ULL * 1024 * 1024;
-  unsigned long long avg_file_size = 1024 * 1024;
-  int qsize = (int)(available_memory / avg_file_size);
+  /* Size the chunk queues from the actual chunk size rather than a fixed 1 MiB
+     average: a chunk holds roughly `chunk_size` bytes of file data, so counting
+     chunks at 1 MiB over-estimated the queue capacity by up to 10x.  The byte
+     budget below is the authoritative bound; this count keeps the unloaded
+     chunks waiting in the scanner queue bounded too. */
+  unsigned long long chunk_size = config->chunk_size > 0 ? config->chunk_size : DEFAULT_CHUNK_SIZE;
+  int qsize = (int)(available_memory / chunk_size);
   if (qsize < 10)
     qsize = 10;
   if (qsize > 1000)
@@ -2223,7 +2236,8 @@ int send_files_multithreaded(Config** config_ptr) {
   }
   context->missing_args = missing_args;
   missing_args = NULL; /* owned by the context from here on */
-  *config_ptr = NULL;  /* context now owns config through all remaining paths */
+  pipeline_context_sender_set_queue_byte_limit(context, SENDER_QUEUE_MAX_BYTES);
+  *config_ptr = NULL; /* context now owns config through all remaining paths */
   struct timespec now_mono;
   if (clock_gettime(CLOCK_MONOTONIC, &now_mono) != 0) {
     now_mono.tv_sec = 0;

src/client/scanner.c

@@ -388,9 +388,13 @@ static int scanner_inspect_entry(const ScannerOptions* options, const char* sour
   goto apply_filters;
 
 regular:
-  if (stat(entry->path, &entry->stats) != 0)
-    goto skip;
-  entry->is_directory = S_ISDIR(entry->stats.st_mode);
+  /* Not a symlink: the lstat() above already described this entry, and lstat
+     and stat are identical for every non-symlink, so reuse that result instead
+     of issuing a redundant stat() on the scanner hot path.  stat() is still
+     used on the dereference paths above/below for actual symlinks (copy-links,
+     safe/copy-unsafe links, and -k symlinks-to-directories). */
+  entry->stats = link_stats;
+  entry->is_directory = S_ISDIR(link_stats.st_mode);
   if (entry->is_directory)
     return 1;
 

src/shared/multiprocessing.c

@@ -11,6 +11,7 @@
 #include "protocol.h"
 #include "queue.h"
 #include "utils.h"
+#include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -26,6 +27,8 @@ PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* que
   context->queue_loader = queue_loader;
   context->scanner_done = false;
   context->loader_done = false;
+  context->queued_bytes = 0;
+  context->max_queue_bytes = 0;
   context->manifest = NULL;
   context->excluded_paths = NULL;
   context->missing_args = NULL;
@@ -97,6 +100,84 @@ fail:
   return NULL;
 }
 
+void pipeline_context_sender_set_queue_byte_limit(PipelineContextSender* context,
+                                                  size_t max_bytes) {
+  if (context == NULL)
+    return;
+  mtx_lock(&context->mutex_loader);
+  context->max_queue_bytes = max_bytes;
+  context->queued_bytes = 0;
+  cnd_broadcast(&context->condition_not_full_loader);
+  mtx_unlock(&context->mutex_loader);
+}
+
+size_t pipeline_context_sender_chunk_bytes(const Chunk* chunk) {
+  if (chunk == NULL || chunk->items == NULL)
+    return 0;
+  size_t total = 0;
+  for (int i = 0; i < chunk->element_count; i++) {
+    const File* file = chunk->items[i];
+    if (file == NULL || file->data == NULL || file->data->data == NULL)
+      continue;
+    if (file->data->size > SIZE_MAX - total)
+      return SIZE_MAX;
+    total += file->data->size;
+  }
+  return total;
+}
+
+void pipeline_context_sender_note_bytes_released(PipelineContextSender* context,
+                                                 size_t released_bytes) {
+  if (context == NULL || context->max_queue_bytes == 0 || released_bytes == 0)
+    return;
+  mtx_lock(&context->mutex_loader);
+  if (released_bytes >= context->queued_bytes)
+    context->queued_bytes = 0;
+  else
+    context->queued_bytes -= released_bytes;
+  cnd_signal(&context->condition_not_full_loader);
+  mtx_unlock(&context->mutex_loader);
+}
+
+bool pipeline_context_sender_enqueue_chunk(PipelineContextSender* context, Chunk* chunk) {
+  if (context == NULL || chunk == NULL)
+    return false;
+  size_t chunk_bytes = pipeline_context_sender_chunk_bytes(chunk);
+  mtx_lock(&context->mutex_loader);
+  while (!atomic_load(&context->cancelled)) {
+    bool blocked_by_count = queue_is_full(context->queue_loader);
+    bool blocked_by_budget = false;
+    if (context->max_queue_bytes > 0) {
+      size_t budget = context->max_queue_bytes;
+      size_t used = context->queued_bytes;
+      if (used >= budget) {
+        blocked_by_budget = true;
+      } else if (chunk_bytes > budget - used) {
+        /* A single payload larger than the whole budget is only admitted to an
+           empty pipeline so the wait can never deadlock. */
+        blocked_by_budget = used != 0;
+      }
+    }
+    if (!blocked_by_count && !blocked_by_budget)
+      break;
+    cnd_wait(&context->condition_not_full_loader, &context->mutex_loader);
+  }
+  if (atomic_load(&context->cancelled)) {
+    mtx_unlock(&context->mutex_loader);
+    chunk_destroy(chunk);
+    return false;
+  }
+  if (!queue_enqueue(context->queue_loader, chunk)) {
+    mtx_unlock(&context->mutex_loader);
+    chunk_destroy(chunk);
+    return false;
+  }
+  context->queued_bytes += chunk_bytes;
+  cnd_signal(&context->condition_not_empty_loader);
+  mtx_unlock(&context->mutex_loader);
+  return true;
+}
+
 void pipeline_context_sender_destroy(PipelineContextSender* context) {
   if (context->manifest) {
     array_list_delete(context->manifest);

src/shared/multiprocessing.h

@@ -5,6 +5,7 @@
 #include <stdatomic.h>
 
 #include "array_list.h"
+#include "chunk.h"
 #include "config.h"
 #include "file.h"
 #include "protocol.h"
@@ -25,6 +26,15 @@ typedef struct {
   cnd_t condition_not_full_loader;
   cnd_t condition_not_empty_loader;
   bool loader_done;
+  /* Aggregate loaded payload bytes queued on queue_loader but not yet released
+     by the sender.  Guarded by `mutex_loader`.  When `max_queue_bytes` is
+     non-zero the loader blocks before enqueueing a chunk that would push this
+     total over it, so the sender buffers a bounded number of bytes rather than
+     an unbounded count of chunks that may each be up to chunk_size (or a single
+     file) in size.  Files streamed straight from disk by sendfile hold no
+     payload, so only in-memory (`data->data`) payloads are counted. */
+  size_t queued_bytes;
+  size_t max_queue_bytes;
   ArrayList* manifest;
   /* Protected prefixes (paths the source scan excluded by user rules) sent
      with the keep-set manifest so --delete leaves them alone unless
@@ -113,6 +123,22 @@ typedef struct PipelineContextReceiver {
 PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,
                                                       Queue* queue_loader);
 void pipeline_context_sender_destroy(PipelineContextSender* context);
+/* Bound the loaded payload bytes the sender may buffer ahead of the network
+   writer (see max_queue_bytes). */
+void pipeline_context_sender_set_queue_byte_limit(PipelineContextSender* context, size_t max_bytes);
+/* Total payload bytes a chunk currently holds in memory (loaded file data
+   only; zero for entries with no payload or data streamed from disk). */
+size_t pipeline_context_sender_chunk_bytes(const Chunk* chunk);
+/* Blocking enqueue used by the sender's loader stage.  Blocks while
+   queue_loader is full by element count or when adding `chunk` would push the
+   queued payload bytes over the configured byte limit; waits until the sender
+   releases bytes.  Takes ownership of `chunk` on success and destroys it on
+   failure/cancel. */
+bool pipeline_context_sender_enqueue_chunk(PipelineContextSender* context, Chunk* chunk);
+/* Account for `released_bytes` of payload memory that the sender freed after
+   destroying a chunk, unblocking a loader waiting on the byte limit. */
+void pipeline_context_sender_note_bytes_released(PipelineContextSender* context,
+                                                 size_t released_bytes);
 PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue* queue_receiver,
                                                           int file_descriptor, SSL* ssl);
 void pipeline_context_receiver_destroy(PipelineContextReceiver* context);