-
-
Notifications
You must be signed in to change notification settings - Fork 150
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: Remove tokio dependency for blocking ureq and curl transpor…
…ts (#422) These transport backends don't need the overhead of a tokio runtime and all the dependency crates that come with it, when they are already blocking on their own. A simple Rust thread and mpsc channel is enough. The transport thread implementation is duplicated between a tokio and non-tokio module that are used by reqwest/surf, and curl/ureq respectively. See also #419 (comment) where this was briefly discussed.
- Loading branch information
Showing
8 changed files
with
146 additions
and
61 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| use std::sync::atomic::{AtomicBool, Ordering}; | ||
| use std::sync::mpsc::{sync_channel, SyncSender}; | ||
| use std::sync::Arc; | ||
| use std::thread::{self, JoinHandle}; | ||
| use std::time::Duration; | ||
|
|
||
| use super::ratelimit::{RateLimiter, RateLimitingCategory}; | ||
| use crate::{sentry_debug, Envelope}; | ||
|
|
||
| enum Task { | ||
| SendEnvelope(Envelope), | ||
| Flush(SyncSender<()>), | ||
| Shutdown, | ||
| } | ||
|
|
||
| pub struct TransportThread { | ||
| sender: SyncSender<Task>, | ||
| shutdown: Arc<AtomicBool>, | ||
| handle: Option<JoinHandle<()>>, | ||
| } | ||
|
|
||
| impl TransportThread { | ||
| pub fn new<SendFn, SendFuture>(mut send: SendFn) -> Self | ||
| where | ||
| SendFn: FnMut(Envelope, RateLimiter) -> SendFuture + Send + 'static, | ||
| // NOTE: returning RateLimiter here, otherwise we are in borrow hell | ||
| SendFuture: std::future::Future<Output = RateLimiter>, | ||
| { | ||
| let (sender, receiver) = sync_channel(30); | ||
| let shutdown = Arc::new(AtomicBool::new(false)); | ||
| let shutdown_worker = shutdown.clone(); | ||
| let handle = thread::Builder::new() | ||
| .name("sentry-transport".into()) | ||
| .spawn(move || { | ||
| // create a runtime on the transport thread | ||
| let rt = tokio::runtime::Builder::new_current_thread() | ||
| .enable_all() | ||
| .build() | ||
| .unwrap(); | ||
|
|
||
| let mut rl = RateLimiter::new(); | ||
|
|
||
| // and block on an async fn in this runtime/thread | ||
| rt.block_on(async move { | ||
| for task in receiver.into_iter() { | ||
| if shutdown_worker.load(Ordering::SeqCst) { | ||
| return; | ||
| } | ||
| let envelope = match task { | ||
| Task::SendEnvelope(envelope) => envelope, | ||
| Task::Flush(sender) => { | ||
| sender.send(()).ok(); | ||
| continue; | ||
| } | ||
| Task::Shutdown => { | ||
| return; | ||
| } | ||
| }; | ||
|
|
||
| if let Some(time_left) = rl.is_disabled(RateLimitingCategory::Any) { | ||
| sentry_debug!( | ||
| "Skipping event send because we're disabled due to rate limits for {}s", | ||
| time_left.as_secs() | ||
| ); | ||
| continue; | ||
| } | ||
| match rl.filter_envelope(envelope) { | ||
| Some(envelope) => { | ||
| rl = send(envelope, rl).await; | ||
| }, | ||
| None => { | ||
| sentry_debug!("Envelope was discarded due to per-item rate limits"); | ||
| }, | ||
| }; | ||
| } | ||
| }) | ||
| }) | ||
| .ok(); | ||
|
|
||
| Self { | ||
| sender, | ||
| shutdown, | ||
| handle, | ||
| } | ||
| } | ||
|
|
||
| pub fn send(&self, envelope: Envelope) { | ||
| let _ = self.sender.send(Task::SendEnvelope(envelope)); | ||
| } | ||
|
|
||
| pub fn flush(&self, timeout: Duration) -> bool { | ||
| let (sender, receiver) = sync_channel(1); | ||
| let _ = self.sender.send(Task::Flush(sender)); | ||
| receiver.recv_timeout(timeout).is_err() | ||
| } | ||
| } | ||
|
|
||
| impl Drop for TransportThread { | ||
| fn drop(&mut self) { | ||
| self.shutdown.store(true, Ordering::SeqCst); | ||
| let _ = self.sender.send(Task::Shutdown); | ||
| if let Some(handle) = self.handle.take() { | ||
| handle.join().unwrap(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters