add tests

src/nvm/file.rs

@@ -487,7 +487,7 @@ mod tests {
         let mut parent = dir.as_ref();
         while let Some(p) = parent.parent() {
             parent = p;
-        };
+        }
         assert!(create_parent_directories(parent).is_ok());
         Ok(())
     }

src/storage/allocator/data_portion_allocator.rs

@@ -195,7 +195,7 @@ impl DataPortionAllocator {
     //    現在の実装では `nth(0)` を用いているため、
     //    フリーリスト内の相異なる部分領域が互いに素であるという前提が必要である。
     //    ただしこの前提は通常のCannyLSの使用であれば成立する。
-    fn is_allocated_portion(&self, portion: &DataPortion) -> bool {
+    pub(crate) fn is_allocated_portion(&self, portion: &DataPortion) -> bool {
         let key = EndBasedFreePortion(FreePortion::new(portion.start, 0));
         if let Some(next) = self.end_to_free.range((Excluded(&key), Unbounded)).nth(0) {
             // 終端位置が `portion.start` を超えるfree portionのうち最小のもの `next` については

src/storage/builder.rs

@@ -207,7 +207,8 @@ impl StorageBuilder {
         ))?;
 
         // データ領域を準備
-        let data_region = DataRegion::new(&self.metrics, allocator, data_nvm);
+        let mut data_region = DataRegion::new(&self.metrics, allocator, data_nvm);
+        data_region.enable_safe_release_mode(self.safe_release_mode);
 
         let metrics = StorageMetrics::new(
             &self.metrics,

src/storage/data_region.rs

@@ -43,8 +43,10 @@ where
 
     /// 安全にリソースを解放するモードに移行するためのメソッド。
     /// * `true`を渡すと、安全な解放モードに入る。
-    ///    * 安全な解放については [issue28](https://github.com/frugalos/cannyls/issue28) を参考のこと。
-    /// * `false`を渡すと、削除されたデータポーションが即座にアロケータから解放される。
+    ///    * 安全な解放については [wiki](https://github.com/frugalos/cannyls/wiki/Safe-Release-Mode) を参考のこと。
+    /// * `false`を渡すと、従来の解放モードに入る。
+    ///    * このモードでは、削除されたデータポーションが即座にアロケータから解放される。
+    ///    * [issue28](https://github.com/frugalos/cannyls/issues28)があるため安全ではない。
     pub fn enable_safe_release_mode(&mut self, enabling: bool) {
         if !enabling && !self.reserved_portions.is_empty() {
             // 削除対象ポーションが存在する状況で即時削除モードに切り替える場合は
@@ -54,6 +56,13 @@ where
         self.safe_release_mode = enabling;
     }
 
+    /// 安全にリソースを解放するモードに入っているかどうかを返す。
+    /// * `true`なら安全な解放モード
+    /// * `false`なら従来の解放モード
+    pub fn is_in_safe_release_mode(&self) -> bool {
+        self.safe_release_mode
+    }
+
     /// データ領域のメトリクスを返す.
     pub fn metrics(&self) -> &DataRegionMetrics {
         &self.metrics
@@ -113,6 +122,12 @@ where
         }
     }
 
+    /// 解放を遅延させているデータポーションをアロケータに送ることで全て解放する。
+    ///
+    /// # 安全解放モードで呼び出す前提条件
+    /// 解放されるデータポーションに対応する削除レコードが、全て永続化されていること。
+    ///
+    /// 通常の解放モードで呼び出しても効果はない。
     pub fn release_pending_portions(&mut self) {
         if self.safe_release_mode {
             for p in ::std::mem::replace(&mut self.reserved_portions, Vec::new()) {
@@ -132,6 +147,11 @@ where
     fn block_count(&self, size: u32) -> u32 {
         (size + u32::from(self.block_size.as_u16()) - 1) / u32::from(self.block_size.as_u16())
     }
+
+    #[cfg(test)]
+    pub fn is_allocated_portion(&self, portion: &DataPortion) -> bool {
+        self.allocator.is_allocated_portion(portion)
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -202,17 +222,23 @@ mod tests {
     use metrics::DataAllocatorMetrics;
     use nvm::MemoryNvm;
 
+    macro_rules! make_data_region_on_memory {
+        ($capacity:expr, $block_size:expr) => {{
+            let metrics = MetricBuilder::new();
+            let allocator = track!(DataPortionAllocator::build(
+                DataAllocatorMetrics::new(&metrics, $capacity, $block_size),
+                iter::empty(),
+            ))?;
+            let nvm = MemoryNvm::new(vec![0; $capacity as usize]);
+            DataRegion::new(&metrics, allocator, nvm)
+        }};
+    }
+
     #[test]
     fn data_region_works() -> TestResult {
         let capacity = 10 * 1024;
         let block_size = BlockSize::min();
-        let metrics = MetricBuilder::new();
-        let allocator = track!(DataPortionAllocator::build(
-            DataAllocatorMetrics::new(&metrics, capacity, block_size),
-            iter::empty(),
-        ))?;
-        let nvm = MemoryNvm::new(vec![0; capacity as usize]);
-        let mut region = DataRegion::new(&metrics, allocator, nvm);
+        let mut region = make_data_region_on_memory!(capacity, block_size);
 
         // put
         let mut data = DataRegionLumpData::new(3, block_size);
@@ -226,4 +252,53 @@ mod tests {
         );
         Ok(())
     }
+
+    #[test]
+    fn enabling_and_confirming_safe_release_mode_work() -> TestResult {
+        let capacity = 10 * 1024;
+        let block_size = BlockSize::min();
+        let mut region = make_data_region_on_memory!(capacity, block_size);
+
+        // デフォルトでは安全解放モードを使わない。
+        assert_eq!(region.is_in_safe_release_mode(), false);
+
+        region.enable_safe_release_mode(true);
+        assert_eq!(region.is_in_safe_release_mode(), true);
+
+        Ok(())
+    }
+
+    #[test]
+    fn delayed_releasing_works() -> TestResult {
+        let capacity = 10 * 1024;
+        let block_size = BlockSize::min();
+        let mut region = make_data_region_on_memory!(capacity, block_size);
+
+        region.enable_safe_release_mode(true);
+
+        // put
+        let mut data = DataRegionLumpData::new(3, block_size);
+        data.as_bytes_mut().copy_from_slice("foo".as_bytes());
+        let portion = track!(region.put(&data))?;
+
+        // get
+        assert_eq!(
+            region.get(portion).ok().map(|d| d.as_bytes().to_owned()),
+            Some("foo".as_bytes().to_owned())
+        );
+
+        region.delete(portion.clone());
+
+        // まだ解放されていない。
+        assert_eq!(region.allocator.is_allocated_portion(&portion), true);
+
+        assert_eq!(region.reserved_portions.len(), 1);
+
+        region.release_pending_portions();
+
+        // 解放された。
+        assert_eq!(region.allocator.is_allocated_portion(&portion), false);
+
+        Ok(())
+    }
 }

src/storage/journal/nvm_buffer.rs

@@ -82,10 +82,6 @@ impl<N: NonVolatileMemory> JournalNvmBuffer<N> {
         &self.inner
     }
 
-    pub fn is_dirty(&self) -> bool {
-        self.maybe_dirty
-    }
-
     fn is_dirty_area(&self, offset: u64, length: usize) -> bool {
         if !self.maybe_dirty || length == 0 || self.write_buf.is_empty() {
             return false;
@@ -184,7 +180,7 @@ impl<N: NonVolatileMemory> Seek for JournalNvmBuffer<N> {
 impl<N: NonVolatileMemory> Read for JournalNvmBuffer<N> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         if self.is_dirty_area(self.position, buf.len()) {
-            track!(self.flush_write_buf())?;
+            track!(self.sync())?;
         }
 
         let aligned_start = self.block_size().floor_align(self.position);

src/storage/journal/region.rs

@@ -44,6 +44,11 @@ impl<N> JournalRegion<N>
 where
     N: NonVolatileMemory,
 {
+    #[cfg(test)]
+    pub fn options(&self) -> &JournalRegionOptions {
+        &self.options
+    }
+
     pub fn journal_entries(&mut self) -> Result<(u64, u64, u64, Vec<JournalEntry>)> {
         self.ring_buffer.journal_entries()
     }
@@ -152,7 +157,6 @@ where
         if self.gc_queue.is_empty() {
             track!(self.fill_gc_queue())?;
         } else if self.sync_countdown != self.options.sync_interval {
-            // バッファにエントリがある場合なので同期してしまう。
             track!(self.sync())?;
         } else {
             for _ in 0..GC_COUNT_IN_SIDE_JOB {
@@ -168,8 +172,18 @@ where
         &self.metrics
     }
 
-    pub fn is_dirty(&self) -> bool {
-        self.ring_buffer.is_dirty()
+    /// ジャーナルバッファが同期され永続化された直後の状態かどうかを返す。
+    ///
+    /// この状態は、厳密には次を意味する。
+    /// 1. 現時点までに以下四種類の操作でバッファに書き込まれたジャーナルエントリは
+    /// 全てDiskに同期されている。
+    ///   (a) records_put
+    ///   (b) records_embed
+    ///   (c) records_delete
+    ///   (d) records_delete_range
+    /// 2. ジャーナルバッファは空である。
+    pub fn is_just_synced(&self) -> bool {
+        self.sync_countdown == self.options.sync_interval
     }
 
     /// GC処理を一単位実行する.
@@ -271,10 +285,11 @@ where
     }
 
     fn try_sync(&mut self) -> Result<()> {
+        assert!(self.sync_countdown >= 1);
+        self.sync_countdown -= 1;
+
         if self.sync_countdown == 0 {
             track!(self.sync())?;
-        } else {
-            self.sync_countdown -= 1;
         }
         Ok(())
     }

src/storage/journal/ring_buffer.rs

@@ -94,10 +94,6 @@ impl<N: NonVolatileMemory> JournalRingBuffer<N> {
         &self.metrics
     }
 
-    pub fn is_dirty(&self) -> bool {
-        self.nvm.is_dirty()
-    }
-
     /// 指定位置に埋め込まれたlumpデータの読み込みを行う.
     ///
     /// データの妥当性検証は`cannyls`内では行わない.

src/storage/mod.rs

@@ -83,18 +83,23 @@ where
     data_region: DataRegion<N>,
     lump_index: LumpIndex,
     metrics: StorageMetrics,
-    safe_release_mode: bool,
 }
 impl<N> Storage<N>
 where
     N: NonVolatileMemory,
 {
     /// [issue28](https://github.com/frugalos/cannyls/issue28)
     pub fn enable_safe_release_mode(&mut self, enabling: bool) {
-        self.safe_release_mode = enabling;
         self.data_region.enable_safe_release_mode(enabling);
     }
 
+    /// 安全にリソースを解放するモードに入っているかどうかを返す。
+    /// * `true`なら安全な解放モード
+    /// * `false`なら従来の解放モード
+    pub fn is_in_safe_release_mode(&self) -> bool {
+        self.data_region.is_in_safe_release_mode()
+    }
+
     pub(crate) fn new(
         header: StorageHeader,
         journal_region: JournalRegion<N>,
@@ -108,7 +113,6 @@ where
             data_region,
             lump_index,
             metrics,
-            safe_release_mode: false,
         }
     }
 
@@ -184,8 +188,14 @@ where
         self.lump_index.list_range(range)
     }
 
+    /// ジャーナル領域が直前の操作に伴って同期済みであるならば
+    /// 遅延させている削除済みポーションを全て解放する。
+    ///
+    /// syncは実際にはバッファのflushとディスク同期の両方を行うため
+    /// is_just_synced() == trueならば
+    /// この時点までの全てのジャーナルレコードが同期済みである。
     fn release_pending_portions(&mut self) {
-        if !self.journal_region.is_dirty() {
+        if self.journal_region.is_just_synced() {
             self.data_region.release_pending_portions();
         }
     }
@@ -226,7 +236,7 @@ where
         }
         self.metrics.put_lumps_at_running.increment();
 
-        if self.safe_release_mode {
+        if self.is_in_safe_release_mode() {
             self.release_pending_portions();
         }
         Ok(!updated)
@@ -243,7 +253,7 @@ where
     /// 以後はこのインスタンスの使用を中止するのが望ましい.
     pub fn delete(&mut self, lump_id: &LumpId) -> Result<bool> {
         let result = track!(self.delete_if_exists(lump_id, true))?;
-        if self.safe_release_mode {
+        if self.is_in_safe_release_mode() {
             self.release_pending_portions();
         }
         Ok(result)
@@ -286,7 +296,7 @@ where
             }
         }
 
-        if self.safe_release_mode {
+        if self.is_in_safe_release_mode() {
             track!(self.journal_region.sync())?;
             self.release_pending_portions();
         }
@@ -873,4 +883,64 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn safe_releasing_works() -> TestResult {
+        let dir = track_io!(TempDir::new("cannyls_test"))?;
+
+        let nvm = track!(FileNvm::create(
+            dir.path().join("test.lusf"),
+            BlockSize::min().ceil_align(512 * 0x1000 * 10)
+        ))?;
+        let mut storage = track!(StorageBuilder::new()
+            .journal_region_ratio(0.5)
+            .enable_safe_release_mode()
+            .create(nvm))?;
+        assert!(storage.is_in_safe_release_mode());
+
+        assert_eq!(storage.journal_region.options().sync_interval, 0x1000);
+
+        let mut portions = Vec::new();
+
+        for i in 0..(0x1000 / 2 - 1) {
+            track!(storage.put(&LumpId::new(i), &zeroed_data(42)))?;
+            let portion = storage
+                .lump_index
+                .get(&LumpId::new(i))
+                .expect("should succeed");
+            portions.push(portion.clone());
+            track!(storage.delete(&LumpId::new(i)))?;
+        }
+        track!(storage.put(&LumpId::new(10000), &zeroed_data(42)))?;
+        let portion = storage
+            .lump_index
+            .get(&LumpId::new(10000))
+            .expect("should succeed");
+        portions.push(portion.clone());
+
+        for p in &portions {
+            if let Portion::Data(portion) = p {
+                assert!(storage.data_region.is_allocated_portion(portion));
+            } else {
+                unreachable!("since we've put unembedded lump data");
+            }
+        }
+
+        assert_eq!(storage.journal_region.is_just_synced(), false);
+
+        // このDeleteによって0x1000個目のエントリが書き込まれるため
+        // ジャーナル領域の同期が行われる。
+        track!(storage.delete(&LumpId::new(10000)))?;
+        assert_eq!(storage.journal_region.is_just_synced(), true);
+
+        for p in &portions {
+            if let Portion::Data(portion) = p {
+                assert!(!storage.data_region.is_allocated_portion(portion));
+            } else {
+                unreachable!("since we've put unembedded lump data");
+            }
+        }
+
+        Ok(())
+    }
 }