krabka_log/log/
verbatim_group.rs

1//! Bounded Direct I/O groups for ordinary producer batches.
2
3use {
4    super::{Log, VerbatimBatch, open::DataProducerAppend},
5    crate::{DeliveryPolicy, LogError, SegmentAllocation, segment::PlacedBatch},
6    krabka_ids::Offset,
7    krabka_protocol::records::{Attributes, CRC_COVERAGE_START, HEADER_LEN, TimestampType},
8    krabka_units::prelude::{ByteSize, ByteSizeExt as _, mebibytes},
9};
10
11impl Log {
12    /// Try to append ordinary producer batches with one aligned Direct I/O write.
13    ///
14    /// `None` leaves the batches unwritten. The caller then uses individual appends.
15    /// Groups are at most 1 MiB and stay in one nonempty active segment. Transactions,
16    /// scheduled delivery, stamps, append-time timestamps, and explicit flush policies
17    /// use individual appends. The caller has already validated the batch CRCs.
18    ///
19    /// A successful group returns each assigned base offset in input order.
20    ///
21    /// # Errors
22    /// A write or bookkeeping failure rolls the whole group back before returning the error.
23    pub fn try_append_verbatim_group(
24        &mut self,
25        batches: &[&VerbatimBatch],
26    ) -> Option<Result<Vec<Offset>, LogError>> {
27        let config = self.config_snapshot();
28        if batches.len() < 2
29            || config.segment_allocation != SegmentAllocation::Preallocate
30            || config.delivery_policy != DeliveryPolicy::Immediate
31            || config.message_timestamp_type != TimestampType::CreateTime
32            || config.flush_on_append
33            || config.flush_messages.is_some()
34            || config.flush_interval.is_some()
35            || self.stamp_source.is_some()
36        {
37            return None;
38        }
39        let base = self.append_at_expected_offset();
40        let mut next = base;
41        let mut size = 0_u64;
42        let mut max_timestamp = i64::MIN;
43        let mut placed = Vec::with_capacity(batches.len());
44        for &batch in batches {
45            if batch.is_transactional || batch.bytes.len() < HEADER_LEN {
46                return None;
47            }
48            let &attributes = batch
49                .bytes
50                .get(CRC_COVERAGE_START..)
51                .and_then(<[u8]>::first_chunk::<2>)?;
52            let attributes = Attributes(i16::from_be_bytes(attributes));
53            if attributes.is_control_batch() || attributes.is_transactional() {
54                return None;
55            }
56            let (_, end) =
57                krabka_verified::local_append_coordinates(next.0, next.0, batch.last_offset_delta)?;
58            Self::data_producer_tail(
59                batch.producer_id,
60                batch.base_sequence,
61                batch.last_offset_delta,
62                next,
63            )
64            .ok()?;
65            placed.push(PlacedBatch { batch, base: next });
66            next = Offset(end);
67            size = size.checked_add(u64::try_from(batch.bytes.len()).ok()?)?;
68            max_timestamp = max_timestamp.max(batch.max_timestamp);
69        }
70        if size > mebibytes(1).bytes_u64() {
71            return None;
72        }
73        let active = self.active.as_ref()?;
74        if active.size() == ByteSize::ZERO
75            || active.indexes_full_after(config.segment_index_size, batches.len())
76        {
77            return None;
78        }
79        let interval = self.jittered_roll_interval(config.segment_roll_interval);
80        if self.should_roll_for_incoming(
81            ByteSize::from_bytes(size),
82            max_timestamp,
83            config.segment_size,
84            interval,
85            config.segment_index_size,
86        ) {
87            return None;
88        }
89        let active = self.active.as_mut()?;
90        active.reserve(config.segment_size);
91        active.write_direct(config.tail_cache_size);
92        if !active.writes_direct() {
93            return None;
94        }
95        Some(self.append_direct_group(&placed, config.index_interval))
96    }
97
98    fn append_direct_group(
99        &mut self,
100        batches: &[PlacedBatch<'_>],
101        index_interval: ByteSize,
102    ) -> Result<Vec<Offset>, LogError> {
103        self.ensure_unfenced()?;
104        self.rollover_flusher.check()?;
105        self.active
106            .as_mut()
107            .ok_or_else(super::no_active_segment)?
108            .append_verbatim_group(batches, index_interval)?;
109        let result = (|| {
110            for &PlacedBatch { batch, base } in batches {
111                self.flush_after_append(batch.last_offset_delta, false)?;
112                self.update_data_producer_entry(DataProducerAppend {
113                    producer_id: batch.producer_id,
114                    producer_epoch: batch.producer_epoch,
115                    base_sequence: batch.base_sequence,
116                    last_offset_delta: batch.last_offset_delta,
117                    base_offset: base,
118                    timestamp: batch.max_timestamp,
119                    is_transactional: false,
120                })?;
121                self.assign_appended_epoch(batch.leader_epoch, base)?;
122            }
123            self.refresh_lso()
124        })();
125        match batches.first() {
126            Some(first) => self.complete_append_state(result, first.base)?,
127            // An empty group appended nothing to roll back.
128            None => result?,
129        }
130        Ok(batches.iter().map(|placed| placed.base).collect())
131    }
132}
krabka_log/log/verbatim_group.rs

krabka_log/log/
verbatim_group.rs
