feat: function v2 (#212)

* feat: function v2

* expose evaluate

* remove async trait, add caching

* fix naming

* add reverse compatibility

* add http module

* update impl

* resolve issues with reqwest

* fix

* fix

* fix ssl

* remove unused deps

* fix: tests

* fix

bindings/python/src/decision.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use anyhow::{anyhow, Context};
 use pyo3::types::PyDict;
 use pyo3::{pyclass, pymethods, PyAny, PyObject, PyResult, Python, ToPyObject};
+use pyo3_asyncio::tokio;
 use pythonize::depythonize;
 
 use zen_engine::{Decision, EvaluationOptions};
@@ -10,6 +11,7 @@ use zen_engine::{Decision, EvaluationOptions};
 use crate::custom_node::PyCustomNode;
 use crate::engine::PyZenEvaluateOptions;
 use crate::loader::PyDecisionLoader;
+use crate::mt::{spawn_worker, spawn_worker_blocking};
 use crate::value::PyValue;
 
 #[pyclass]
@@ -33,15 +35,19 @@ impl PyZenDecision {
         };
 
         let decision = self.0.clone();
-        let result = futures::executor::block_on(decision.evaluate_with_opts(
-            &context,
-            EvaluationOptions {
-                max_depth: options.max_depth,
-                trace: options.trace,
-            },
-        ))
-        .map_err(|e| {
-            anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
+        let result = spawn_worker_blocking(move || async move {
+            decision
+                .evaluate_with_opts(
+                    &context,
+                    EvaluationOptions {
+                        max_depth: options.max_depth,
+                        trace: options.trace,
+                    },
+                )
+                .await
+                .map_err(|e| {
+                    anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
+                })
         })?;
 
         let value = serde_json::to_value(&result).context("Fail")?;
@@ -62,22 +68,27 @@ impl PyZenDecision {
         };
 
         let decision = self.0.clone();
-        pyo3_asyncio::tokio::future_into_py(py, async move {
-            let result = futures::executor::block_on(decision.evaluate_with_opts(
-                &context,
-                EvaluationOptions {
-                    max_depth: options.max_depth,
-                    trace: options.trace,
-                },
-            ))
-            .map_err(|e| {
-                anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
-            })?;
 
-            let value = serde_json::to_value(result).context("Failed to serialize result")?;
+        tokio::future_into_py(
+            py,
+            spawn_worker(move || async move {
+                let result = decision
+                    .evaluate_with_opts(
+                        &context,
+                        EvaluationOptions {
+                            max_depth: options.max_depth,
+                            trace: options.trace,
+                        },
+                    )
+                    .await
+                    .map_err(|e| {
+                        anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
+                    })?;
 
-            Python::with_gil(|py| Ok(PyValue(value).to_object(py)))
-        })
+                let value = serde_json::to_value(result).context("Failed to serialize result")?;
+                Python::with_gil(|py| Ok(PyValue(value).to_object(py)))
+            }),
+        )
     }
 
     pub fn validate(&self) -> PyResult<()> {

bindings/python/src/engine.rs

@@ -1,16 +1,20 @@
-use crate::custom_node::PyCustomNode;
-use crate::decision::PyZenDecision;
-use crate::loader::PyDecisionLoader;
-use crate::value::PyValue;
+use std::sync::Arc;
+
 use anyhow::{anyhow, Context};
 use pyo3::types::PyDict;
 use pyo3::{pyclass, pymethods, PyAny, PyObject, PyResult, Python, ToPyObject};
 use pythonize::depythonize;
 use serde::{Deserialize, Serialize};
-use std::sync::Arc;
+
 use zen_engine::model::DecisionContent;
 use zen_engine::{DecisionEngine, EvaluationOptions};
 
+use crate::custom_node::PyCustomNode;
+use crate::decision::PyZenDecision;
+use crate::loader::PyDecisionLoader;
+use crate::mt::{spawn_worker, spawn_worker_blocking};
+use crate::value::PyValue;
+
 #[pyclass]
 #[pyo3(name = "ZenEngine")]
 pub struct PyZenEngine {
@@ -86,16 +90,20 @@ impl PyZenEngine {
         };
 
         let graph = self.graph.clone();
-        let result = futures::executor::block_on(graph.evaluate_with_opts(
-            key,
-            &context,
-            EvaluationOptions {
-                max_depth: options.max_depth,
-                trace: options.trace,
-            },
-        ))
-        .map_err(|e| {
-            anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
+        let result = spawn_worker_blocking(move || async move {
+            graph
+                .evaluate_with_opts(
+                    key,
+                    &context,
+                    EvaluationOptions {
+                        max_depth: options.max_depth,
+                        trace: options.trace,
+                    },
+                )
+                .await
+                .map_err(|e| {
+                    anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
+                })
         })?;
 
         let value = serde_json::to_value(&result).context("Failed to serialize result")?;
@@ -117,23 +125,28 @@ impl PyZenEngine {
         };
 
         let graph = self.graph.clone();
-        pyo3_asyncio::tokio::future_into_py(py, async move {
-            let result = futures::executor::block_on(graph.evaluate_with_opts(
-                key,
-                &context,
-                EvaluationOptions {
-                    max_depth: options.max_depth,
-                    trace: options.trace,
-                },
-            ))
-            .map_err(|e| {
-                anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
-            })?;
+        pyo3_asyncio::tokio::future_into_py(
+            py,
+            spawn_worker(move || async move {
+                let result = graph
+                    .evaluate_with_opts(
+                        key,
+                        &context,
+                        EvaluationOptions {
+                            max_depth: options.max_depth,
+                            trace: options.trace,
+                        },
+                    )
+                    .await
+                    .map_err(|e| {
+                        anyhow!(serde_json::to_string(e.as_ref()).unwrap_or_else(|_| e.to_string()))
+                    })?;
 
-            let value = serde_json::to_value(result).context("Failed to serialize result")?;
+                let value = serde_json::to_value(result).context("Failed to serialize result")?;
 
-            Python::with_gil(|py| Ok(PyValue(value).to_object(py)))
-        })
+                Python::with_gil(|py| Ok(PyValue(value).to_object(py)))
+            }),
+        )
     }
 
     pub fn create_decision(&self, content: String) -> PyResult<PyZenDecision> {
@@ -145,8 +158,13 @@ impl PyZenEngine {
     }
 
     pub fn get_decision(&self, key: String) -> PyResult<PyZenDecision> {
-        let decision = futures::executor::block_on(self.graph.get_decision(&key))
-            .context("Failed to find decision with given key")?;
+        let graph = self.graph.clone();
+        let decision = spawn_worker_blocking(move || async move {
+            graph
+                .get_decision(&key)
+                .await
+                .context("Failed to find decision with given key")
+        })?;
 
         Ok(PyZenDecision::from(decision))
     }

bindings/python/src/lib.rs

@@ -9,6 +9,7 @@ mod decision;
 mod engine;
 mod expression;
 mod loader;
+mod mt;
 mod types;
 mod value;
 

bindings/python/src/loader.rs

@@ -1,8 +1,10 @@
-use anyhow::anyhow;
-use async_trait::async_trait;
-use pyo3::{PyObject, Python};
+use std::future::Future;
 use std::sync::Arc;
-use zen_engine::loader::{DecisionLoader, LoaderError, LoaderResult};
+
+use anyhow::anyhow;
+use pyo3::{PyObject, Python};
+
+use zen_engine::loader::{DecisionLoader, LoaderError, LoaderResponse};
 use zen_engine::model::DecisionContent;
 
 #[derive(Default)]
@@ -35,15 +37,16 @@ impl PyDecisionLoader {
     }
 }
 
-#[async_trait]
 impl DecisionLoader for PyDecisionLoader {
-    async fn load(&self, key: &str) -> LoaderResult<Arc<DecisionContent>> {
-        self.load_element(key).map_err(|e| {
-            LoaderError::Internal {
-                source: e,
-                key: key.to_string(),
-            }
-            .into()
-        })
+    fn load<'a>(&'a self, key: &'a str) -> impl Future<Output = LoaderResponse> + 'a {
+        async move {
+            self.load_element(key).map_err(|e| {
+                LoaderError::Internal {
+                    source: e,
+                    key: key.to_string(),
+                }
+                .into()
+            })
+        }
     }
 }

bindings/python/src/mt.rs

@@ -0,0 +1,48 @@
+use std::future::Future;
+use std::sync::OnceLock;
+use std::thread::available_parallelism;
+use tokio::runtime::Handle;
+use tokio_util::task::LocalPoolHandle;
+
+fn parallelism() -> usize {
+    available_parallelism().map(Into::into).unwrap_or(1)
+}
+
+pub(crate) fn worker_pool() -> LocalPoolHandle {
+    static LOCAL_POOL: OnceLock<LocalPoolHandle> = OnceLock::new();
+    LOCAL_POOL
+        .get_or_init(|| LocalPoolHandle::new(parallelism()))
+        .clone()
+}
+
+pub(crate) fn spawn_worker<F, Fut>(create_task: F) -> impl Future<Output = Fut::Output>
+where
+    F: FnOnce() -> Fut,
+    F: Send + 'static,
+    Fut: Future + 'static,
+    Fut::Output: Send + 'static,
+{
+    async move {
+        worker_pool()
+            .spawn_pinned(create_task)
+            .await
+            .expect("Thread panicked")
+    }
+}
+
+pub(crate) fn spawn_worker_blocking<F, Fut>(create_task: F) -> Fut::Output
+where
+    F: FnOnce() -> Fut,
+    F: Send + 'static,
+    Fut: Future + 'static,
+    Fut::Output: Send + 'static,
+{
+    tokio::task::block_in_place(move || {
+        Handle::current().block_on(async move {
+            worker_pool()
+                .spawn_pinned(create_task)
+                .await
+                .expect("Thread panicked")
+        })
+    })
+}