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

Code before:
                payload_header.serialize_into(header_bytes);
                key_writer(remaining)
            },
        )?;

        Ok(())
    }
}

/// Try to reserve a frame grant of `total_length` by spin
/// looping on the BBQueue buffer, panics if the receiver
/// has been disconnected or send a WakeUp message if necessary.
fn reserve_and_write_grant<F>(
    producer: &mut FrameProducer,
    total_length: usize,
    sender: &flume::Sender<ReceiverAction>,
    sent_messages_attempts: &AtomicUsize,
    blocking_sent_messages_attempts: &AtomicUsize,
    f: F,
) -> crate::Result<()>
where
    F: FnOnce(&mut [u8]) -> crate::Result<()>,
{
    loop {
        // An attempt means trying multiple times
        // and succeeded to send or not.
        sent_messages_attempts.fetch_add(1, atomic::Ordering::Relaxed);

        for _ in 0..10_000 {
            match producer.grant(total_length) {
                Ok(mut grant) => {
                    // We could commit only the used memory.
                    f(&mut grant)?;
                    grant.commit(total_length);

                    // We only send a wake up message when the channel is empty
                    // so that we don't fill the channel with too many WakeUps.
                    if sender.is_empty() {
                        sender.send(ReceiverAction::WakeUp).unwrap();
                    }

                    return Ok(());
                }
                Err(bbqueue::Error::InsufficientSize) => continue,
                Err(e) => unreachable!("{e:?}"),
            }
        }
        if sender.is_disconnected() {
            return Err(Error::InternalError(InternalError::AbortedIndexation));
        }


Review comment (suggestion):
```suggestion
        // whether is succeeded or not.
```

Code after:
                payload_header.serialize_into(header_bytes);
                key_writer(remaining)
            },
        )?;

        Ok(())
    }
}

/// Try to reserve a frame grant of `total_length` by spin
/// looping on the BBQueue buffer, panics if the receiver
/// has been disconnected or send a WakeUp message if necessary.
fn reserve_and_write_grant<F>(
    producer: &mut FrameProducer,
    total_length: usize,
    sender: &flume::Sender<ReceiverAction>,
    sent_messages_attempts: &AtomicUsize,
    blocking_sent_messages_attempts: &AtomicUsize,
    f: F,
) -> crate::Result<()>
where
    F: FnOnce(&mut [u8]) -> crate::Result<()>,
{
    loop {
        // An attempt means trying multiple times
        // whether is succeeded or not.
        sent_messages_attempts.fetch_add(1, atomic::Ordering::Relaxed);

        for _ in 0..10_000 {
            match producer.grant(total_length) {
                Ok(mut grant) => {
                    // We could commit only the used memory.
                    f(&mut grant)?;
                    grant.commit(total_length);

                    // We only send a wake up message when the channel is empty
                    // so that we don't fill the channel with too many WakeUps.
                    if sender.is_empty() {
                        sender.send(ReceiverAction::WakeUp).unwrap();
                    }

                    return Ok(());
                }
                Err(bbqueue::Error::InsufficientSize) => continue,
                Err(e) => unreachable!("{e:?}"),
            }
        }
        if sender.is_disconnected() {
            return Err(Error::InternalError(InternalError::AbortedIndexation));
        }