fix: #254 bound receiver queue by aggregate payload bytes
The per-connection memory budget (MAX_CONNECTION_MEMORY, 256 MiB) only charged wire buffers via receive_data_limited. Decompression buffers and per-file chunk copies were not accounted for, and the multithreaded receiver could enqueue up to 100 files (each up to 64 MiB uncompressed) ahead of a slow disk writer, retaining ~6.4 GiB per connection. A client sending highly compressible chunks with little bandwidth could OOM the host while the reserve never tripped. Bound the receive pipeline by aggregate payload bytes instead of item count alone: - Export MAX_CONNECTION_MEMORY from protocol.h. - PipelineContextReceiver tracks queued_bytes (payload bytes received but not yet released by the disk writer, i.e. queued or in the writer's hand) under the existing mutex. - receiver enqueue now blocks while the queue is full by count OR when adding the file would push queued_bytes over the configured byte limit, applying backpressure to the sender instead of failing the transfer. - The disk writer releases the byte budget after each file is freed and signals the not-full condition. - The server sets the byte ceiling to MAX_CONNECTION_MEMORY - 2*MAX_CHUNK_SIZE so that the queued payloads plus the transient wire/decompression buffers of the one in-flight chunk stay within the per-connection budget. The single-threaded receive path is already bounded: it writes files to disk before reading the next chunk, so its transient is at most one chunk's wire + decompressed + copied payload (~3 * MAX_CHUNK_SIZE, below the budget). Wire buffers remain charged exactly once by receive_data_limited; this change does not double charge them. Adds a deterministic unit test in test_multiprocessing.c proving that an enqueue which would exceed the byte budget blocks until the writer releases bytes.
This commit is contained in:
@@ -45,6 +45,14 @@ typedef struct PipelineContextReceiver {
|
||||
cnd_t condition_not_empty;
|
||||
bool receiver_done;
|
||||
atomic_bool cancelled;
|
||||
/* Aggregate payload bytes that have been received but not yet released by
|
||||
the disk writer (queued or in the writer's hand). Guarded by `mutex`.
|
||||
When `max_queue_bytes` is non-zero the receiver blocks before enqueuing
|
||||
once this total would exceed it, so decompressed/copied file payloads
|
||||
buffered ahead of a slow disk writer respect the per-connection memory
|
||||
budget instead of growing without bound. */
|
||||
size_t queued_bytes;
|
||||
size_t max_queue_bytes;
|
||||
} PipelineContextReceiver;
|
||||
|
||||
PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,
|
||||
@@ -53,6 +61,18 @@ void pipeline_context_sender_destroy(PipelineContextSender* context);
|
||||
PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue* queue_receiver,
|
||||
int file_descriptor, SSL* ssl);
|
||||
void pipeline_context_receiver_destroy(PipelineContextReceiver* context);
|
||||
/* Bound the bytes buffered ahead of the disk writer (see max_queue_bytes). */
|
||||
void pipeline_context_receiver_set_queue_byte_limit(PipelineContextReceiver* context,
|
||||
size_t max_bytes);
|
||||
/* Blocking enqueue used by the receive pipeline sink. Blocks while the queue
|
||||
is full by element count or when adding `file` would push queued_bytes over
|
||||
the configured byte limit; waits until the disk writer releases bytes.
|
||||
Takes ownership of `file` on success and destroys it on failure/cancel. */
|
||||
bool pipeline_context_receiver_enqueue_file(PipelineContextReceiver* context, File* file);
|
||||
/* Account for `released_bytes` of payload memory that has been freed by the
|
||||
disk writer, unblocking a receiver that is waiting on the byte limit. */
|
||||
void pipeline_context_receiver_note_bytes_released(PipelineContextReceiver* context,
|
||||
size_t released_bytes);
|
||||
int receive_thread(void* pipeline_context);
|
||||
int write_thread(void* pipeline_context);
|
||||
#endif
|
||||
|
||||
Reference in New Issue
Block a user