octagonal/oxydsp
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Starts synchronous block interface

Browse Source
This commit is contained in:
Albin Chaboissier
2026-03-19 15:28:31 +01:00
parent f727c119b8
commit 2baee8d28b
11 changed files with 348 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
oxydsp-flowgraph/oxydsp-flowgraph-macros/src/lib.rs
View File

@ -4,8 +4,11 @@ use zyn::ext::AttrExt;
use zyn::ext::FieldsExt;
use zyn::format_ident;
use zyn::parse_input;
use zyn::syn::GenericParam;
use zyn::syn::Index;
use zyn::syn::Lit;
use zyn::syn::TypeGenerics;
use zyn::syn::parse_quote;
use zyn::syn::spanned::Spanned;
#[zyn::derive("BlockIO", attributes(input, output), debug = "pretty")]
@ -146,7 +149,7 @@ fn block_io_set_streams(fields: zyn::syn::Fields) -> zyn::TokenStream
fn set_anonymous_out_stream(
&mut self,
output_index: usize,
producer: oxydsp_flowgraph::edge::AnonymousStreamProducer,
producer: oxydsp_flowgraph::io::AnonymousStreamProducer,
)
{
match output_index
@ -160,7 +163,7 @@ fn block_io_set_streams(fields: zyn::syn::Fields) -> zyn::TokenStream
}
#[allow(unreachable_code)]
fn set_anonymous_in_stream(&mut self, input_index: usize, consumer: oxydsp_flowgraph::edge::AnonymousStreamConsumer)
fn set_anonymous_in_stream(&mut self, input_index: usize, consumer: oxydsp_flowgraph::io::AnonymousStreamConsumer)
{
match input_index
{
@ -178,29 +181,26 @@ fn block_io_set_streams(fields: zyn::syn::Fields) -> zyn::TokenStream
fn block_io_create_stream(fields: zyn::syn::Fields) -> zyn::TokenStream
{
zyn::zyn!(
#[allow(unreachable_code)]
fn create_anonymous_stream_for(
#[allow(unreachable_code)]
fn create_anonymous_stream_for(
&mut self,
output_index: usize,
capacity: usize
) -> (oxydsp_flowgraph::edge::AnonymousStreamProducer, oxydsp_flowgraph::edge::AnonymousStreamConsumer)
{
let (tx, rx): (oxydsp_flowgraph::edge::AnonymousStreamProducer, oxydsp_flowgraph::edge::AnonymousStreamConsumer)
= match output_index
capacity: usize,
) -> (
oxydsp_flowgraph::io::AnonymousStreamProducer,
oxydsp_flowgraph::io::AnonymousStreamConsumer,
)
{
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("output"))).enumerate())
let output = match output_index
{
{{ field.0 }} =>
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("output"))).enumerate())
{
let (tx, rx) = oxydsp_flowgraph::stream::bounded_queue::<@out_inner_type(ty = field.1.ty.clone())>(capacity);
(tx.into(), rx.into())
},
}
_ => panic!("output_index out of bounds.")
};
(tx, rx)
}
{{ field.0 }} => self.{{ field.1.ident }}.create_anonymous_stream(capacity),
}
_ => panic!("output_index out of bounds."),
};
return output;
}
)
}
@ -234,81 +234,37 @@ fn out_inner_type(ty: zyn::syn::Type) -> zyn::TokenStream
#[zyn::attribute]
pub fn sync_block(#[zyn(input)] item: zyn::syn::ItemStruct) -> zyn::TokenStream
{
let mut strcut_item = item.clone();
let (impl_generics, type_generics, where_clause) = item.generics.split_for_impl();
let lifetime: GenericParam = parse_quote!('view);
let mut generics = item.generics.clone();
generics.params.insert(0, lifetime);
let (_impl_generics, type_generics, where_clause) = generics.split_for_impl();
let fields = &item.fields.as_named().unwrap().named;
// Get state fields
let mut state_fields = vec![];
for field in strcut_item.fields.iter_mut()
for field in fields.iter()
{
let attr_index = field.attrs.iter().enumerate().find_map(|(i, attr)| {
if attr.is("sync_state") { Some(i) } else { None }
});
if let Some(state_field) = &field.ident
&& let Some(attr_index) = attr_index
let mut f = field.clone();
if f.attrs.iter().any(|x| x.is("input") || x.is("output"))
{
state_fields.push(state_field.clone());
field.attrs.remove(attr_index);
continue;
}
let tk = field.ty.clone().into_token_stream();
f.ty = parse_quote!(&'view mut #tk);
state_fields.push(f);
}
zyn::zyn!(
{{ strcut_item }}
{{ item }}
impl {{ impl_generics }} oxydsp_flowgraph::block::Block for {{ strcut_item.ident }} {{ type_generics }}
where {{ where_clause }}
struct {{ item.ident | ident:"{}View" }} {{ type_generics }}
{{ where_clause }}
{
fn work(&mut self) -> oxydsp_flowgraph::block::BlockResult
@for (field in state_fields.iter())
{
let mut len = usize::MAX;
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("input"))).enumerate())
{
let mut {{ field.1.ident.clone().unwrap() | ident:"{}_reader" }} = self.{{field.1.ident}}.read();
len = len.min({{ field.1.ident.clone().unwrap() | ident:"{}_reader" }}.len());
}
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("output"))).enumerate())
{
let mut {{ field.1.ident.clone().unwrap() | ident: "{}_writer" }} = self.{{field.1.ident}}.write();
len = len.min({{ field.1.ident.clone().unwrap() | ident: "{}_writer" }}.len());
}
for _ in 0..len
{
if let Some((
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("output"))).enumerate())
{
{{ field.1.ident.clone().unwrap() | ident: "{}_out" }},
}
)) = Self::sync_work(
(
@for (state_field in state_fields)
{
&mut self.{{ state_field }},
}
),
(
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("input"))).enumerate())
{
{{ field.1.ident.clone().unwrap() | ident: "{}_reader" }}.pop().unwrap(),
}
)
)
{
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("output"))).enumerate())
{
let _ = {{ field.1.ident.clone().unwrap() | ident: "{}_writer" }}.push({{ field.1.ident.clone().unwrap() | ident: "{}_out" }});
}
} else
{
return oxydsp_flowgraph::block::BlockResult::Terminated;
}
}
oxydsp_flowgraph::block::BlockResult::Ok
{{ field }},
}
_sync_block_phantom: std::marker::PhantomData<'view>,
}
)
}
@ -373,7 +329,7 @@ pub fn generate_pop_iterable_tuple_impl(input: TokenStream) -> TokenStream
type Output = (
@for (i in 0..count)
{
StreamReader<'a, {{ generics[i] }}>,
InReader<'a, {{ generics[i] }}>,
}
);
fn pop_iter(&'a mut self) -> PopIter<Self::Output>
@ -457,14 +413,14 @@ pub fn impl_iterator_for_pop_iter_tuple(input: TokenStream) -> TokenStream
> Iterator for PopIter<(
@for (i in 0..count)
{
StreamReader<'a, {{ generics[i] }}>,
InReader<'a, {{ generics[i] }}>,
}
)>
{
type Item = (
@for (i in 0..count)
{
{{ generics[i] }},
({{ generics[i] }}, Option<crate::tag::Tag>),
}
);
fn next(&mut self) -> Option<Self::Item>
@ -476,7 +432,7 @@ pub fn impl_iterator_for_pop_iter_tuple(input: TokenStream) -> TokenStream
Some((
@for (i in 0..count)
{
self.reader.{{ Index::from(i) }}.pop().unwrap(),
self.reader.{{ Index::from(i) }}.pop_tagged().unwrap(),
}
))
}

16
oxydsp-flowgraph/src/block.rs
View File

@ -1,4 +1,7 @@
use crate::edge::{AnonymousStreamConsumer, AnonymousStreamProducer, BlockIOIndex};
use crate::{
edge::BlockIOIndex,
io::{AnonymousStreamConsumer, AnonymousStreamProducer},
};
pub enum BlockResult
{
@ -46,13 +49,18 @@ pub trait Block
fn work(&mut self) -> BlockResult;
}
pub trait SyncBlock
// Represents the input, output, state types
// that a SyncBlock will have to interacti with
pub trait SyncBlockIO
{
type StateView;
type Input;
type Output;
type State;
}
fn sync_work(state: &mut Self::State, input: Self::Input) -> Option<Self::Output>;
pub trait SyncBlock: SyncBlockIO
{
fn sync_work(state: Self::StateView, input: Self::Input) -> Option<Self::Output>;
}
pub trait GraphableBlock: Block + BlockIO {}

138
oxydsp-flowgraph/src/edge.rs
View File

@ -1,17 +1,4 @@
use std::any::Any;
use std::collections::binary_heap::Iter;
use std::sync::Arc;
use std::sync::Mutex;
use oxydsp_flowgraph_macros::generate_pop_iterable_tuple_impl;
use oxydsp_flowgraph_macros::impl_iterator_for_pop_iter_tuple;
use crate::stream;
use crate::stream::StreamConsumer;
use crate::stream::StreamProducer;
use crate::stream::StreamReader;
use crate::stream::StreamWriter;
use crate::tag::Tag;
#[derive(Default)]
pub struct Edge
@ -31,128 +18,3 @@ pub struct BlockIOIndex
pub block_index: usize,
pub port_index: usize,
}
// Needed for graph to be able to manipulate
// stream endings without knowing the generic type
pub struct AnonymousStreamProducer
{
inner: Box<dyn Any>,
}
pub struct AnonymousStreamConsumer
{
inner: Box<dyn Any>,
}
impl<T: 'static> From<StreamProducer<T>> for AnonymousStreamProducer
{
fn from(value: StreamProducer<T>) -> Self
{
AnonymousStreamProducer {
inner: Box::new(value),
}
}
}
impl<T: 'static> From<StreamConsumer<T>> for AnonymousStreamConsumer
{
fn from(value: StreamConsumer<T>) -> Self
{
AnonymousStreamConsumer {
inner: Box::new(value),
}
}
}
impl AnonymousStreamProducer
{
pub fn downcast<T: 'static>(self) -> StreamProducer<T>
{
*self.inner.downcast::<StreamProducer<T>>().unwrap()
}
}
impl AnonymousStreamConsumer
{
pub fn downcast<T: 'static>(self) -> StreamConsumer<T>
{
*self.inner.downcast::<StreamConsumer<T>>().unwrap()
}
}
pub struct PopIter<T>
{
len: usize,
popped: usize,
reader: T,
}
pub trait PopIterable<'a>
{
type Output;
fn pop_iter(&'a mut self) -> PopIter<Self::Output>;
}
impl<'a, T: 'static> PopIterable<'a> for In<T>
{
type Output = StreamReader<'a, T>;
fn pop_iter(&'a mut self) -> PopIter<StreamReader<'a, T>>
{
let reader = self.read();
PopIter {
popped: 0,
len: reader.len(),
reader,
}
}
}
generate_pop_iterable_tuple_impl! {2}
generate_pop_iterable_tuple_impl! {3}
generate_pop_iterable_tuple_impl! {4}
generate_pop_iterable_tuple_impl! {5}
generate_pop_iterable_tuple_impl! {6}
generate_pop_iterable_tuple_impl! {7}
generate_pop_iterable_tuple_impl! {8}
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<StreamReader<'a, T>>
{
type Item = T;
fn next(&mut self) -> Option<Self::Item>
{
self.reader.pop()
}
}
impl_iterator_for_pop_iter_tuple! {2}
impl_iterator_for_pop_iter_tuple! {3}
impl_iterator_for_pop_iter_tuple! {4}
impl_iterator_for_pop_iter_tuple! {5}
impl_iterator_for_pop_iter_tuple! {6}
impl_iterator_for_pop_iter_tuple! {7}
impl_iterator_for_pop_iter_tuple! {8}
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}

207
oxydsp-flowgraph/src/io.rs
View File

@ -1,8 +1,10 @@
use std::any::Any;
use std::sync::Arc;
use std::sync::Mutex;
use crate::edge::AnonymousStreamConsumer;
use crate::edge::AnonymousStreamProducer;
use oxydsp_flowgraph_macros::generate_pop_iterable_tuple_impl;
use oxydsp_flowgraph_macros::impl_iterator_for_pop_iter_tuple;
use crate::edge::BlockIOIndex;
use crate::edge::Edge;
use crate::stream::StreamConsumer;
@ -73,7 +75,9 @@ impl<T: 'static> In<T>
pub fn set_anonymous_stream(&mut self, consumer: AnonymousStreamConsumer)
{
self.stream = Some(consumer.downcast::<T>())
let (stream, tag_stream) = consumer.downcast::<T>();
self.stream = Some(stream);
self.tag_stream = Some(tag_stream);
}
pub fn read<'a>(&'a mut self) -> InReader<'a, T>
@ -101,7 +105,9 @@ impl<T: 'static> Out<T>
pub fn set_anonymous_stream(&mut self, producer: AnonymousStreamProducer)
{
self.stream = Some(producer.downcast::<T>())
let (stream, tag_stream) = producer.downcast::<T>();
self.stream = Some(stream);
self.tag_stream = Some(tag_stream);
}
// Delegate stream creation to Out object
@ -112,7 +118,8 @@ impl<T: 'static> Out<T>
) -> (AnonymousStreamProducer, AnonymousStreamConsumer)
{
let (tx, rx) = stream::bounded_queue::<T>(capacity);
(tx.into(), rx.into())
let (tx_tag, rx_tag) = stream::bounded_queue::<Tag>(capacity);
((tx, tx_tag).into(), (rx, rx_tag).into())
}
pub fn write<'a>(&'a mut self) -> OutWriter<'a, T>
@ -123,7 +130,7 @@ impl<T: 'static> Out<T>
}
}
pub fn push_iter<I: Iterator<Item = T>>(&mut self, mut iter: I) -> bool
pub fn push_iter<I: Iterator<Item = (T, Option<Tag>)>>(&mut self, mut iter: I) -> bool
{
let writer = self.write();
let len = writer.len();
@ -132,7 +139,17 @@ impl<T: 'static> Out<T>
{
if let Some(elt) = iter.next()
{
let _ = writer.push(elt);
match elt.1
{
Some(tag) =>
{
let _ = writer.push_tagged(elt.0, tag);
}
None =>
{
let _ = writer.push_no_tag(elt.0);
}
}
}
else
{
@ -182,3 +199,179 @@ impl<T> InReader<'_, T>
self.pop_tagged().map(|(data, _)| data)
}
}
impl<T> OutWriter<'_, T>
{
pub fn len(&self) -> usize
{
self.data_writer.len().min(self.tag_writer.len())
}
pub fn push(&self, (data, tag): (T, Option<Tag>)) -> Result<(), (T, Option<Tag>)>
{
match self.data_writer.push(data)
{
Ok(_) if tag.is_some() =>
{
let _ = self.tag_writer.push(tag.unwrap());
Ok(())
}
Ok(_) => Ok(()),
Err(data) => Err((data, tag)),
}
}
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)
}
}
pub struct PopIter<T>
{
len: usize,
popped: usize,
reader: T,
}
pub trait PopIterable<'a>
{
type Output;
fn pop_iter(&'a mut self) -> PopIter<Self::Output>;
}
impl<'a, T: 'static> PopIterable<'a> for In<T>
{
type Output = InReader<'a, T>;
fn pop_iter(&'a mut self) -> PopIter<InReader<'a, T>>
{
let reader = self.read();
PopIter {
popped: 0,
len: reader.len(),
reader,
}
}
}
generate_pop_iterable_tuple_impl! {2}
generate_pop_iterable_tuple_impl! {3}
generate_pop_iterable_tuple_impl! {4}
generate_pop_iterable_tuple_impl! {5}
generate_pop_iterable_tuple_impl! {6}
generate_pop_iterable_tuple_impl! {7}
generate_pop_iterable_tuple_impl! {8}
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<InReader<'a, T>>
{
type Item = (T, Option<Tag>);
fn next(&mut self) -> Option<Self::Item>
{
self.reader.pop_tagged()
}
}
impl_iterator_for_pop_iter_tuple! {2}
impl_iterator_for_pop_iter_tuple! {3}
impl_iterator_for_pop_iter_tuple! {4}
impl_iterator_for_pop_iter_tuple! {5}
impl_iterator_for_pop_iter_tuple! {6}
impl_iterator_for_pop_iter_tuple! {7}
impl_iterator_for_pop_iter_tuple! {8}
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
pub struct AnonymousStreamProducer
{
inner: Box<dyn Any>,
inner_tag: StreamProducer<Tag>,
}
pub struct AnonymousStreamConsumer
{
inner: Box<dyn Any>,
inner_tag: StreamConsumer<Tag>,
}
impl<T: 'static> From<(StreamProducer<T>, StreamProducer<Tag>)> for AnonymousStreamProducer
{
fn from(value: (StreamProducer<T>, StreamProducer<Tag>)) -> Self
{
AnonymousStreamProducer {
inner: Box::new(value.0),
inner_tag: value.1,
}
}
}
impl<T: 'static> From<(StreamConsumer<T>, StreamConsumer<Tag>)> for AnonymousStreamConsumer
{
fn from(value: (StreamConsumer<T>, StreamConsumer<Tag>)) -> Self
{
AnonymousStreamConsumer {
inner: Box::new(value.0),
inner_tag: value.1,
}
}
}
impl AnonymousStreamProducer
{
pub fn downcast<T: 'static>(self) -> (StreamProducer<T>, StreamProducer<Tag>)
{
(
*self.inner.downcast::<StreamProducer<T>>().unwrap(),
self.inner_tag,
)
}
}
impl AnonymousStreamConsumer
{
pub fn downcast<T: 'static>(self) -> (StreamConsumer<T>, StreamConsumer<Tag>)
{
(
*self.inner.downcast::<StreamConsumer<T>>().unwrap(),
self.inner_tag,
)
}
}

46
oxydsp-flowgraph/src/tag.rs
View File

@ -17,7 +17,51 @@ pub struct Tag
// TODO: Make it such that when a tag is duplicated, the data seems to be too:
// When adding on a duplicate, it should not replicate on others, but without
// requiring a deep copy.
pub data: Arc<Mutex<HashMap<String, Box<dyn Any>>>>,
pub data: Arc<Mutex<HashMap<String, Arc<dyn Any + Send + Sync>>>>,
}
pub trait TagValue: Clone {}
impl<T> TagValue for T where T: Clone {}
pub trait TagMergable<T>
{
fn merge(&self, other: &T) -> Self;
}
impl TagMergable<Tag> for Tag
{
fn merge(&self, other: &Self) -> Self
{
// TODO: More performant merge
let mut new = other.clone();
new.position = self.position;
{
let mut data_locked = new.data.lock().unwrap();
for (k, v) in self.data.lock().unwrap().iter()
{
data_locked.insert(k.clone(), v.clone());
}
}
new
}
}
impl TagMergable<Option<Tag>> for Option<Tag>
{
fn merge(&self, other: &Self) -> Self
{
match self
{
Some(first) => match other
{
Some(other) => Some(first.merge(other)),
None => Some(first.clone()),
},
None => other.clone(),
}
}
}
/// Represents a data, with a potential tag attached to it.