Module batching 

Re-chunking and bounded concurrency for connector writes and reads (RFC 0013, #880).

Every sink re-chunks the page it is handed into backend-sized requests, and many run those requests concurrently under a limit. chunks honours the batch_size contract (0 = no re-chunking, the whole page is one request) and for_each_concurrent / map_concurrent run work with at most n in flight, stopping at the first error.

use faucet_core::batching::{chunks, map_concurrent};

let page: Vec<u32> = (0..10).collect();
let sums = map_concurrent(chunks(&page, 4), 2, |chunk| async move {
    Ok(chunk.iter().sum::<u32>())
})
.await?;
assert_eq!(sums, vec![6, 22, 17]);

Structs

ConcurrencyLimit

A concurrency limit shared across calls: a semaphore with n permits (0 is treated as 1). Cheap to clone; every clone shares the permits. Use it when the work to bound is spread over separate calls (requests spawned from several write_batch invocations); within one call prefer map_concurrent.

Functions

chunks

Split items into slices of at most batch_size. batch_size == 0 means “don’t re-chunk”: the whole slice is one chunk. An empty slice yields no chunks.

for_each_concurrent

Run f on every item with at most limit in flight (0 is treated as 1). Stops at, and returns, the first error; work already in flight is dropped.

into_chunks

Split an owned vector the same way as chunks, moving the items.

map_concurrent

Run f on every item with at most limit in flight (0 is treated as 1) and return the results in input order. Stops at, and returns, the first error.