You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by ya...@apache.org on 2022/12/27 13:50:35 UTC
[dubbo-rust] branch main updated: Support timeout after waiting RPC response for a maximum time. #47 (#92)
This is an automated email from the ASF dual-hosted git repository.
yangyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/dubbo-rust.git
The following commit(s) were added to refs/heads/main by this push:
new b3fce7f Support timeout after waiting RPC response for a maximum time. #47 (#92)
b3fce7f is described below
commit b3fce7fe3c4c7ec94c5b09c7805d116ad758892d
Author: daedalus2022 <11...@users.noreply.github.com>
AuthorDate: Tue Dec 27 21:50:30 2022 +0800
Support timeout after waiting RPC response for a maximum time. #47 (#92)
* Support timeout after waiting RPC response for a maximum time. #47
* timeout key by define const
Co-authored-by: qunwei <qu...@prevailcloud.com>
---
dubbo/src/filter/context.rs | 71 ++++++++++++++++++++++++++++++++++++++++
dubbo/src/filter/mod.rs | 6 ++++
dubbo/src/filter/timeout.rs | 65 ++++++++++++++++++++++++++++++++++++
examples/echo/src/echo/server.rs | 7 +++-
4 files changed, 148 insertions(+), 1 deletion(-)
diff --git a/dubbo/src/filter/context.rs b/dubbo/src/filter/context.rs
new file mode 100644
index 0000000..3f73fd0
--- /dev/null
+++ b/dubbo/src/filter/context.rs
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use serde_json::Value;
+
+use crate::{
+ codegen::Request,
+ context::{Context, RpcContext},
+ filter::{TIMEOUT_COUNTDOWN, TIMEOUT_DEFAULT, TRI_TIMEOUT_DEADLINE_IN_NANOS},
+ status::Status,
+};
+
+use super::Filter;
+
+#[derive(Clone)]
+pub struct ContextFilter {}
+
+impl Filter for ContextFilter {
+ fn call(&mut self, req: Request<()>) -> Result<Request<()>, Status> {
+ let headers = &mut req.metadata.into_headers();
+
+ let timeout = headers.get(TIMEOUT_COUNTDOWN);
+
+ let time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .unwrap()
+ .as_nanos();
+
+ let mut dead_line_in_nanos = 0_u128;
+
+ if let Some(t) = timeout {
+ let timeout: u128 = t.to_str().unwrap().parse().unwrap();
+ if timeout > 0_u128 {
+ dead_line_in_nanos = time + timeout * 1000000;
+ }
+ } else {
+ let timeout: u128 = TIMEOUT_DEFAULT * 1000000;
+ dead_line_in_nanos = time + timeout;
+ }
+
+ tracing::debug!(
+ "ContextFilter tri-timeout-deadline-in-nanos : {}",
+ dead_line_in_nanos
+ );
+ if let Some(at) = RpcContext::get_attachments() {
+ let mut attachments = at.lock().unwrap();
+ attachments.insert(
+ String::from(TRI_TIMEOUT_DEADLINE_IN_NANOS),
+ Value::from(dead_line_in_nanos.to_string()),
+ );
+ }
+
+ Ok(req)
+ }
+}
diff --git a/dubbo/src/filter/mod.rs b/dubbo/src/filter/mod.rs
index 075781a..9523020 100644
--- a/dubbo/src/filter/mod.rs
+++ b/dubbo/src/filter/mod.rs
@@ -15,10 +15,16 @@
* limitations under the License.
*/
+pub mod context;
pub mod service;
+pub mod timeout;
use crate::invocation::Request;
+pub const TRI_TIMEOUT_DEADLINE_IN_NANOS: &str = "tri-timeout-deadline-in-nanos";
+pub const TIMEOUT_COUNTDOWN: &str = "timeout-countdown";
+pub const TIMEOUT_DEFAULT: u128 = 1000;
+
pub trait Filter {
fn call(&mut self, req: Request<()>) -> Result<Request<()>, crate::status::Status>;
}
diff --git a/dubbo/src/filter/timeout.rs b/dubbo/src/filter/timeout.rs
new file mode 100644
index 0000000..353585d
--- /dev/null
+++ b/dubbo/src/filter/timeout.rs
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use crate::{
+ codegen::Request,
+ context::{Context, RpcContext},
+ filter::TRI_TIMEOUT_DEADLINE_IN_NANOS,
+ status::{Code, Status},
+};
+
+use super::Filter;
+
+#[derive(Clone)]
+pub struct TimeoutFilter {}
+
+/// timeout count
+/// 1. ContextFilter 初始化 timeout 时间，初始化后将 tri-timeout-deadline-in-nanos 放入 context 中
+/// 2. TimeoutFilter read context tri-timeout-deadline-in-nanos
+/// 3. 响应时计算 tri-timeout-deadline-in-nanos - current_nanos <= 0
+///
+impl Filter for TimeoutFilter {
+ fn call(&mut self, req: Request<()>) -> Result<Request<()>, Status> {
+ if let Some(attachments) = RpcContext::get_attachments() {
+ let current_nanos = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .unwrap()
+ .as_nanos();
+
+ let attachments = attachments.lock().unwrap();
+ let tri_timeout_deadline_in_nanos =
+ attachments.get(TRI_TIMEOUT_DEADLINE_IN_NANOS).unwrap();
+ let tri_timeout_deadline_in_nanos: u128 = tri_timeout_deadline_in_nanos
+ .as_str()
+ .unwrap()
+ .parse()
+ .unwrap();
+
+ tracing::debug!(
+ "TimeoutFilter tri-timeout-deadline-in-nanos : {}, current-nanos:{}",
+ tri_timeout_deadline_in_nanos,
+ current_nanos
+ );
+ if tri_timeout_deadline_in_nanos - current_nanos <= 0 {
+ return Err(Status::new(Code::DeadlineExceeded, String::from("Timeout")));
+ }
+ }
+
+ Ok(req)
+ }
+}
diff --git a/examples/echo/src/echo/server.rs b/examples/echo/src/echo/server.rs
index 2be45b0..6f2d610 100644
--- a/examples/echo/src/echo/server.rs
+++ b/examples/echo/src/echo/server.rs
@@ -19,6 +19,8 @@ use std::io::ErrorKind;
use std::pin::Pin;
use async_trait::async_trait;
+use dubbo::filter::context::ContextFilter;
+use dubbo::filter::timeout::TimeoutFilter;
use futures_util::Stream;
use futures_util::StreamExt;
use tokio::sync::mpsc;
@@ -52,12 +54,15 @@ async fn main() {
});
let server = EchoServerImpl::default();
let s = EchoServer::<EchoServerImpl>::with_filter(server, FakeFilter {});
+ let timeout_filter = FilterService::new(s, TimeoutFilter {});
+ let context_filter = FilterService::new(timeout_filter, ContextFilter {});
+
dubbo::protocol::triple::TRIPLE_SERVICES
.write()
.unwrap()
.insert(
"grpc.examples.echo.Echo".to_string(),
- dubbo::utils::boxed_clone::BoxCloneService::new(s),
+ dubbo::utils::boxed_clone::BoxCloneService::new(context_filter),
);
// Dubbo::new().start().await;