Merge pull request ipfs-force-community#74 from ipfs-force-community/…

…feat/dtynn/support_multiple_attached_stores

Feat/dtynn/support multiple attached stores

docs/zh/03.venus-worker的配置解析.md

@@ -44,7 +44,12 @@ location = "./mock-tmp/store2"
 [[sealing_thread]]
 location = "./mock-tmp/store3"
 
-[remote_store]
+# deprecated
+# [remote_store]
+# name = "persist-store1"
+# location = "./mock-tmp/remote"
+
+[[attached]]
 # name = "persist-store1"
 location = "./mock-tmp/remote"
 
@@ -314,7 +319,7 @@ sealing.allowed_sizes = ["64GiB"]
 
 
 
-## [remote_store]
+## [remote_store] 已废弃
 
 `remote_store` 用于配置已完成的扇区持久化数据保存的位置。
 
@@ -339,6 +344,35 @@ location = "/mnt/remote/10.0.0.14/store"
 如果持久化存储目录在所有机器上的挂载路径都统一的话，配置时也可以选择在 `venus-worker` 和`venus-sector-manager` 两侧都不配置 `name`。这种情况下，两者都会使用绝对路径作为 `name`，也能匹配。
 
 
+## [[attached]]
+
+`attached` 用于配置已完成的扇区持久化数据保存的位置，允许同时配置多个。
+
+
+
+### 基础配置范例
+
+```
+[attached]
+# 名称， 选填项，字符串类型
+# 默认为路径对应的绝对路径
+# name = "remote-store1"
+
+# 路径，必填项，字符串类型
+# 建议直接填写绝对路径
+location = "/mnt/remote/10.0.0.14/store"
+
+# 只读，选填项，布尔类型
+# 默认值为 false
+# readonly = true
+
+```
+
+由于需要在 `venus-worker` 和 `venus-sector-manager` 之间协调存储位置信息，而在很多情况下，同一个持久化存储目录在`venus-worker` 机器和 `venus-sector-manager` 机器上的挂载路径不完全一致，因此我们决定使用 `name` 作为协调的基础信息.
+
+如果持久化存储目录在所有机器上的挂载路径都统一的话，配置时也可以选择在 `venus-worker` 和`venus-sector-manager` 两侧都不配置 `name`。这种情况下，两者都会使用绝对路径作为 `name`，也能匹配。
+
+
 
 ## [processors]
 

venus-worker/assets/venus-worker.mock.toml

@@ -35,7 +35,12 @@ location = "./mock-tmp/store2"
 [[sealing_thread]]
 location = "./mock-tmp/store3"
 
-[remote_store]
+# deprecated
+# [remote_store]
+# name = "persist-store1"
+# location = "./mock-tmp/remote"
+
+[[attached]]
 # name = "persist-store1"
 location = "./mock-tmp/remote"
 

venus-worker/src/config.rs

@@ -96,10 +96,12 @@ pub struct SealingOptional {
 
 /// configuration for remote store
 #[derive(Debug, Default, Serialize, Deserialize)]
-pub struct Remote {
+pub struct Attached {
     pub name: Option<String>,
     /// store path, if we are using fs based store
-    pub location: Option<String>,
+    pub location: String,
+
+    pub readonly: Option<bool>,
 }
 
 /// configurations for local sealing store
@@ -173,8 +175,11 @@ pub struct Config {
     /// section for list of local sealing stores
     pub sealing_thread: Vec<SealingThread>,
 
-    /// section for remote store
-    pub remote_store: Remote,
+    /// section for remote store, deprecated
+    pub remote_store: Option<Attached>,
+
+    /// section for attached store
+    pub attached: Option<Vec<Attached>>,
 
     /// section for processors
     pub processors: Processors,

venus-worker/src/infra/objstore/attached.rs

@@ -0,0 +1,53 @@
+//! manages multiple attached stores
+
+use std::collections::HashMap;
+
+use anyhow::{anyhow, Result};
+
+use super::ObjectStore;
+
+/// manages all attached stores
+pub struct AttachedManager {
+    stores: HashMap<String, Box<dyn ObjectStore>>,
+}
+
+impl AttachedManager {
+    /// init AttachedManager with given stores
+    pub fn init(attached: Vec<Box<dyn ObjectStore>>) -> Result<Self> {
+        let mut stores = HashMap::new();
+        for astore in attached {
+            if let Some(prev) = stores.insert(astore.instance(), astore) {
+                return Err(anyhow!("duplicate instance name {}", prev.instance()));
+            };
+        }
+
+        Ok(AttachedManager { stores })
+    }
+
+    /// get a named store instance
+    pub fn get(&self, instance: &str) -> Option<&dyn ObjectStore> {
+        self.stores.get(instance).map(|b| b.as_ref())
+    }
+
+    /// acquire an available store for sector persistence
+    pub fn acquire_persist(
+        &self,
+        _size: u64,
+        prev_instance: Option<String>,
+    ) -> Option<&dyn ObjectStore> {
+        if let Some(ins) = prev_instance
+            .as_ref()
+            .and_then(|name| self.stores.get(name))
+        {
+            if !ins.readonly() {
+                return Some(ins.as_ref());
+            }
+        };
+
+        // TODO: depends on the free space
+        self.stores
+            .values()
+            .find(|s| !s.readonly())
+            .map(|ins| ins.as_ref())
+    }
+}

venus-worker/src/infra/objstore/filestore.rs

@@ -1,16 +1,14 @@
 //! ObjectStore implemented based on fs
 
-use std::fs::{create_dir_all, File, OpenOptions};
+use std::fs::{create_dir_all, remove_file, File, OpenOptions};
 use std::io::{copy, BufReader, Read, Seek, SeekFrom};
+use std::os::unix::fs::symlink;
 use std::path::{Path, PathBuf, MAIN_SEPARATOR};
 
 use anyhow::{anyhow, Context, Result};
 
 use super::{ObjResult, ObjectStore, Range};
-use crate::{
-    infra::util::PlaceHolder,
-    logging::{debug_field, trace},
-};
+use crate::logging::{debug_field, trace};
 
 const LOG_TARGET: &str = "filestore";
 
@@ -19,20 +17,19 @@ pub struct FileStore {
     sep: String,
     local_path: PathBuf,
     instance: String,
-    _holder: PlaceHolder,
+    readonly: bool,
 }
 
 impl FileStore {
     /// init filestore, create a placeholder file in its root dir
     pub fn init<P: AsRef<Path>>(p: P) -> Result<()> {
         create_dir_all(p.as_ref())?;
-        let _holder = PlaceHolder::init(p.as_ref())?;
 
         Ok(())
     }
 
     /// open the file store at given path
-    pub fn open<P: AsRef<Path>>(p: P, ins: Option<String>) -> Result<Self> {
+    pub fn open<P: AsRef<Path>>(p: P, ins: Option<String>, readonly: bool) -> Result<Self> {
         let dir_path = p.as_ref().canonicalize().context("canonicalize dir path")?;
         if !dir_path
             .metadata()
@@ -42,8 +39,6 @@ impl FileStore {
             return Err(anyhow!("base path of the file store should a dir"));
         };
 
-        let _holder = PlaceHolder::open(&dir_path).context("open placeholder")?;
-
         let instance = match ins.or(dir_path.to_str().map(|s| s.to_owned())) {
             Some(i) => i,
             None => {
@@ -58,7 +53,7 @@ impl FileStore {
             sep: MAIN_SEPARATOR.to_string(),
             local_path: dir_path,
             instance,
-            _holder,
+            readonly,
         })
     }
 
@@ -147,6 +142,30 @@ impl ObjectStore for FileStore {
 
         Ok(iter)
     }
+
+    fn copy_to(&self, path: &Path, dst: &Path, allow_sym: bool) -> ObjResult<()> {
+        if allow_sym {
+            let src_path = self.path(path)?;
+            remove_file(dst)?;
+            symlink(src_path, dst)?;
+            return Ok(());
+        }
+
+        let mut r = self.get(path)?;
+        let mut f = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(dst)?;
+
+        copy(&mut r, &mut f)?;
+        Ok(())
+    }
+
+    fn readonly(&self) -> bool {
+        self.readonly
+    }
 }
 
 struct ChunkReader {

venus-worker/src/infra/objstore/mod.rs

@@ -5,6 +5,7 @@ use std::fmt;
 use std::io::{self, Read};
 use std::path::Path;
 
+pub mod attached;
 pub mod filestore;
 
 /// errors in object storage usage
@@ -77,4 +78,10 @@ pub trait ObjectStore: Send + Sync {
         path: &Path,
         ranges: &[Range],
     ) -> ObjResult<Box<dyn Iterator<Item = ObjResult<Box<dyn Read>>>>>;
+
+    /// copy an object to a local path
+    fn copy_to(&self, path: &Path, dest: &Path, allow_sym: bool) -> ObjResult<()>;
+
+    /// if this instance is read-only
+    fn readonly(&self) -> bool;
 }