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

Stream rework, qpsk example, splitter/merger

Browse Source
This commit is contained in:
Albin Chaboissier
2026-04-12 00:46:41 +02:00
parent 87921968b4
commit dad09cddcf
17 changed files with 474 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -141,9 +141,10 @@ fn block_io_get_meta(ident: zyn::syn::Ident, fields: zyn::syn::Fields) -> zyn::T
fn get_output_type_names(&self) -> Vec<&'static str>
{
let mut output = Vec::new();
use oxydsp_flowgraph::block::BlockOutput;
@for (field in fields.iter().filter(|x| x.attrs.iter().any(|x| x.is("output"))).enumerate())
{
output.push(self.{{ field.1.ident.clone() }}.get_type_name());
output.extend(self.{{ field.1.ident.clone() }}.get_type_names());
}
return output;
}

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

@ -23,7 +23,7 @@ pub trait BlockInput
fn get_inputs(&self) -> Vec<&dyn AnonymousIn>;
// Meta information
fn get_types_names(&self) -> Vec<&'static str>;
fn get_type_names(&self) -> Vec<&'static str>;
}
pub trait BlockOutput
@ -32,7 +32,7 @@ pub trait BlockOutput
fn get_outputs(&self) -> Vec<&dyn AnonymousOut>;
// Meta information
fn get_types_names(&self) -> Vec<&'static str>;
fn get_type_names(&self) -> Vec<&'static str>;
}
pub trait BlockIO
@ -114,7 +114,7 @@ impl<T: 'static> BlockInput for In<T>
vec![self]
}
fn get_types_names(&self) -> Vec<&'static str>
fn get_type_names(&self) -> Vec<&'static str>
{
vec![std::any::type_name::<T>()]
}
@ -146,11 +146,11 @@ impl<I: BlockInput> BlockInput for Option<I>
}
}
fn get_types_names(&self) -> Vec<&'static str>
fn get_type_names(&self) -> Vec<&'static str>
{
if let Some(input) = self
{
input.get_types_names()
input.get_type_names()
}
else
{
@ -181,7 +181,7 @@ impl<I: BlockInput, const N: usize> BlockInput for [I; N]
output
}
fn get_types_names(&self) -> Vec<&'static str>
fn get_type_names(&self) -> Vec<&'static str>
{
vec![std::any::type_name::<I>(); N]
}
@ -199,7 +199,7 @@ impl<T: 'static> BlockOutput for Out<T>
vec![self]
}
fn get_types_names(&self) -> Vec<&'static str>
fn get_type_names(&self) -> Vec<&'static str>
{
vec![std::any::type_name::<T>()]
}
@ -231,11 +231,11 @@ impl<I: BlockOutput> BlockOutput for Option<I>
}
}
fn get_types_names(&self) -> Vec<&'static str>
fn get_type_names(&self) -> Vec<&'static str>
{
if let Some(input) = self
{
input.get_types_names()
input.get_type_names()
}
else
{
@ -266,7 +266,7 @@ impl<I: BlockOutput, const N: usize> BlockOutput for [I; N]
result
}
fn get_types_names(&self) -> Vec<&'static str>
fn get_type_names(&self) -> Vec<&'static str>
{
vec![std::any::type_name::<I>(); N]
}

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

@ -1,3 +1,4 @@
use std::any::Any;
use std::mem::ManuallyDrop;
use std::mem::MaybeUninit;
use std::sync::Arc;
@ -48,7 +49,7 @@ pub struct OutPush<'a, T>
start_index: usize,
}
impl<'a, T> OutPush<'a, T>
impl<'a, T: 'static> OutPush<'a, T>
{
pub fn len(&self) -> usize
{
@ -68,6 +69,7 @@ impl<'a, T> OutPush<'a, T>
pub fn push(&mut self, data: Tagged<T>) -> Result<(), Tagged<T>>
{
// println!("\n\n\n");
if self.written_data >= self.total_length
{
return Err(data);
@ -120,10 +122,12 @@ impl<'a, T> Drop for OutPush<'a, T>
{
fn drop(&mut self)
{
let data_writer =
let mut data_writer =
unsafe { ManuallyDrop::<StreamWriter<'a, T>>::take(&mut self.data_writer) };
let tag_writer =
let mut tag_writer =
unsafe { ManuallyDrop::<StreamWriter<'a, TagSlot>>::take(&mut self.tag_writer) };
tag_writer.produce(self.written_tags);
data_writer.produce(self.written_data);
}
@ -216,9 +220,9 @@ impl<'a, T> Drop for InIter<'a, T>
{
fn drop(&mut self)
{
let data_reader =
let mut data_reader =
unsafe { ManuallyDrop::<StreamReader<'a, T>>::take(&mut self.data_reader) };
let tag_reader =
let mut tag_reader =
unsafe { ManuallyDrop::<StreamReader<'a, TagSlot>>::take(&mut self.tag_reader) };
tag_reader.consume(self.read_tags);
data_reader.consume(self.read_data);
@ -290,7 +294,9 @@ impl<T: 'static> Out<T>
len -= 1;
match iter.next()
{
Some(element) => {let _ = pusher.push(element); },
Some(element) => {
let _ = pusher.push(element);
},
None => return false,
}
}

32
oxydsp-flowgraph/src/stream.rs
View File

@ -124,7 +124,7 @@ pub struct StreamReader<'a, T>
slices: (&'a mut [Takable<T>], &'a mut [Takable<T>]),
// UNSAFE !
consumer: &'a mut StreamConsumer<T>,
inner: &'a mut StreamConsumer<T>,
}
pub struct StreamWriter<'a, T>
@ -132,7 +132,7 @@ pub struct StreamWriter<'a, T>
slices: (&'a mut [MaybeUninit<T>], &'a mut [MaybeUninit<T>]),
// UNSAFE !
producer: &'a mut StreamProducer<T>,
inner: &'a mut StreamProducer<T>,
}
impl<'a, T> StreamReader<'a, T>
@ -147,9 +147,9 @@ impl<'a, T> StreamReader<'a, T>
(self.slices.0, self.slices.1)
}
pub fn consume(self, amount: usize)
pub fn consume(&mut self, read: usize)
{
self.consumer.consume(amount);
self.inner.consume(read);
}
}
@ -165,9 +165,9 @@ impl<'a, T> StreamWriter<'a, T>
(self.slices.0, self.slices.1)
}
pub fn produce(self, amount: usize)
pub fn produce(&mut self, written: usize)
{
self.producer.produce(amount);
self.inner.produce(written);
}
}
@ -228,8 +228,8 @@ impl<T> StreamProducer<T>
// This functions borrows the stream mutably. As such, only one instance
// of these slices can exist for a given stream.
StreamWriter {
slices: (start_to_head, head_to_end),
producer: self,
slices: (head_to_end, start_to_head),
inner: self,
}
}
}
@ -262,7 +262,7 @@ impl<T> StreamProducer<T>
StreamWriter {
slices: (head_to_end, start_to_tail),
producer: self,
inner: self,
}
}
}
@ -292,12 +292,13 @@ impl<T> StreamProducer<T>
unsafe {
let k = &mut *self.inner.buffer.get();
let (start_to_tail, _tail_to_end) = k.split_at_mut_unchecked(wrapped_tail);
let (_start_to_head, head_to_tail) = start_to_tail.split_at_mut_unchecked(wrapped_head);
let (_start_to_head, head_to_tail) =
start_to_tail.split_at_mut_unchecked(wrapped_head);
let (empty_slice, head_to_tail) = head_to_tail.split_at_mut_unchecked(0);
StreamWriter {
slices: (head_to_tail, empty_slice),
producer: self,
inner: self,
}
}
}
@ -349,7 +350,7 @@ impl<T> StreamConsumer<T>
takable_slice_from_maybe_uninitt(empty_1),
takable_slice_from_maybe_uninitt(empty_2),
),
consumer: self,
inner: self,
}
}
}
@ -383,7 +384,7 @@ impl<T> StreamConsumer<T>
takable_slice_from_maybe_uninitt(tail_to_head),
takable_slice_from_maybe_uninitt(empty_slice),
),
consumer: self,
inner: self,
}
}
}
@ -424,7 +425,7 @@ impl<T> StreamConsumer<T>
takable_slice_from_maybe_uninitt(tail_to_end),
takable_slice_from_maybe_uninitt(start_to_head),
),
consumer: self,
inner: self,
}
}
}
@ -505,8 +506,7 @@ mod test
assert_eq!(a.len(), 3);
assert_eq!(b.len(), 0);
unsafe
{
unsafe {
assert_eq!(a[0].take(), 0);
assert_eq!(a[1].take(), 1);
assert_eq!(a[2].take(), 2);

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

@ -197,24 +197,29 @@ impl Tag
///
/// If all the tag options are None, None is returned
/// Otherwise it is Some of the combination of all of the tags which are Some
pub fn from_tag_opts<const N: usize>(tag_opts: &[&Option<Tag>; N]) -> Option<Tag>
pub fn from_tag_opts<'a>(mut tag_opts: impl Iterator<Item = &'a Option<Tag>>) -> Option<Tag>
{
if tag_opts.iter().all(|t| t.is_none())
{
return None;
}
let new_tag = Self::default();
let mut some_tags = 0;
{
let mut writer = new_tag.data.write().unwrap();
for tag in tag_opts.iter().filter(|t| t.is_some())
for tag in tag_opts.filter(|t| t.is_some())
{
some_tags += 1;
let reader = tag.as_ref().unwrap().data.read().unwrap();
writer.extend(reader.iter().map(|x| (*x.0, x.1.clone())));
}
}
Some(new_tag)
if some_tags > 0
{
Some(new_tag)
}
else
{
None
}
}
/// Adds a new entry in the tag. If it already exists, it is overwritten
@ -264,7 +269,7 @@ impl TagMergable<Option<Tag>> for Option<Tag>
{
fn merge(&self, other: &Self) -> Self
{
Tag::from_tag_opts(&[self, other])
Tag::from_tag_opts([self, other].into_iter())
}
}