refactor(object store): make object store a separate crate (#2939)

* refactor(object store): make object store common

* create a new crate for object store

Author: wenym1

Cargo.lock

@@ -12,6 +12,7 @@ members = [
   "src/frontend",
   "src/frontend/test_runner",
   "src/meta",
+  "src/object_store",
   "src/prost",
   "src/prost/helpers",
   "src/risedevtool",

Cargo.toml

@@ -20,11 +20,12 @@ mod utils;
 use clap::Parser;
 use operations::*;
 use risingwave_common::config::StorageConfig;
+use risingwave_common::monitor::Print;
 use risingwave_meta::hummock::test_utils::setup_compute_env;
 use risingwave_meta::hummock::MockHummockMetaClient;
 use risingwave_storage::hummock::compaction_executor::CompactionExecutor;
 use risingwave_storage::hummock::compactor::{get_remote_sstable_id_generator, CompactorContext};
-use risingwave_storage::monitor::{ObjectStoreMetrics, Print, StateStoreMetrics};
+use risingwave_storage::monitor::{ObjectStoreMetrics, StateStoreMetrics};
 use risingwave_storage::{dispatch_state_store, StateStoreImpl};
 
 #[derive(Parser, Debug)]

src/bench/ss_bench/main.rs

@@ -12,7 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use risingwave_storage::monitor::{MyHistogram, StateStoreMetrics};
+use risingwave_common::monitor::my_stats::MyHistogram;
+use risingwave_storage::monitor::StateStoreMetrics;
 
 use super::my_stats::MyStateStoreStats;
 

src/bench/ss_bench/utils/display_stats.rs

@@ -13,7 +13,8 @@
 // limitations under the License.
 
 use prometheus::core::Metric;
-use risingwave_storage::monitor::{MyHistogram, StateStoreMetrics};
+use risingwave_common::monitor::my_stats::MyHistogram;
+use risingwave_storage::monitor::StateStoreMetrics;
 
 #[derive(Clone, Default)]
 pub(crate) struct MyStateStoreStats {

src/bench/ss_bench/utils/my_stats.rs

@@ -31,6 +31,7 @@ risingwave_pb = { path = "../prost" }
 rust_decimal = "1"
 serde = { version = "1", features = ["derive"] }
 smallvec = "1"
+spin = "0.9"
 thiserror = "1"
 tokio = { version = "=0.2.0-alpha.3", package = "madsim-tokio", features = ["rt", "rt-multi-thread", "sync", "macros", "time", "signal"] }
 tokio-stream = "0.1"

src/common/Cargo.toml

@@ -20,17 +20,16 @@
 //! `LruCache` implementation port from github.com/facebook/rocksdb. The class `LruCache` is
 //! thread-safe, because every operation on cache will be protected by a spin lock.
 use std::collections::HashMap;
+use std::error::Error;
 use std::future::Future;
 use std::hash::Hash;
 use std::ptr::null_mut;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
-use futures::channel::oneshot::{channel, Receiver, Sender};
+use futures::channel::oneshot::{channel, Canceled, Receiver, Sender};
 use spin::Mutex;
 
-use crate::hummock::{HummockError, HummockResult};
-
 const IN_CACHE: u8 = 1;
 const REVERSE_IN_CACHE: u8 = !IN_CACHE;
 
@@ -520,20 +519,22 @@ impl<K: LruKey, T: LruValue> LruCacheShard<K, T> {
     }
 
     // Clears the content of the cache.
-    // This method only works if no cache entries are referenced outside.
-    fn clear(&mut self) {
+    // This method is safe to use only if no cache entries are referenced outside.
+    unsafe fn clear(&mut self) {
         while !std::ptr::eq(self.lru.next, self.lru.as_mut()) {
             let handle = self.lru.next;
-            unsafe {
-                self.erase((*handle).hash, (*handle).get_key());
-            }
+            self.erase((*handle).hash, (*handle).get_key());
         }
     }
 }
 
 impl<K: LruKey, T: LruValue> Drop for LruCacheShard<K, T> {
     fn drop(&mut self) {
-        self.clear();
+        // Since the shard is being drop, there must be no cache entries referenced outside. So we
+        // are safe to call clear.
+        unsafe {
+            self.clear();
+        }
     }
 }
 
@@ -674,8 +675,10 @@ impl<K: LruKey, T: LruValue> LruCache<K, T> {
         hash as usize % self.shards.len()
     }
 
-    #[cfg(test)]
-    pub fn clear(&self) {
+    /// # Safety
+    ///
+    /// This method can only be called when no cache entry are referenced outside.
+    pub unsafe fn clear(&self) {
         for shard in &self.shards {
             let mut shard = shard.lock();
             shard.clear();
@@ -684,30 +687,31 @@ impl<K: LruKey, T: LruValue> LruCache<K, T> {
 }
 
 impl<K: LruKey + Clone, T: LruValue> LruCache<K, T> {
-    pub async fn lookup_with_request_dedup<F, VC>(
+    pub async fn lookup_with_request_dedup<F, E, VC>(
         self: &Arc<Self>,
         hash: u64,
         key: K,
         fetch_value: F,
-    ) -> HummockResult<CachableEntry<K, T>>
+    ) -> Result<Result<CachableEntry<K, T>, E>, Canceled>
     where
         F: FnOnce() -> VC,
-        VC: Future<Output = HummockResult<(T, usize)>>,
+        E: Error,
+        VC: Future<Output = Result<(T, usize), E>>,
     {
         match self.lookup_for_request(hash, key.clone()) {
-            LookupResult::Cached(entry) => Ok(entry),
+            LookupResult::Cached(entry) => Ok(Ok(entry)),
             LookupResult::WaitPendingRequest(recv) => {
-                let entry = recv.await.map_err(HummockError::other)?;
-                Ok(entry)
+                let entry = recv.await?;
+                Ok(Ok(entry))
             }
             LookupResult::Miss => match fetch_value().await {
                 Ok((value, charge)) => {
                     let entry = self.insert(key, hash, charge, value);
-                    Ok(entry)
+                    Ok(Ok(entry))
                 }
                 Err(e) => {
                     self.clear_pending_request(&key, hash);
-                    Err(e)
+                    Ok(Err(e))
                 }
             },
         }
@@ -753,8 +757,6 @@ mod tests {
     use rand::{RngCore, SeedableRng};
 
     use super::*;
-    use crate::hummock::cache::LruHandle;
-    use crate::hummock::LruCache;
 
     pub struct Block {
         pub offset: u64,

src/storage/src/hummock/cache.rs src/common/src/cache.rs

@@ -43,10 +43,12 @@ pub mod array;
 #[macro_use]
 pub mod util;
 pub mod buffer;
+pub mod cache;
 pub mod catalog;
 pub mod collection;
 pub mod config;
 pub mod hash;
+pub mod monitor;
 pub mod service;
 #[cfg(test)]
 pub mod test_utils;

src/common/src/lib.rs

@@ -0,0 +1,63 @@
+// Copyright 2022 Singularity Data
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod my_stats;
+
+use prometheus::core::{AtomicU64, Collector, GenericCounter, GenericCounterVec, Metric};
+use prometheus::{Histogram, HistogramVec};
+
+use crate::monitor::my_stats::MyHistogram;
+
+/// Define extension method `print` used in `print_statistics`.
+pub trait Print {
+    fn print(&self);
+}
+
+impl Print for GenericCounter<AtomicU64> {
+    fn print(&self) {
+        let desc = &self.desc()[0].fq_name;
+        let counter = self.metric().get_counter().get_value() as u64;
+        println!("{desc} COUNT : {counter}");
+    }
+}
+
+impl Print for Histogram {
+    fn print(&self) {
+        let desc = &self.desc()[0].fq_name;
+
+        let histogram = MyHistogram::from_prom_hist(self.metric().get_histogram());
+        let p50 = histogram.get_percentile(50.0);
+        let p95 = histogram.get_percentile(95.0);
+        let p99 = histogram.get_percentile(99.0);
+        let p100 = histogram.get_percentile(100.0);
+
+        let sample_count = self.get_sample_count();
+        let sample_sum = self.get_sample_sum();
+        println!("{desc} P50 : {p50} P95 : {p95} P99 : {p99} P100 : {p100} COUNT : {sample_count} SUM : {sample_sum}");
+    }
+}
+
+impl Print for HistogramVec {
+    fn print(&self) {
+        let desc = &self.desc()[0].fq_name;
+        println!("{desc} {:?}", self);
+    }
+}
+
+impl Print for GenericCounterVec<AtomicU64> {
+    fn print(&self) {
+        let desc = &self.desc()[0].fq_name;
+        println!("{desc} {:?}", self);
+    }
+}

src/common/src/monitor/mod.rs

@@ -0,0 +1,27 @@
+[package]
+name = "risingwave_object_store"
+version = "0.1.7"
+edition = "2021"
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+async-trait = "0.1"
+aws-config = { version = "0.12", default-features = false, features = ["rt-tokio", "native-tls"] }
+aws-endpoint = { version = "0.12", default-features = false }
+aws-sdk-s3 = { version = "0.12", default-features = false, features = ["rt-tokio", "native-tls"] }
+aws-smithy-http = "0.42"
+aws-smithy-types = "0.42"
+aws-types = { version = "0.12", features = ["hardcoded-credentials"] }
+bytes = { version = "1", features = ["serde"] }
+fail = "0.5"
+futures = { version = "0.3", default-features = false, features = ["alloc"] }
+itertools = "0.10"
+prometheus = { version = "0.13", features = ["process"] }
+risingwave_common = { path = "../common" }
+spin = "0.9"
+tempfile = "3"
+thiserror = "1"
+tokio = { version = "=0.2.0-alpha.3", package = "madsim-tokio", features = [
+    "fs",
+] }
+tracing = { version = "0.1" }

src/storage/src/monitor/my_stats.rs src/common/src/monitor/my_stats.rs

@@ -0,0 +1,17 @@
+// Copyright 2022 Singularity Data
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#![feature(backtrace)]
+
+pub mod object;

src/object_store/Cargo.toml

@@ -22,9 +22,9 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use futures::future::try_join_all;
+use risingwave_common::cache::{CachableEntry, LruCache};
 use tokio::io::AsyncWriteExt;
 
-use crate::hummock::{CachableEntry, HummockError, LruCache};
 use crate::object::{
     strip_path_local, BlockLocation, ObjectError, ObjectMetadata, ObjectResult, ObjectStore,
 };
@@ -131,24 +131,21 @@ impl LocalDiskObjectStore {
         };
         let entry = self
             .opened_read_file_cache
-            .lookup_with_request_dedup(hash, path.clone(), || async {
+            .lookup_with_request_dedup::<_, ObjectError, _>(hash, path.clone(), || async {
                 let file = utils::open_file(&path, true, false, false)
-                    .await
-                    .map_err(HummockError::object_io_error)?
+                    .await?
                     .into_std()
                     .await;
                 Ok((file, 1))
             })
             .await
             .map_err(|e| {
-                ObjectError::disk(
-                    "".to_string(),
-                    std::io::Error::new(
-                        ErrorKind::Other,
-                        format!("Failed to open file {:?}. Err{:?}", path.to_str(), e),
-                    ),
-                )
-            })?;
+                ObjectError::internal(format!(
+                    "open file cache request dedup get canceled {:?}. Err{:?}",
+                    path.to_str(),
+                    e
+                ))
+            })??;
         Ok(Arc::new(entry))
     }
 }