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

Jump to bottom

refactor: strip down the iroh runtime #1114

Merged
merged 1 commit into from
Jun 21, 2023
Merged

refactor: strip down the iroh runtime #1114

Show file tree
Hide file tree
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
16 changes: 16 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ tokio = { version = "1", features = ["full"] }
tokio-rustls = { version = "0.24", optional = true }
tokio-rustls-acme = { version = "0.1", optional = true }
tokio-stream = "0.1"
tokio-util = { version = "0.7", features = ["io-util", "io"] }
tokio-util = { version = "0.7", features = ["io-util", "io", "rt"] }
toml = { version = "0.7.3", optional = true }
tracing = "0.1"
tracing-futures = "0.2.5"
Expand Down
30 changes: 15 additions & 15 deletions src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,8 +70,8 @@ mod tests {

/// Pick up the tokio runtime from the thread local and add a
/// thread per core runtime.
fn test_runtime() -> crate::runtime::Runtime {
crate::runtime::Runtime::from_currrent("test", 1).unwrap()
fn test_runtime() -> crate::runtime::Handle {
crate::runtime::Handle::from_currrent(1).unwrap()
}

#[tokio::test]
Expand All @@ -80,7 +80,7 @@ mod tests {
let rt = test_runtime();
transfer_data(
vec![("hello_world", "hello world!".as_bytes().to_vec())],
rt.handle(),
&rt,
)
.await
}
Expand All @@ -97,7 +97,7 @@ mod tests {
// overkill, but it works! Just annoying to wait for
// ("4", 1024 * 1024 * 90),
];
transfer_random_data(file_opts, rt.handle()).await
transfer_random_data(file_opts, &rt).await
}

#[tokio::test]
Expand All @@ -114,7 +114,7 @@ mod tests {
(name, 10)
})
.collect();
transfer_random_data(file_opts, rt.handle()).await?;
transfer_random_data(file_opts, &rt).await?;
}
Ok(())
}
Expand All @@ -138,7 +138,7 @@ mod tests {

for size in sizes {
let now = Instant::now();
transfer_random_data(vec![("hello_world", size)], rt.handle()).await?;
transfer_random_data(vec![("hello_world", size)], &rt).await?;
println!(" took {}ms", now.elapsed().as_millis());
}

Expand All @@ -155,7 +155,7 @@ mod tests {
for i in 0..num_files {
file_opts.push((i.to_string(), 0));
}
transfer_random_data(file_opts, rt.handle()).await
transfer_random_data(file_opts, &rt).await
}

#[tokio::test(flavor = "multi_thread")]
Expand All @@ -176,7 +176,7 @@ mod tests {

let rt = test_runtime();
let provider = provider::Provider::builder(db)
.runtime(rt.handle())
.runtime(&rt)
.bind_addr(addr)
.spawn()
.await?;
Expand Down Expand Up @@ -388,7 +388,7 @@ mod tests {
let (db, hash) = create_collection(vec![src.into()]).await.unwrap();
let mut provider = Provider::builder(db)
.bind_addr("127.0.0.1:0".parse().unwrap())
.runtime(rt.handle())
.runtime(&rt)
.spawn()
.await
.unwrap();
Expand Down Expand Up @@ -460,7 +460,7 @@ mod tests {
let (db, hash) = create_collection(vec![src0.into(), src1.into()]).await?;
let provider = Provider::builder(db)
.bind_addr("127.0.0.1:0".parse().unwrap())
.runtime(rt.handle())
.runtime(&rt)
.spawn()
.await?;
let provider_addr = provider.local_endpoint_addresses().await?;
Expand Down Expand Up @@ -501,7 +501,7 @@ mod tests {
let (db, hash) = create_collection(vec![readme.into()]).await.unwrap();
let provider = match Provider::builder(db)
.bind_addr((Ipv6Addr::UNSPECIFIED, 0).into())
.runtime(rt.handle())
.runtime(&rt)
.spawn()
.await
{
Expand Down Expand Up @@ -540,7 +540,7 @@ mod tests {
let (db, hash) = create_collection(vec![readme.into()]).await.unwrap();
let provider = Provider::builder(db)
.bind_addr((Ipv4Addr::UNSPECIFIED, 0).into())
.runtime(rt.handle())
.runtime(&rt)
.spawn()
.await
.unwrap();
Expand Down Expand Up @@ -614,7 +614,7 @@ mod tests {
let (db, hash) = create_collection(vec![readme.into()]).await.unwrap();
let provider = match Provider::builder(db)
.bind_addr("[::1]:0".parse().unwrap())
.runtime(rt.handle())
.runtime(&rt)
.spawn()
.await
{
Expand Down Expand Up @@ -703,7 +703,7 @@ mod tests {
let db = Database::default();
let provider = Provider::builder(db)
.bind_addr("127.0.0.1:0".parse().unwrap())
.runtime(rt.handle())
.runtime(&rt)
.custom_get_handler(BlobCustomHandler)
.spawn()
.await
Expand Down Expand Up @@ -741,7 +741,7 @@ mod tests {
let db = Database::default();
let provider = Provider::builder(db)
.bind_addr("127.0.0.1:0".parse().unwrap())
.runtime(rt.handle())
.runtime(&rt)
.custom_get_handler(CollectionCustomHandler)
.spawn()
.await
Expand Down
10 changes: 5 additions & 5 deletions src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -486,7 +486,7 @@ fn init_metrics_collection(
init_metrics();
// doesn't start the server if the address is None
if let Some(metrics_addr) = metrics_addr {
return Some(rt.spawn(async move {
return Some(rt.main().spawn(async move {
iroh::metrics::start_metrics_server(metrics_addr)
.await
.unwrap_or_else(|e| {
Expand Down Expand Up @@ -514,8 +514,8 @@ fn main() -> Result<()> {

async fn main_impl() -> Result<()> {
let tokio = tokio::runtime::Handle::current();
let tpc = iroh::runtime::tpc::Runtime::new("io", num_cpus::get());
let rt = iroh::runtime::Runtime::new(tokio, tpc);
let tpc = tokio_util::task::LocalPoolHandle::new(num_cpus::get());
let rt = iroh::runtime::Handle::new(tokio, tpc);
tracing_subscriber::registry()
.with(tracing_subscriber::fmt::layer().with_writer(std::io::stderr))
.with(EnvFilter::from_default_env())
Expand All @@ -535,7 +535,7 @@ async fn main_impl() -> Result<()> {
)?;

#[cfg(feature = "metrics")]
let metrics_fut = init_metrics_collection(cli.metrics_addr, rt.handle());
let metrics_fut = init_metrics_collection(cli.metrics_addr, &rt);

let r = match cli.command {
Commands::Get {
Expand Down Expand Up @@ -610,7 +610,7 @@ async fn main_impl() -> Result<()> {
cli.keylog,
rpc_port.into(),
config.derp_map(),
rt.handle(),
&rt,
)
.await?;
let controller = provider.controller();
Expand Down
22 changes: 10 additions & 12 deletions src/provider.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,7 +353,7 @@ where
let handler = RpcHandler {
inner: inner.clone(),
};
rt2.spawn(async move {
rt2.main().spawn(async move {
Self::run(
endpoint,
events_sender,
Expand Down Expand Up @@ -431,7 +431,7 @@ where
let events = events.clone();
let custom_get_handler = custom_get_handler.clone();
let rt2 = rt.clone();
rt.spawn(handle_connection(connecting, db, events, custom_get_handler, rt2));
rt.main().spawn(handle_connection(connecting, db, events, custom_get_handler, rt2));
}
else => break,
}
Expand Down Expand Up @@ -686,7 +686,7 @@ impl RpcHandler {
) -> impl Stream<Item = ValidateProgress> + Send + 'static {
let (tx, rx) = mpsc::channel(1);
let tx2 = tx.clone();
self.rt().spawn(async move {
self.rt().main().spawn(async move {
if let Err(e) = self.inner.db.validate(tx).await {
tx2.send(ValidateProgress::Abort(e.into())).await.unwrap();
}
Expand All @@ -697,7 +697,7 @@ impl RpcHandler {
fn provide(self, msg: ProvideRequest) -> impl Stream<Item = ProvideProgress> {
let (tx, rx) = mpsc::channel(1);
let tx2 = tx.clone();
self.rt().spawn(async move {
self.rt().main().spawn(async move {
if let Err(e) = self.provide0(msg, tx).await {
tx2.send(ProvideProgress::Abort(e.into())).await.unwrap();
}
Expand Down Expand Up @@ -808,7 +808,7 @@ fn handle_rpc_request<C: ServiceEndpoint<ProviderService>>(
rt: &crate::runtime::Handle,
) {
let handler = handler.clone();
rt.spawn(async move {
rt.main().spawn(async move {
use ProviderRequest::*;
match msg {
ListBlobs(msg) => {
Expand Down Expand Up @@ -868,16 +868,14 @@ async fn handle_connection<C: CustomGetHandler>(
events.send(Event::ClientConnected { connection_id }).ok();
let db = db.clone();
let custom_get_handler = custom_get_handler.clone();
rt.spawn_tpc(|| {
rt.local_pool().spawn_pinned(|| {
async move {
if let Err(err) = handle_stream(db, reader, writer, custom_get_handler).await {
warn!("error: {err:#?}",);
}
}
.instrument(span)
.boxed_local()
})
.await;
});
}
}
.instrument(span)
Expand Down Expand Up @@ -1287,8 +1285,8 @@ mod tests {

/// Pick up the tokio runtime from the thread local and add a
/// thread per core runtime.
fn test_runtime() -> crate::runtime::Runtime {
crate::runtime::Runtime::from_currrent("test", 1).unwrap()
fn test_runtime() -> crate::runtime::Handle {
crate::runtime::Handle::from_currrent(1).unwrap()
}

fn blob(size: usize) -> impl Strategy<Value = Bytes> {
Expand Down Expand Up @@ -1420,7 +1418,7 @@ mod tests {
let (db, hash) = create_collection(vec![readme.into()]).await.unwrap();
let provider = Provider::builder(db)
.bind_addr((Ipv4Addr::UNSPECIFIED, 0).into())
.runtime(rt.handle())
.runtime(&rt)
.spawn()
.await
.unwrap();
Expand Down
Loading