Made detach function Send

Author: ankane

ext/polars/src/dataframe/general.rs

@@ -337,6 +337,7 @@ impl RbDataFrame {
         lambda: Value,
         maintain_order: bool,
     ) -> RbResult<Self> {
+        let lambda = Opaque::from(lambda);
         rb.enter_polars_df(|| {
             let df = self_.df.read().clone(); // Clone so we can't deadlock on re-entrance from lambda.
             let gb = if maintain_order {
@@ -345,7 +346,6 @@ impl RbDataFrame {
                 df.group_by(by.iter().map(|x| &**x))
             }?;
 
-            let lambda = Opaque::from(lambda);
             let function = move |df: DataFrame| {
                 Ruby::attach(|rb| {
                     let lambda = rb.get_inner(lambda);

ext/polars/src/expr/array.rs

@@ -119,7 +119,7 @@ impl RbExpr {
     }
 
     pub fn arr_to_struct(&self, name_gen: Option<Value>) -> Self {
-        let name_gen = name_gen.map(|o| PlanCallback::new_ruby(RubyObject(o)));
+        let name_gen = name_gen.map(|o| PlanCallback::new_ruby(RubyObject::from(o)));
         self.inner.clone().arr().to_struct(name_gen).into()
     }
 

ext/polars/src/expr/name.rs

@@ -14,7 +14,7 @@ impl RbExpr {
         self.inner
             .clone()
             .name()
-            .map(PlanCallback::new_ruby(RubyObject(lambda)))
+            .map(PlanCallback::new_ruby(RubyObject::from(lambda)))
             .into()
     }
 
@@ -46,7 +46,7 @@ impl RbExpr {
         self.inner
             .clone()
             .name()
-            .map_fields(PlanCallback::new_ruby(RubyObject(name_mapper)))
+            .map_fields(PlanCallback::new_ruby(RubyObject::from(name_mapper)))
             .into()
     }
 

ext/polars/src/functions/lazy.rs

@@ -214,7 +214,7 @@ pub fn cum_fold(
     include_init: bool,
 ) -> RbResult<RbExpr> {
     let exprs = exprs.to_exprs()?;
-    let func = PlanCallback::new_ruby(RubyObject(lambda));
+    let func = PlanCallback::new_ruby(RubyObject::from(lambda));
     Ok(dsl::cum_fold_exprs(
         acc.inner.clone(),
         func,
@@ -233,7 +233,7 @@ pub fn cum_reduce(
     return_dtype: Option<&RbDataTypeExpr>,
 ) -> RbResult<RbExpr> {
     let exprs = exprs.to_exprs()?;
-    let func = PlanCallback::new_ruby(RubyObject(lambda));
+    let func = PlanCallback::new_ruby(RubyObject::from(lambda));
     Ok(dsl::cum_reduce_exprs(
         func,
         exprs,
@@ -360,7 +360,7 @@ pub fn fold(
     return_dtype: Option<&RbDataTypeExpr>,
 ) -> RbResult<RbExpr> {
     let exprs = exprs.to_exprs()?;
-    let func = PlanCallback::new_ruby(RubyObject(lambda));
+    let func = PlanCallback::new_ruby(RubyObject::from(lambda));
     Ok(dsl::fold_exprs(
         acc.inner.clone(),
         func,
@@ -444,7 +444,7 @@ pub fn reduce(
     return_dtype: Option<&RbDataTypeExpr>,
 ) -> RbResult<RbExpr> {
     let exprs = exprs.to_exprs()?;
-    let func = PlanCallback::new_ruby(RubyObject(lambda));
+    let func = PlanCallback::new_ruby(RubyObject::from(lambda));
     Ok(dsl::reduce_exprs(
         func,
         exprs,

ext/polars/src/lazyframe/general.rs

@@ -403,12 +403,11 @@ impl RbLazyFrame {
         rb.enter_polars(|| {
             let ldf = self_.ldf.read().clone();
 
-            let collect_batches = ldf
-                .clone()
-                .collect_batches(engine.0, maintain_order, chunk_size, lazy)
-                .map_err(RbPolarsErr::from)?;
+            let collect_batches =
+                ldf.clone()
+                    .collect_batches(engine.0, maintain_order, chunk_size, lazy)?;
 
-            RbResult::Ok(RbCollectBatches {
+            PolarsResult::Ok(RbCollectBatches {
                 inner: Arc::new(Mutex::new(collect_batches)),
                 _ldf: ldf,
             })

ext/polars/src/lazygroupby.rs

@@ -51,7 +51,7 @@ impl RbLazyGroupBy {
                 .map_err(RbPolarsErr::from)?,
         };
 
-        let function = RubyObject(lambda);
+        let function = RubyObject::from(lambda);
 
         Ok(lgb.apply(PlanCallback::new_ruby(function), schema).into())
     }

ext/polars/src/ruby/gvl.rs

@@ -12,7 +12,7 @@ pub trait GvlExt {
 
     fn detach<T, F>(&self, func: F) -> T
     where
-        F: FnOnce() -> T;
+        F: FnOnce() -> T + Send;
 }
 
 unsafe extern "C" {
@@ -51,7 +51,7 @@ impl GvlExt for Ruby {
 
     fn detach<T, F>(&self, func: F) -> T
     where
-        F: FnOnce() -> T,
+        F: Send + FnOnce() -> T,
     {
         if std::env::var("POLARS_GVL").is_ok() {
             func()

ext/polars/src/ruby/lazy.rs

@@ -30,7 +30,7 @@ impl RubyUdfLazyFrameExt for LazyFrame {
         };
 
         // handle non-Ruby threads
-        start_background_ruby_thread(&Ruby::get_with(function.0));
+        start_background_ruby_thread(&Ruby::get().unwrap());
         let udf = ArcValue::new(function.0);
         let f = move |df| {
             if is_non_ruby_thread() {

ext/polars/src/ruby/plan_callback.rs

@@ -186,7 +186,7 @@ impl<Args: PlanCallbackArgs + Send + 'static, Out: PlanCallbackOut + Send + 'sta
         };
 
         // handle non-Ruby threads
-        start_background_ruby_thread(&Ruby::get_with(rbfn.0));
+        start_background_ruby_thread(&Ruby::get().unwrap());
         let udf = ArcValue::new(rbfn.0);
         let f = move |args: Args| {
             if is_non_ruby_thread() {

ext/polars/src/ruby/ruby_function.rs

@@ -1,12 +1,11 @@
-use magnus::Value;
+use magnus::{Value, value::Opaque};
 
-#[derive(Debug)]
-pub struct RubyObject(pub Value);
+pub struct RubyObject(pub Opaque<Value>);
 
 pub type RubyFunction = RubyObject;
 
 impl From<Value> for RubyObject {
     fn from(value: Value) -> Self {
-        Self(value)
+        Self(value.into())
     }
 }