[ISSUE #2419] Adding #[inline] for RocketMQRuntime method #2420
Conversation
WalkthroughThe pull request introduces several new methods to the Changes
Assessment against linked issues
Possibly related PRs
Suggested labels
Suggested reviewers
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
Documentation and Community
|
|
🔊@nakul-py 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
rocketmq-rust-bot
added
AI review first
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🔭 Outside diff range comments (2)
rocketmq-runtime/src/lib.rs (2)
Line range hint
67-100: Consider improvements to the task scheduling implementationSeveral concerns with the current implementation:
- The #[inline] attribute might not be beneficial for this large method
- The scheduling logic doesn't account for task execution time, which could lead to drift
- No mechanism to cancel scheduled tasks
- No error handling for task panics
Consider implementing a more robust solution:
pub fn schedule_at_fixed_rate<F>( &self, task: F, initial_delay: Option<Duration>, period: Duration, - ) where - F: Fn() + Send + 'static, + ) -> tokio::task::JoinHandle<()> where + F: Fn() + Send + Sync + 'static, { match self { RocketMQRuntime::Multi(runtime) => { - runtime.handle().spawn(async move { + runtime.handle().spawn(async move { + let mut interval = match initial_delay { + Some(delay) => { + tokio::time::sleep(delay).await; + tokio::time::interval(period) + } + None => tokio::time::interval(period), + }; + // initial delay if let Some(initial_delay_inner) = initial_delay { tokio::time::sleep(initial_delay_inner).await; } loop { - // record current execution time - let current_execution_time = tokio::time::Instant::now(); - // execute task - task(); - // Calculate the time of the next execution - let next_execution_time = current_execution_time + period; - - // Wait until the next execution - let delay = next_execution_time - .saturating_duration_since(tokio::time::Instant::now()); - tokio::time::sleep(delay).await; + interval.tick().await; + if let Err(e) = std::panic::catch_unwind(|| task()) { + eprintln!("Scheduled task panicked: {:?}", e); + // Optionally break the loop or implement retry logic + } } }); } } }This improvement:
- Uses tokio's built-in interval timing to prevent drift
- Returns a JoinHandle for cancellation
- Adds basic error handling for panics
- Removes the #[inline] attribute as it's unlikely to be beneficial for this size
Line range hint
102-137: Reduce code duplication with the non-mut versionThis method is almost identical to
schedule_at_fixed_rate, leading to unnecessary code duplication. Consider implementing a common internal method that both public methods can use.- #[inline] - pub fn schedule_at_fixed_rate_mut<F>( + pub fn schedule_at_fixed_rate_mut<F>( &self, mut task: F, initial_delay: Option<Duration>, period: Duration, - ) where - F: FnMut() + Send + 'static, + ) -> tokio::task::JoinHandle<()> where + F: FnMut() + Send + 'static, { - match self { - RocketMQRuntime::Multi(runtime) => { - runtime.handle().spawn(async move { - // initial delay - if let Some(initial_delay_inner) = initial_delay { - tokio::time::sleep(initial_delay_inner).await; - } - - loop { - // record current execution time - let current_execution_time = tokio::time::Instant::now(); - // execute task - task(); - // Calculate the time of the next execution - let next_execution_time = current_execution_time + period; - - // Wait until the next execution - let delay = next_execution_time - .saturating_duration_since(tokio::time::Instant::now()); - tokio::time::sleep(delay).await; - } - }); - } - } + // Convert FnMut to Fn by moving it into a RefCell + let task = std::cell::RefCell::new(task); + self.schedule_at_fixed_rate(move || task.borrow_mut()(), initial_delay, period) }
🧹 Nitpick comments (3)
rocketmq-runtime/src/lib.rs (3)
Line range hint
25-35: Consider implementing proper error handling instead of using unwrap()The
unwrap()call could cause a panic if the runtime creation fails. Consider propagating the error usingResult.- pub fn new_multi(threads: usize, name: &str) -> Self { + pub fn new_multi(threads: usize, name: &str) -> std::io::Result<Self> { Self::Multi( tokio::runtime::Builder::new_multi_thread() .worker_threads(threads) .thread_name(name) .enable_all() - .build() - .unwrap(), + .build()?, ) }
53-58: Consider adding documentation for shutdown behaviorWhile the implementation is correct, it would be helpful to add documentation explaining:
- What happens to running tasks
- Whether it's a blocking or non-blocking operation
- Any potential side effects
+ /// Initiates a non-blocking shutdown of the runtime. + /// This method drops the runtime, causing any unfinished tasks to be dropped. + /// For a graceful shutdown with timeout, use `shutdown_timeout` instead. #[inline] pub fn shutdown(self) {
60-65: Add documentation and consider validating the timeout valueThe implementation would benefit from:
- Documentation explaining the shutdown behavior
- Validation of the timeout value to prevent unreasonable values
+ /// Initiates a graceful shutdown of the runtime with a timeout. + /// - Waits for up to `timeout` duration for all tasks to complete + /// - Returns immediately if timeout is zero + /// - Tasks that don't complete within the timeout will be dropped #[inline] pub fn shutdown_timeout(self, timeout: Duration) { + if timeout.is_zero() { + return self.shutdown(); + } match self { Self::Multi(runtime) => runtime.shutdown_timeout(timeout), } }
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
rocketmq-runtime/src/lib.rs(3 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (9)
- GitHub Check: build (windows-latest, nightly)
- GitHub Check: build (windows-latest, stable)
- GitHub Check: build (macos-latest, nightly)
- GitHub Check: build (macos-latest, stable)
- GitHub Check: test
- GitHub Check: build (ubuntu-latest, nightly)
- GitHub Check: build (ubuntu-latest, stable)
- GitHub Check: build
- GitHub Check: auto-approve
🔇 Additional comments (2)
rocketmq-runtime/src/lib.rs (2)
39-44: LGTM! Appropriate use of #[inline] for this getter methodThe implementation is correct and the #[inline] attribute is well-suited for this simple getter method.
46-51: LGTM! Appropriate use of #[inline] for this getter methodThe implementation is correct and the #[inline] attribute is well-suited for this simple getter method.
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## main #2420 +/- ##
=======================================
Coverage 28.59% 28.59%
=======================================
Files 508 508
Lines 73450 73450
=======================================
Hits 21005 21005
Misses 52445 52445 ☔ View full report in Codecov by Sentry. |
rocketmq-rust-bot
added
approved
Which Issue(s) This PR Fixes(Closes)
Add #[inline] for RocketMQRuntime methods.
Fixes #2419
Summary by CodeRabbit