Skip to content

fix(iroh): get rid of flume channels #2536

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

Closed
wants to merge 10 commits into from
+82 −164
Closed

fix(iroh): get rid of flume channels #2536

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
4d6d1c8
get rid of a few flume channels
rklaehn Jul 23, 2024
1945183
more async_channel
rklaehn Jul 23, 2024
8145cd6
more async_channel
rklaehn Jul 23, 2024
961a1e4
more purge
rklaehn Jul 23, 2024
350b342
More flume purge
rklaehn Jul 23, 2024
6ef111d
rewrite our workaround same_channel thing
rklaehn Jul 24, 2024
b9d7ace
Remove flume from iroh-docs, iroh-cli and iroh as a direct dependency
rklaehn Jul 24, 2024
ed37ce6
remove double async-channel dep
Aug 1, 2024
6f263e9
Merge remote-tracking branch 'origin/main' into flume-purge
dignifiedquire Aug 5, 2024
2d8dd9a
purge more
dignifiedquire Aug 5, 2024
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
2 changes: 0 additions & 2 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.

1 change: 0 additions & 1 deletion iroh-blobs/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ bao-tree = { version = "0.13", features = ["tokio_fsm", "validate"], default-fe
bytes = { version = "1.4", features = ["serde"] }
chrono = "0.4.31"
derive_more = { version = "=1.0.0-beta.7", features = ["debug", "display", "deref", "deref_mut", "from", "try_into", "into"] }
flume = "0.11"
futures-buffered = "0.2.4"
futures-lite = "2.3"
genawaiter = { version = "0.99.1", features = ["futures03"] }
Expand Down
108 changes: 49 additions & 59 deletions iroh-blobs/src/store/fs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@
//! Design:
//!
//! The redb store is accessed in a single threaded way by an actor that runs
//! on its own std thread. Communication with this actor is via a flume channel,
//! on its own std thread. Communication with this actor is via a async_channel,
//! with oneshot channels for the return values if needed.
//!
//! Errors:
Expand Down Expand Up @@ -116,7 +116,7 @@ use crate::{
};
use tables::{ReadOnlyTables, ReadableTables, Tables};

use self::{tables::DeleteSet, util::PeekableFlumeReceiver};
use self::{tables::DeleteSet, util::PeekableAsyncChannelReceiver};

use self::test_support::EntryData;

Expand Down Expand Up @@ -534,25 +534,25 @@ pub(crate) enum ActorMessage {
/// Query method: get the rough entry status for a hash. Just complete, partial or not found.
EntryStatus {
hash: Hash,
tx: flume::Sender<ActorResult<EntryStatus>>,
tx: async_channel::Sender<ActorResult<EntryStatus>>,
},
#[cfg(test)]
/// Query method: get the full entry state for a hash, both in memory and in redb.
/// This is everything we got about the entry, including the actual inline outboard and data.
EntryState {
hash: Hash,
tx: flume::Sender<ActorResult<test_support::EntryStateResponse>>,
tx: async_channel::Sender<ActorResult<test_support::EntryStateResponse>>,
},
/// Query method: get the full entry state for a hash.
GetFullEntryState {
hash: Hash,
tx: flume::Sender<ActorResult<Option<EntryData>>>,
tx: async_channel::Sender<ActorResult<Option<EntryData>>>,
},
/// Modification method: set the full entry state for a hash.
SetFullEntryState {
hash: Hash,
entry: Option<EntryData>,
tx: flume::Sender<ActorResult<()>>,
tx: async_channel::Sender<ActorResult<()>>,
},
/// Modification method: get or create a file handle for a hash.
///
Expand All @@ -575,7 +575,7 @@ pub(crate) enum ActorMessage {
/// At this point the size, hash and outboard must already be known.
Import {
cmd: Import,
tx: flume::Sender<ActorResult<(TempTag, u64)>>,
tx: async_channel::Sender<ActorResult<(TempTag, u64)>>,
},
/// Modification method: export data from a redb store
///
Expand Down Expand Up @@ -772,7 +772,7 @@ impl Store {

#[derive(Debug)]
struct StoreInner {
tx: flume::Sender<ActorMessage>,
tx: async_channel::Sender<ActorMessage>,
temp: Arc<RwLock<TempCounterMap>>,
handle: Option<std::thread::JoinHandle<()>>,
path_options: Arc<PathOptions>,
Expand Down Expand Up @@ -827,15 +827,13 @@ impl StoreInner {

pub async fn get(&self, hash: Hash) -> OuterResult<Option<BaoFileHandle>> {
let (tx, rx) = oneshot::channel();
self.tx.send_async(ActorMessage::Get { hash, tx }).await?;
self.tx.send(ActorMessage::Get { hash, tx }).await?;
Ok(rx.await??)
}

async fn get_or_create(&self, hash: Hash) -> OuterResult<BaoFileHandle> {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::GetOrCreate { hash, tx })
.await?;
self.tx.send(ActorMessage::GetOrCreate { hash, tx }).await?;
Ok(rx.await??)
}

Expand All @@ -849,9 +847,7 @@ impl StoreInner {
None
}
});
self.tx
.send_async(ActorMessage::Blobs { filter, tx })
.await?;
self.tx.send(ActorMessage::Blobs { filter, tx }).await?;
let blobs = rx.await?;
let res = blobs?
.into_iter()
Expand All @@ -873,9 +869,7 @@ impl StoreInner {
None
}
});
self.tx
.send_async(ActorMessage::Blobs { filter, tx })
.await?;
self.tx.send(ActorMessage::Blobs { filter, tx }).await?;
let blobs = rx.await?;
let res = blobs?
.into_iter()
Expand All @@ -891,9 +885,7 @@ impl StoreInner {
let (tx, rx) = oneshot::channel();
let filter: FilterPredicate<Tag, HashAndFormat> =
Box::new(|_i, k, v| Some((k.value(), v.value())));
self.tx
.send_async(ActorMessage::Tags { filter, tx })
.await?;
self.tx.send(ActorMessage::Tags { filter, tx }).await?;
let tags = rx.await?;
// transform the internal error type into io::Error
let tags = tags?
Expand All @@ -906,51 +898,47 @@ impl StoreInner {
async fn set_tag(&self, tag: Tag, value: Option<HashAndFormat>) -> OuterResult<()> {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::SetTag { tag, value, tx })
.send(ActorMessage::SetTag { tag, value, tx })
.await?;
Ok(rx.await??)
}

async fn create_tag(&self, hash: HashAndFormat) -> OuterResult<Tag> {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::CreateTag { hash, tx })
.await?;
self.tx.send(ActorMessage::CreateTag { hash, tx }).await?;
Ok(rx.await??)
}

async fn delete(&self, hashes: Vec<Hash>) -> OuterResult<()> {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::Delete { hashes, tx })
.await?;
self.tx.send(ActorMessage::Delete { hashes, tx }).await?;
Ok(rx.await??)
}

async fn gc_start(&self) -> OuterResult<()> {
let (tx, rx) = oneshot::channel();
self.tx.send_async(ActorMessage::GcStart { tx }).await?;
self.tx.send(ActorMessage::GcStart { tx }).await?;
Ok(rx.await?)
}

async fn entry_status(&self, hash: &Hash) -> OuterResult<EntryStatus> {
let (tx, rx) = flume::bounded(1);
let (tx, rx) = async_channel::bounded(1);
self.tx
.send_async(ActorMessage::EntryStatus { hash: *hash, tx })
.send(ActorMessage::EntryStatus { hash: *hash, tx })
.await?;
Ok(rx.into_recv_async().await??)
Ok(rx.recv().await??)
}

fn entry_status_sync(&self, hash: &Hash) -> OuterResult<EntryStatus> {
let (tx, rx) = flume::bounded(1);
let (tx, rx) = async_channel::bounded(1);
self.tx
.send(ActorMessage::EntryStatus { hash: *hash, tx })?;
Ok(rx.recv()??)
.send_blocking(ActorMessage::EntryStatus { hash: *hash, tx })?;
Ok(rx.recv_blocking()??)
}

async fn complete(&self, entry: Entry) -> OuterResult<()> {
self.tx
.send_async(ActorMessage::OnComplete { handle: entry })
.send(ActorMessage::OnComplete { handle: entry })
.await?;
Ok(())
}
Expand Down Expand Up @@ -985,7 +973,7 @@ impl StoreInner {
let temp_tag = self.temp.temp_tag(HashAndFormat::raw(hash));
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::Export {
.send(ActorMessage::Export {
cmd: Export {
temp_tag,
target,
Expand All @@ -1005,7 +993,7 @@ impl StoreInner {
) -> OuterResult<()> {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::Fsck {
.send(ActorMessage::Fsck {
repair,
progress,
tx,
Expand All @@ -1017,7 +1005,7 @@ impl StoreInner {
async fn import_flat_store(&self, paths: FlatStorePaths) -> OuterResult<bool> {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::ImportFlatStore { paths, tx })
.send(ActorMessage::ImportFlatStore { paths, tx })
.await?;
Ok(rx.await?)
}
Expand All @@ -1029,7 +1017,7 @@ impl StoreInner {
) -> OuterResult<()> {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::UpdateInlineOptions {
.send(ActorMessage::UpdateInlineOptions {
inline_options,
reapply,
tx,
Expand All @@ -1039,13 +1027,13 @@ impl StoreInner {
}

async fn dump(&self) -> OuterResult<()> {
self.tx.send_async(ActorMessage::Dump).await?;
self.tx.send(ActorMessage::Dump).await?;
Ok(())
}

async fn sync(&self) -> OuterResult<()> {
let (tx, rx) = oneshot::channel();
self.tx.send_async(ActorMessage::Sync { tx }).await?;
self.tx.send(ActorMessage::Sync { tx }).await?;
Ok(rx.await?)
}

Expand Down Expand Up @@ -1141,8 +1129,8 @@ impl StoreInner {
let tag = self.temp.temp_tag(HashAndFormat { hash, format });
let hash = *tag.hash();
// blocking send for the import
let (tx, rx) = flume::bounded(1);
self.tx.send(ActorMessage::Import {
let (tx, rx) = async_channel::bounded(1);
self.tx.send_blocking(ActorMessage::Import {
cmd: Import {
content_id: HashAndFormat { hash, format },
source: file,
Expand All @@ -1151,7 +1139,7 @@ impl StoreInner {
},
tx,
})?;
Ok(rx.recv()??)
Ok(rx.recv_blocking()??)
}

fn temp_file_name(&self) -> PathBuf {
Expand All @@ -1161,7 +1149,7 @@ impl StoreInner {
async fn shutdown(&self) {
let (tx, rx) = oneshot::channel();
self.tx
.send_async(ActorMessage::Shutdown { tx: Some(tx) })
.send(ActorMessage::Shutdown { tx: Some(tx) })
.await
.ok();
rx.await.ok();
Expand All @@ -1171,7 +1159,9 @@ impl StoreInner {
impl Drop for StoreInner {
fn drop(&mut self) {
if let Some(handle) = self.handle.take() {
self.tx.send(ActorMessage::Shutdown { tx: None }).ok();
self.tx
.send_blocking(ActorMessage::Shutdown { tx: None })
.ok();
handle.join().ok();
}
}
Expand All @@ -1181,7 +1171,7 @@ struct ActorState {
handles: BTreeMap<Hash, BaoFileHandleWeak>,
protected: BTreeSet<Hash>,
temp: Arc<RwLock<TempCounterMap>>,
msgs: flume::Receiver<ActorMessage>,
msgs: async_channel::Receiver<ActorMessage>,
create_options: Arc<BaoFileConfig>,
options: Options,
rt: tokio::runtime::Handle,
Expand Down Expand Up @@ -1243,13 +1233,13 @@ pub(crate) enum OuterError {
#[error("inner error: {0}")]
Inner(#[from] ActorError),
#[error("send error: {0}")]
Send(#[from] flume::SendError<ActorMessage>),
Send1(#[from] async_channel::SendError<ActorMessage>),
#[error("progress send error: {0}")]
ProgressSend(#[from] ProgressSendError),
#[error("recv error: {0}")]
Recv(#[from] oneshot::error::RecvError),
#[error("recv error: {0}")]
FlumeRecv(#[from] flume::RecvError),
AsyncChannelRecv(#[from] async_channel::RecvError),
#[error("join error: {0}")]
JoinTask(#[from] tokio::task::JoinError),
}
Expand Down Expand Up @@ -1434,7 +1424,7 @@ impl Actor {
options: Options,
temp: Arc<RwLock<TempCounterMap>>,
rt: tokio::runtime::Handle,
) -> ActorResult<(Self, flume::Sender<ActorMessage>)> {
) -> ActorResult<(Self, async_channel::Sender<ActorMessage>)> {
let db = match redb::Database::create(path) {
Ok(db) => db,
Err(DatabaseError::UpgradeRequired(1)) => {
Expand All @@ -1451,11 +1441,11 @@ impl Actor {
txn.commit()?;
// make the channel relatively large. there are some messages that don't
// require a response, it's fine if they pile up a bit.
let (tx, rx) = flume::bounded(1024);
let (tx, rx) = async_channel::bounded(1024);
let tx2 = tx.clone();
let on_file_create: CreateCb = Arc::new(move |hash| {
// todo: make the callback allow async
tx2.send(ActorMessage::OnMemSizeExceeded { hash: *hash })
tx2.send_blocking(ActorMessage::OnMemSizeExceeded { hash: *hash })
.ok();
Ok(())
});
Expand All @@ -1482,7 +1472,7 @@ impl Actor {
}

fn run_batched(mut self) -> ActorResult<()> {
let mut msgs = PeekableFlumeReceiver::new(self.state.msgs.clone());
let mut msgs = PeekableAsyncChannelReceiver::new(self.state.msgs.clone());
while let Some(msg) = msgs.recv() {
if let ActorMessage::Shutdown { tx } = msg {
// Make sure the database is dropped before we send the reply.
Expand Down Expand Up @@ -2217,7 +2207,7 @@ impl ActorState {
}
ActorMessage::EntryStatus { hash, tx } => {
let res = self.entry_status(tables, hash);
tx.send(res).ok();
tx.send_blocking(res).ok();
}
ActorMessage::Blobs { filter, tx } => {
let res = self.blobs(tables, filter);
Expand All @@ -2237,11 +2227,11 @@ impl ActorState {
}
#[cfg(test)]
ActorMessage::EntryState { hash, tx } => {
tx.send(self.entry_state(tables, hash)).ok();
tx.send_blocking(self.entry_state(tables, hash)).ok();
}
ActorMessage::GetFullEntryState { hash, tx } => {
let res = self.get_full_entry_state(tables, hash);
tx.send(res).ok();
tx.send_blocking(res).ok();
}
x => return Ok(Err(x)),
}
Expand All @@ -2256,7 +2246,7 @@ impl ActorState {
match msg {
ActorMessage::Import { cmd, tx } => {
let res = self.import(tables, cmd);
tx.send(res).ok();
tx.send_blocking(res).ok();
}
ActorMessage::SetTag { tag, value, tx } => {
let res = self.set_tag(tables, tag, value);
Expand Down Expand Up @@ -2287,7 +2277,7 @@ impl ActorState {
}
ActorMessage::SetFullEntryState { hash, entry, tx } => {
let res = self.set_full_entry_state(tables, hash, entry);
tx.send(res).ok();
tx.send_blocking(res).ok();
}
msg => {
// try to handle it as readonly
Expand Down
Loading
Loading