hubcio commented on code in PR #4169:
URL: https://github.com/apache/iggy/pull/4169#discussion_r4003132218
##########
core/journal/src/partition_journal.rs:
##########
@@ -1075,6 +1074,24 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
Ok(())
}
+ /// Whether [`Self::reclaim_obsolete`] has anything left to remove.
+ #[must_use]
+ pub fn has_obsolete(&self) -> bool {
+ !self.obsolete.is_empty() || self.cleanup_directory_dirty
+ }
+
+ /// Remove a bounded batch of the files no published generation retains.
+ ///
+ /// Separate from the append path so an acknowledgement never waits on
+ /// unlinks and a directory barrier for history it does not depend on. Only
+ /// a checkpoint and the boot scan ever queue work here, and both do so
+ /// after the publication that excludes those files, so the queue holds
+ /// nothing a reader or a recovery could still need. The writer owns the
+ /// journal, so this runs between mutations and never beside one.
+ pub async fn reclaim_obsolete(&mut self) {
Review Comment:
simplification: `reclaim_obsolete` only forwards to `cleanup_obsolete`, and
`has_obsolete` guards a call that is already a no-op on an empty queue. make
`cleanup_obsolete` public under this name and drop both.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java:
##########
@@ -268,69 +283,112 @@ public static ByteBuf toMessagesBatch(List<Message>
messages) {
message.payload(),
readAllBytes(toBytes(message.userHeaders()))));
}
- return encodeBatch(rawMessages);
+ return rawMessages;
}
static ByteBuf encodeBatch(List<RawMessage> messages) {
- var batchOriginTimestamp = messages.stream()
- .map(RawMessage::originTimestamp)
- .min(BigInteger::compareTo)
- .orElseThrow(() -> new IggyInvalidArgumentException("Cannot
encode an empty message batch"));
- var blobLength = 0;
- for (RawMessage message : messages) {
- blobLength += MessageHeader.SIZE + message.payload().length +
message.userHeaders().length;
+ var batch = Unpooled.buffer(BATCH_HEADER_SIZE);
+ try {
+ encodeBatchInto(batch, messages);
+ return batch;
+ } catch (RuntimeException | Error error) {
+ batch.release();
+ throw error;
}
+ }
- var batch = Unpooled.buffer(BATCH_HEADER_SIZE + blobLength);
- batch.writeZero(BATCH_HEADER_SIZE);
- for (int index = 0; index < messages.size(); index++) {
- RawMessage message = messages.get(index);
- var timestampDelta =
message.originTimestamp().subtract(batchOriginTimestamp);
- if (timestampDelta.compareTo(MAX_TIMESTAMP_DELTA_MICROS) > 0) {
- throw new IggyInvalidArgumentException("Message origin
timestamp exceeds the batch origin by "
- + timestampDelta + " microseconds, more than the
timestamp delta field can hold");
+ private static BatchExtent measureBatch(List<RawMessage> messages, long
capacityAllowance) {
+ var originTimestamp = messages.get(0).originTimestamp();
+ var latestTimestamp = originTimestamp;
+ long length = BATCH_HEADER_SIZE;
+ for (RawMessage message : messages) {
+ var timestamp = message.originTimestamp();
+ if (timestamp.signum() < 0 || timestamp.bitLength() > Long.SIZE) {
+ throw new IggyInvalidArgumentException("Message origin
timestamp is outside unsigned 64-bit range");
+ }
+ originTimestamp = originTimestamp.min(timestamp);
+ latestTimestamp = latestTimestamp.max(timestamp);
+ length += (long) MessageHeader.SIZE + message.payload().length +
message.userHeaders().length;
+ if (length > capacityAllowance) {
+ throw new IggyInvalidArgumentException("Message batch exceeds
the output buffer capacity");
}
- var frameStart = batch.writerIndex();
- batch.writeLongLE(0); // checksum, backpatched below
- batch.writeBytes(message.id());
- batch.writeIntLE(index); // offset_delta
- batch.writeIntLE(timestampDelta.intValue());
- batch.writeIntLE(message.userHeaders().length);
- batch.writeIntLE(message.payload().length);
- batch.writeLongLE(0); // reserved
- batch.writeBytes(message.payload());
- batch.writeBytes(message.userHeaders());
- batch.setLongLE(
- frameStart, xxHash3(batch, frameStart + Long.BYTES,
batch.writerIndex() - frameStart - Long.BYTES));
}
+ if
(latestTimestamp.subtract(originTimestamp).compareTo(MAX_TIMESTAMP_DELTA_MICROS)
> 0) {
+ throw new IggyInvalidArgumentException("Message origin timestamp
delta exceeds unsigned 32-bit range");
Review Comment:
nit: the old message and the server's `InvalidMessageTimestampDelta` error
both name the offending delta, this one drops it. append the delta and the
message index.
##########
core/journal/src/partition_journal.rs:
##########
@@ -1075,6 +1074,24 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
Ok(())
}
+ /// Whether [`Self::reclaim_obsolete`] has anything left to remove.
+ #[must_use]
+ pub fn has_obsolete(&self) -> bool {
+ !self.obsolete.is_empty() || self.cleanup_directory_dirty
+ }
+
+ /// Remove a bounded batch of the files no published generation retains.
+ ///
+ /// Separate from the append path so an acknowledgement never waits on
Review Comment:
nit: an ack can wait on the unlinks, because the writer loop reclaims before
it pops the next append. and `rewrite` queues the old generation for
`truncate_from`, `reset` and `reset_with_prepare` too, not only the checkpoint
and boot scan.
##########
core/partitions/src/poll_plan.rs:
##########
@@ -383,16 +387,82 @@ pub enum DiskReadOutcome {
Faulted,
}
+/// Largest first read of a disk poll, and the size every poll used to read
+/// whatever it asked for. A batch wider than this still grows past it through
+/// the re-read path below; this bounds only where a walk starts.
+const DISK_POLL_CHUNK_MAX: u64 = 1 << 20;
+
+/// Smallest first read of a disk poll. Below this the syscall and the segment
+/// walk cost more than the bytes the smaller read saves, and a poll for a
+/// handful of messages would issue a read per batch.
+const DISK_POLL_CHUNK_MIN: u64 = 64 << 10;
+
+/// How the chunk loop over one segment ended.
+enum SegmentWalk {
+ /// The segment is exhausted or the requested count is filled. The walk
+ /// may continue into the next segment.
+ Done,
+ /// Fail-closed: the segment may hold present-but-unreadable or corrupt
+ /// data, so no later segment may be served over it.
+ Faulted,
+}
+
+/// The state one disk walk carries across its segments.
+struct DiskWalk {
+ /// Byte offset into the segment being walked; reset at each boundary.
+ position: u64,
+ matched: u32,
+ fragments: PollFragments<4096>,
+ last_matching_offset: Option<u64>,
+ #[cfg(feature = "poll-diagnostics")]
+ requested_bytes: u64,
+ #[cfg(feature = "poll-diagnostics")]
+ chunk_reads: u32,
+}
+
+impl DiskWalk {
+ fn starting_at(position: u64) -> Self {
+ Self {
+ position,
+ matched: 0,
+ fragments: PollFragments::new(),
+ last_matching_offset: None,
+ #[cfg(feature = "poll-diagnostics")]
+ requested_bytes: 0,
+ #[cfg(feature = "poll-diagnostics")]
+ chunk_reads: 0,
+ }
+ }
+}
+
impl DiskReadPlan {
+ /// Bytes to read for the next `remaining` messages.
+ ///
+ /// A poll asks for a message count, and the walk reads bytes, so the two
+ /// are bridged by the partition's own mean encoded size. Reading a fixed
+ /// megabyte instead costs a poll for a thousand hundred-byte messages ten
+ /// times the bytes it returns, and the sparse-selection copy that follows
+ /// scales with the chunk rather than with the selection.
+ ///
+ /// The estimate is deliberately not a bound. Messages vary in size, the
+ /// starting offset can sit inside a batch the index resolved before it,
+ /// and an underestimate only costs another read of the next chunk.
+ fn chunk_len(&self, remaining: u32) -> u64 {
+ let Some(bytes_per_message) = self.bytes_per_message else {
+ return DISK_POLL_CHUNK_MAX;
+ };
+ u64::from(bytes_per_message)
+ .saturating_mul(u64::from(remaining))
+ .saturating_add(COMMAND_HEADER_SIZE as u64)
+ .clamp(DISK_POLL_CHUNK_MIN, DISK_POLL_CHUNK_MAX)
Review Comment:
warning: the decoder already returns the batch length in `UnexpectedEof {
need, .. }`, and line 898 throws it away before the `*4` ladder. return `need`
from `walk_disk_chunk` and grow to `max(chunk_len * 4, need)`, so the second
read is exactly the batch.
##########
core/partitions/src/poll_plan.rs:
##########
@@ -532,6 +574,74 @@ impl DiskReadPlan {
}
}
+ /// Read one segment from `walk.position` until the count is filled, the
+ /// segment is exhausted, or the walk must fail closed.
+ ///
+ /// The chunk length is already bounded by `DISK_POLL_CHUNK_MAX` and by the
+ /// segment's persisted bytes, so narrowing it to a `usize` cannot
truncate.
+ #[allow(clippy::cast_possible_truncation)]
+ async fn walk_segment(
+ &self,
+ file: &compio::fs::File,
+ query: MessageLookup,
+ count: u32,
+ persisted: u64,
+ walk: &mut DiskWalk,
+ ) -> SegmentWalk {
+ let mut chunk_len = self.chunk_len(count - walk.matched);
Review Comment:
warning: the first read ignores the gap from the index entry to the target
offset, so a poll deep in a flush group walks it in 64 KiB reads. add that gap
to the first read size.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java:
##########
@@ -354,9 +412,13 @@ private static long xxHash3(ByteBuf buffer, int index, int
length) {
}
private static byte[] readAllBytes(ByteBuf buffer) {
- var bytes = new byte[buffer.readableBytes()];
- buffer.readBytes(bytes);
- return bytes;
+ try {
+ var bytes = new byte[buffer.readableBytes()];
+ buffer.readBytes(bytes);
+ return bytes;
+ } finally {
+ buffer.release();
Review Comment:
nit: `readAllBytes` now releases the shared `Unpooled.EMPTY_BUFFER` that
`toBytes(Map)` returns for empty headers, which only works because that release
is a no-op. return an empty `byte[]` for empty headers before any buffer exists.
##########
core/partitions/src/poll_plan.rs:
##########
@@ -532,6 +574,74 @@ impl DiskReadPlan {
}
}
+ /// Read one segment from `walk.position` until the count is filled, the
+ /// segment is exhausted, or the walk must fail closed.
+ ///
+ /// The chunk length is already bounded by `DISK_POLL_CHUNK_MAX` and by the
Review Comment:
nit: this says `DISK_POLL_CHUNK_MAX` bounds the chunk, but line 636 grows
`chunk_len` past it. keep only the `persisted` bound here.
##########
core/partitions/src/persistence.rs:
##########
@@ -1140,49 +1148,66 @@ impl<S: DurableStorage> PartitionPersistence<S> {
break;
}
if epoch == self.epoch.get() && !self.retired.get() {
- let mut references = self.segment_references.borrow_mut();
- if rebuild_references {
- references.clear();
- references.extend(journal.written_segment_references(0));
- } else if let Some(from_op) =
self.written_head.get().checked_add(1) {
-
references.extend(journal.written_segment_references(from_op));
- }
- drop(references);
- self.disk_bytes.set(journal.size_bytes());
- self.retained_bytes.set(journal.retained_bytes());
- self.segment_checkpoint.set(journal.segment_checkpoint());
- let advanced = journal.durable_op() != self.durable_head.get()
- || journal.checkpoint_op() != self.checkpoint.get()
- || journal.certified_log_view() !=
self.certified_log_view.get()
- || (journal.segment_checkpoint().is_some()
- && journal.head() != self.written_head.get());
- self.certified_log_view.set(journal.certified_log_view());
- if self
- .requested_log_view
- .get()
- .is_some_and(|(view, _, _)| Some(view) ==
self.certified_log_view.get())
- {
- self.requested_log_view.set(None);
- }
- self.written_head.set(journal.head());
- self.durable_head.set(journal.durable_op());
- if journal.checkpoint_op() > self.checkpoint.get() {
- self.accepted
- .borrow_mut()
- .checkpoint(journal.checkpoint_op());
- }
- self.checkpoint.set(journal.checkpoint_op());
- self.checkpoint_checksum.set(journal.checkpoint_checksum());
- self.purge_generation.set(journal.purge_marker().0);
- self.purge_floor.set(journal.purge_marker().1);
- if advanced {
- self.notify();
- }
+ self.publish_mutation(journal, rebuild_references);
+ }
+ // After the notification, never before the barrier it would delay.
+ // A checkpoint queues the generation it replaced, so reclaiming it
+ // here keeps the unlinks and the directory barrier out of the
+ // acknowledgement the next append is waiting on, while still
+ // running once per mutation so a busy partition reclaims as
+ // promptly as an idle one.
+ if journal.has_obsolete() {
Review Comment:
nit: this runs before the next append is popped, so that append still waits
on the unlinks and the directory fsync, contrary to the comment above. reclaim
only on an empty queue, with a mutation-count bound for busy partitions.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2987,12 +2993,18 @@ where
self.check_local_poll_key(kind, consumer_id)
.map_err(|error| self.poll_capacity_error(error))?;
let consensus = self.consensus();
- if !consensus.is_primary()
- || !consensus.is_normal()
- || consensus.is_transferring()
- || self
- .durable_consumer_offsets
- .covers(kind, consumer_id, offset)
+ // A replica that cannot originate the prepare cannot record this
+ // progress anywhere a peer will ever see. `Ok(None)` would leave
+ // `complete_poll` applying the offset to local state alone, so the
+ // poll would report progress the group never agreed, and a later
+ // read on the primary would hand the same messages out again.
+ // Refusing keeps the outcome retriable on a replica that can commit.
+ if !consensus.is_primary() || !consensus.is_normal() ||
consensus.is_transferring() {
+ return Err(IggyError::TransientNotAccepted);
Review Comment:
nit: the role check runs before `covers`, so a caught-up backup refuses a
poll whose offset the durable table already covers. test `covers` first and
keep the refusal for uncovered offsets only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
