diff --git a/example/src/main.rs b/example/src/main.rs index d70505e..a6c4804 100644 --- a/example/src/main.rs +++ b/example/src/main.rs @@ -28,10 +28,8 @@ use oxydsp_flowgraph::io::Out; use oxydsp_flowgraph::io::PopIterable; use oxydsp_flowgraph::sync_block; -use crate::printer_synchronous_block::PrinterView; - #[derive(BlockIO)] -#[sync_block] +#[sync_block(tagged)] pub struct Printer { #[input] @@ -40,6 +38,25 @@ pub struct Printer n: usize, } +// impl Block for Printer +// where +// T: Display, +// { +// fn work(&mut self) -> oxydsp_flowgraph::block::BlockResult +// { +// for x in self.input.pop_iter() +// { +// if self.n.is_multiple_of(2usize.pow(20)) +// { +// println!("{}", x.0); +// self.n = 0; +// } +// self.n += 1; +// } +// BlockResult::Ok +// } +// } + impl<'view, T: 'static + Display> SyncBlock<'view> for Printer { fn sync_work(state: Self::StateView, input: Self::Input) -> Option diff --git a/oxydsp-dsp/src/blocks/math/basic.rs b/oxydsp-dsp/src/blocks/math/basic.rs index 772af36..ede9223 100644 --- a/oxydsp-dsp/src/blocks/math/basic.rs +++ b/oxydsp-dsp/src/blocks/math/basic.rs @@ -2,12 +2,14 @@ use std::ops::{Add, Mul}; use oxydsp_flowgraph::{ BlockIO, - block::{Block, BlockResult}, + block::{Block, BlockResult, SyncBlock}, io::{In, Out, PopIterable}, + sync_block, tag::TagMergable, }; #[derive(BlockIO)] +#[sync_block] pub struct Adder where Ia: Add + 'static, @@ -44,23 +46,35 @@ where } } -impl Block for Adder +impl<'view, Ia, Ib, O> SyncBlock<'view> for Adder where Ia: Add + 'static, Ib: 'static, O: 'static, { - fn work(&mut self) -> BlockResult + fn sync_work(_state: Self::StateView, input: Self::Input) -> Option { - self.output.push_iter( - (&mut self.input_a, &mut self.input_b) - .pop_iter() - .map(|(a, b)| (a.0 + b.0, a.1.merge(&b.1))), - ); - BlockResult::Ok + Some(input.0 + input.1) } } +// impl Block for Adder +// where +// Ia: Add + 'static, +// Ib: 'static, +// O: 'static, +// { +// fn work(&mut self) -> BlockResult +// { +// self.output.push_iter( +// (&mut self.input_a, &mut self.input_b) +// .pop_iter() +// .map(|(a, b)| (a.0 + b.0, a.1.merge(&b.1))), +// ); +// BlockResult::Ok +// } +// } + #[derive(BlockIO)] pub struct Multiplier where @@ -109,7 +123,7 @@ where self.output.push_iter( (&mut self.input_a, &mut self.input_b) .pop_iter() - .map(|(a, b)| (a.0 * b.0, a.1.merge(&b.1))), + .map(|(a, b)| (a.0 * b.0, a.1.merge(&b.1)).into()), ); BlockResult::Ok } diff --git a/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/lib.rs b/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/lib.rs index b3ff381..bd2f2b4 100644 --- a/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/lib.rs +++ b/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/lib.rs @@ -1,25 +1,40 @@ use proc_macro::TokenStream; +use zyn::FromInput; use zyn::ToTokens; use zyn::ext::AttrExt; use zyn::ext::FieldsExt; +use zyn::ext::ItemExt; use zyn::format_ident; use zyn::ident; use zyn::parse_input; +use zyn::syn::Attribute; use zyn::syn::GenericParam; use zyn::syn::Index; use zyn::syn::Lit; use zyn::syn::TypeGenerics; use zyn::syn::parse_quote; +use zyn::syn::punctuated::Punctuated; use zyn::syn::spanned::Spanned; mod sync; -#[zyn::attribute] -pub fn sync_block(#[zyn(input)] item: zyn::syn::ItemStruct) -> zyn::TokenStream +#[derive(zyn::Attribute, Clone, Copy)] +#[zyn("sync_block", about = "Convenient derivation for synchronous blocks")] +struct SyncBlockConfig { + #[zyn(default)] + tagged: bool, +} + +#[zyn::attribute] +pub fn sync_block(#[zyn(input)] item: zyn::syn::ItemStruct, args: zyn::Args) -> zyn::TokenStream +{ + let config = SyncBlockConfig { + tagged: args.iter().any(|x| x.as_flag() == "tagged"), + }; use sync::SyncBlockImpl; zyn::zyn!( - @sync_block_impl(item = item) + @sync_block_impl(item = item, config = config) ) } @@ -393,7 +408,7 @@ pub fn impl_iterator_for_pop_iter_tuple(input: TokenStream) -> TokenStream type Item = ( @for (i in 0..count) { - ({{ generics[i] }}, Option), + Tagged<{{ generics[i] }}>, } ); fn next(&mut self) -> Option @@ -405,7 +420,7 @@ pub fn impl_iterator_for_pop_iter_tuple(input: TokenStream) -> TokenStream Some(( @for (i in 0..count) { - self.reader.{{ Index::from(i) }}.pop_tagged().unwrap(), + self.reader.{{ Index::from(i) }}.pop().unwrap(), } )) } @@ -419,3 +434,95 @@ pub fn impl_iterator_for_pop_iter_tuple(input: TokenStream) -> TokenStream .to_token_stream() .into() } +// #[proc_macro] +// pub fn generate_push_iterable_tuple_impl(input: TokenStream) -> TokenStream +// { +// let count = parse_input!(input as Lit); +// let count: usize = match count +// { +// Lit::Int(lit_int) => lit_int.base10_parse::().unwrap(), +// _ => +// { +// return zyn::syn::Error::new(count.span(), "Must be an integer") +// .to_compile_error() +// .into(); +// } +// }; +// let generics = [ +// format_ident!("A"), +// format_ident!("B"), +// format_ident!("C"), +// format_ident!("D"), +// format_ident!("E"), +// format_ident!("F"), +// format_ident!("G"), +// format_ident!("H"), +// format_ident!("I"), +// format_ident!("J"), +// format_ident!("K"), +// format_ident!("L"), +// format_ident!("M"), +// format_ident!("N"), +// format_ident!("O"), +// format_ident!("P"), +// format_ident!("Q"), +// format_ident!("R"), +// format_ident!("S"), +// format_ident!("T"), +// format_ident!("U"), +// format_ident!("V"), +// format_ident!("W"), +// format_ident!("X"), +// format_ident!("Y"), +// format_ident!("Z"), +// ]; +// +// let iterator_item = zyn::zyn!( +// ( +// @for (i in 0..count) +// { +// ({{ generics[i] }}, Option), +// } +// ) +// ) +// .to_token_stream(); +// +// zyn::zyn!( +// impl<'a, +// @for (i in 0..count) +// { +// {{ generics[i] }}: 'static, +// } +// > PushIterable<'a, {{ iterator_item }}> for ( +// @for (i in 0..count) +// { +// &mut Out<{{ generics[i] }}>, +// } +// ) +// { +// fn push_iter>( +// &'a mut self, +// mut iter: I, +// ) -> bool +// { +// @for (i in 0..count) +// { +// let {{ i | ident:"writer_{}" }} = self.{{ Index::from(i) }}.write(); +// } +// let len = [ +// @for (i in 0..count) +// { +// {{ i | ident:"reader_{}" }}.len(), +// } +// ].into_iter().min().unwrap(); +// +// for _ in 0..len +// { +// if let Some() +// } +// } +// } +// ) +// .to_token_stream() +// .into() +// } diff --git a/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/sync.rs b/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/sync.rs index 0a44e93..21e0a7d 100644 --- a/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/sync.rs +++ b/oxydsp-flowgraph/oxydsp-flowgraph-macros/src/sync.rs @@ -1,4 +1,5 @@ use zyn::Fields; +use zyn::FromInput; use zyn::ToTokens; use zyn::ast::at; use zyn::ext::AttrExt; @@ -10,10 +11,12 @@ use zyn::syn::Field; use zyn::syn::GenericParam; use zyn::syn::parse_quote; +use crate::SyncBlockConfig; + // Sync block #[zyn::element] -pub fn sync_block_impl(item: zyn::syn::ItemStruct) -> zyn::TokenStream +pub fn sync_block_impl(item: zyn::syn::ItemStruct, config: SyncBlockConfig) -> zyn::TokenStream { zyn::zyn!( {{ item }} @@ -24,12 +27,13 @@ pub fn sync_block_impl(item: zyn::syn::ItemStruct) -> zyn::TokenStream @sync_block_view_struct(item = item.clone()) } - @sync_block_syncio_impl(item = item.clone()) + @sync_block_syncio_impl(item = item.clone(), config = *config) + @sync_block_impl_block(item = item.clone()) ) } #[zyn::element] -fn sync_block_syncio_impl(item: zyn::syn::ItemStruct) -> zyn::TokenStream +fn sync_block_syncio_impl(item: zyn::syn::ItemStruct, config: SyncBlockConfig) -> zyn::TokenStream { let view_lifetime: GenericParam = parse_quote!('view); let mut view_generics = item.generics.clone(); @@ -45,13 +49,17 @@ fn sync_block_syncio_impl(item: zyn::syn::ItemStruct) -> zyn::TokenStream { // Path within module type StateView = {{ item.ident | snake | ident: "{}_synchronous_block" }}::{{ item.ident | ident:"{}View" }} {{ view_type_generics }}; - type Input = {{ sync_block_io_types(item.clone(), "input") }}; - type Output = {{ sync_block_io_types(item.clone(), "output") }}; + type Input = {{ sync_block_io_types(item.clone(), "input", config.tagged) }}; + type Output = {{ sync_block_io_types(item.clone(), "output", config.tagged) }}; } ) } -fn sync_block_io_types(item: zyn::syn::ItemStruct, io: &'static str) -> zyn::TokenStream +fn sync_block_io_types( + item: zyn::syn::ItemStruct, + io: &'static str, + tagged: bool, +) -> zyn::TokenStream { let field_types = item .fields @@ -71,7 +79,18 @@ fn sync_block_io_types(item: zyn::syn::ItemStruct, io: &'static str) -> zyn::Tok .arguments .clone() { - zyn::syn::PathArguments::AngleBracketed(args) => args.args.to_token_stream(), + zyn::syn::PathArguments::AngleBracketed(args) => + { + let args = args.args.to_token_stream(); + if tagged + { + quote!(oxydsp_flowgraph::tag::Tagged<#args>).into_token_stream() + } + else + { + args + } + } zyn::syn::PathArguments::None => panic!(), zyn::syn::PathArguments::Parenthesized(_) => panic!(), } @@ -151,9 +170,7 @@ fn sync_block_view_struct(item: zyn::syn::ItemStruct) -> zyn::TokenStream .filter(|tokens| !tokens.is_empty()); zyn::zyn!( - #[derive(Clone, Copy)] pub struct {{ item.ident | ident:"{}View" }} {{ type_generics }} - {{ where_clause }} { @for (field in state_fields.iter()) { @@ -194,7 +211,6 @@ fn sync_block_impl_block(item: zyn::syn::ItemStruct) -> zyn::TokenStream { fn work(&mut self) -> oxydsp_flowgraph::block::BlockResult { - let state = {{ sync_block_make_view_struct(item.clone()) }}; // Get writers from outputs let mut max_len = usize::MAX; @@ -208,6 +224,10 @@ fn sync_block_impl_block(item: zyn::syn::ItemStruct) -> zyn::TokenStream { @sync_block_block_impl_with_inputs(item = item.clone(), input_fields = input_fields.clone(), output_fields = output_fields.clone()) } + @else + { + @sync_block_block_impl_without_inputs(item = item.clone(), output_fields = output_fields.clone()) + } oxydsp_flowgraph::block::BlockResult::Ok } @@ -216,105 +236,40 @@ fn sync_block_impl_block(item: zyn::syn::ItemStruct) -> zyn::TokenStream } #[zyn::element] -fn sync_block_block_impl_with_inputs( +fn sync_block_block_impl_without_inputs( item: zyn::syn::ItemStruct, output_fields: Vec, ) -> zyn::TokenStream { zyn::zyn!( - for _ in 0..max_len { + // Get outputs @if (output_fields.len() == 1) { let {{output_fields[0].ident.clone().unwrap() | ident:"{}_element"}} } @else { + let state = {{ sync_block_make_view_struct(item.clone()) }}; let (@for (out_field in output_fields.iter()) { {{out_field.ident.clone().unwrap() | ident:"{}_element"}}, } ) } - = ::sync_work(state, ()).unwrap(); - } - - // Iterate on inputs - ( - @for (in_field in input_fields.iter()) - { - &mut self.{{ in_field.ident }}, - } - ).pop_iter() - .zip(0..max_len) - .for_each( - // Deconstruct foreach arguments - | - ( - (@for (in_field in input_fields.iter()) - { - ({{in_field.ident.clone().unwrap() | ident:"{}_element"}}, - {{in_field.ident.clone().unwrap() | ident:"{}_tag_opt"}}), - }), - _ // Ignore index - ) - | - { - // Create output tag - let tag = oxydsp_flowgraph::tag::merge_tag_opts([ - @for (in_field in input_fields.iter()) - { - {{in_field.ident.clone().unwrap() | ident:"{}_tag_opt"}}, - }, - ]); - - // Compute output sample - @if (output_fields.is_empty()) - { - let _ - } - @else if (output_fields.len() == 1) - { - let {{output_fields[0].ident.clone().unwrap() | ident:"{}_element"}} - } - @else - { - let (@for (out_field in output_fields.iter()) - { - {{out_field.ident.clone().unwrap() | ident:"{}_element"}}, - } - ) - } - = ::sync_work(state, - @if (input_fields.len() == 1) - { - {{input_fields[0].ident.clone().unwrap() | ident:"{}_element"}}, - } - @else - { - (@for (in_field in input_fields.iter()) - { - {{in_field.ident.clone().unwrap() | ident:"{}_element"}}, - } - ) - } - ).unwrap(); + = ::sync_work(state, ()).unwrap(); // Now the output samples must be sent to their resepective outputs @for (out_field in output_fields.iter()) { {{ out_field.ident.clone().unwrap() | ident: "{}_writer"}}.push( ( - {{ out_field.ident.clone().unwrap() | ident: "{}_element"}}, - tag.clone() + {{ out_field.ident.clone().unwrap() | ident: "{}_element"}}, None ) ); } - // - } - ); ) } @@ -349,13 +304,14 @@ fn sync_block_block_impl_with_inputs( | { // Create output tag - let tag = oxydsp_flowgraph::tag::merge_tag_opts([ + let tag = oxydsp_flowgraph::tag::Tag::merge_tag_opts([ @for (in_field in input_fields.iter()) { {{in_field.ident.clone().unwrap() | ident:"{}_tag_opt"}}, - }, + } ]); + let state = {{ sync_block_make_view_struct(item.clone()) }}; // Compute output sample @if (output_fields.is_empty()) { @@ -373,7 +329,7 @@ fn sync_block_block_impl_with_inputs( } ) } - = ::sync_work(state, + = ::sync_work(state, @if (input_fields.len() == 1) { {{input_fields[0].ident.clone().unwrap() | ident:"{}_element"}}, @@ -424,7 +380,7 @@ fn sync_block_make_view_struct(item: zyn::syn::ItemStruct) -> zyn::TokenStream } zyn::zyn!( - {{ item.ident | ident:"{}View" }} { + {{item.ident | snake | ident:"{}_synchronous_block" }}::{{ item.ident | ident:"{}View" }} { @for (field in state_fields) { {{field.ident}}: &mut self.{{ field.ident }}, diff --git a/oxydsp-flowgraph/src/io.rs b/oxydsp-flowgraph/src/io.rs index 0e4935c..cd2df74 100644 --- a/oxydsp-flowgraph/src/io.rs +++ b/oxydsp-flowgraph/src/io.rs @@ -13,6 +13,7 @@ use crate::stream::StreamReader; use crate::stream::StreamWriter; use crate::stream::{self}; use crate::tag::Tag; +use crate::tag::Tagged; pub struct In { @@ -173,7 +174,7 @@ impl InReader<'_, T> self.data_reader.len() } - pub fn pop_tagged(&self) -> Option<(T, Option)> + pub fn pop(&self) -> Option> { let data = self.data_reader.pop_with_index(); if let Some((data, index)) = data @@ -186,7 +187,7 @@ impl InReader<'_, T> { tag = self.tag_reader.pop(); } - Some((data, tag)) + Some((data, tag).into()) } else { @@ -194,9 +195,9 @@ impl InReader<'_, T> } } - pub fn pop_drop_tag(&self) -> Option + pub fn pop_untag(&self) -> Option { - self.pop_tagged().map(|(data, _)| data) + self.pop().map(|data| data.into_inner()) } } @@ -207,8 +208,9 @@ impl OutWriter<'_, T> self.data_writer.len().min(self.tag_writer.len()) } - pub fn push(&self, (data, tag): (T, Option)) -> Result<(), (T, Option)> + pub fn push(&self, data: Tagged) -> Result<(), (T, Option)> { + let (data, tag) = data.into(); match self.data_writer.push(data) { Ok(_) if tag.is_some() => @@ -221,23 +223,9 @@ impl OutWriter<'_, T> } } - pub fn push_tagged(&self, data: T, tag: Tag) -> Result<(), (T, Tag)> - { - let res = self.data_writer.push(data); - match res - { - Ok(_) => - { - let _ = self.tag_writer.push(tag); - Ok(()) - } - Err(t) => Err((t, tag)), - } - } - pub fn push_no_tag(&self, data: T) -> Result<(), T> { - self.data_writer.push(data) + self.data_writer.push(data.into()) } } @@ -268,6 +256,7 @@ impl<'a, T: 'static> PopIterable<'a> for In } } +generate_pop_iterable_tuple_impl! {1} generate_pop_iterable_tuple_impl! {2} generate_pop_iterable_tuple_impl! {3} generate_pop_iterable_tuple_impl! {4} @@ -279,25 +268,18 @@ generate_pop_iterable_tuple_impl! {9} generate_pop_iterable_tuple_impl! {10} generate_pop_iterable_tuple_impl! {11} generate_pop_iterable_tuple_impl! {12} -generate_pop_iterable_tuple_impl! {13} -generate_pop_iterable_tuple_impl! {14} -generate_pop_iterable_tuple_impl! {15} -generate_pop_iterable_tuple_impl! {16} -generate_pop_iterable_tuple_impl! {17} -generate_pop_iterable_tuple_impl! {18} -generate_pop_iterable_tuple_impl! {19} -generate_pop_iterable_tuple_impl! {20} impl<'a, T> Iterator for PopIter> { - type Item = (T, Option); + type Item = Tagged; fn next(&mut self) -> Option { - self.reader.pop_tagged() + self.reader.pop() } } +impl_iterator_for_pop_iter_tuple! {1} impl_iterator_for_pop_iter_tuple! {2} impl_iterator_for_pop_iter_tuple! {3} impl_iterator_for_pop_iter_tuple! {4} @@ -309,14 +291,6 @@ impl_iterator_for_pop_iter_tuple! {9} impl_iterator_for_pop_iter_tuple! {10} impl_iterator_for_pop_iter_tuple! {11} impl_iterator_for_pop_iter_tuple! {12} -impl_iterator_for_pop_iter_tuple! {13} -impl_iterator_for_pop_iter_tuple! {14} -impl_iterator_for_pop_iter_tuple! {15} -impl_iterator_for_pop_iter_tuple! {16} -impl_iterator_for_pop_iter_tuple! {17} -impl_iterator_for_pop_iter_tuple! {18} -impl_iterator_for_pop_iter_tuple! {19} -impl_iterator_for_pop_iter_tuple! {20} // Needed for graph to be able to manipulate // stream endings without knowing the generic type @@ -375,3 +349,10 @@ impl AnonymousStreamConsumer ) } } + +// pub trait PushIterable<'a, T, I> +// where +// I: Iterator, +// { +// fn push_iter(&'a mut self, iter: I) -> bool; +// } diff --git a/oxydsp-flowgraph/src/tag.rs b/oxydsp-flowgraph/src/tag.rs index e9e13b2..2a667c9 100644 --- a/oxydsp-flowgraph/src/tag.rs +++ b/oxydsp-flowgraph/src/tag.rs @@ -77,35 +77,44 @@ impl TagMergable> for Option } } +impl TagMergable> for Tagged +{ + fn merge(&self, other: &Option) -> Self + { + Tagged::new(self.0.clone(), self.1.merge(other)) + } +} + /// Represents a data, with a potential tag attached to it. #[derive(Clone)] -pub struct Tagged -{ - inner: T, - tag: Option, -} +pub struct Tagged(pub T, pub Option); impl Tagged { pub fn new(inner: T, tag: Option) -> Self { - Self { inner, tag } + Self(inner, tag) } pub fn has_tag(&self) -> bool { - self.tag.is_some() + self.1.is_some() } pub fn strip(&mut self) { - self.tag = None; + self.1 = None; + } + + pub fn into_inner(self) -> T + { + self.0 } pub fn tag(&mut self, tag: Tag) -> Option { - let t = self.tag.take(); - self.tag = Some(tag); + let t = self.1.take(); + self.1 = Some(tag); t } } @@ -114,12 +123,12 @@ impl Tagged { pub fn stripped(&self) -> Self { - self.inner.clone().into() + self.0.clone().into() } pub fn tagged(&self, tag: Tag) -> Self { - (self.inner.clone(), tag).into() + (self.0.clone(), tag).into() } } @@ -139,11 +148,27 @@ impl From<(T, Tag)> for Tagged } } +impl From<(T, Option)> for Tagged +{ + fn from((value, tag): (T, Option)) -> Self + { + Self::new(value, tag) + } +} + +impl Into<(T, Option)> for Tagged +{ + fn into(self) -> (T, Option) + { + (self.0, self.1) + } +} + impl DerefMut for Tagged { fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.inner + &mut self.0 } } @@ -153,6 +178,6 @@ impl Deref for Tagged fn deref(&self) -> &Self::Target { - &self.inner + &self.0 } }