refactor: use encoded_size as memory usage (#1213)

## Rationale
In one of our cluster, we find the cost of calculate memory usage cost
too much CPU, and it's rwlock contention to be blamed.

<img width="1089" alt="image"
src="https://github.com/CeresDB/ceresdb/assets/3848910/a9b77fdf-92b7-4fa9-9a35-7cfa15a4dfee">

## Detailed Changes
- Use atomic usize to record written row size.

## Test Plan

---------

Co-authored-by: WEI Xikai <ShiKaiWi@users.noreply.github.com>

analytic_engine/src/memtable/skiplist/mod.rs

@@ -17,10 +17,7 @@
 pub mod factory;
 pub mod iter;
 
-use std::{
-    convert::TryInto,
-    sync::atomic::{self, AtomicU64, AtomicUsize},
-};
+use std::sync::atomic::{self, AtomicU64, AtomicUsize};
 
 use arena::{Arena, BasicStats};
 use bytes_ext::Bytes;
@@ -153,12 +150,14 @@ impl<A: Arena<Stats = BasicStats> + Clone + Sync + Send + 'static> MemTable
     }
 
     fn approximate_memory_usage(&self) -> usize {
-        // Mem size of skiplist is u32, need to cast to usize
-        match self.skiplist.mem_size().try_into() {
-            Ok(v) => v,
-            // The skiplist already use bytes larger than usize
-            Err(_) => usize::MAX,
-        }
+        let encoded_size = self
+            .metrics
+            .row_encoded_size
+            .load(atomic::Ordering::Relaxed);
+        let arena_block_size = self.skiplist.arena_block_size();
+
+        // Ceil to block_size
+        (encoded_size + arena_block_size - 1) / arena_block_size * arena_block_size
     }
 
     fn set_last_sequence(&self, sequence: SequenceNumber) -> Result<()> {

components/arena/src/arena_trait.rs

@@ -36,6 +36,9 @@ pub trait Arena {
     /// Get arena's statistics.
     fn stats(&self) -> Self::Stats;
 
+    /// Get arena's block size.
+    fn block_size(&self) -> usize;
+
     // provided methods
 
     /// Allocate required memory. Panic if failed.

components/arena/src/mono_inc.rs

@@ -38,21 +38,24 @@ const DEFAULT_ALIGN: usize = 8;
 #[derive(Clone)]
 pub struct MonoIncArena {
     core: Arc<RwLock<ArenaCore>>,
+    block_size: usize,
 }
 
 impl MonoIncArena {
-    pub fn new(regular_block_size: usize) -> Self {
+    pub fn new(block_size: usize) -> Self {
         Self {
             core: Arc::new(RwLock::new(ArenaCore::new(
-                regular_block_size,
+                block_size,
                 Arc::new(NoopCollector {}),
             ))),
+            block_size,
         }
     }
 
-    pub fn with_collector(regular_block_size: usize, collector: CollectorRef) -> Self {
+    pub fn with_collector(block_size: usize, collector: CollectorRef) -> Self {
         Self {
-            core: Arc::new(RwLock::new(ArenaCore::new(regular_block_size, collector))),
+            core: Arc::new(RwLock::new(ArenaCore::new(block_size, collector))),
+            block_size,
         }
     }
 }
@@ -71,6 +74,10 @@ impl Arena for MonoIncArena {
     fn alloc(&self, layout: Layout) -> NonNull<u8> {
         self.core.write().unwrap().alloc(layout)
     }
+
+    fn block_size(&self) -> usize {
+        self.block_size
+    }
 }
 
 struct ArenaCore {

components/skiplist/src/list.rs

@@ -252,6 +252,10 @@ impl<C, A: Arena<Stats = BasicStats> + Clone> Skiplist<C, A> {
     fn height(&self) -> usize {
         self.core.height.load(Ordering::SeqCst)
     }
+
+    pub fn arena_block_size(&self) -> usize {
+        self.core.arena.block_size()
+    }
 }
 
 impl<C: KeyComparator, A: Arena<Stats = BasicStats> + Clone> Skiplist<C, A> {