Repository: meilisearch/meilisearch
File: crates/milli/src/update/new/channel.rs (Rust)

Code before:
/// The `total_bbbuffer_capacity` represents the number of bytes
/// allocated to all BBQueue buffers. It will be split by the
/// number of threads.
///
/// The `channel_capacity` parameter defines the number of
/// too-large-to-fit-in-BBQueue entries that can be sent through
/// a flume channel. This parameter must stay low to make
/// sure we do not use too much memory.
///
/// Note that the channel is also used to wake-up the receiver
/// when new stuff is available in any BBQueue buffer but we send
/// a message in this queue only if it is empty to avoid filling
/// the channel *and* the BBQueue.
pub fn extractor_writer_bbqueue(
    bbbuffers: &mut Vec<BBBuffer>,
    total_bbbuffer_capacity: usize,
    channel_capacity: usize,
) -> (ExtractorBbqueueSender, WriterBbqueueReceiver) {
    let current_num_threads = rayon::current_num_threads();
    let bbbuffer_capacity = total_bbbuffer_capacity.checked_div(current_num_threads).unwrap();
    bbbuffers.resize_with(current_num_threads, || BBBuffer::new(bbbuffer_capacity));

    let capacity = bbbuffers.first().unwrap().capacity();
    // Read the const description to understand this
    let max_grant = capacity.saturating_div(2).checked_sub(MAX_FRAME_HEADER_SIZE).unwrap();

    let producers = ThreadLocal::with_capacity(bbbuffers.len());
    let consumers = rayon::broadcast(|bi| {
        let bbqueue = &bbbuffers[bi.index()];
        let (producer, consumer) = bbqueue.try_split_framed().unwrap();
        producers.get_or(|| FullySend(RefCell::new(producer)));
        consumer
    });

    let (sender, receiver) = flume::bounded(channel_capacity);
    let sender = ExtractorBbqueueSender { sender, producers, max_grant };
    let receiver = WriterBbqueueReceiver {
        receiver,
        look_at_consumer: (0..consumers.len()).cycle(),
        consumers,
    };
    (sender, receiver)
}

pub struct ExtractorBbqueueSender<'a> {
    /// This channel is used to wake-up the receiver and
    /// send large entries that cannot fit in the BBQueue.
    sender: flume::Sender<ReceiverAction>,
    /// A memory buffer, one by thread, is used to serialize
    /// the entries directly in this shared, lock-free space.
    producers: ThreadLocal<FullySend<RefCell<FrameProducer<'a>>>>,

Review comment (suggestion):
```suggestion
    // 1. Due to fragmentation in the bbbuffer, we can only accept up to half the capacity in a single message.
    // 2. Read the documentation for `MAX_FRAME_HEADER_SIZE` for more information about why it is here.
    let max_grant = capacity.saturating_div(2).checked_sub(MAX_FRAME_HEADER_SIZE).unwrap();

```

Code after:
/// The `total_bbbuffer_capacity` represents the number of bytes
/// allocated to all BBQueue buffers. It will be split by the
/// number of threads.
///
/// The `channel_capacity` parameter defines the number of
/// too-large-to-fit-in-BBQueue entries that can be sent through
/// a flume channel. This parameter must stay low to make
/// sure we do not use too much memory.
///
/// Note that the channel is also used to wake-up the receiver
/// when new stuff is available in any BBQueue buffer but we send
/// a message in this queue only if it is empty to avoid filling
/// the channel *and* the BBQueue.
pub fn extractor_writer_bbqueue(
    bbbuffers: &mut Vec<BBBuffer>,
    total_bbbuffer_capacity: usize,
    channel_capacity: usize,
) -> (ExtractorBbqueueSender, WriterBbqueueReceiver) {
    let current_num_threads = rayon::current_num_threads();
    let bbbuffer_capacity = total_bbbuffer_capacity.checked_div(current_num_threads).unwrap();
    bbbuffers.resize_with(current_num_threads, || BBBuffer::new(bbbuffer_capacity));

    let capacity = bbbuffers.first().unwrap().capacity();
    // 1. Due to fragmentation in the bbbuffer, we can only accept up to half the capacity in a single message.
    // 2. Read the documentation for `MAX_FRAME_HEADER_SIZE` for more information about why it is here.
    let max_grant = capacity.saturating_div(2).checked_sub(MAX_FRAME_HEADER_SIZE).unwrap();

    let producers = ThreadLocal::with_capacity(bbbuffers.len());
    let consumers = rayon::broadcast(|bi| {
        let bbqueue = &bbbuffers[bi.index()];
        let (producer, consumer) = bbqueue.try_split_framed().unwrap();
        producers.get_or(|| FullySend(RefCell::new(producer)));
        consumer
    });

    let (sender, receiver) = flume::bounded(channel_capacity);
    let sender = ExtractorBbqueueSender { sender, producers, max_grant };
    let receiver = WriterBbqueueReceiver {
        receiver,
        look_at_consumer: (0..consumers.len()).cycle(),
        consumers,
    };
    (sender, receiver)
}

pub struct ExtractorBbqueueSender<'a> {
    /// This channel is used to wake-up the receiver and
    /// send large entries that cannot fit in the BBQueue.
    sender: flume::Sender<ReceiverAction>,
    /// A memory buffer, one by thread, is used to serialize
    /// the entries directly in this shared, lock-free space.