From 1935d40c37924c3f47b2faa3398f5140264403b6 Mon Sep 17 00:00:00 2001 From: yoshidan Date: Fri, 8 Mar 2024 09:00:53 +0900 Subject: [PATCH 1/3] set same ack_deadline_seconds as subscription config --- pubsub/src/client.rs | 8 +++++--- pubsub/src/subscriber.rs | 20 ++++++++++++++++---- pubsub/src/subscription.rs | 37 ++++++++++++++++++++++++------------- 3 files changed, 45 insertions(+), 20 deletions(-) diff --git a/pubsub/src/client.rs b/pubsub/src/client.rs index 00bab54c..f6dae5a4 100644 --- a/pubsub/src/client.rs +++ b/pubsub/src/client.rs @@ -325,10 +325,10 @@ mod tests { let config = ReceiveConfig { worker_count: 2, channel_capacity: None, - subscriber_config: SubscriberConfig { + subscriber_config: Some(SubscriberConfig { ping_interval: Duration::from_secs(1), ..Default::default() - }, + }), }; let cancel_receiver = cancellation_token.clone(); let (s, mut r) = tokio::sync::mpsc::channel(100); @@ -631,6 +631,8 @@ mod tests_in_gcp { break; } let msg_id = &message.message.message_id; + // heavy task + tokio::time::sleep(Duration::from_secs(1)).await; *msgs.entry(msg_id.clone()).or_insert(0) += 1; message.ack().await.unwrap(); } @@ -638,7 +640,7 @@ mod tests_in_gcp { msgs }); - tokio::time::sleep(Duration::from_secs(30)).await; + tokio::time::sleep(Duration::from_secs(60)).await; // check redelivered messages ctx.cancel(); diff --git a/pubsub/src/subscriber.rs b/pubsub/src/subscriber.rs index d8446035..2d4161fe 100644 --- a/pubsub/src/subscriber.rs +++ b/pubsub/src/subscriber.rs @@ -90,7 +90,21 @@ pub struct SubscriberConfig { /// ping interval for Bi Directional Streaming pub ping_interval: Duration, pub retry_setting: Option, + /// It is important for exactly_once_delivery + /// The ack deadline to use for the stream. This must be provided in + /// the first request on the stream, but it can also be updated on subsequent + /// requests from client to server. The minimum deadline you can specify is 10 + /// seconds. The maximum deadline you can specify is 600 seconds (10 minutes). pub stream_ack_deadline_seconds: i32, + /// Flow control settings for the maximum number of outstanding messages. When + /// there are `max_outstanding_messages` or more currently sent to the + /// streaming pull client that have not yet been acked or nacked, the server + /// stops sending more messages. The sending of messages resumes once the + /// number of outstanding messages is less than this value. If the value is + /// <= 0, there is no limit to the number of outstanding messages. This + /// property can only be set on the initial StreamingPullRequest. If it is set + /// on a subsequent request, the stream will be aborted with status + /// `INVALID_ARGUMENT`. pub max_outstanding_messages: i64, pub max_outstanding_bytes: i64, } @@ -101,7 +115,7 @@ impl Default for SubscriberConfig { ping_interval: std::time::Duration::from_secs(10), retry_setting: Some(default_retry_setting()), stream_ack_deadline_seconds: 60, - max_outstanding_messages: 1000, + max_outstanding_messages: 50, max_outstanding_bytes: 1000 * 1000 * 1000, } } @@ -119,10 +133,8 @@ impl Subscriber { subscription: String, client: SubscriberClient, queue: async_channel::Sender, - opt: Option, + config: SubscriberConfig, ) -> Self { - let config = opt.unwrap_or_default(); - let (ping_sender, ping_receiver) = async_channel::unbounded(); // ping request diff --git a/pubsub/src/subscription.rs b/pubsub/src/subscription.rs index b9031af6..96b4fa06 100644 --- a/pubsub/src/subscription.rs +++ b/pubsub/src/subscription.rs @@ -83,7 +83,7 @@ pub struct SubscriptionConfigToUpdate { pub struct SubscribeConfig { enable_multiple_subscriber: bool, channel_capacity: Option, - subscriber_config: SubscriberConfig, + subscriber_config: Option, } impl SubscribeConfig { @@ -92,7 +92,7 @@ impl SubscribeConfig { self } pub fn with_subscriber_config(mut self, v: SubscriberConfig) -> Self { - self.subscriber_config = v; + self.subscriber_config = Some(v); self } pub fn with_channel_capacity(mut self, v: usize) -> Self { @@ -105,14 +105,14 @@ impl SubscribeConfig { pub struct ReceiveConfig { pub worker_count: usize, pub channel_capacity: Option, - pub subscriber_config: SubscriberConfig, + pub subscriber_config: Option, } impl Default for ReceiveConfig { fn default() -> Self { Self { worker_count: 10, - subscriber_config: SubscriberConfig::default(), + subscriber_config: None, channel_capacity: None, } } @@ -385,6 +385,7 @@ impl Subscription { let opt = opt.unwrap_or_default(); let (tx, rx) = create_channel(opt.channel_capacity); let cancel = CancellationToken::new(); + let sub_opt = self.unwrap_subscribe_config(opt.subscriber_config).await?; // spawn a separate subscriber task for each connection in the pool let subscribers = if opt.enable_multiple_subscriber { @@ -398,7 +399,7 @@ impl Subscription { self.fqsn.clone(), self.subc.clone(), tx.clone(), - Some(opt.subscriber_config.clone()), + sub_opt.clone(), ); } @@ -420,9 +421,10 @@ impl Subscription { let op = config.unwrap_or_default(); let mut receivers = Vec::with_capacity(op.worker_count); let mut senders = Vec::with_capacity(receivers.len()); + let sub_opt = self.unwrap_subscribe_config(op.subscriber_config).await?; if self - .config(op.subscriber_config.retry_setting.clone()) + .config(sub_opt.retry_setting.clone()) .await? .1 .enable_message_ordering @@ -444,13 +446,7 @@ impl Subscription { let subscribers: Vec = senders .into_iter() .map(|queue| { - Subscriber::start( - cancel.clone(), - self.fqsn.clone(), - self.subc.clone(), - queue, - Some(op.subscriber_config.clone()), - ) + Subscriber::start(cancel.clone(), self.fqsn.clone(), self.subc.clone(), queue, sub_opt.clone()) }) .collect(); @@ -602,6 +598,21 @@ impl Subscription { let _ = self.subc.delete_snapshot(req, retry).await?; Ok(()) } + + async fn unwrap_subscribe_config(&self, cfg: Option) -> Result { + if let Some(cfg) = cfg { + return Ok(cfg); + } + let cfg = self.config(None).await?; + let mut default_cfg = SubscriberConfig { + stream_ack_deadline_seconds: cfg.1.ack_deadline_seconds, + ..Default::default() + }; + if cfg.1.enable_exactly_once_delivery { + default_cfg.max_outstanding_messages = 5; + } + Ok(default_cfg) + } } fn create_channel( From 9dce72f66b999b6c3302b7997cb746ded969c7ed Mon Sep 17 00:00:00 2001 From: yoshidan Date: Fri, 8 Mar 2024 09:34:42 +0900 Subject: [PATCH 2/3] set same ack_deadline_seconds as subscription config --- pubsub/src/subscription.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pubsub/src/subscription.rs b/pubsub/src/subscription.rs index 96b4fa06..9d24ab4a 100644 --- a/pubsub/src/subscription.rs +++ b/pubsub/src/subscription.rs @@ -1,3 +1,4 @@ +use std::cmp::{max, min}; use std::collections::HashMap; use std::future::Future; use std::pin::Pin; @@ -605,7 +606,7 @@ impl Subscription { } let cfg = self.config(None).await?; let mut default_cfg = SubscriberConfig { - stream_ack_deadline_seconds: cfg.1.ack_deadline_seconds, + stream_ack_deadline_seconds: max(min(cfg.1.ack_deadline_seconds, 600), 10), ..Default::default() }; if cfg.1.enable_exactly_once_delivery { From 1db19ab16af45dfd729d68fad642ae9f20ddf922 Mon Sep 17 00:00:00 2001 From: yoshidan Date: Fri, 8 Mar 2024 16:15:41 +0900 Subject: [PATCH 3/3] make set_call_option to pub --- spanner/src/reader.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spanner/src/reader.rs b/spanner/src/reader.rs index cbe0aa47..7b0d9a3f 100644 --- a/spanner/src/reader.rs +++ b/spanner/src/reader.rs @@ -213,7 +213,7 @@ where }) } - pub(crate) fn set_call_options(&mut self, option: CallOptions) { + pub fn set_call_options(&mut self, option: CallOptions) { self.reader_option = Some(option); }