cyfs_base/objects/object_map/
path.rs

1use super::cache::*;
2use super::check::*;
3use super::iterator::*;
4use super::object_map::*;
5use super::op::*;
6use crate::*;
7
8use std::sync::{Arc, Mutex};
9
10// 基于路径管理的ObjectMap集合，共享同一个root，每级子路径对应一个ObjectMap
11pub struct ObjectMapPath {
12    root: Arc<Mutex<ObjectId>>,
13    obj_map_cache: ObjectMapOpEnvCacheRef,
14
15    // 用以暂存所有写入操作
16    write_ops: Option<ObjectMapOpList>,
17}
18
19struct ObjectMapPathSeg {
20    obj_map: ObjectMap,
21    seg: Option<String>,
22}
23
24impl std::fmt::Debug for ObjectMapPathSeg {
25    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        write!(f, "[{:?},{:?}]", self.seg, self.obj_map.cached_object_id())
27    }
28}
29
30impl ObjectMapPath {
31    pub fn new(
32        root: ObjectId,
33        obj_map_cache: ObjectMapOpEnvCacheRef,
34        enable_transaction: bool,
35    ) -> Self {
36        Self {
37            root: Arc::new(Mutex::new(root)),
38            obj_map_cache,
39            write_ops: if enable_transaction {
40                Some(ObjectMapOpList::new())
41            } else {
42                None
43            },
44        }
45    }
46
47    // 获取当前的root
48    pub fn root(&self) -> ObjectId {
49        self.root.lock().unwrap().clone()
50    }
51
52    pub fn update_root(&self, root_id: ObjectId, prev_id: &ObjectId) -> BuckyResult<()> {
53        let mut root = self.root.lock().unwrap();
54        if *root != *prev_id {
55            let msg = format!(
56                "update root but unmatch! current={}, prev={}, new={}",
57                *root, prev_id, root_id
58            );
59            error!("{}", msg);
60            return Err(BuckyError::new(BuckyErrorCode::Unmatch, msg));
61        }
62
63        info!("objectmap path root updated! {} -> {}", *root, root_id);
64        *root = root_id;
65        Ok(())
66    }
67
68    async fn get_root(&self) -> BuckyResult<ObjectMapRef> {
69        let root_id = self.root();
70        let ret = self.obj_map_cache.get_object_map(&root_id).await?;
71        if ret.is_none() {
72            let msg = format!("load root object but not found! id={}", root_id);
73            error!("{}", msg);
74            return Err(BuckyError::new(BuckyErrorCode::NotFound, msg));
75        }
76
77        Ok(ret.unwrap())
78    }
79
80    /*
81    /a/b/ -> /a/b
82    / -> /
83    */
84    fn fix_path(path: &str) -> BuckyResult<&str> {
85        let path = path.trim();
86        if path == "/" {
87            return Ok(path);
88        }
89
90        // 末尾的/需要去除
91        let path_ret = path.trim_end_matches("/");
92        if !path_ret.starts_with("/") {
93            let msg = format!("invalid objectmap path format! path={}", path);
94            error!("{}", msg);
95            return Err(BuckyError::new(BuckyErrorCode::InvalidFormat, msg));
96        }
97
98        Ok(path_ret)
99    }
100
101    // 获取path对应的obj_map叶子节点
102    async fn get_object_map(&self, path: &str) -> BuckyResult<Option<ObjectMapRef>> {
103        let mut current = self.get_root().await?;
104
105        let path = Self::fix_path(path)?;
106        // 判断是不是root
107        if path == "/" {
108            return Ok(Some(current));
109        }
110
111        // 依次获取每级子路径
112        let parts = path.split("/").skip(1);
113        for part in parts {
114            ObjectMapChecker::check_key_value(part)?;
115
116            let sub = current
117                .lock()
118                .await
119                .get_or_create_child_object_map(
120                    &self.obj_map_cache,
121                    part,
122                    ObjectMapSimpleContentType::Map,
123                    ObjectMapCreateStrategy::NotCreate,
124                    None,
125                )
126                .await
127                .map_err(|e| {
128                    let msg = format!(
129                        "get object by path error! path={}, part={}, {}",
130                        path, part, e
131                    );
132                    error!("{}", msg);
133                    BuckyError::new(e.code(), msg)
134                })?;
135
136            if sub.is_none() {
137                let msg = format!(
138                    "get object by path but not found! path={}, part={}",
139                    path, part
140                );
141                warn!("{}", msg);
142                return Ok(None);
143            }
144
145            current = sub.unwrap();
146            debug!(
147                "get objectmap path seg: {}={:?}",
148                part,
149                current.lock().await.cached_object_id()
150            );
151        }
152
153        Ok(Some(current))
154    }
155
156    // 以/开头的路径
157    async fn create_object_map(
158        &self,
159        path: &str,
160        content_type: ObjectMapSimpleContentType,
161        auto_create: ObjectMapCreateStrategy,
162    ) -> BuckyResult<Option<Vec<ObjectMapPathSeg>>> {
163        let root = self.get_root().await?;
164        let current = root.lock().await.clone();
165
166        let path = Self::fix_path(path)?;
167
168        let root_seg = ObjectMapPathSeg {
169            obj_map: current,
170            seg: None,
171        };
172
173        let mut obj_list = vec![root_seg];
174
175        // 判断是不是root
176        if path == "/" {
177            trace!("object map path list: path={}, list={:?}", path, obj_list);
178            return Ok(Some(obj_list));
179        }
180
181        // 依次获取每级子路径
182        let parts: Vec<&str> = path.split("/").skip(1).collect();
183        for (index, &part) in parts.iter().enumerate() {
184            ObjectMapChecker::check_key_value(part)?;
185
186            let is_last_part = index == parts.len() - 1;
187            // 最后一级使用目标类型， 中间子目录统一使用map
188            let content_type = if is_last_part {
189                content_type.clone()
190            } else {
191                ObjectMapSimpleContentType::Map
192            };
193
194            let create_strategy = match auto_create {
195                ObjectMapCreateStrategy::CreateIfNotExists => {
196                    ObjectMapCreateStrategy::CreateIfNotExists
197                }
198                ObjectMapCreateStrategy::NotCreate => ObjectMapCreateStrategy::NotCreate,
199                ObjectMapCreateStrategy::CreateNew => {
200                    // only use createNew for the last seg
201                    if is_last_part {
202                        ObjectMapCreateStrategy::CreateNew
203                    } else {
204                        ObjectMapCreateStrategy::CreateIfNotExists
205                    }
206                }
207            };
208
209            let sub = obj_list
210                .last_mut()
211                .unwrap()
212                .obj_map
213                .get_or_create_child_object_map(&self.obj_map_cache, part, content_type, create_strategy, None)
214                .await
215                .map_err(|e| {
216                    let msg = format!(
217                        "get or create object by path error! path={}, part={}, create_strategy={:?}, {}",
218                        path, part, create_strategy, e
219                    );
220                    error!("{}", msg);
221                    BuckyError::new(e.code(), msg)
222                })?;
223
224            if sub.is_none() {
225                let msg = format!(
226                    "get object by path but not found! path={}, part={}",
227                    path, part
228                );
229                warn!("{}", msg);
230                return Ok(None);
231            }
232
233            // 可能涉及到修改操作,所以路径上的objectmap都clone一份
234            let current = sub.unwrap().lock().await.clone();
235            let current_seq = ObjectMapPathSeg {
236                obj_map: current,
237                seg: Some(part.to_owned()),
238            };
239
240            obj_list.push(current_seq);
241        }
242
243        debug!("object map path list: path={}, list={:?}", path, obj_list);
244
245        Ok(Some(obj_list))
246    }
247
248    async fn update_path_obj_map_list(
249        &self,
250        mut obj_map_list: Vec<ObjectMapPathSeg>,
251    ) -> BuckyResult<Vec<(ObjectMap, ObjectId)>> {
252        assert!(!obj_map_list.is_empty());
253
254        let mut current_obj_map = obj_map_list.pop().unwrap();
255        let mut new_obj_map_list = vec![];
256
257        // 更新路径上的所有obj_map
258        loop {
259            // 刷新当前obj_map的id
260            let prev_id = current_obj_map.obj_map.cached_object_id().unwrap();
261            let current_id = current_obj_map.obj_map.flush_id();
262            assert_ne!(prev_id, current_id);
263
264            trace!(
265                "update objectmap path seg: seg={:?}, {} -> {}",
266                current_obj_map.seg, prev_id, current_id
267            );
268
269            // 更新此段的obj_map(id发生了变化)
270            new_obj_map_list.push((current_obj_map.obj_map, prev_id.clone()));
271
272            if obj_map_list.is_empty() {
273                break;
274            }
275
276            // 如果存在父一级，那么需要更新到父级obj_map
277            let seg = current_obj_map.seg.unwrap();
278            assert!(seg.len() > 0);
279
280            let mut parent_obj_map = obj_map_list.pop().unwrap();
281            parent_obj_map
282                .obj_map
283                .set_with_key(
284                    &self.obj_map_cache,
285                    &seg,
286                    &current_id,
287                    &Some(prev_id),
288                    false,
289                )
290                .await
291                .map_err(|e| e)?;
292
293            current_obj_map = parent_obj_map;
294        }
295
296        Ok(new_obj_map_list)
297    }
298
299    // 新的path刷新到缓存，并更新root
300    fn flush_path_obj_map_list(&self, obj_map_list: Vec<(ObjectMap, ObjectId)>) -> BuckyResult<()> {
301        let count = obj_map_list.len();
302
303        // 从叶子节点向根节点依次更新，最后更新root
304        for (index, (obj_map, prev_id)) in obj_map_list.into_iter().enumerate() {
305            // 最新的map必须已经计算过id了
306            let current_id = obj_map.cached_object_id().unwrap();
307            assert_ne!(current_id, prev_id);
308
309            self.obj_map_cache
310                .put_object_map(&current_id, obj_map, None)?;
311
312            if index + 1 == count {
313                self.update_root(current_id, &prev_id)?;
314            }
315
316            // TODO 如果之前的object在pending区，那么尝试移除
317            // 这里不能简单的移除，因为可能别的path操作里面长生的新路径存在相同对象id，会引用，这里移除会导致后续的查找失败
318            // 需要依赖统一的GC逻辑
319            // remove_list.insert(prev_id);
320        }
321
322        Ok(())
323    }
324
325    pub async fn metadata(&self, path: &str) -> BuckyResult<ObjectMapMetaData> {
326        let ret = self.get_object_map(path).await?;
327        if ret.is_none() {
328            let msg = format!("get value from path but objectmap not found! path={}", path);
329            warn!("{}", msg);
330            return Err(BuckyError::new(BuckyErrorCode::NotFound, msg));
331        }
332
333        let ret = ret.unwrap();
334        let obj = ret.lock().await;
335        Ok(obj.metadata())
336    }
337
338    pub async fn list(&self, path: &str) -> BuckyResult<ObjectMapContentList> {
339        let ret = self.get_object_map(path).await?;
340        if ret.is_none() {
341            let msg = format!("get value from path but objectmap not found! path={}", path);
342            warn!("{}", msg);
343            return Err(BuckyError::new(BuckyErrorCode::NotFound, msg));
344        }
345
346        let item = ret.unwrap();
347        let obj = item.lock().await;
348        let mut list = ObjectMapContentList::new(obj.count() as usize);
349        obj.list(&self.obj_map_cache, &mut list).await?;
350        Ok(list)
351    }
352
353    pub fn parse_path_allow_empty_key(full_path: &str) -> BuckyResult<(&str, &str)> {
354        let full_path = Self::fix_path(full_path)?;
355
356        if full_path == "/" {
357            return Ok((full_path, ""));
358        }
359
360        let mut path_segs: Vec<&str> = full_path.split("/").collect();
361        if path_segs.len() < 2 {
362            let msg = format!("invalid objectmap full path: {}", full_path);
363            error!("{}", msg);
364            return Err(BuckyError::new(BuckyErrorCode::InvalidFormat, msg));
365        }
366
367        let key = path_segs.pop().unwrap();
368        let trim_len = if path_segs.len() > 1 {
369            key.len() + 1
370        } else {
371            key.len()
372        };
373
374        let path = &full_path[..(full_path.len() - trim_len)];
375        if path.len() == 0 {
376            let msg = format!("invalid objectmap full path: {}", full_path);
377            error!("{}", msg);
378            return Err(BuckyError::new(BuckyErrorCode::InvalidFormat, msg));
379        }
380
381        Ok((path, key))
382    }
383
384    // map methods with full_path
385    // key should not been empty string
386    // 用来解析全路径，提取path和key
387    /*
388    /a -> / + a
389    /a/b/ -> /a + b
390    / -> Err
391    */
392    pub fn parse_full_path(full_path: &str) -> BuckyResult<(&str, &str)> {
393        let (path, key) = Self::parse_path_allow_empty_key(full_path)?;
394
395        let full_path = Self::fix_path(full_path)?;
396
397        if key.len() == 0 {
398            let msg = format!("invalid objectmap full path: {}", full_path);
399            error!("{}", msg);
400            return Err(BuckyError::new(BuckyErrorCode::InvalidFormat, msg));
401        }
402
403        Ok((path, key))
404    }
405
406    pub async fn get_by_path(&self, full_path: &str) -> BuckyResult<Option<ObjectId>> {
407        let (path, key) = Self::parse_path_allow_empty_key(full_path)?;
408
409        self.get_by_key(path, key).await
410    }
411
412    pub async fn create_new_with_path(
413        &self,
414        full_path: &str,
415        content_type: ObjectMapSimpleContentType,
416    ) -> BuckyResult<()> {
417        let (path, key) = Self::parse_full_path(full_path)?;
418
419        self.create_new(path, key, content_type).await
420    }
421
422    pub async fn insert_with_path(&self, full_path: &str, value: &ObjectId) -> BuckyResult<()> {
423        let (path, key) = Self::parse_full_path(full_path)?;
424
425        self.insert_with_key(path, key, value).await
426    }
427
428    pub async fn set_with_path(
429        &self,
430        full_path: &str,
431        value: &ObjectId,
432        prev_value: &Option<ObjectId>,
433        auto_insert: bool,
434    ) -> BuckyResult<Option<ObjectId>> {
435        let (path, key) = Self::parse_full_path(full_path)?;
436
437        self.set_with_key(path, key, value, prev_value, auto_insert)
438            .await
439    }
440
441    pub async fn remove_with_path(
442        &self,
443        full_path: &str,
444        prev_value: &Option<ObjectId>,
445    ) -> BuckyResult<Option<ObjectId>> {
446        let (path, key) = Self::parse_full_path(full_path)?;
447
448        self.remove_with_key(path, key, prev_value).await
449    }
450
451    // map methods
452    pub async fn create_new(
453        &self,
454        path: &str,
455        key: &str,
456        content_type: ObjectMapSimpleContentType,
457    ) -> BuckyResult<()> {
458        // 创建事务
459        let param = CreateNewParam {
460            key: key.to_owned(),
461            content_type,
462        };
463        let op_data = CreateNewOpData {
464            path: path.to_owned(),
465            param,
466            state: None,
467        };
468
469        let ret = self.create_new_op(&op_data).await?;
470
471        // insert不需要保存状态，只要插入成功，那么状态就认为是一致的
472
473        if let Some(write_ops) = &self.write_ops {
474            write_ops.append_op(ObjectMapWriteOp::CreateNew(op_data));
475        }
476
477        Ok(ret)
478    }
479
480    async fn create_new_op(&self, op_data: &CreateNewOpData) -> BuckyResult<()> {
481        // 首先获取路径上的所有ObjectMap(空目录自动创建)
482        let ret = self
483            .create_object_map(
484                &op_data.path,
485                ObjectMapSimpleContentType::Map,
486                ObjectMapCreateStrategy::CreateIfNotExists,
487            )
488            .await?;
489        let mut obj_map_list = ret.unwrap();
490        assert!(obj_map_list.len() > 0);
491
492        // create_new不需要保存旧值，因为如果存在旧值，那么会直接失败；只有为空才可以创建成功
493        obj_map_list
494            .last_mut()
495            .unwrap()
496            .obj_map
497            .get_or_create_child_object_map(
498                &self.obj_map_cache,
499                &op_data.param.key,
500                op_data.param.content_type,
501                ObjectMapCreateStrategy::CreateNew,
502                None,
503            )
504            .await?;
505
506        let list = self.update_path_obj_map_list(obj_map_list).await?;
507        self.flush_path_obj_map_list(list)?;
508
509        Ok(())
510    }
511
512    pub async fn get_by_key(&self, path: &str, key: &str) -> BuckyResult<Option<ObjectId>> {
513        let ret = self.get_object_map(path).await?;
514        if ret.is_none() {
515            info!(
516                "get value from path but objectmap not found! path={}, key={}",
517                path, key
518            );
519            return Ok(None);
520        }
521
522        // without key, return the path last node
523        if key.len() == 0 {
524            let obj_map = ret.as_ref().unwrap().lock().await;
525            return Ok(obj_map.cached_object_id());
526        }
527
528        let ret = ret.unwrap();
529        let obj_map = ret.lock().await;
530        obj_map.get_by_key(&self.obj_map_cache, key).await
531    }
532
533    pub async fn insert_with_key(
534        &self,
535        path: &str,
536        key: &str,
537        value: &ObjectId,
538    ) -> BuckyResult<()> {
539        // 创建事务
540        let param = InsertWithKeyParam {
541            key: key.to_owned(),
542            value: value.to_owned(),
543        };
544        let op_data = InsertWithKeyOpData {
545            path: path.to_owned(),
546            param,
547            state: None,
548        };
549
550        let ret = self.insert_with_key_op(&op_data).await?;
551
552        // insert不需要保存状态，只要插入成功，那么状态就认为是一致的
553
554        if let Some(write_ops) = &self.write_ops {
555            write_ops.append_op(ObjectMapWriteOp::InsertWithKey(op_data));
556        }
557
558        Ok(ret)
559    }
560
561    async fn insert_with_key_op(&self, op_data: &InsertWithKeyOpData) -> BuckyResult<()> {
562        // 首先获取路径上的所有ObjectMap(空目录自动创建)
563        let ret = self
564            .create_object_map(
565                &op_data.path,
566                ObjectMapSimpleContentType::Map,
567                ObjectMapCreateStrategy::CreateIfNotExists,
568            )
569            .await?;
570        let mut obj_map_list = ret.unwrap();
571        assert!(obj_map_list.len() > 0);
572
573        // insert_with_key不需要保存旧值，因为如果存在旧值，那么会直接失败；只有为空才可以插入成功
574        obj_map_list
575            .last_mut()
576            .unwrap()
577            .obj_map
578            .insert_with_key(
579                &self.obj_map_cache,
580                &op_data.param.key,
581                &op_data.param.value,
582            )
583            .await?;
584
585        let list = self.update_path_obj_map_list(obj_map_list).await?;
586        self.flush_path_obj_map_list(list)?;
587
588        Ok(())
589    }
590
591    pub async fn set_with_key(
592        &self,
593        path: &str,
594        key: &str,
595        value: &ObjectId,
596        prev_value: &Option<ObjectId>,
597        auto_insert: bool,
598    ) -> BuckyResult<Option<ObjectId>> {
599        // 创建事务
600        let param = SetWithKeyParam {
601            key: key.to_owned(),
602            value: value.to_owned(),
603            prev_value: prev_value.to_owned(),
604            auto_insert,
605        };
606
607        let mut op_data = SetWithKeyOpData {
608            path: path.to_owned(),
609            param,
610            state: None,
611        };
612
613        let ret = self.set_with_key_op(&op_data).await?;
614
615        // 保存状态
616        if let Some(write_ops) = &self.write_ops {
617            let state = ObjectMapKeyState { value: ret.clone() };
618            op_data.state = Some(state);
619
620            write_ops.append_op(ObjectMapWriteOp::SetWithKey(op_data));
621        }
622
623        Ok(ret)
624    }
625
626    async fn set_with_key_op(&self, op_data: &SetWithKeyOpData) -> BuckyResult<Option<ObjectId>> {
627        // 首先获取路径上的所有ObjectMap(空目录自动创建)
628
629        let create_strategy = if op_data.param.auto_insert {
630            ObjectMapCreateStrategy::CreateIfNotExists
631        } else {
632            ObjectMapCreateStrategy::NotCreate
633        };
634
635        let obj_map_list = self
636            .create_object_map(
637                &op_data.path,
638                ObjectMapSimpleContentType::Map,
639                create_strategy,
640            )
641            .await?;
642        if obj_map_list.is_none() {
643            // 如果auto_insert=false，并且路径不存在，那么直接返回Err(NotFound)
644            let msg = format!(
645                "set_with_key but path not found! path={}, value={}",
646                op_data.path, op_data.param.value,
647            );
648            error!("{}", msg);
649            return Err(BuckyError::new(BuckyErrorCode::NotFound, msg));
650        }
651
652        let mut obj_map_list = obj_map_list.unwrap();
653        assert!(obj_map_list.len() > 0);
654
655        // set_with_key存在以下几种情况:
656        // 1. 当前slot为空，auto_insert=false，那么直接返回Err(NotFound)
657        // 2. 当前slot为空，auto_insert=true，那么操作成功，返回Ok(None)
658        // 3. 当前slot不为空，prev_value=None, 那么操作成功，返回Ok(prev_value)
659        // 4. 当前slot不为空, prev_value!=None, 那么只有当前value和prev_value匹配，才成功，并且返回当前值；否则返回Err(Unmatch)
660        let ret = obj_map_list
661            .last_mut()
662            .unwrap()
663            .obj_map
664            .set_with_key(
665                &self.obj_map_cache,
666                &op_data.param.key,
667                &op_data.param.value,
668                &op_data.param.prev_value,
669                op_data.param.auto_insert,
670            )
671            .await?;
672
673        // 判断状态是否一致
674        if let Some(state) = &op_data.state {
675            if ret != state.value {
676                let msg = format!(
677                    "set_with_key with path commit but state conflict! op_data={:?}, ret={:?}",
678                    op_data, ret,
679                );
680                warn!("{}", msg);
681                return Err(BuckyError::new(BuckyErrorCode::Conflict, msg));
682            }
683        }
684
685        if ret != Some(op_data.param.value) {
686            let list = self.update_path_obj_map_list(obj_map_list).await?;
687            self.flush_path_obj_map_list(list)?;
688        }
689
690        Ok(ret)
691    }
692
693    pub async fn remove_with_key(
694        &self,
695        path: &str,
696        key: &str,
697        prev_value: &Option<ObjectId>,
698    ) -> BuckyResult<Option<ObjectId>> {
699        // 创建事务
700        let param = RemoveWithKeyParam {
701            key: key.to_owned(),
702            prev_value: prev_value.to_owned(),
703        };
704        let mut op_data = RemoveWithKeyOpData {
705            path: path.to_owned(),
706            param,
707            state: None,
708        };
709
710        let ret = self.remove_with_key_op(&op_data).await?;
711
712        // 保存状态
713        if let Some(write_ops) = &self.write_ops {
714            let state = ObjectMapKeyState { value: ret.clone() };
715            op_data.state = Some(state);
716
717            write_ops.append_op(ObjectMapWriteOp::RemoveWithKey(op_data));
718        }
719
720        Ok(ret)
721    }
722
723    async fn remove_with_key_op(
724        &self,
725        op_data: &RemoveWithKeyOpData,
726    ) -> BuckyResult<Option<ObjectId>> {
727        let (ret, obj_map_list) = loop {
728            let ret = self
729                .create_object_map(
730                    &op_data.path,
731                    ObjectMapSimpleContentType::Map,
732                    ObjectMapCreateStrategy::NotCreate,
733                )
734                .await?;
735
736            // 所在目录不存在，那么直接返回不存在即可
737            if ret.is_none() {
738                debug!(
739                    "objectmap path remove_with_key but path not found! root={}, path={}, key={}",
740                    self.root(),
741                    op_data.path,
742                    op_data.param.key,
743                );
744
745                break (None, None);
746            }
747
748            let mut obj_map_list = ret.unwrap();
749            assert!(obj_map_list.len() > 0);
750
751            // 发起真正的remove操作
752            let ret = obj_map_list
753                .last_mut()
754                .unwrap()
755                .obj_map
756                .remove_with_key(
757                    &self.obj_map_cache,
758                    &op_data.param.key,
759                    &op_data.param.prev_value,
760                )
761                .await?;
762
763            info!(
764                "objectmap path remove_with_key success! root={}, path={}, key={}, value={:?}",
765                self.root(),
766                op_data.path,
767                op_data.param.key,
768                ret
769            );
770            break (ret, Some(obj_map_list));
771        };
772
773        // 判断状态是否一致
774        if let Some(state) = &op_data.state {
775            if ret != state.value {
776                let msg = format!(
777                    "remove_with_key from path commit but state conflict! op_data={:?}, ret={:?}",
778                    op_data, ret,
779                );
780                warn!("{}", msg);
781                return Err(BuckyError::new(BuckyErrorCode::Conflict, msg));
782            }
783        }
784
785        if ret.is_none() {
786            return Ok(None);
787        }
788
789        // 内容改变了，需要更新整个路径
790        let list = self.update_path_obj_map_list(obj_map_list.unwrap()).await?;
791        self.flush_path_obj_map_list(list)?;
792
793        Ok(ret)
794    }
795
796    // set methods
797    pub async fn contains(&self, path: &str, object_id: &ObjectId) -> BuckyResult<bool> {
798        let ret = self.get_object_map(path).await?;
799
800        if ret.is_none() {
801            let msg = format!(
802                "contains from path but objectmap not found! path={}, value={}",
803                path, object_id,
804            );
805            error!("{}", msg);
806            return Err(BuckyError::new(BuckyErrorCode::NotFound, msg));
807        }
808
809        let ret = ret.unwrap();
810        let obj_map = ret.lock().await;
811        obj_map.contains(&self.obj_map_cache, object_id).await
812    }
813
814    pub async fn insert(&self, path: &str, object_id: &ObjectId) -> BuckyResult<bool> {
815        // 创建事务
816        let param = InsertParam {
817            value: object_id.to_owned(),
818        };
819        let mut op_data = InsertOpData {
820            path: path.to_owned(),
821            param,
822            state: None,
823        };
824
825        let ret = self.insert_op(&op_data).await?;
826
827        // 保存现有状态
828        if let Some(write_ops) = &self.write_ops {
829            op_data.state = Some(ret);
830
831            write_ops.append_op(ObjectMapWriteOp::Insert(op_data));
832        }
833
834        Ok(ret)
835    }
836
837    async fn insert_op(&self, op_data: &InsertOpData) -> BuckyResult<bool> {
838        let obj_map_list = self
839            .create_object_map(
840                &op_data.path,
841                ObjectMapSimpleContentType::Set,
842                ObjectMapCreateStrategy::CreateIfNotExists,
843            )
844            .await?;
845
846        let mut obj_map_list = obj_map_list.unwrap();
847        assert!(obj_map_list.len() > 0);
848
849        // 发起真正的insert操作
850        let ret = obj_map_list
851            .last_mut()
852            .unwrap()
853            .obj_map
854            .insert(&self.obj_map_cache, &op_data.param.value)
855            .await?;
856        // 如果事务是带状态的，那么需要校验一次状态
857        if let Some(state) = &op_data.state {
858            if *state != ret {
859                let msg = format!(
860                    "insert to path commit but state conflict! op_data={:?}",
861                    op_data,
862                );
863                warn!("{}", msg);
864                return Err(BuckyError::new(BuckyErrorCode::Conflict, msg));
865            }
866        }
867
868        // 值不存在，插入成功，需要更新路径
869        if ret {
870            // 内容改变了，需要更新整个路径
871            let list = self.update_path_obj_map_list(obj_map_list).await?;
872            self.flush_path_obj_map_list(list)?;
873        }
874
875        Ok(ret)
876    }
877
878    pub async fn remove(&self, path: &str, object_id: &ObjectId) -> BuckyResult<bool> {
879        // 创建事务
880        let param = RemoveParam {
881            value: object_id.to_owned(),
882        };
883        let mut op_data = RemoveOpData {
884            path: path.to_owned(),
885            param,
886            state: None,
887        };
888
889        let ret = self.remove_op(&op_data).await?;
890
891        // 保存状态
892        if let Some(write_ops) = &self.write_ops {
893            op_data.state = Some(ret);
894
895            write_ops.append_op(ObjectMapWriteOp::Remove(op_data));
896        }
897
898        Ok(ret)
899    }
900
901    async fn remove_op(&self, op_data: &RemoveOpData) -> BuckyResult<bool> {
902        let ret = self
903            .create_object_map(
904                &op_data.path,
905                ObjectMapSimpleContentType::Set,
906                ObjectMapCreateStrategy::NotCreate,
907            )
908            .await?;
909
910        // 所在目录不存在，那么直接返回错误
911        if ret.is_none() {
912            let msg = format!(
913                "remove but path not found! path={}, value={}",
914                op_data.path, op_data.param.value,
915            );
916            error!("{}", msg);
917            return Err(BuckyError::new(BuckyErrorCode::NotFound, msg));
918        }
919
920        let mut obj_map_list = ret.unwrap();
921        assert!(obj_map_list.len() > 0);
922
923        // 发起真正的remove操作
924        let ret = obj_map_list
925            .last_mut()
926            .unwrap()
927            .obj_map
928            .remove(&self.obj_map_cache, &op_data.param.value)
929            .await?;
930
931        // 如果事务是带状态的，那么需要校验一次状态
932        if let Some(state) = &op_data.state {
933            if *state != ret {
934                let msg = format!(
935                    "remove from path commit but state conflict! op_data={:?}",
936                    op_data,
937                );
938                warn!("{}", msg);
939                return Err(BuckyError::new(BuckyErrorCode::Conflict, msg));
940            }
941        }
942
943        if ret {
944            // 内容改变了，需要更新整个路径
945            let list = self.update_path_obj_map_list(obj_map_list).await?;
946            self.flush_path_obj_map_list(list)?;
947        }
948
949        Ok(ret)
950    }
951
952    pub fn clear_op_list(&self) {
953        if let Some(write_ops) = &self.write_ops {
954            let _ = write_ops.fetch_all();
955        }
956    }
957
958    // 提交操作列表，用以实现事务的commit
959    pub async fn commit_op_list(&self) -> BuckyResult<()> {
960        let op_list = self.write_ops.as_ref().unwrap().fetch_all();
961
962        for op_data in op_list {
963            self.commit_op(op_data).await?;
964        }
965
966        Ok(())
967    }
968
969    async fn commit_op(&self, op: ObjectMapWriteOp) -> BuckyResult<()> {
970        match op {
971            ObjectMapWriteOp::CreateNew(op_data) => {
972                self.create_new_op(&op_data).await?;
973            }
974            ObjectMapWriteOp::InsertWithKey(op_data) => {
975                self.insert_with_key_op(&op_data).await?;
976            }
977            ObjectMapWriteOp::SetWithKey(op_data) => {
978                self.set_with_key_op(&op_data).await?;
979            }
980            ObjectMapWriteOp::RemoveWithKey(op_data) => {
981                self.remove_with_key_op(&op_data).await?;
982            }
983
984            ObjectMapWriteOp::Insert(op_data) => {
985                self.insert_op(&op_data).await?;
986            }
987            ObjectMapWriteOp::Remove(op_data) => {
988                self.remove_op(&op_data).await?;
989            }
990        }
991
992        Ok(())
993    }
994}
995
996#[cfg(test)]
997mod test_path {
998    use super::super::cache::*;
999    use super::super::path_iterator::*;
1000    use super::*;
1001
1002    use std::str::FromStr;
1003
1004    async fn dump_path(item: &ObjectMapPath, path: &str) {
1005        let list = item.list(path).await.unwrap();
1006        info!("dump path={} as follows:", path);
1007        info!("{}", list);
1008    }
1009
1010    async fn test_path1(path: &ObjectMapPath) {
1011        let x1_value = ObjectId::from_str("5aSixgPg3hDa1oU9eAtRcKTyVKg5X2bVXWPVhk3U5c7G").unwrap();
1012        let x1_value2 = ObjectId::from_str("5aSixgPCivmQfASRbjAvBiwgxhU8LrNtYtC2D6Lis2NQ").unwrap();
1013
1014        path.insert_with_key("/", "x1", &x1_value).await.unwrap();
1015
1016        let ret = path.get_by_key("/a/b/c", "x1").await.unwrap();
1017        assert!(ret.is_none());
1018
1019        let ret = path.get_by_path("/a/b/c/x1").await.unwrap();
1020        assert!(ret.is_none());
1021
1022        path.insert_with_key("/a/b/c", "x1", &x1_value)
1023            .await
1024            .unwrap();
1025        let ret = path.insert_with_path("/a/b/c/x1", &x1_value).await;
1026        let e = ret.unwrap_err();
1027        assert_eq!(e.code(), BuckyErrorCode::AlreadyExists);
1028
1029        let ret = path.get_by_key("/a/b/c", "x1").await.unwrap();
1030        assert_eq!(ret, Some(x1_value));
1031        let ret = path.get_by_path("/a/b/c/x1").await.unwrap();
1032        assert_eq!(ret, Some(x1_value));
1033
1034        dump_path(path, "/").await;
1035        dump_path(path, "/a").await;
1036        dump_path(path, "/a/b").await;
1037        dump_path(path, "/a/b/c").await;
1038        let ret = path.get_by_key("/a/b/c", "x1").await.unwrap();
1039        assert_eq!(ret, Some(x1_value));
1040
1041        // 插入已经存在的key，返回错误
1042        let ret = path.insert_with_key("/a/b/c", "x1", &x1_value).await;
1043        let err = ret.unwrap_err();
1044        assert_eq!(err.code(), BuckyErrorCode::AlreadyExists);
1045
1046        // 测试set_with_key
1047        let ret = path
1048            .set_with_key("/a/b/c", "x1", &x1_value2, &Some(x1_value2), false)
1049            .await;
1050        assert!(ret.is_err());
1051        let err = ret.unwrap_err();
1052        assert_eq!(err.code(), BuckyErrorCode::Unmatch);
1053
1054        let ret = path
1055            .set_with_key("/a/b/c", "x1", &x1_value2, &Some(x1_value), false)
1056            .await
1057            .unwrap();
1058        assert_eq!(ret, Some(x1_value));
1059
1060        // 测试删除
1061        let ret = path.remove_with_key("/a/b/c", "x1", &Some(x1_value)).await;
1062        assert!(ret.is_err());
1063        let err = ret.unwrap_err();
1064        assert_eq!(err.code(), BuckyErrorCode::Unmatch);
1065
1066        let ret = path.remove_with_key("/a/b/c", "x1", &None).await.unwrap();
1067        assert_eq!(ret, Some(x1_value2));
1068
1069        // 再次测试set_with_key
1070        let ret = path
1071            .set_with_key("/a/b/c", "x1", &x1_value2, &None, false)
1072            .await;
1073        assert!(ret.is_err());
1074        let err = ret.unwrap_err();
1075        assert_eq!(err.code(), BuckyErrorCode::NotFound);
1076
1077        // 自动插入x1
1078        let ret = path
1079            .set_with_key("/a/b/c", "x1", &x1_value2, &None, true)
1080            .await
1081            .unwrap();
1082        assert_eq!(ret, None);
1083
1084        let ret = path.get_by_key("/a/b/c", "x1").await.unwrap();
1085        assert_eq!(ret, Some(x1_value2));
1086
1087        let ret = path.remove_with_key("/a/b/c", "x1", &None).await.unwrap();
1088        assert_eq!(ret, Some(x1_value2));
1089
1090        let ret = path.get_by_key("/a/b/c", "x1").await.unwrap();
1091        assert!(ret.is_none());
1092
1093        let ret = path.get_by_key("/a/b", "c").await.unwrap();
1094        assert!(ret.is_some());
1095        let c_id = ret.unwrap();
1096        info!("/a/b/c={}", c_id);
1097
1098        dump_path(path, "/").await;
1099        dump_path(path, "/a").await;
1100        dump_path(path, "/a/b").await;
1101        dump_path(path, "/a/b/c").await;
1102
1103        let ret = path.remove_with_key("/a/b", "c", &None).await.unwrap();
1104        assert_eq!(ret, Some(c_id));
1105
1106        let ret = path.get_by_key("/a/b/c", "x1").await.unwrap();
1107        assert!(ret.is_none());
1108
1109        let ret = path.get_by_key("/a/b", "c").await.unwrap();
1110        assert!(ret.is_none());
1111
1112        let ret = path.get_by_key("/a/b/c", "x1").await.unwrap();
1113        assert!(ret.is_none());
1114
1115        let ret = path.get_by_path("/").await.unwrap();
1116        assert!(ret.is_some());
1117
1118        path.create_new("/a/b", "c", ObjectMapSimpleContentType::Set)
1119            .await
1120            .unwrap();
1121        if let Err(e) = path
1122            .create_new("/a/b", "c", ObjectMapSimpleContentType::Set)
1123            .await
1124        {
1125            assert!(e.code() == BuckyErrorCode::AlreadyExists);
1126        } else {
1127            unreachable!();
1128        }
1129        if let Err(e) = path
1130            .create_new("/a/b", "c", ObjectMapSimpleContentType::Set)
1131            .await
1132        {
1133            assert!(e.code() == BuckyErrorCode::AlreadyExists);
1134        } else {
1135            unreachable!();
1136        }
1137
1138        let ret = path.get_by_key("/a/b", "c").await.unwrap();
1139        assert!(ret.is_some());
1140    }
1141
1142    async fn test_path() {
1143        let noc = ObjectMapMemoryNOCCache::new();
1144        let root_cache = ObjectMapRootMemoryCache::new_default_ref(None, noc);
1145        let cache = ObjectMapOpEnvMemoryCache::new_ref(root_cache.clone());
1146
1147        // 创建一个空的objectmap作为root
1148        let owner = ObjectId::default();
1149        let root = ObjectMap::new(
1150            ObjectMapSimpleContentType::Map,
1151            Some(owner.clone()),
1152            Some(owner.clone()),
1153        )
1154        .no_create_time()
1155        .build();
1156        let root_id = root.flush_id();
1157        cache.put_object_map(&root_id, root, None).unwrap();
1158        info!("new root: {}", root_id);
1159
1160        let path = ObjectMapPath::new(root_id.clone(), cache.clone(), true);
1161        test_path1(&path).await;
1162
1163        let opt = ObjectMapPathIteratorOption::new(true, true);
1164        let root = path.root();
1165        let root_obj = cache.get_object_map(&root).await.unwrap();
1166        let mut it =
1167            ObjectMapPathIterator::new(root_obj.unwrap(), cache.clone(), opt.clone()).await;
1168        while !it.is_end() {
1169            let list = it.next(5).await.unwrap();
1170            info!("list: {} {:?}", 1, list.list);
1171        }
1172
1173        let root_id = path.root();
1174        info!("result root: {}", root_id);
1175
1176        cache.gc(false, &root_id).await.unwrap();
1177
1178        let root_obj = cache.get_object_map(&root_id).await.unwrap();
1179        let mut it =
1180            ObjectMapPathIterator::new(root_obj.unwrap(), cache.clone(), opt.clone()).await;
1181        while !it.is_end() {
1182            let list = it.next(5).await.unwrap();
1183            info!("list: {} {:?}", 1, list.list);
1184        }
1185    }
1186
1187    #[test]
1188    fn test_full_path() {
1189        ObjectMapPath::parse_full_path("/").unwrap_err();
1190        let (path, key) = ObjectMapPath::parse_full_path("/a").unwrap();
1191        assert_eq!(path, "/");
1192        assert_eq!(key, "a");
1193
1194        let (path, key) = ObjectMapPath::parse_full_path("/a/").unwrap();
1195        assert_eq!(path, "/");
1196        assert_eq!(key, "a");
1197
1198        let (path, key) = ObjectMapPath::parse_full_path("/a/b").unwrap();
1199        assert_eq!(path, "/a");
1200        assert_eq!(key, "b");
1201
1202        let (path, key) = ObjectMapPath::parse_full_path("/eeee/eeee").unwrap();
1203        assert_eq!(path, "/eeee");
1204        assert_eq!(key, "eeee");
1205
1206        let (path, key) = ObjectMapPath::parse_full_path("/eeee/eeee/").unwrap();
1207        assert_eq!(path, "/eeee");
1208        assert_eq!(key, "eeee");
1209    }
1210
1211    #[test]
1212    fn test() {
1213        crate::init_simple_log("test-object-map-path", Some("debug"));
1214        test_full_path();
1215        async_std::task::block_on(async move {
1216            test_path().await;
1217        });
1218    }
1219}
cyfs_base/objects/object_map/path.rs

cyfs_base/objects/object_map/
path.rs
