Queue optimize related updates

Cargo.toml

@@ -3,15 +3,15 @@ members = ["codegen", "tools/create-data-file", "tools/dump-data-file"]
 
 [package]
 name = "data_bucket"
-version = "0.2.3"
+version = "0.2.4"
 edition = "2021"
 authors = ["Handy-caT"]
 license = "MIT"
 repository = "https://github.com/pathscale/DataBucket"
 description = "DataBucket is container for WorkTable's data"
 
 [dependencies]
-data_bucket_derive = { path = "codegen", version = "0.2.2" }
+data_bucket_derive = { path = "codegen", version = "0.2.4" }
 
 eyre = "0.6.12"
 derive_more = { version = "1.0.0", features = ["from", "error", "display", "into"] }

codegen/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "data_bucket_derive"
-version = "0.2.2"
+version = "0.2.4"
 edition = "2021"
 authors = ["Handy-caT"]
 license = "MIT"

codegen/src/persistable/generator/obj_impl.rs

@@ -106,7 +106,11 @@ impl Generator {
                     quote! {
                         Self::#fn_ident(#size_ident)
                     }
-                } else if gens.contains(&field_type_str) && self.is_generic_unsized {
+                } else if field_type_str
+                    .split("<")
+                    .any(|v| gens.contains(&v.replace(">", "").trim().to_string()))
+                    && self.is_generic_unsized
+                {
                     quote! {
                         #size_ident
                     }
@@ -144,7 +148,11 @@ impl Generator {
                 self.gen_vec_size_fns(f)
             } else if field_type_str.contains("String") {
                 self.gen_string_size_fn(f)
-            } else if gens.contains(&field_type_str) && self.is_generic_unsized {
+            } else if field_type_str
+                .split("<")
+                .any(|v| gens.contains(&v.replace(">", "").trim().to_string()))
+                && self.is_generic_unsized
+            {
                 self.gen_generic_size_fn(f)
             } else {
                 self.gen_primitive_size_fn(f)

codegen/src/persistable/generator/persistable_impl.rs

@@ -190,7 +190,11 @@ impl Generator {
                     self.gen_from_bytes_for_vec(&f.ty, ident, &size_fields)
                 } else if field_type_str.contains("String") {
                     self.gen_from_bytes_for_string(&f.ty, ident, &size_fields)
-                } else if gens.contains(&field_type_str) && self.is_generic_unsized {
+                } else if field_type_str
+                    .split("<")
+                    .any(|v| gens.contains(&v.replace(">", "").trim().to_string()))
+                    && self.is_generic_unsized
+                {
                     self.gen_from_bytes_for_unsized_generic(&f.ty, ident, &size_fields)
                 } else {
                     self.gen_from_bytes_for_primitive(&f.ty, ident)
@@ -208,7 +212,7 @@ impl Generator {
             let size_type = &f.ty;
             let size_ident = f.ident.as_ref().unwrap();
             quote! {
-                let size_length = #size_type::default().aligned_size();
+                let size_length = <#size_type as Default>::default().aligned_size();
                 let archived =
                     unsafe { rkyv::access_unchecked::<<#size_type as Archive>::Archived>(&bytes[offset..offset + size_length]) };
                 let #size_ident =
@@ -233,7 +237,7 @@ impl Generator {
 
     fn gen_from_bytes_for_primitive(&self, ty: &Type, ident: &Ident) -> TokenStream {
         quote! {
-            let length = #ty::default().aligned_size();
+            let length = <#ty as Default>::default().aligned_size();
             let mut v = rkyv::util::AlignedVec::<4>::new();
             v.extend_from_slice(&bytes[offset..offset + length]);
             let archived = unsafe { rkyv::access_unchecked::<<#ty as Archive>::Archived>(&v[..]) };

codegen/src/size_measure/generator.rs

@@ -30,12 +30,30 @@ impl Generator {
             })
             .collect::<Vec<_>>();
 
+        let align_check = self
+            .struct_def
+            .fields
+            .iter()
+            .map(|f| {
+                let t = &f.ty;
+                quote! {
+                    if #t::align() == Some(8) {
+                        return Some(8)
+                    }
+                }
+            })
+            .collect::<Vec<_>>();
+
         quote! {
             impl SizeMeasurable for #struct_ident {
                 fn aligned_size(&self) -> usize {
                     let len = #(#sum+)* 0;
                     align(len)
                 }
+                fn align() -> Option<usize> {
+                    #(#align_check)*
+                    None
+                }
             }
         }
     }

src/lib.rs

@@ -11,10 +11,11 @@ pub use link::Link;
 pub use data_bucket_codegen::{SizeMeasure, VariableSizeMeasure};
 pub use page::{
     get_index_page_size_from_data_length, map_data_pages_to_general, parse_data_page,
-    parse_general_header_by_index, parse_page, persist_page, seek_by_link, seek_to_page_start,
-    update_at, DataPage, GeneralHeader, GeneralPage, IndexPage, IndexPageUtility, IndexValue,
-    Interval, PageType, SpaceInfoPage, TableOfContentsPage, UnsizedIndexPage,
-    UnsizedIndexPageUtility, DATA_VERSION, GENERAL_HEADER_SIZE, INNER_PAGE_SIZE, PAGE_SIZE,
+    parse_data_pages_batch, parse_general_header_by_index, parse_page, parse_pages_batch,
+    persist_page, persist_pages_batch, seek_by_link, seek_to_page_start, update_at, DataPage,
+    GeneralHeader, GeneralPage, IndexPage, IndexPageUtility, IndexValue, Interval, PageType,
+    SpaceInfoPage, TableOfContentsPage, UnsizedIndexPage, UnsizedIndexPageUtility, DATA_VERSION,
+    GENERAL_HEADER_SIZE, INNER_PAGE_SIZE, PAGE_SIZE,
 };
 pub use persistence::{PersistableIndex, PersistableTable};
 pub use space::Id as SpaceId;

src/link.rs

@@ -18,6 +18,7 @@ pub const LINK_LENGTH: usize = 12;
     PartialOrd,
     Serialize,
 )]
+#[rkyv(derive(Debug, PartialOrd, PartialEq, Eq, Ord))]
 pub struct Link {
     pub page_id: page::PageId,
     pub offset: u32,

src/page/index/mod.rs

@@ -1,3 +1,4 @@
+use std::fmt::Debug;
 use std::io::SeekFrom;
 
 use indexset::core::multipair::MultiPair;
@@ -6,10 +7,15 @@ use rkyv::{Archive, Deserialize, Serialize};
 use tokio::fs::File;
 use tokio::io::{AsyncSeekExt, AsyncWriteExt};
 
-use crate::{seek_to_page_start, Link, Persistable, SizeMeasurable, GENERAL_HEADER_SIZE};
+use crate::{
+    align, align8, seek_to_page_start, Link, Persistable, SizeMeasurable, VariableSizeMeasurable,
+    GENERAL_HEADER_SIZE,
+};
 
 mod page;
+mod page_cdc_impl;
 mod page_for_unsized;
+mod page_for_unsized_cdc_impl;
 mod table_of_contents_page;
 
 use crate::page::PageId;
@@ -56,7 +62,21 @@ where
     T: SizeMeasurable,
 {
     fn aligned_size(&self) -> usize {
-        self.key.aligned_size() + self.link.aligned_size()
+        if let Some(align) = T::align() {
+            if align % 8 == 0 {
+                return align8(self.key.aligned_size() + self.link.aligned_size());
+            }
+        }
+        align(self.key.aligned_size() + self.link.aligned_size())
+    }
+}
+
+impl<T> VariableSizeMeasurable for IndexValue<T>
+where
+    T: VariableSizeMeasurable,
+{
+    fn aligned_size(length: usize) -> usize {
+        length
     }
 }
 

src/page/index/page.rs

@@ -53,7 +53,7 @@ where
 #[persistable(by_parts)]
 pub struct IndexPage<T: Default + SizeMeasurable> {
     pub size: u16,
-    pub node_id: T,
+    pub node_id: IndexValue<T>,
     pub current_index: u16,
     pub current_length: u16,
     pub slots: Vec<u16>,
@@ -66,7 +66,7 @@ pub struct IndexPage<T: Default + SizeMeasurable> {
 #[persistable(by_parts)]
 pub struct SizedIndexPageUtility<T: Default + SizeMeasurable> {
     pub size: u16,
-    pub node_id: T,
+    pub node_id: IndexValue<T>,
     pub current_index: u16,
     pub current_length: u16,
     pub slots: Vec<u16>,
@@ -115,7 +115,7 @@ where
 }
 
 impl<T: Default + SizeMeasurable> IndexPage<T> {
-    pub fn new(node_id: T, size: usize) -> Self
+    pub fn new(node_id: IndexValue<T>, size: usize) -> Self
     where
         T: Clone,
     {
@@ -133,7 +133,7 @@ impl<T: Default + SizeMeasurable> IndexPage<T> {
 
     pub fn split(&mut self, index: usize) -> IndexPage<T>
     where
-        T: Clone,
+        T: Clone + Debug,
     {
         let mut new_page = IndexPage::new(self.node_id.clone(), self.size as usize);
         let mut first_empty_value = u16::MAX;
@@ -151,9 +151,7 @@ impl<T: Default + SizeMeasurable> IndexPage<T> {
         new_page.current_length = self.current_length - index as u16;
 
         self.current_index = first_empty_value;
-        self.node_id = self.index_values[self.slots[index - 1] as usize]
-            .key
-            .clone();
+        self.node_id = self.index_values[self.slots[index - 1] as usize].clone();
         self.current_length = index as u16;
 
         new_page
@@ -284,8 +282,7 @@ impl<T: Default + SizeMeasurable> IndexPage<T> {
                 node.last()
                     .expect("should contain at least one key")
                     .clone(),
-            )
-            .key,
+            ),
             size,
         );
 
@@ -308,7 +305,13 @@ mod tests {
     #[test]
     fn test_bytes() {
         let size: usize = get_index_page_size_from_data_length::<u64>(INNER_PAGE_SIZE);
-        let page = IndexPage::<u64>::new(1, size);
+        let page = IndexPage::<u64>::new(
+            IndexValue {
+                key: 1,
+                link: Default::default(),
+            },
+            size,
+        );
         let bytes = page.as_bytes();
         let new_page = IndexPage::<u64>::from_bytes(bytes.as_ref());
 
@@ -321,7 +324,13 @@ mod tests {
 
     #[test]
     fn test_split() {
-        let mut page = IndexPage::<u64>::new(7, 8);
+        let mut page = IndexPage::<u64>::new(
+            IndexValue {
+                key: 7,
+                link: Default::default(),
+            },
+            8,
+        );
         page.slots = vec![0, 1, 2, 3, 4, 5, 6, 7];
         page.current_index = 8;
         page.current_length = 8;