fxfs/object_store/allocator/

merge.rs

// Copyright 2021 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use crate::lsm_tree::merge::ItemOp::{Discard, Keep, Replace};
use crate::lsm_tree::merge::{MergeLayerIterator, MergeResult};
use crate::lsm_tree::types::{Item, LayerIterator};
use crate::object_store::allocator::{AllocatorKey, AllocatorValue};
use anyhow::Error;
use std::collections::HashSet;

pub fn merge(
    left: &MergeLayerIterator<'_, AllocatorKey, AllocatorValue>,
    right: &MergeLayerIterator<'_, AllocatorKey, AllocatorValue>,
) -> MergeResult<AllocatorKey, AllocatorValue> {
    // Wherever Replace is used below, it must not extend the *end* of the range for whichever item
    // is returned i.e. if replacing the left item, replacement.end <= left.end because otherwise we
    // might not merge records that come after that end point because the merger won't merge records
    // in the same layer

    /*  Case 1: Disjoint
     *    L:    |------------|
     *    R:                      |-----------|
     */
    if left.key().device_range.end < right.key().device_range.start {
        return MergeResult::EmitLeft;
    }

    /*  Case 2: Touching
     *    L:    |------------|
     *    R:                 |-----------|
     */
    if left.key().device_range.end == right.key().device_range.start {
        // We can only merge the range if the values are an exact match.
        if *left.value() == *right.value() {
            return MergeResult::Other {
                emit: None,
                left: Discard,
                right: Replace(
                    Item {
                        key: AllocatorKey {
                            device_range: left.key().device_range.start
                                ..right.key().device_range.end,
                        },
                        value: left.value().clone(),
                        sequence: std::cmp::min(left.sequence(), right.sequence()),
                    }
                    .boxed(),
                ),
            };
        } else {
            return MergeResult::EmitLeft;
        }
    }
    if left.key().device_range.start == right.key().device_range.start {
        /*  Case 3: Overlap with same start
         *    L:    |------------|
         *    R:    |-----------------|
         */
        if left.key().device_range.end < right.key().device_range.end {
            // The newer value eclipses the older.
            if left.layer_index < right.layer_index {
                return MergeResult::Other {
                    emit: None,
                    left: Keep,
                    right: if left.key().device_range.end == right.key().device_range.end {
                        Discard
                    } else {
                        Replace(
                            Item {
                                key: AllocatorKey {
                                    device_range: left.key().device_range.end
                                        ..right.key().device_range.end,
                                },
                                value: right.value().clone(),
                                sequence: right.sequence(),
                            }
                            .boxed(),
                        )
                    },
                };
            } else {
                // right is a newer Abs/None than left
                return MergeResult::Other { emit: None, left: Discard, right: Keep };
            }

        /*  Case 4: Overlap with same start
         *    L:    |-----------------|
         *    R:    |------------|
         */
        } else {
            // The newer value eclipses the older.
            if right.layer_index < left.layer_index {
                return MergeResult::Other {
                    emit: None,
                    left: if right.key().device_range.end == left.key().device_range.end {
                        Discard
                    } else {
                        Replace(
                            Item {
                                key: AllocatorKey {
                                    device_range: right.key().device_range.end
                                        ..left.key().device_range.end,
                                },
                                value: left.value().clone(),
                                sequence: left.sequence(),
                            }
                            .boxed(),
                        )
                    },
                    right: Keep,
                };
            } else {
                // right is a newer Abs/None than left
                return MergeResult::Other { emit: None, left: Keep, right: Discard };
            }
        }
    }
    /*  Case 5: Split off left prefix
     *    L:    |-----...
     *    R:         |-----...
     */
    debug_assert!(left.key().device_range.end >= right.key().device_range.start);
    MergeResult::Other {
        emit: Some(
            Item {
                key: AllocatorKey {
                    device_range: left.key().device_range.start..right.key().device_range.start,
                },
                value: left.value().clone(),
                sequence: left.sequence(),
            }
            .boxed(),
        ),
        left: Replace(
            Item {
                key: AllocatorKey {
                    device_range: right.key().device_range.start..left.key().device_range.end,
                },
                value: left.value().clone(),
                sequence: left.sequence(),
            }
            .boxed(),
        ),
        right: Keep,
    }
}

pub async fn filter_tombstones(
    iter: impl LayerIterator<AllocatorKey, AllocatorValue>,
) -> Result<impl LayerIterator<AllocatorKey, AllocatorValue>, Error> {
    Ok(iter.filter(|i| *i.value != AllocatorValue::None).await?)
}

pub async fn filter_marked_for_deletion(
    iter: impl LayerIterator<AllocatorKey, AllocatorValue>,
    marked_for_deletion: HashSet<u64>,
) -> Result<impl LayerIterator<AllocatorKey, AllocatorValue>, Error> {
    Ok(iter
        .filter(move |i| {
            if let AllocatorValue::Abs { owner_object_id, .. } = i.value {
                !marked_for_deletion.contains(owner_object_id)
            } else {
                true
            }
        })
        .await?)
}

#[cfg(test)]
mod tests {
    use crate::lsm_tree::cache::NullCache;
    use crate::lsm_tree::types::{Item, ItemRef, LayerIterator};
    use crate::lsm_tree::{LSMTree, Query};
    use crate::object_handle::INVALID_OBJECT_ID;
    use crate::object_store::allocator::merge::{filter_tombstones, merge};
    use crate::object_store::allocator::{AllocatorKey, AllocatorValue};
    use std::ops::Range;

    // Tests merge logic given (range, delta and object_id) for left, right and expected output.
    async fn test_merge(
        left: (Range<u64>, AllocatorValue),
        right: (Range<u64>, AllocatorValue),
        expected: &[(Range<u64>, AllocatorValue)],
    ) {
        let tree = LSMTree::new(merge, Box::new(NullCache {}));
        tree.insert(Item::new(AllocatorKey { device_range: right.0 }, right.1))
            .expect("insert error");
        tree.seal();
        tree.insert(Item::new(AllocatorKey { device_range: left.0 }, left.1))
            .expect("insert error");
        let layer_set = tree.layer_set();
        let mut merger = layer_set.merger();
        let mut iter = filter_tombstones(merger.query(Query::FullScan).await.expect("seek failed"))
            .await
            .expect("filter failed");
        for e in expected {
            let ItemRef { key, value, .. } = iter.get().expect("get failed");
            assert_eq!((key, value), (&AllocatorKey { device_range: e.0.clone() }, &e.1));
            iter.advance().await.expect("advance failed");
        }
        assert!(iter.get().is_none());
    }

    #[fuchsia::test]
    async fn test_no_overlap() {
        test_merge(
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (200..300, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &[
                (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
                (200..300, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            ],
        )
        .await;
    }

    #[fuchsia::test]
    async fn test_touching() {
        test_merge(
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (100..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &[(0..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 })],
        )
        .await;
    }

    #[fuchsia::test]
    async fn test_identical() {
        test_merge(
            (0..100, AllocatorValue::Abs { count: 2, owner_object_id: 1 }),
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &[(0..100, AllocatorValue::Abs { count: 2, owner_object_id: 1 })],
        )
        .await;
        test_merge(
            (0..100, AllocatorValue::None),
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &[],
        )
        .await;
    }

    #[fuchsia::test]
    async fn test_left_smaller_than_right_with_same_start() {
        test_merge(
            (0..100, AllocatorValue::Abs { count: 2, owner_object_id: 1 }),
            (0..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &[
                (0..100, AllocatorValue::Abs { count: 2, owner_object_id: 1 }),
                (100..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            ],
        )
        .await;
        test_merge(
            (0..100, AllocatorValue::None),
            (0..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &[(100..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 })],
        )
        .await;
    }

    #[fuchsia::test]
    async fn test_left_starts_before_right_with_overlap() {
        test_merge(
            (0..200, AllocatorValue::Abs { count: 2, owner_object_id: 1 }),
            (100..150, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &[
                (0..100, AllocatorValue::Abs { count: 2, owner_object_id: 1 }),
                (100..200, AllocatorValue::Abs { count: 2, owner_object_id: 1 }),
            ],
        )
        .await;
    }

    #[fuchsia::test]
    async fn test_different_object_id() {
        // Case 1
        test_merge(
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (200..300, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            &[
                (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
                (200..300, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            ],
        )
        .await;
        // Case 2
        test_merge(
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (100..200, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            &[
                (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
                (100..200, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            ],
        )
        .await;
        // Case 3
        test_merge(
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            &[(0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 })],
        )
        .await;
        // Case 4
        test_merge(
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (0..200, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            &[
                (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
                (100..200, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            ],
        )
        .await;
        // Case 5
        test_merge(
            (0..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            &[(0..200, AllocatorValue::Abs { count: 1, owner_object_id: 1 })],
        )
        .await;
        // Case 6
        test_merge(
            (0..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            (50..150, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            &[
                (0..50, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
                (50..100, AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
                (100..150, AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            ],
        )
        .await;
    }

    #[fuchsia::test]
    async fn test_tombstones() {
        // We have to make sure we don't prematurely discard records. seal() may be called at
        // any time and the resulting layer tree must remain valid.
        // Here we test absolute allocation counts and reuse of allocated space.
        //
        //  1. Alloc object_id A, write layer file.
        //  2. Dealloc object_id A, Alloc object_id B, write layer file.
        //  3. Dealloc object_id B, Alloc object_id A.
        let key = AllocatorKey { device_range: 0..100 };
        let lower_bound = AllocatorKey::lower_bound_for_merge_into(&key);
        let tree = LSMTree::new(merge, Box::new(NullCache {}));
        tree.merge_into(
            Item::new(key.clone(), AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &lower_bound,
        );
        tree.seal();
        tree.merge_into(
            Item::new(key.clone(), AllocatorValue::Abs { count: 2, owner_object_id: 1 }),
            &lower_bound,
        );
        tree.seal();
        tree.merge_into(Item::new(key.clone(), AllocatorValue::None), &lower_bound);
        tree.merge_into(
            Item::new(key.clone(), AllocatorValue::Abs { count: 1, owner_object_id: 2 }),
            &lower_bound,
        );
        tree.seal();
        tree.merge_into(Item::new(key.clone(), AllocatorValue::None), &lower_bound);
        tree.merge_into(
            Item::new(key.clone(), AllocatorValue::Abs { count: 1, owner_object_id: 1 }),
            &lower_bound,
        );
        let layer_set = tree.layer_set();
        let mut merger = layer_set.merger();
        let mut iter = merger.query(Query::FullScan).await.expect("seek failed");
        let ItemRef { key: k, value, .. } = iter.get().expect("get failed");
        assert_eq!((k, value), (&key, &AllocatorValue::Abs { count: 1, owner_object_id: 1 }));
        iter.advance().await.expect("advance failed");
        assert!(iter.get().is_none());
    }

    #[fuchsia::test]
    async fn test_merge_preserves_sequences() {
        let tree = LSMTree::new(merge, Box::new(NullCache {}));
        // |1-1-1-1|
        tree.insert(Item {
            key: AllocatorKey { device_range: 0..100 },
            value: AllocatorValue::Abs { count: 1, owner_object_id: INVALID_OBJECT_ID },
            sequence: 1u64,
        })
        .expect("insert error");
        tree.seal();
        // |1|0|1-1|
        tree.insert(Item {
            key: AllocatorKey { device_range: 25..50 },
            value: AllocatorValue::None,
            sequence: 2u64,
        })
        .expect("insert error");
        // |1|0|1|2|
        tree.insert(Item {
            key: AllocatorKey { device_range: 75..100 },
            value: AllocatorValue::Abs { count: 2, owner_object_id: INVALID_OBJECT_ID },
            sequence: 3u64,
        })
        .expect("insert error");
        let layer_set = tree.layer_set();
        let mut merger = layer_set.merger();
        let mut iter = filter_tombstones(merger.query(Query::FullScan).await.expect("seek failed"))
            .await
            .expect("filter failed");
        assert_eq!(iter.get().unwrap().key, &AllocatorKey { device_range: 0..25 });
        assert_eq!(
            iter.get().unwrap().value,
            &AllocatorValue::Abs { count: 1, owner_object_id: INVALID_OBJECT_ID }
        );
        assert_eq!(iter.get().unwrap().sequence, 1u64);
        iter.advance().await.expect("advance failed");
        assert_eq!(iter.get().unwrap().key, &AllocatorKey { device_range: 50..75 });
        assert_eq!(
            iter.get().unwrap().value,
            &AllocatorValue::Abs { count: 1, owner_object_id: INVALID_OBJECT_ID }
        );
        assert_eq!(iter.get().unwrap().sequence, 1u64);
        iter.advance().await.expect("advance failed");
        assert_eq!(iter.get().unwrap().key, &AllocatorKey { device_range: 75..100 });
        assert_eq!(
            iter.get().unwrap().value,
            &AllocatorValue::Abs { count: 2, owner_object_id: INVALID_OBJECT_ID }
        );
        assert_eq!(iter.get().unwrap().sequence, 3u64);
        iter.advance().await.expect("advance failed");
        assert!(iter.get().is_none());
    }
}