use super::super::plumbing::*;
use crate::SendPtr;
use std::marker::PhantomData;
use std::ptr;
use std::slice;
pub(super) struct CollectConsumer<'c, T: Send> {
start: SendPtr<T>,
len: usize,
marker: PhantomData<&'c mut T>,
}
impl<T: Send> CollectConsumer<'_, T> {
pub(super) fn appender(vec: &mut Vec<T>, len: usize) -> CollectConsumer<'_, T> {
let start = vec.len();
assert!(vec.capacity() - start >= len);
unsafe { CollectConsumer::new(vec.as_mut_ptr().add(start), len) }
}
}
impl<'c, T: Send + 'c> CollectConsumer<'c, T> {
unsafe fn new(start: *mut T, len: usize) -> Self {
CollectConsumer {
start: SendPtr(start),
len,
marker: PhantomData,
}
}
}
#[must_use]
pub(super) struct CollectResult<'c, T> {
start: SendPtr<T>,
total_len: usize,
initialized_len: usize,
invariant_lifetime: PhantomData<&'c mut &'c mut [T]>,
}
unsafe impl<'c, T> Send for CollectResult<'c, T> where T: Send {}
impl<'c, T> CollectResult<'c, T> {
pub(super) fn len(&self) -> usize {
self.initialized_len
}
pub(super) fn release_ownership(mut self) -> usize {
let ret = self.initialized_len;
self.initialized_len = 0;
ret
}
}
impl<'c, T> Drop for CollectResult<'c, T> {
fn drop(&mut self) {
unsafe {
ptr::drop_in_place(slice::from_raw_parts_mut(
self.start.0,
self.initialized_len,
));
}
}
}
impl<'c, T: Send + 'c> Consumer<T> for CollectConsumer<'c, T> {
type Folder = CollectResult<'c, T>;
type Reducer = CollectReducer;
type Result = CollectResult<'c, T>;
fn split_at(self, index: usize) -> (Self, Self, CollectReducer) {
let CollectConsumer { start, len, .. } = self;
unsafe {
assert!(index <= len);
(
CollectConsumer::new(start.0, index),
CollectConsumer::new(start.0.add(index), len - index),
CollectReducer,
)
}
}
fn into_folder(self) -> Self::Folder {
CollectResult {
start: self.start,
total_len: self.len,
initialized_len: 0,
invariant_lifetime: PhantomData,
}
}
fn full(&self) -> bool {
false
}
}
impl<'c, T: Send + 'c> Folder<T> for CollectResult<'c, T> {
type Result = Self;
fn consume(mut self, item: T) -> Self {
assert!(
self.initialized_len < self.total_len,
"too many values pushed to consumer"
);
unsafe {
self.start.0.add(self.initialized_len).write(item);
self.initialized_len += 1;
}
self
}
fn complete(self) -> Self::Result {
self
}
fn full(&self) -> bool {
false
}
}
impl<'c, T: Send + 'c> UnindexedConsumer<T> for CollectConsumer<'c, T> {
fn split_off_left(&self) -> Self {
unreachable!("CollectConsumer must be indexed!")
}
fn to_reducer(&self) -> Self::Reducer {
CollectReducer
}
}
pub(super) struct CollectReducer;
impl<'c, T> Reducer<CollectResult<'c, T>> for CollectReducer {
fn reduce(
self,
mut left: CollectResult<'c, T>,
right: CollectResult<'c, T>,
) -> CollectResult<'c, T> {
unsafe {
let left_end = left.start.0.add(left.initialized_len);
if left_end == right.start.0 {
left.total_len += right.total_len;
left.initialized_len += right.release_ownership();
}
left
}
}
}