Skip to content

Commit 41a9f19

Browse files
gzgz
committed
Switch from crossbeam-channel to std::sync::mpsc::channel.
The code is identical except that rust-lang/rust#107466 is fixed in std but not in crossbeam-channel. I originally reworked this because the bug in 80ff3ff made me desparate to suspect something is wrong with the channel. While it didn't turn out to be an issue with the channels I still think it's good to depend on std code instead of a library. Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>
1 parent 80ff3ff commit 41a9f19

4 files changed

Lines changed: 23 additions & 19 deletions

File tree

‎crates/dbsp/src/circuit/runtime.rs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -13,12 +13,12 @@ use crate::{
1313
},
1414
DetailedError,
1515
};
16-
use crossbeam::channel::{bounded, Receiver, Sender};
1716
use crossbeam_utils::sync::{Parker, Unparker};
1817
use lazy_static::lazy_static;
1918
use once_cell::sync::Lazy;
2019
use pipeline_types::config::StorageCacheConfig;
2120
use serde::Serialize;
21+
use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
2222
use std::{
2323
backtrace::Backtrace,
2424
borrow::Cow,
@@ -327,7 +327,7 @@ fn mk_background_thread(
327327
runtime: Option<Runtime>,
328328
worker_index: usize,
329329
bg_work_receiver: Receiver<BackgroundOperation>,
330-
init_sender: Option<Sender<(Unparker, Arc<AtomicBool>)>>,
330+
init_sender: Option<SyncSender<(Unparker, Arc<AtomicBool>)>>,
331331
) -> JoinHandle<()> {
332332
Builder::new()
333333
.name(thread_name)
@@ -451,8 +451,8 @@ impl Runtime {
451451
let mut background_handles = Vec::with_capacity(nworkers);
452452
background_handles.extend(workers.clone().map(|worker_index| {
453453
let cloned_runtime = runtime.clone();
454-
let (init_sender, init_receiver) = bounded(1);
455-
let (bg_work_sender, bg_work_receiver) = bounded(BatchMerger::RX_QUEUE_SIZE);
454+
let (init_sender, init_receiver) = sync_channel(1);
455+
let (bg_work_sender, bg_work_receiver) = sync_channel(BatchMerger::RX_QUEUE_SIZE);
456456
let join_handle = mk_background_thread(
457457
format!("dbsp-background-{}", worker_index),
458458
Some(cloned_runtime),
@@ -479,7 +479,7 @@ impl Runtime {
479479
let runtime = runtime.clone();
480480
let build_circuit = circuit.clone();
481481

482-
let (init_sender, init_receiver) = bounded(1);
482+
let (init_sender, init_receiver) = sync_channel(1);
483483
let join_handle = Builder::new()
484484
.name(format!("dbsp-worker-{worker_index}"))
485485
.spawn(move || {
@@ -524,15 +524,15 @@ impl Runtime {
524524
///
525525
/// This is currently only used for file compaction but could be extended for
526526
/// more generic background work in the future.
527-
pub(crate) fn background_channel() -> Arc<Sender<BackgroundOperation>> {
527+
pub(crate) fn background_channel() -> Arc<SyncSender<BackgroundOperation>> {
528528
let worker_index = Runtime::worker_index();
529529
if let Some(rt) = Runtime::runtime() {
530530
rt.local_store()
531531
.get(&BackgroundChannel(worker_index))
532532
.unwrap()
533533
.clone()
534534
} else {
535-
let (bg_work_sender, bg_work_receiver) = bounded(BatchMerger::RX_QUEUE_SIZE);
535+
let (bg_work_sender, bg_work_receiver) = sync_channel(BatchMerger::RX_QUEUE_SIZE);
536536
let _join_handle = mk_background_thread(
537537
String::from("dbsp-background-no-runtime"),
538538
None,
@@ -880,7 +880,7 @@ impl TypedMapKey<LocalStoreMarker> for WorkerId {
880880
struct BackgroundChannel(usize);
881881

882882
impl TypedMapKey<LocalStoreMarker> for BackgroundChannel {
883-
type Value = Arc<Sender<BackgroundOperation>>;
883+
type Value = Arc<SyncSender<BackgroundOperation>>;
884884
}
885885

886886
#[derive(Hash, PartialEq, Eq)]

‎crates/dbsp/src/storage/backend/mod.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -196,11 +196,11 @@ pub enum StorageError {
196196

197197
/// Unable to receive a message from the thread that does storage compaction.
198198
#[error("Unable to receive a message from the merge-thread.")]
199-
TryRx(#[from] crossbeam::channel::TryRecvError),
199+
TryRx(#[from] std::sync::mpsc::TryRecvError),
200200

201201
/// Error when sending a message to the thread that merges batches.
202202
#[error("Unable to receive a message from the compactor-thread.")]
203-
Rx(#[from] crossbeam::channel::RecvError),
203+
Rx(#[from] std::sync::mpsc::RecvError),
204204

205205
/// The thread that merges batches exited unexpectedly.
206206
#[error("Compactor Thread exited unexpectedly.")]

‎crates/dbsp/src/trace/spine_async/merger.rs‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,8 @@ use crate::storage::backend::StorageError as Error;
55
use crate::trace::spine_async::{BatchIdent, MAX_LEVELS};
66
use crate::trace::Batch;
77
use crate::Runtime;
8-
use crossbeam::channel::internal::SelectHandle;
9-
use crossbeam::channel::{Receiver, RecvError};
108
use metrics::gauge;
9+
use std::sync::mpsc::{Receiver, RecvError};
1110

1211
pub(crate) enum BackgroundOperation {
1312
/// This is a closure that will be called by the compactor thread which ultimately
@@ -51,7 +50,7 @@ impl BatchMerger {
5150
pub(crate) fn run(&mut self) {
5251
let label = vec![("compactor", Runtime::background_index().to_string())];
5352
while !Runtime::kill_in_progress() {
54-
gauge!(COMPACTION_QUEUE_LENGTH, &label).set(self.receiver.len() as f64);
53+
gauge!(COMPACTION_QUEUE_LENGTH, &label).set(self.receiver.try_iter().count() as f64);
5554
let op: Result<BackgroundOperation, RecvError> = self.receiver.recv();
5655
match op {
5756
Ok(BackgroundOperation::Merge(merge_fun)) => {
@@ -75,8 +74,13 @@ impl BatchMerger {
7574
fuel <= 0
7675
});
7776

78-
if self.in_progress.len() < Self::CONCURRENT_MERGES && self.receiver.is_ready() {
79-
break;
77+
if self.in_progress.len() < Self::CONCURRENT_MERGES {
78+
self.in_progress.extend(
79+
self.receiver
80+
.try_iter()
81+
.take(Self::CONCURRENT_MERGES - self.in_progress.len())
82+
.map(|BackgroundOperation::Merge(merge_fun)| merge_fun),
83+
);
8084
}
8185
}
8286
}

‎crates/dbsp/src/trace/spine_async/mod.rs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,13 @@ use crate::storage::{checkpoint_path, write_commit_metadata};
1818
use crate::trace::spine_async::merger::{BackgroundOperation, MergeResult};
1919
use crate::trace::spine_fueled::CommittedSpine;
2020
use crate::trace::Merger;
21-
use crossbeam::channel::{unbounded, Receiver, Sender, TryRecvError};
2221
use metrics::{counter, histogram};
2322
use rand::Rng;
2423
use rkyv::{ser::Serializer, Archive, Archived, Deserialize, Fallible, Serialize};
2524
use size_of::SizeOf;
2625
use std::collections::BTreeMap;
2726
use std::path::{Path, PathBuf};
27+
use std::sync::mpsc::{channel, Receiver, Sender, SyncSender, TryRecvError};
2828
use std::sync::Arc;
2929
use std::time::Instant;
3030
use std::{
@@ -213,7 +213,7 @@ where
213213
value_filter: Option<Filter<B::Val>>,
214214
/// The channel where we send merge requests to the compactor thread.
215215
#[size_of(skip)]
216-
merger_tx: Arc<Sender<BackgroundOperation>>,
216+
merger_tx: Arc<SyncSender<BackgroundOperation>>,
217217
/// The closure of the compactor thread uses this channel to send completed
218218
/// merges back to us.
219219
#[size_of(skip)]
@@ -574,7 +574,7 @@ where
574574

575575
/// Starts a new merge.
576576
fn enqueue(
577-
sender: &Arc<Sender<BackgroundOperation>>,
577+
sender: &Arc<SyncSender<BackgroundOperation>>,
578578
key: BatchIdent,
579579
mut batches: Vec<Arc<B>>,
580580
key_filter: Option<Filter<B::Key>>,
@@ -966,7 +966,7 @@ where
966966
/// The `effort` parameter is that multiplier. This value should be at
967967
/// least one for the merging to happen; a value of zero is not helpful.
968968
pub fn with_effort(factories: &B::Factories, _effort: usize) -> Self {
969-
let (tx, rx) = unbounded();
969+
let (tx, rx) = channel();
970970
Spine {
971971
factories: factories.clone(),
972972
lower: Antichain::from_elem(B::Time::minimum()),

0 commit comments

Comments
 (0)