Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: don't create static variables containing a tokio::runtime::Handle #38

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix: don't create static variables containing a tokio::runtime::Handle #38

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 29 additions & 44 deletions src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,7 @@ pin_project! {
if this.inner.is_some() {
// If the inner future wasn't moved out using into_inner,
// enter the tokio context while the inner value is dropped.
let _guard = TOKIO1.handle.enter();
let _guard = get_runtime_handle().enter();
this.project().inner.set(None);
}
}
Expand Down Expand Up @@ -326,7 +326,7 @@ impl<T: Future> Future for Compat<T> {
type Output = T::Output;

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let _guard = TOKIO1.handle.enter();
let _guard = get_runtime_handle().enter();
self.get_pin_mut().poll(cx)
}
}
Expand Down Expand Up @@ -453,38 +453,21 @@ impl<T: futures_io::AsyncSeek> tokio::io::AsyncSeek for Compat<T> {
}
}

static TOKIO1: Lazy<GlobalRuntime> = Lazy::new(|| {
let mut fallback_rt = None;
let handle = tokio::runtime::Handle::try_current().unwrap_or_else(|_| {
thread::Builder::new()
.name("async-compat/tokio-1".into())
.spawn(move || TOKIO1.fallback_rt.as_ref().unwrap().block_on(Pending))
.unwrap();
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("cannot start tokio-1 runtime");

let handle = rt.handle().clone();

fallback_rt = Some(rt);

handle
});
fn get_runtime_handle() -> tokio::runtime::Handle {
tokio::runtime::Handle::try_current().unwrap_or_else(|_| TOKIO1.handle().clone())
}

GlobalRuntime {
handle,
fallback_rt,
}
static TOKIO1: Lazy<tokio::runtime::Runtime> = Lazy::new(|| {
thread::Builder::new()
.name("async-compat/tokio-1".into())
.spawn(|| TOKIO1.block_on(Pending))
.unwrap();
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("cannot start tokio-1 runtime")
});

struct GlobalRuntime {
/// The handle used for all `Compat` futures.
handle: tokio::runtime::Handle,
/// Only used if we couldn't acquire a handle to a runtime on creation.
fallback_rt: Option<tokio::runtime::Runtime>,
}

struct Pending;

impl Future for Pending {
Expand All @@ -497,29 +480,31 @@ impl Future for Pending {

#[cfg(test)]
mod tests {
use super::Lazy;
use crate::{CompatExt, TOKIO1};

#[test]
fn existing_tokio_runtime_is_reused_by_compat() {
fn fallback_runtime_is_created_if_and_only_if_outside_tokio_context() {
// Use compat inside of a tokio context.
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
.block_on(async { println!("foo") }.compat());
.block_on(use_tokio().compat());

assert!(TOKIO1.fallback_rt.is_none());
}
// We didn't need to create the fallback runtime, because we used compat
// inside of an existing tokio context.
assert!(Lazy::get(&TOKIO1).is_none());

#[test]
fn tokio_runtime_is_reused_even_after_it_exits() {
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap()
.block_on(async { println!("foo") });
// Use compat outside of a tokio context.
futures::executor::block_on(use_tokio().compat());

futures::executor::block_on(async { println!("foo") }.compat());
// We must have created the fallback runtime, because we used compat
// outside of a tokio context.
assert!(Lazy::get(&TOKIO1).is_some());
}

assert!(TOKIO1.fallback_rt.is_none());
async fn use_tokio() {
tokio::time::sleep(std::time::Duration::from_micros(1)).await
}
}
Loading