You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by er...@apache.org on 2025/10/17 20:06:46 UTC
(opendal) branch debug-gdrive-ci-test-2 updated (b1935dca5 -> f7e345785)
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a change to branch debug-gdrive-ci-test-2
in repository https://gitbox.apache.org/repos/asf/opendal.git
discard b1935dca5 fixup! Random wait to test gdrive in CI
discard 0f9def0a4 fixup! Change PathQuery to HashMap
discard 0e2e2f4f0 Random wait to test gdrive in CI
discard 8320d0e84 Change PathQuery to HashMap
new 6d0259960 Change PathQuery to HashMap
new f7e345785 Random wait to test gdrive in CI
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (b1935dca5)
\
N -- N -- N refs/heads/debug-gdrive-ci-test-2 (f7e345785)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
(opendal) 01/02: Change PathQuery to HashMap
Posted by er...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a commit to branch debug-gdrive-ci-test-2
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 6d0259960ca6f705d838e18616c510da17f5a639
Author: Erick Guan <29...@users.noreply.github.com>
AuthorDate: Fri Oct 17 21:39:43 2025 +0200
Change PathQuery to HashMap
---
core/src/raw/path_cache.rs | 95 ++++++++++++++++++++-----------------
core/src/services/gdrive/builder.rs | 2 +-
2 files changed, 52 insertions(+), 45 deletions(-)
diff --git a/core/src/raw/path_cache.rs b/core/src/raw/path_cache.rs
index ef8f51253..9ac98385d 100644
--- a/core/src/raw/path_cache.rs
+++ b/core/src/raw/path_cache.rs
@@ -19,6 +19,7 @@ use std::collections::VecDeque;
use futures::Future;
use moka::sync::Cache;
+use std::collections::HashMap;
use tokio::sync::Mutex;
use tokio::sync::MutexGuard;
@@ -56,13 +57,12 @@ pub trait PathQuery {
/// then the path of file `a/b` in cache will be `/root/a/b`.
pub struct PathCacher<Q: PathQuery> {
query: Q,
- cache: Cache<String, String>,
-
- /// This optional lock here is used to prevent concurrent insertions of the same path.
- ///
- /// Some services like gdrive allows the same name to exist in the same directory. We need to introduce
- /// a global lock to prevent concurrent insertions of the same path.
- lock: Option<Mutex<()>>,
+ cache: Mutex<HashMap<String, String>>,
+ // This optional lock here is used to prevent concurrent insertions of the same path.
+ //
+ // Some services like gdrive allows the same name to exist in the same directory. We need to introduce
+ // a global lock to prevent concurrent insertions of the same path.
+ // lock: Option<Mutex<()>>,
}
impl<Q: PathQuery> PathCacher<Q> {
@@ -70,54 +70,58 @@ impl<Q: PathQuery> PathCacher<Q> {
pub fn new(query: Q) -> Self {
Self {
query,
- cache: Cache::new(64 * 1024),
- lock: None,
+ cache: Mutex::new(HashMap::new()),
+ // lock: None,
}
}
- /// Enable the lock for the path cacher.
- pub fn with_lock(mut self) -> Self {
- self.lock = Some(Mutex::default());
- self
- }
+ // Enable the lock for the path cacher.
+ // pub fn with_lock(mut self) -> Self {
+ // self.lock = Some(Mutex::default());
+ // self
+ // }
- async fn lock(&self) -> Option<MutexGuard<'_, ()>> {
- if let Some(l) = &self.lock {
- Some(l.lock().await)
- } else {
- None
- }
- }
+ // async fn lock(&self) -> Option<MutexGuard<'_, ()>> {
+ // if let Some(l) = &self.lock {
+ // Some(l.lock().await)
+ // } else {
+ // None
+ // }
+ // }
/// Insert a new cache entry.
pub async fn insert(&self, path: &str, id: &str) {
- let _guard = self.lock().await;
+ let mut cache = self.cache.lock().await;
+ // let _guard = self.lock().await;
// This should never happen, but let's ignore the insert if happened.
- if self.cache.contains_key(path) {
+ if cache.contains_key(path) {
debug_assert!(
- self.cache.get(path) == Some(id.to_string()),
+ cache.get(path) == Some(&id.to_string()),
"path {path} exists but it's value is inconsistent"
);
return;
}
- self.cache.insert(path.to_string(), id.to_string());
+ cache.insert(path.to_string(), id.to_string());
}
/// Remove a cache entry.
pub async fn remove(&self, path: &str) {
- let _guard = self.lock().await;
+ let mut cache = self.cache.lock().await;
+
+ // let _guard = self.lock().await;
- self.cache.invalidate(path)
+ cache.remove(path);
}
/// Get the id for the given path.
pub async fn get(&self, path: &str) -> Result<Option<String>> {
- let _guard = self.lock().await;
+ // let _guard = self.lock().await;
+ let mut cache = self.cache.lock().await;
- if let Some(id) = self.cache.get(path) {
- return Ok(Some(id));
+ if let Some(id) = cache.get(path) {
+ return Ok(Some(id.to_string()));
}
let mut paths = VecDeque::new();
@@ -126,13 +130,14 @@ impl<Q: PathQuery> PathCacher<Q> {
while current_path != "/" && !current_path.is_empty() {
paths.push_front(current_path.to_string());
current_path = get_parent(current_path);
- if let Some(id) = self.cache.get(current_path) {
+ if let Some(id) = cache.get(current_path) {
+ let id = id.to_string();
return self.query_down(&id, paths).await;
}
}
let root_id = self.query.root().await?;
- self.cache.insert("/".to_string(), root_id.clone());
+ cache.insert("/".to_string(), root_id.clone());
self.query_down(&root_id, paths).await
}
@@ -141,12 +146,13 @@ impl<Q: PathQuery> PathCacher<Q> {
///
/// We should fetch the next `file_id` by sending `query`.
async fn query_down(&self, start_id: &str, paths: VecDeque<String>) -> Result<Option<String>> {
+ let mut cache = self.cache.lock().await;
let mut current_id = start_id.to_string();
for path in paths.into_iter() {
let name = get_basename(&path);
current_id = match self.query.query(¤t_id, name).await? {
Some(id) => {
- self.cache.insert(path, id.clone());
+ cache.insert(path, id.clone());
id
}
None => return Ok(None),
@@ -157,7 +163,7 @@ impl<Q: PathQuery> PathCacher<Q> {
/// Ensure input dir exists.
pub async fn ensure_dir(&self, path: &str) -> Result<String> {
- let _guard = self.lock().await;
+ let mut cache = self.cache.lock().await;
let mut tmp = "".to_string();
// All parents that need to check.
@@ -172,13 +178,13 @@ impl<Q: PathQuery> PathCacher<Q> {
parents.push(tmp.to_string());
}
- let mut parent_id = match self.cache.get("/") {
- Some(v) => v,
- None => self.query.root().await?,
+ let mut parent_id = match cache.get("/") {
+ Some(v) => v.to_string(),
+ None => self.query.root().await?.to_string(),
};
for parent in parents {
- parent_id = match self.cache.get(&parent) {
- Some(value) => value,
+ parent_id = match cache.get(&parent) {
+ Some(value) => value.to_string(),
None => {
let value = match self.query.query(&parent_id, get_basename(&parent)).await? {
Some(value) => value,
@@ -188,13 +194,13 @@ impl<Q: PathQuery> PathCacher<Q> {
.await?
}
};
- self.cache.insert(parent, value.clone());
- value
+ cache.insert(parent, value.clone());
+ value.to_string()
}
}
}
- Ok(parent_id)
+ Ok(parent_id.to_string())
}
}
@@ -237,8 +243,9 @@ mod tests {
for (name, input, expect) in cases {
let cache = PathCacher::new(TestQuery {});
- let actual = cache.get(input).await.unwrap();
- assert_eq!(actual.as_deref(), expect, "{name}")
+ let actual_string = cache.get(input).await.unwrap();
+ let actual = actual_string.as_deref();
+ assert_eq!(actual, expect, "{name}")
}
}
}
diff --git a/core/src/services/gdrive/builder.rs b/core/src/services/gdrive/builder.rs
index e519e175f..4254af746 100644
--- a/core/src/services/gdrive/builder.rs
+++ b/core/src/services/gdrive/builder.rs
@@ -208,7 +208,7 @@ impl Builder for GdriveBuilder {
root,
signer: signer.clone(),
path_cache: PathCacher::new(GdrivePathQuery::new(accessor_info, signer))
- .with_lock(),
+ // .with_lock(),
}),
})
}
(opendal) 02/02: Random wait to test gdrive in CI
Posted by er...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a commit to branch debug-gdrive-ci-test-2
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit f7e345785d7e83a30c46ed023d510545916a01d5
Author: Erick Guan <29...@users.noreply.github.com>
AuthorDate: Fri Oct 17 22:00:12 2025 +0200
Random wait to test gdrive in CI
Locally, it looks like a timing issue
---
core/Cargo.toml | 2 +-
core/src/services/gdrive/builder.rs | 10 ++++++++--
core/src/services/gdrive/core.rs | 19 +++++++++++++++++--
3 files changed, 26 insertions(+), 5 deletions(-)
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 78474f7c3..cc7a183bf 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -155,7 +155,7 @@ services-gcs = [
"reqsign?/services-google",
"reqsign?/reqwest_request",
]
-services-gdrive = ["internal-path-cache"]
+services-gdrive = ["internal-path-cache", "dep:rand"]
services-ghac = ["dep:ghac", "dep:prost", "services-azblob"]
services-github = []
services-gridfs = ["dep:mongodb", "dep:mongodb-internal-macros"]
diff --git a/core/src/services/gdrive/builder.rs b/core/src/services/gdrive/builder.rs
index 4254af746..225b99f57 100644
--- a/core/src/services/gdrive/builder.rs
+++ b/core/src/services/gdrive/builder.rs
@@ -16,6 +16,7 @@
// under the License.
use log::debug;
+use rand::rngs::StdRng;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
@@ -35,6 +36,7 @@ use crate::raw::Timestamp;
use crate::raw::normalize_root;
use crate::services::GdriveConfig;
use crate::*;
+use rand::SeedableRng;
/// [GoogleDrive](https://drive.google.com/) backend support.
#[derive(Default)]
@@ -201,14 +203,18 @@ impl Builder for GdriveBuilder {
};
let signer = Arc::new(Mutex::new(signer));
+ let seed = [
+ 42, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 00, 0, 0,
+ 0, 0, 0, 42,
+ ];
Ok(GdriveBackend {
core: Arc::new(GdriveCore {
info: accessor_info.clone(),
root,
+ rng: Arc::new(Mutex::new(StdRng::from_seed(seed))),
signer: signer.clone(),
- path_cache: PathCacher::new(GdrivePathQuery::new(accessor_info, signer))
- // .with_lock(),
+ path_cache: PathCacher::new(GdrivePathQuery::new(accessor_info, signer)), // .with_lock(),
}),
})
}
diff --git a/core/src/services/gdrive/core.rs b/core/src/services/gdrive/core.rs
index 94b7e340f..089fe413a 100644
--- a/core/src/services/gdrive/core.rs
+++ b/core/src/services/gdrive/core.rs
@@ -22,8 +22,11 @@ use http::Request;
use http::Response;
use http::StatusCode;
use http::header;
+use rand::Rng;
+use rand::rngs::StdRng;
use serde::Deserialize;
use serde_json::json;
+use std::borrow::BorrowMut;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::sync::Arc;
@@ -36,6 +39,7 @@ use crate::*;
pub struct GdriveCore {
pub info: Arc<AccessorInfo>,
+ pub rng: Arc<Mutex<StdRng>>,
pub root: String,
@@ -54,6 +58,12 @@ impl Debug for GdriveCore {
}
impl GdriveCore {
+ pub async fn sleep_random(&self) {
+ let mut added = Arc::clone(&self.rng);
+ let mut rng = added.borrow_mut().lock().await;
+ let range = rng.gen_range(200..1500);
+ tokio::time::sleep(Duration::from_millis(range)).await;
+ }
pub async fn gdrive_stat(&self, path: &str) -> Result<Response<Buffer>> {
let path = build_abs_path(&self.root, path);
let file_id = self.path_cache.get(&path).await?.ok_or(Error::new(
@@ -248,8 +258,13 @@ impl GdriveCore {
}
pub async fn sign<T>(&self, req: &mut Request<T>) -> Result<()> {
- let mut signer = self.signer.lock().await;
- signer.sign(req).await
+ let signed = {
+ let mut signer = self.signer.lock().await;
+ signer.sign(req).await
+ };
+ self.sleep_random().await;
+
+ signed
}
pub async fn gdrive_copy(&self, from: &str, to: &str) -> Result<Response<Buffer>> {