use crossbeam instead of mpsc

polars/polars-lazy/polars-pipe/Cargo.toml

@@ -9,6 +9,7 @@ description = "Lazy query engine for the Polars DataFrame library"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+crossbeam-channel = "0.5"
 enum_dispatch = "0.3"
 hashbrown.workspace = true
 num.workspace = true

polars/polars-lazy/polars-pipe/src/executors/sinks/parquet_sink.rs

@@ -1,8 +1,9 @@
 use std::any::Any;
 use std::path::Path;
-use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
+// use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
 use std::thread::JoinHandle;
 
+use crossbeam_channel::{bounded, Receiver, Sender};
 use polars_core::prelude::*;
 use polars_io::parquet::{BatchedWriter, ParquetWriter};
 use polars_plan::prelude::ParquetWriteOptions;
@@ -57,7 +58,7 @@ fn init_writer_thread(
 // Ensure the data is return in the order it was streamed
 #[derive(Clone)]
 pub struct ParquetSink {
-    sender: SyncSender<Option<DataChunk>>,
+    sender: Sender<Option<DataChunk>>,
     io_thread_handle: Arc<Option<JoinHandle<()>>>,
 }
 
@@ -73,7 +74,7 @@ impl ParquetSink {
 
         let morsels_per_sink = morsels_per_sink();
         let backpressure = morsels_per_sink * 2;
-        let (sender, receiver) = sync_channel(backpressure);
+        let (sender, receiver) = bounded(backpressure);
 
         let io_thread_handle = Arc::new(Some(init_writer_thread(
             receiver,

polars/polars-lazy/polars-pipe/src/executors/sinks/sort/io.rs

@@ -1,9 +1,9 @@
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::mpsc::SyncSender;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+use crossbeam_channel::{bounded, Sender};
 use polars_core::prelude::*;
 use polars_core::{POOL, PROCESS_ID};
 use polars_io::prelude::*;
@@ -13,7 +13,7 @@ pub(super) type DfIter = Box<dyn ExactSizeIterator<Item = DataFrame> + Sync + Se
 type Payload = (Option<IdxCa>, DfIter);
 
 pub(super) struct IOThread {
-    sender: SyncSender<Payload>,
+    sender: Sender<Payload>,
     pub(super) dir: PathBuf,
     pub(super) sent: Arc<AtomicUsize>,
     pub(super) total: Arc<AtomicUsize>,
@@ -53,8 +53,7 @@ impl IOThread {
         std::fs::create_dir_all(&dir)?;
 
         // we need some pushback otherwise we still could go OOM.
-        let (sender, receiver) =
-            std::sync::mpsc::sync_channel::<Payload>(POOL.current_num_threads() * 2);
+        let (sender, receiver) = bounded::<Payload>(POOL.current_num_threads() * 2);
 
         let sent: Arc<AtomicUsize> = Default::default();
         let total: Arc<AtomicUsize> = Default::default();

py-polars/Cargo.toml

@@ -164,8 +164,8 @@ name = "polars"
 crate-type = ["cdylib"]
 
 [profile.release]
-codegen-units = 1
-lto = "fat"
+#codegen-units = 1
+#lto = "fat"
 
 # This is ignored here; would be set in .cargo/config.toml.
 # Should not be used when packaging