chore: rename unwritten -> pending

Author: dpc

loglog/src/std.rs

@@ -179,7 +179,7 @@ impl RawClient {
             size: EntrySize(u32::expect_from(raw_entry.len())),
         };
 
-        // TODO: instead of copy, use `write_vectored_all` when it stabilizes
+        // TODO(perf): instead of copy, use `write_vectored_all` when it stabilizes
         // https://github.com/rust-lang/rust/issues/70436
         args.write(&mut NoSeek::new(&mut buf)).expect("can't fail");
 

loglogd/src/node.rs

@@ -107,12 +107,12 @@ impl SealedSegments {
     }
 }
 #[derive(Debug)]
-pub struct EntriesInFlight {
+pub struct PendingEntries {
     /// Next log offset to give out to incoming entry
     next_available_log_offset: LogOffset,
 
     /// Entries that were already allocated but were not yet written to storage
-    unwritten: BTreeSet<LogOffset>,
+    entries: BTreeSet<LogOffset>,
 }
 
 #[derive(Error, Debug)]
@@ -141,7 +141,7 @@ pub struct NodeShared {
     entry_buffer_pool: Mutex<Vec<Vec<u8>>>,
 
     /// Entries already received and allocated, but not yet written
-    entries_in_flight: RwLock<EntriesInFlight>,
+    pending_entries: RwLock<PendingEntries>,
 
     /// Segment currently being written to
     open_segments: RwLock<OpenSegments>,
@@ -192,10 +192,10 @@ impl NodeShared {
 
     /// Allocate a space in the event stream and return allocation id
     pub fn allocate_new_entry(self: &Arc<Self>, len: EntrySize) -> AllocationId {
-        let mut write_in_flight = self.entries_in_flight.write().expect("locking failed");
+        let mut write_in_flight = self.pending_entries.write().expect("locking failed");
 
         let offset = write_in_flight.next_available_log_offset;
-        let was_inserted = write_in_flight.unwritten.insert(offset);
+        let was_inserted = write_in_flight.entries.insert(offset);
         debug_assert!(was_inserted);
         let alloc = AllocationId {
             offset,
@@ -210,24 +210,24 @@ impl NodeShared {
         alloc
     }
 
-    pub fn get_first_unwritten_log_offset(&self) -> LogOffset {
-        let read_in_flight = self.entries_in_flight.read().expect("Locking failed");
+    pub fn get_first_pending_log_offset(&self) -> LogOffset {
+        let read_in_flight = self.pending_entries.read().expect("Locking failed");
         read_in_flight
-            .unwritten
+            .entries
             .iter()
             .next()
             .copied()
             .unwrap_or(read_in_flight.next_available_log_offset)
     }
 
-    pub fn get_first_unwritten_entry_offset_ge(
+    pub fn get_first_pending_entry_offset_ge(
         &self,
         offset_inclusive: LogOffset,
     ) -> Option<LogOffset> {
-        self.entries_in_flight
+        self.pending_entries
             .read()
             .expect("Locking failed")
-            .unwritten
+            .entries
             .range(offset_inclusive..)
             .next()
             .copied()
@@ -335,12 +335,12 @@ impl Node {
             current_term: TermId(0),
             entry_buffer_pool: Mutex::new(vec![]),
             params: params.clone(),
-            entries_in_flight: RwLock::new(EntriesInFlight {
+            pending_entries: RwLock::new(PendingEntries {
                 next_available_log_offset: segments
                     .last()
                     .map(|s| s.content_meta.end_log_offset)
                     .unwrap_or_default(),
-                unwritten: BTreeSet::new(),
+                entries: BTreeSet::new(),
             }),
             sealed_segments: RwLock::new(SealedSegments::from_iter(segments)),
             open_segments: RwLock::new(OpenSegments {

loglogd/src/node/request_handler.rs

@@ -413,7 +413,7 @@ impl RequestHandlerInner {
 
             let stream_fd = stream.as_raw_fd();
             let bytes_written = tokio::task::spawn_blocking(move || -> io::Result<u64> {
-                // TODO(perf): should we cache these somewhere in some LRU or something?
+                // TODO(perf): we should cache FDs to open sealed files somewhere in some LRU
                 let file = std::fs::File::open(segment.file_meta.path())?;
                 let file_fd = file.as_raw_fd();
                 send_file_to_stream(file_fd, file_offset, stream_fd, bytes_to_send)

loglogd/src/node/segment_sealer.rs

@@ -36,10 +36,10 @@ impl SegmentSealer {
                     let _last_written_entry_log_offset =
                         last_written_entry_log_offset_rx.wait_timeout(Duration::from_secs(1));
 
-                    let first_unwritten_log_offset = shared.get_first_unwritten_log_offset();
+                    let first_pending_log_offset = shared.get_first_pending_log_offset();
 
                     // Termination condition. We sealed all the pending data and...
-                    if fsynced_log_offset == first_unwritten_log_offset {
+                    if fsynced_log_offset == first_pending_log_offset {
                         // if we're done with all the pending work, and the writting loop
                         // is done too, we can finish.
                         if shared.is_segment_writer_done.load(Ordering::SeqCst) {
@@ -48,19 +48,19 @@ impl SegmentSealer {
                         continue;
                     }
 
-                    Self::seal_segments_up_to(&shared, first_unwritten_log_offset);
+                    Self::seal_segments_up_to(&shared, first_pending_log_offset);
 
                     // Only after successfully looping and fsyncing and/or closing all matching open segments
                     // we update `last_fsynced_log_offset`
-                    fsynced_log_offset = first_unwritten_log_offset;
-                    shared.update_fsynced_log_offset(first_unwritten_log_offset);
+                    fsynced_log_offset = first_pending_log_offset;
+                    shared.update_fsynced_log_offset(first_pending_log_offset);
                 }
             }),
         }
     }
 
-    fn seal_segments_up_to(shared: &Arc<NodeShared>, first_unwritten_log_offset: LogOffset) {
-        'inner: loop {
+    fn seal_segments_up_to(shared: &Arc<NodeShared>, first_pending_log_offset: LogOffset) {
+        loop {
             let (first_segment, second_segment_start) = {
                 let read_open_segments = shared.open_segments.read().expect("Locking failed");
 
@@ -75,30 +75,31 @@ impl SegmentSealer {
             };
 
             let Some((open_segment_start, open_segment)) = first_segment else {
-                break 'inner;
+                return;
             };
 
             debug!(
                 segment_id = %open_segment.id,
                 segment_start_log_offset = %open_segment_start,
-                first_unwritten_log_offset = %first_unwritten_log_offset,
+                first_unwritten_log_offset = %first_pending_log_offset,
                 "fsync"
             );
             if let Err(e) = nix::unistd::fsync(open_segment.fd.as_raw_fd()) {
                 warn!(error = %e, "Could not fsync opened segment file");
-                break 'inner;
+                return;
             }
 
             let Some(open_segment_end ) = second_segment_start else {
-                // Until we have at lest two open segments, we won't close the first one.
-                // It's not a big deal, and avoids having to calculate the end of first
-                // segment.
-                break 'inner
+                // Until we have the second open segments, we do not know what is the end
+                // of the first one.
+                // It's not a big deal, we can wait. Nothing depends on it and we will
+                // happily serve data from open segments to the readers.
+                return;
             };
 
             // We can't seal a segment if there are still any pending writes to it.
-            if first_unwritten_log_offset < open_segment_end {
-                break 'inner;
+            if first_pending_log_offset < open_segment_end {
+                return;
             }
 
             let open_segment_offset_size = open_segment_end - open_segment_start;
@@ -129,7 +130,7 @@ impl SegmentSealer {
                 debug_assert!(removed.is_some());
             }
 
-            // continue 'inner - there might be more
+            // continue - there might be more
         }
     }
 }

loglogd/src/node/segment_writer.rs

@@ -177,13 +177,13 @@ impl WriteLoopInner {
             let next_segment_start_log_offset = if let Some(last_segment_info) = last_segment_info {
                 // It would be a mistake to consider current `entry_log_offset` as a beginning of
                 // next segment as we might have arrived here out of order. Instead - we can use
-                // `unwritten` to find first unwritten entry for new segment, and thus its starting
+                // `pending` to find first not yet written entry for new segment, and thus its starting
                 // byte.
                 self.shared
-                    .get_first_unwritten_entry_offset_ge(
+                    .get_first_pending_entry_offset_ge(
                         last_segment_info.end_of_allocation_log_offset,
                     )
-                    .expect("at very least this entry should be in `unwritten`")
+                    .expect("at very least this entry should be in `pending`")
             } else {
                 self.shared
                     .get_sealed_segments_end_log_offset()
@@ -244,12 +244,9 @@ impl WriteLoopInner {
 
     /// Mark the entry as written to disk (thought possibly not yet fsynced)
     pub fn mark_entry_written_no_notify(self: &Arc<Self>, log_offset: LogOffset) {
-        let mut write_in_flight = self
-            .shared
-            .entries_in_flight
-            .write()
-            .expect("Locking failed");
-        let was_removed = write_in_flight.unwritten.remove(&log_offset);
+        let mut pending_entries_write =
+            self.shared.pending_entries.write().expect("Locking failed");
+        let was_removed = pending_entries_write.entries.remove(&log_offset);
         debug_assert!(was_removed);
     }