0.2.8: background asynchronous batches

Author: hackermondev

Cargo.lock

@@ -30,16 +30,23 @@ cargo add batched
 Or add this to your `Cargo.toml`:
 ```toml
 [dependencies]
-batched = "0.2.7"
+batched = "0.2.8"
 ```
 
+### Nightly Rust
+Due to the use of advanced features, `batched` requires a nightly Rust compiler. 
+
+
 ## #[batched]
-- **limit**: Maximum amount of items that can be grouped and processed in a single batch.
+- **limit**: Maximum amount of items that can be grouped and processed in a single batch. (required)
 - **concurrent**: Maximum amount of concurrent batched tasks running (default: `Infinity`)
-- **window**: Maximum amount of time (in milliseconds) the background thread waits after the first call before processing a batch.
-- **window[x]**: Maximum amount of time (in milliseconds) the background thread waits after the first call before processing a batch, when the buffer size is <= x
+- **asynchronous**: If true, the caller does not wait for the batch to complete, and the return value is `()`. (default: `false`).
+- **window**: Maximum amount of time (in milliseconds) the background thread waits after the first call before processing a batch. (required)
+- **window[x]**: Maximum amount of time (in milliseconds) the background thread waits after the first call before processing a batch, when the buffer size is <= x. (This allows for more granular control of the batching window based on the current load. For example, you might want to use a shorter window when there are fewer items in the buffer to reduce latency, and a longer window when there are more items to maximize batching efficiency.)
+
+
 
-The target function must have a single argument, a vector of items (`Vec<T>`). 
+The target function must have a single input argument, a vector of items (`Vec<T>`). 
 
 The return value of the batched function is propagated (cloned) to all async calls of the batch, unless the batched function returns a `Vec<T>`, in which case the return value for each call is pulled from the iterator in the same order of the input.
 
@@ -48,7 +55,7 @@ If the return value is not an iterator, The target function return type must imp
 
 ## Prerequisites 
 - Built for async environments (tokio), will not work without a tokio async runtime
-- Target function must have async
+- The target function must be an async function
 - Not supported inside structs:
 ```rust
 struct A;
@@ -65,7 +72,7 @@ impl A {
 ### [`tracing_span`]
 This feature automatically adds tracing spans to call functions for batched requests (`x`, `x_multiple`).
 
-## [`tracing_opentelemetry`]
+### [`tracing_opentelemetry`]
 This feature adds support for linking spans from callers to the inner batched call when using OpenTelemetry. Depending on whether your OpenTelemetry client supports it, you should be able to see the linked span to the batched call. 
 
 ## Examples

README.md

@@ -1,7 +1,7 @@
 [package]
 name = "batched"
 description = "rust macro util for batching expensive operations"
-version = "0.2.7"
+version = "0.2.8"
 edition = "2024"
 license = "MIT"
 readme = "../README.md"
@@ -15,7 +15,7 @@ tracing_opentelemetry = ["opentelemetry", "tracing-opentelemetry"]
 
 [dependencies]
 anyhow = "1.0.98"
-batched_derive = { version = "0.2.7", path = "../batched_derive" }
+batched_derive = { version = "0.2.8", path = "../batched_derive" }
 opentelemetry = { version = "0.30.0", optional = true }
 tracing = "0.1.41"
 tracing-opentelemetry = { version = "0.31.0", optional = true }

batched/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "batched_derive"
 description = "rust macro util for batching expensive operations"
-version = "0.2.7"
+version = "0.2.8"
 edition = "2024"
 license = "MIT"
 readme = "../README.md"

batched_derive/Cargo.toml

@@ -33,9 +33,22 @@ fn build_identifiers(call_function: &Function) -> Identifiers {
     }
 }
 
-pub fn build_code(call_function: Function, options: Attributes) -> TokenStream {
-    let identifiers = build_identifiers(&call_function);
+pub fn build_code(function: Function, options: Attributes) -> TokenStream {
+    let identifiers = build_identifiers(&function);
+    let executor = build_executor(&identifiers, &function, &options);
+    let public_interface = build_public_interface(&identifiers, &function, &options);
 
+    quote! {
+        #executor
+        #public_interface
+    }
+}
+
+fn build_public_interface(
+    identifiers: &Identifiers,
+    call_function: &Function,
+    options: &Attributes,
+) -> TokenStream {
     let macros = &call_function.macros;
     let visibility = &call_function.visibility;
     let arg = &call_function.batched_arg;
@@ -45,8 +58,9 @@ pub fn build_code(call_function: Function, options: Attributes) -> TokenStream {
     let returned = &call_function.returned.tokens;
 
     let (is_result, is_vec) = function_flags(&call_function);
+    let asynchronous = options.asynchronous;
 
-    let public_interface_returned_type = match &call_function.returned.result_type {
+    let return_type = match &call_function.returned.result_type {
         FunctionResultType::Raw(token) => token.clone(),
         FunctionResultType::VectorRaw(token) => token.clone(),
         FunctionResultType::Result(output, error, _) => {
@@ -57,7 +71,7 @@ pub fn build_code(call_function: Function, options: Attributes) -> TokenStream {
             }
         }
     };
-    let public_interface_return_result = if is_result {
+    let return_result = if is_result {
         if is_vec {
             quote! {
                 let mut result = result?;
@@ -81,7 +95,7 @@ pub fn build_code(call_function: Function, options: Attributes) -> TokenStream {
         }
     };
 
-    let public_interface_multiple_returned_type = match &call_function.returned.result_type {
+    let return_type_multiple = match &call_function.returned.result_type {
         FunctionResultType::Raw(token) => token.clone(),
         FunctionResultType::VectorRaw(token) => quote! { Vec<#token> },
         FunctionResultType::Result(output, error, _) => {
@@ -92,7 +106,7 @@ pub fn build_code(call_function: Function, options: Attributes) -> TokenStream {
             }
         }
     };
-    let public_interface_multiple_return_result = if is_result {
+    let return_result_multiple = if is_result {
         quote! {
             let result = result?;
             Ok(result)
@@ -109,58 +123,82 @@ pub fn build_code(call_function: Function, options: Attributes) -> TokenStream {
             }),
     };
 
-    let executor = build_executor(&identifiers, &call_function, &options);
-    let executor_producer_channel = identifiers.executor_producer_channel;
-    let executor_background_fn = identifiers.executor_background_fn;
-    let inner_batched = identifiers.inner_batched;
-    let public_interface = identifiers.public_interface;
-    let public_interface_multiple = identifiers.public_interface_multiple;
+    let executor_producer_channel = &identifiers.executor_producer_channel;
+    let executor_background_fn = &identifiers.executor_background_fn;
+    let inner_batched = &identifiers.inner_batched;
+    let public_interface = &identifiers.public_interface;
+    let public_interface_multiple = &identifiers.public_interface_multiple;
 
     #[cfg(feature = "tracing_span")]
     let tracing_span = quote! { #[tracing::instrument(skip_all)] };
     #[cfg(not(feature = "tracing_span"))]
     let tracing_span = quote! {};
 
-    quote! {
-        #executor
-
+    let inner_batched = quote! {
         #(#macros)*
         async fn #inner_batched(#arg) -> #returned {
             let result = async { #inner_body };
             let result = result.await;
             #cast_result_error
             result
         }
+    };
 
-        #tracing_span
-        #visibility async fn #public_interface(#arg_name: #arg_type) -> #public_interface_returned_type {
-            let mut result = #public_interface_multiple(vec![#arg_name]).await;
-            #public_interface_return_result
+    if asynchronous {
+        quote! {
+            #inner_batched
+
+            #tracing_span
+            #visibility async fn #public_interface(#arg_name: #arg_type) {
+                #public_interface_multiple(vec![#arg_name]).await;
+            }
+
+            #tracing_span
+            #visibility async fn #public_interface_multiple(#arg_name: Vec<#arg_type>) {
+                let channel = &#executor_producer_channel;
+                let channel = channel.get_or_init(async || { #executor_background_fn().await }).await;
+
+                let span = ::batched::tracing::Span::current();
+                channel.send((#arg_name, span, None)).await
+                    .expect("batched function panicked (send)");
+            }
         }
+    } else {
+        quote! {
+            #inner_batched
 
-        #tracing_span
-        #visibility async fn #public_interface_multiple(#arg_name: Vec<#arg_type>) -> #public_interface_multiple_returned_type {
-            let channel = &#executor_producer_channel;
-            let channel = channel.get_or_init(async || { #executor_background_fn().await }).await;
+            #tracing_span
+            #visibility async fn #public_interface(#arg_name: #arg_type) -> #return_type {
+                let mut result = #public_interface_multiple(vec![#arg_name]).await;
+                #return_result
+            }
+
+            #tracing_span
+            #visibility async fn #public_interface_multiple(#arg_name: Vec<#arg_type>) -> #return_type_multiple {
+                let channel = &#executor_producer_channel;
+                let channel = channel.get_or_init(async || { #executor_background_fn().await }).await;
 
-            let (response_channel_sender, mut response_channel_recv) = ::tokio::sync::mpsc::channel(1);
-            let span = ::batched::tracing::Span::current();
-            channel.send((#arg_name, span, response_channel_sender)).await
-                .expect("batched function panicked (send)");
+                let (response_channel_sender, mut response_channel_recv) = ::tokio::sync::mpsc::channel(1);
+                let span = ::batched::tracing::Span::current();
+                channel.send((#arg_name, span, Some(response_channel_sender))).await
+                    .expect("batched function panicked (send)");
 
-            let result = response_channel_recv.recv().await
-                .expect("batched function panicked (recv)");
-            #public_interface_multiple_return_result
+                let result = response_channel_recv.recv().await
+                    .expect("batched function panicked (recv)");
+                #return_result_multiple
+            }
         }
     }
+    
 }
 
-const SEMAPHORE_MAX_PERMITS: usize = 2305843009213693951;
 fn build_executor(
     identifiers: &Identifiers,
     call_function: &Function,
     options: &Attributes,
 ) -> TokenStream {
+    const SEMAPHORE_MAX_PERMITS: usize = 2305843009213693951;
+
     let capacity = options.limit;
     let concurrent_limit = options.concurrent_limit.unwrap_or(SEMAPHORE_MAX_PERMITS);
     let default_window = options.default_window;
@@ -208,7 +246,13 @@ fn build_executor(
             let result = result.clone();
         }
     };
-    let channel_type = quote! { (Vec<#arg_type>, ::batched::tracing::Span, ::tokio::sync::mpsc::Sender<#returned_type_plural>) };
+    let handle_result = if options.asynchronous { 
+        quote! {}
+    } else {
+        handle_result
+    };
+
+    let channel_type = quote! { (Vec<#arg_type>, ::batched::tracing::Span, Option<::tokio::sync::mpsc::Sender<#returned_type_plural>>) };
 
     let inner_batched = &identifiers.inner_batched;
     let batched_span_name = inner_batched.to_string();
@@ -230,8 +274,8 @@ fn build_executor(
                 let semaphore = ::std::sync::Arc::new(::tokio::sync::Semaphore::new(#concurrent_limit));
 
                 loop {
-                    let mut data_buffer = Vec::with_capacity(capacity);
-                    let mut return_channels: Vec<(::tokio::sync::mpsc::Sender<#returned_type_plural>, usize)> = vec![];
+                    let mut data_buffer = Vec::new();
+                    let mut return_channels: Vec<(Option<::tokio::sync::mpsc::Sender<#returned_type_plural>>, usize)> = vec![];
                     let mut waiting_spans: Vec<::batched::tracing::Span> = vec![];
 
                     let window_start = ::std::time::Instant::now();
@@ -294,7 +338,9 @@ fn build_executor(
                         let mut result = ::batched::tracing::Instrument::instrument(future, batched_span).await;
                         for (channel, count) in channels {
                             #handle_result
-                            let _ = channel.try_send(result);
+                            if let Some(channel) = channel {
+                                let _ = channel.try_send(result);
+                            }
                         }
                     });
                 }

batched_derive/src/builder.rs

@@ -31,7 +31,7 @@ pub struct FunctionResult {
 pub enum FunctionResultType {
     Raw(TokenStream),
     VectorRaw(TokenStream),
-    Result(Box<FunctionResult>, TokenStream, Option<TokenStream>),
+    Result(Box<FunctionResult>, TokenStream, Option<TokenStream>)
 }
 
 fn inner_shared_error(_type: &Type) -> Option<TokenStream> {
@@ -115,7 +115,7 @@ impl Function {
         }
         let returned = match function.sig.output {
             ReturnType::Default => FunctionResult {
-                tokens: function.sig.output.into_token_stream(),
+                tokens: syn::parse_str("()").unwrap(),
                 result_type: FunctionResultType::Raw(syn::parse_str("()").unwrap()),
             },
             ReturnType::Type(_, _type) => parsed_returned(&_type),
@@ -181,6 +181,7 @@ impl Function {
 pub struct Attributes {
     pub limit: usize,
     pub concurrent_limit: Option<usize>,
+    pub asynchronous: bool,
     pub default_window: u64,
     pub windows: BTreeMap<u64, u64>,
 }
@@ -189,12 +190,14 @@ impl Attributes {
     pub fn parse(tokens: TokenStream) -> Self {
         let mut limit: Option<usize> = None;
         let mut concurrent_limit: Option<usize> = None;
+        let mut asynchronous = false;
         let mut default_window: Option<u64> = None;
         let mut windows = BTreeMap::new();
 
         static WINDOW_ATTR: &str = "window";
         static LIMIT_ATTR: &str = "limit";
         static CONCURRENT_LIMIT_ATTR: &str = "concurrent";
+        static ASYNCHRONOUS_ATTR: &str = "asynchronous";
 
         let parser = Punctuated::<Meta, Token![,]>::parse_separated_nonempty;
         let attributes = parser.parse(tokens.into()).unwrap();
@@ -216,6 +219,8 @@ impl Attributes {
                 };
 
                 concurrent_limit = expr_to_u64(value).map(|u| u as usize);
+            } else if path.is_ident(ASYNCHRONOUS_ATTR) {
+                asynchronous = true;
             } else if path.is_ident(WINDOW_ATTR) {
                 let value = match attr {
                     Meta::NameValue(attr) => &attr.value,
@@ -253,8 +258,9 @@ impl Attributes {
         Self {
             limit,
             concurrent_limit,
+            asynchronous,
             default_window,
             windows,
         }
     }
-}
+}