Fix for #151. (#152)

* Fix for #151.

* CR feedback.

src/llfs/basic_log_storage_reader.hpp

@@ -52,15 +52,15 @@ class BasicLogStorageReader : public LogDevice::Reader
     BATT_CHECK(!slot_less_than(this->offset_, this->driver_.get_trim_pos()))
         << "offset=" << this->offset_ << " trim_pos=" << this->driver_.get_trim_pos();
 
-    return data_;
+    return this->data_;
   }
 
   slot_offset_type slot_offset() override
   {
-    return offset_;
+    return this->offset_;
   }
 
-  void consume(std::size_t byte_count) override
+  void consume(usize byte_count) override
   {
     BATT_CHECK_LE(byte_count, this->data_.size());
 
@@ -100,6 +100,15 @@ class BasicLogStorageReader : public LogDevice::Reader
         });
   }
 
+  std::function<void(std::ostream&)> debug_info() override
+  {
+    return [this](std::ostream& out) {
+      out << BATT_INSPECT(this->mode_) << BATT_INSPECT(this->driver_.get_trim_pos())
+          << BATT_INSPECT(this->driver_.get_flush_pos())
+          << BATT_INSPECT(this->driver_.get_commit_pos());
+    };
+  }
+
   //=#=#==#==#===============+=+=+=+=++=++++++++++++++-++-+--+-+----+---------------
  private:
   slot_offset_type get_upper_bound() const
@@ -138,11 +147,8 @@ class BasicLogStorageReader : public LogDevice::Reader
 
   void refresh_view(slot_offset_type upper_bound)
   {
-    data_ = [&] {
-      ConstBuffer b = this->context_.buffer_.get(this->offset_);
-
-      return ConstBuffer{b.data(), slot_distance(this->offset_, upper_bound)};
-    }();
+    const ConstBuffer b = this->context_.buffer_.get(this->offset_);
+    this->data_ = ConstBuffer{b.data(), slot_clamp_distance(this->offset_, upper_bound)};
   }
 
   //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -

src/llfs/basic_ring_buffer_log_device.hpp

@@ -126,15 +126,15 @@ inline u64 BasicRingBufferLogDevice<Impl>::capacity() const
 template <class Impl>
 inline u64 BasicRingBufferLogDevice<Impl>::size() const
 {
-  return slot_distance(this->driver_.get_trim_pos(), this->driver_.get_commit_pos());
+  return slot_clamp_distance(this->driver_.get_trim_pos(), this->driver_.get_commit_pos());
 }
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -
 //
 template <class Impl>
 inline Status BasicRingBufferLogDevice<Impl>::trim(slot_offset_type slot_lower_bound)
 {
-  BATT_CHECK_LE(this->driver_.get_trim_pos(), slot_lower_bound);
+  LLFS_CHECK_SLOT_GE(slot_lower_bound, this->driver_.get_trim_pos());
 
   return this->driver_.set_trim_pos(slot_lower_bound);
 }
@@ -226,7 +226,7 @@ class BasicRingBufferLogDevice<Impl>::WriterImpl : public LogDevice::Writer
     const slot_offset_type readable_end = this->device_->driver_.get_commit_pos();
 
     const usize space_available =
-        this->device_->buffer_.size() - slot_distance(readable_begin, readable_end);
+        this->device_->buffer_.size() - LLFS_CHECKED_SLOT_DISTANCE(readable_begin, readable_end);
 
     return space_available;
   }

src/llfs/file_log_driver.cpp

@@ -119,7 +119,7 @@ StatusOr<std::unique_ptr<FileLogDevice>> FileLogDriver::recover(const Location&
 
   // Whatever wasn't consumed by the scan function must be truncated from the end of the log.
   //
-  const auto bytes_to_truncate = slot_distance(*scan_status, driver.get_flush_pos());
+  const auto bytes_to_truncate = LLFS_CHECKED_SLOT_DISTANCE(*scan_status, driver.get_flush_pos());
   auto truncate_status = active_file->truncate(bytes_to_truncate);
   BATT_REQUIRE_OK(truncate_status);
 

src/llfs/file_log_driver/flush_task_main.cpp

@@ -31,7 +31,7 @@ void FileLogDriver::FlushTaskMain::operator()()
     auto local_flush_pos = this->shared_state_.flush_pos.get_value();
     for (;;) {
       ConstBuffer bytes_to_flush{this->buffer_.get(local_flush_pos).data(),
-                                 slot_distance(local_flush_pos, local_commit_pos)};
+                                 slot_clamp_distance(local_flush_pos, local_commit_pos)};
 
       if (bytes_to_flush.size() == 0) {
         // We've caught up!  Put this task to sleep awaiting more data to flush.

src/llfs/log_device.hpp

@@ -299,6 +299,16 @@ class LogDevice::Reader
   /** \brief Wait for the log to reach the specified state.
    */
   virtual Status await(LogDevice::ReaderEvent event) = 0;
+
+  /** \brief (Optional) Returns a function that prints human-readable debug information about the
+   * Reader.
+   */
+  virtual std::function<void(std::ostream&)> debug_info()
+  {
+    return [](std::ostream& out) {
+      out << "(Not Implemented)";
+    };
+  }
 };
 
 /** \brief Open the log without scanning its contents.

src/llfs/log_device_snapshot.hpp

@@ -46,8 +46,9 @@ class LogDeviceSnapshot : public boost::equality_comparable<LogDeviceSnapshot>
       return device.driver().get_commit_pos();
     }();
 
-    ConstBuffer src = batt::resize_buffer(device.buffer_.get(snapshot.trim_pos_),
-                                          slot_distance(snapshot.trim_pos_, snapshot.commit_pos_));
+    ConstBuffer src =
+        batt::resize_buffer(device.buffer_.get(snapshot.trim_pos_),
+                            LLFS_CHECKED_SLOT_DISTANCE(snapshot.trim_pos_, snapshot.commit_pos_));
 
     snapshot.byte_storage_.reset(new u8[src.size()]);
 
@@ -87,7 +88,7 @@ class LogDeviceSnapshot : public boost::equality_comparable<LogDeviceSnapshot>
 
   usize size() const
   {
-    return slot_distance(this->trim_pos(), this->commit_pos());
+    return LLFS_CHECKED_SLOT_DISTANCE(this->trim_pos(), this->commit_pos());
   }
 
   const u8* bytes() const

src/llfs/memory_log_device.cpp

@@ -56,7 +56,7 @@ usize MemoryLogStorageDriver::unflushed_size() noexcept
   if (this->is_auto_flush()) {
     return 0;
   }
-  return slot_distance(this->get_flush_pos(), this->get_commit_pos());
+  return slot_clamp_distance(this->get_flush_pos(), this->get_commit_pos());
 }
 
 //==#==========+==+=+=++=+++++++++++-+-+--+----- --- -- -  -  -   -

src/llfs/page_recycler_options.cpp

@@ -110,8 +110,8 @@ u64 PageRecyclerOptions::recycle_task_target() const
 bool PageRecyclerOptions::info_needs_refresh(slot_offset_type last_info_refresh_slot_lower_bound,
                                              LogDevice& log_device) const
 {
-  return (slot_distance(last_info_refresh_slot_lower_bound,
-                        log_device.slot_range(LogReadMode::kSpeculative).upper_bound) +
+  return (slot_clamp_distance(last_info_refresh_slot_lower_bound,
+                              log_device.slot_range(LogReadMode::kSpeculative).upper_bound) +
           this->info_slot_size()) >= (log_device.capacity() / this->info_refresh_rate());
 }
 

src/llfs/slot.hpp

@@ -110,14 +110,41 @@ inline slot_offset_type slot_max(slot_offset_type first, slot_offset_type second
   return first;
 }
 
-inline std::size_t slot_distance(slot_offset_type x, slot_offset_type y)
+/** \brief Returns the (signed) difference between two slot offsets.
+ *
+ * If slot_less_than(y, x), the returned value will be negative; otherwise, it will be non-negative.
+ */
+inline isize slot_difference(slot_offset_type x, slot_offset_type y)
+{
+  return (isize)y - (isize)x;
+}
+
+/** \brief Returns the (unsigned) absolute value of the difference between two slot offsets.
+ */
+inline usize slot_abs_distance(slot_offset_type x, slot_offset_type y)
+{
+  if (slot_less_than(y, x)) {
+    return x - y;
+  }
+  return y - x;
+}
+
+/** \brief Returns the distance from x to y, if y is not before x; otherwise returns 0.
+ */
+inline usize slot_clamp_distance(slot_offset_type x, slot_offset_type y)
 {
   if (slot_less_than(y, x)) {
-    return slot_distance(y, x);
+    return 0;
   }
   return y - x;
 }
 
+[[deprecated("Use one of these instead: slot_abs_distance, slot_difference, slot_clamp_distance")]]
+inline usize slot_distance(slot_offset_type x, slot_offset_type y)
+{
+  return slot_abs_distance(x, y);
+}
+
 // Sets `active_offset` to at least `min_offset`.  Returns the distance between the old offset and
 // the new offset, if any.
 //
@@ -128,7 +155,7 @@ inline slot_offset_type clamp_min_slot(batt::Watch<slot_offset_type>& active_off
   active_offset.modify_if(
       [min_offset, &delta](slot_offset_type current_offset) -> Optional<slot_offset_type> {
         if (slot_less_than(current_offset, min_offset)) {
-          delta = slot_distance(current_offset, min_offset);
+          delta = min_offset - current_offset;
           return min_offset;
         }
         delta = 0;
@@ -295,6 +322,31 @@ struct SlotRangeOrder {
   }
 };
 
+//=#=#==#==#===============+=+=+=+=++=++++++++++++++-++-+--+-+----+---------------
+
+#define LLFS_CHECK_SLOT_LT(first, second)                                                          \
+  BATT_CHECK(::llfs::slot_less_than((first), (second)))                                            \
+      << BATT_INSPECT(first) << BATT_INSPECT(second)
+
+#define LLFS_CHECK_SLOT_GE(first, second)                                                          \
+  BATT_CHECK(!::llfs::slot_less_than((first), (second)))                                           \
+      << BATT_INSPECT(first) << BATT_INSPECT(second)
+
+#define LLFS_CHECK_SLOT_GT(first, second) LLFS_CHECK_SLOT_LT(second, first)
+#define LLFS_CHECK_SLOT_LE(first, second) LLFS_CHECK_SLOT_GE(second, first)
+
+/** \brief Panics if second is slot_less_than first; otherwise, evaluates to slot_distance(first,
+ * second).
+ */
+#define LLFS_CHECKED_SLOT_DISTANCE(first, second)                                                  \
+  ([&]() -> ::llfs::slot_offset_type {                                                             \
+    const ::llfs::slot_offset_type first_offset = (first);                                         \
+    const ::llfs::slot_offset_type second_offset = (second);                                       \
+    BATT_CHECK(!::llfs::slot_less_than(second_offset, first_offset))                               \
+        << " " << #first << " == " << first_offset << " " << #second << " == " << second_offset;   \
+    return (second) - (first);                                                                     \
+  })()
+
 }  // namespace llfs
 
 #endif  // LLFS_SLOT_HPP

src/llfs/slot.test.cpp

@@ -51,4 +51,114 @@ TEST(SlotRangeSpecTest, ToString)
               ::testing::StrEq("SlotRangeSpec{.lower_bound=100, .upper_bound=200,}"));
 }
 
+TEST(SlotClampDistanceTest, Test)
+{
+  llfs::slot_offset_type a = 0;
+  llfs::slot_offset_type b = a - 100;
+  llfs::slot_offset_type c = a + 100;
+
+  EXPECT_EQ(llfs::slot_clamp_distance(a, a), 0);
+
+  EXPECT_EQ(llfs::slot_clamp_distance(a, c), 100);
+  EXPECT_EQ(llfs::slot_clamp_distance(b, c), 200);
+  EXPECT_EQ(llfs::slot_clamp_distance(b, a), 100);
+
+  EXPECT_EQ(llfs::slot_clamp_distance(a, b), 0);
+  EXPECT_EQ(llfs::slot_clamp_distance(c, b), 0);
+  EXPECT_EQ(llfs::slot_clamp_distance(c, a), 0);
+}
+
+TEST(SlotAbsDistanceTest, Test)
+{
+  llfs::slot_offset_type a = 0;
+  llfs::slot_offset_type b = a - 100;
+  llfs::slot_offset_type c = a + 100;
+
+  EXPECT_EQ(llfs::slot_abs_distance(a, a), 0);
+
+  EXPECT_EQ(llfs::slot_abs_distance(a, c), 100);
+  EXPECT_EQ(llfs::slot_abs_distance(b, c), 200);
+  EXPECT_EQ(llfs::slot_abs_distance(b, a), 100);
+
+  EXPECT_EQ(llfs::slot_abs_distance(a, b), 100);
+  EXPECT_EQ(llfs::slot_abs_distance(c, b), 200);
+  EXPECT_EQ(llfs::slot_abs_distance(c, a), 100);
+}
+
+TEST(SlotDifferenceTest, Test)
+{
+  llfs::slot_offset_type a = 0;
+  llfs::slot_offset_type b = a - 100;
+  llfs::slot_offset_type c = a + 100;
+
+  EXPECT_EQ(llfs::slot_difference(a, a), 0);
+
+  EXPECT_EQ(llfs::slot_difference(a, c), 100);
+  EXPECT_EQ(llfs::slot_difference(b, c), 200);
+  EXPECT_EQ(llfs::slot_difference(b, a), 100);
+
+  EXPECT_EQ(llfs::slot_difference(a, b), -100);
+  EXPECT_EQ(llfs::slot_difference(c, b), -200);
+  EXPECT_EQ(llfs::slot_difference(c, a), -100);
+}
+
+TEST(SlotCheckSlotTest, OkTests)
+{
+  llfs::slot_offset_type a = 0;
+  llfs::slot_offset_type b = a - 100;
+  llfs::slot_offset_type c = a + 100;
+
+  LLFS_CHECK_SLOT_LT(b, a);
+  LLFS_CHECK_SLOT_LT(a, c);
+  LLFS_CHECK_SLOT_LT(b, c);
+
+  LLFS_CHECK_SLOT_GT(a, b);
+  LLFS_CHECK_SLOT_GT(c, a);
+  LLFS_CHECK_SLOT_GT(c, b);
+
+  LLFS_CHECK_SLOT_LE(a, a);
+  LLFS_CHECK_SLOT_LE(b, a);
+  LLFS_CHECK_SLOT_LE(a, c);
+  LLFS_CHECK_SLOT_LE(b, c);
+
+  LLFS_CHECK_SLOT_GE(a, a);
+  LLFS_CHECK_SLOT_GE(a, b);
+  LLFS_CHECK_SLOT_GE(c, a);
+  LLFS_CHECK_SLOT_GE(c, b);
+
+  EXPECT_EQ(LLFS_CHECKED_SLOT_DISTANCE(a, a), 0);
+  EXPECT_EQ(LLFS_CHECKED_SLOT_DISTANCE(a, c), 100);
+  EXPECT_EQ(LLFS_CHECKED_SLOT_DISTANCE(b, c), 200);
+  EXPECT_EQ(LLFS_CHECKED_SLOT_DISTANCE(b, a), 100);
+}
+
+TEST(SlotCheckSlotTest, DeathTests)
+{
+  llfs::slot_offset_type a = 0;
+  llfs::slot_offset_type b = a - 100;
+  llfs::slot_offset_type c = a + 100;
+
+  EXPECT_DEATH(LLFS_CHECK_SLOT_LT(a, a) << " A_lt_A", "a == 0 a == 0 A_lt_A");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_LT(a, b) << " A_lt_B", "a == 0 b == 18446744073709551516 A_lt_B");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_LT(c, a) << " C_lt_A", "c == 100 a == 0 C_lt_A");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_LT(c, b) << " C_lt_B", "c == 100 b == 18446744073709551516 C_lt_B");
+
+  EXPECT_DEATH(LLFS_CHECK_SLOT_GT(a, a) << " A_gt_A", "a == 0 a == 0 A_gt_A");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_GT(b, a) << " B_gt_A", "a == 0 b == 18446744073709551516 B_gt_A");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_GT(a, c) << " A_gt_C", "c == 100 a == 0 A_gt_C");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_GT(b, c) << " B_gt_C", "c == 100 b == 18446744073709551516 B_gt_C");
+
+  EXPECT_DEATH(LLFS_CHECK_SLOT_LE(a, b) << " A_le_B", "b == 18446744073709551516 a == 0 A_le_B");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_LE(c, a) << " C_le_A", "a == 0 c == 100 C_le_A");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_LE(c, b) << " C_le_B", "b == 18446744073709551516 c == 100 C_le_B");
+
+  EXPECT_DEATH(LLFS_CHECK_SLOT_GE(b, a) << " B_ge_A", "b == 18446744073709551516 a == 0 B_ge_A");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_GE(a, c) << " A_ge_C", "a == 0 c == 100 A_ge_C");
+  EXPECT_DEATH(LLFS_CHECK_SLOT_GE(b, c) << " B_ge_C", "b == 18446744073709551516 c == 100 B_ge_C");
+
+  EXPECT_DEATH(LLFS_CHECKED_SLOT_DISTANCE(c, a), "c == 100 a == 0");
+  EXPECT_DEATH(LLFS_CHECKED_SLOT_DISTANCE(c, b), "c == 100 b == 18446744073709551516");
+  EXPECT_DEATH(LLFS_CHECKED_SLOT_DISTANCE(a, b), "a == 0 b == 18446744073709551516");
+}
+
 }  // namespace

src/llfs/slot_reader.cpp

@@ -104,11 +104,13 @@ StatusOr<SlotParse> SlotReader::parse_next(batt::WaitForResource wait_for_data)
     //
     const usize bytes_available_before = data_reader.bytes_available();
 
-    if (bytes_available_before != 0) {
+    if (bytes_available_before != 0 && !data_reader.at_end()) {
       BATT_CHECK_NE((int)(*data_reader.unread_begin()), 0)
           << BATT_INSPECT(bytes_available_before) << BATT_INSPECT(current_slot)
           << BATT_INSPECT(data.size()) << BATT_INSPECT(current_slot + data.size())
-          << BATT_INSPECT(this->slots_parsed_count_);
+          << BATT_INSPECT(this->slots_parsed_count_)
+          << BATT_INSPECT(this->log_reader_.slot_offset())
+          << BATT_INSPECT(this->log_reader_.data().size()) << this->log_reader_.debug_info();
     }
 
     Optional<u64> slot_body_size = data_reader.read_varint();