use std::rc::Rc;
use std::cell::RefCell;
use crate::Container;
use crate::container::{ContainerBuilder, CapacityContainerBuilder};
use crate::scheduling::{Schedule, ActivateOnDrop};
use crate::progress::frontier::Antichain;
use crate::progress::{Operate, operate::SharedProgress, Timestamp};
use crate::progress::Source;
use crate::progress::ChangeBatch;
use crate::dataflow::channels::pushers::{Counter, Tee};
use crate::dataflow::channels::pushers::buffer::{Buffer as PushBuffer, AutoflushSession};
use crate::dataflow::operators::{ActivateCapability, Capability};
use crate::dataflow::{Scope, StreamCore};
pub trait UnorderedInput<G: Scope> {
fn new_unordered_input<CB: ContainerBuilder>(&mut self) -> ((UnorderedHandle<G::Timestamp, CB>, ActivateCapability<G::Timestamp>), StreamCore<G, CB::Container>);
}
impl<G: Scope> UnorderedInput<G> for G {
fn new_unordered_input<CB: ContainerBuilder>(&mut self) -> ((UnorderedHandle<G::Timestamp, CB>, ActivateCapability<G::Timestamp>), StreamCore<G, CB::Container>) {
let (output, registrar) = Tee::<G::Timestamp, CB::Container>::new();
let internal = Rc::new(RefCell::new(ChangeBatch::new()));
let cap = Capability::new(G::Timestamp::minimum(), internal.clone());
let counter = Counter::new(output);
let produced = counter.produced().clone();
let peers = self.peers();
let index = self.allocate_operator_index();
let address = self.addr_for_child(index);
let cap = ActivateCapability::new(cap, address.clone(), self.activations());
let helper = UnorderedHandle::new(counter);
self.add_operator_with_index(Box::new(UnorderedOperator {
name: "UnorderedInput".to_owned(),
address,
shared_progress: Rc::new(RefCell::new(SharedProgress::new(0, 1))),
internal,
produced,
peers,
}), index);
((helper, cap), StreamCore::new(Source::new(index, 0), registrar, self.clone()))
}
}
struct UnorderedOperator<T:Timestamp> {
name: String,
address: Rc<[usize]>,
shared_progress: Rc<RefCell<SharedProgress<T>>>,
internal: Rc<RefCell<ChangeBatch<T>>>,
produced: Rc<RefCell<ChangeBatch<T>>>,
peers: usize,
}
impl<T:Timestamp> Schedule for UnorderedOperator<T> {
fn name(&self) -> &str { &self.name }
fn path(&self) -> &[usize] { &self.address[..] }
fn schedule(&mut self) -> bool {
let shared_progress = &mut *self.shared_progress.borrow_mut();
self.internal.borrow_mut().drain_into(&mut shared_progress.internals[0]);
self.produced.borrow_mut().drain_into(&mut shared_progress.produceds[0]);
false
}
}
impl<T:Timestamp> Operate<T> for UnorderedOperator<T> {
fn inputs(&self) -> usize { 0 }
fn outputs(&self) -> usize { 1 }
fn get_internal_summary(&mut self) -> (Vec<Vec<Antichain<<T as Timestamp>::Summary>>>, Rc<RefCell<SharedProgress<T>>>) {
let mut borrow = self.internal.borrow_mut();
for (time, count) in borrow.drain() {
self.shared_progress.borrow_mut().internals[0].update(time, count * (self.peers as i64));
}
(Vec::new(), self.shared_progress.clone())
}
fn notify_me(&self) -> bool { false }
}
#[derive(Debug)]
pub struct UnorderedHandle<T: Timestamp, CB: ContainerBuilder> {
buffer: PushBuffer<T, CB, Counter<T, CB::Container, Tee<T, CB::Container>>>,
}
impl<T: Timestamp, CB: ContainerBuilder> UnorderedHandle<T, CB> {
fn new(pusher: Counter<T, CB::Container, Tee<T, CB::Container>>) -> UnorderedHandle<T, CB> {
UnorderedHandle {
buffer: PushBuffer::new(pusher),
}
}
#[inline]
pub fn session_with_builder(&mut self, cap: ActivateCapability<T>) -> ActivateOnDrop<AutoflushSession<T, CB, Counter<T, CB::Container, Tee<T, CB::Container>>>> {
ActivateOnDrop::new(self.buffer.autoflush_session_with_builder(cap.capability.clone()), cap.address.clone(), cap.activations.clone())
}
}
impl<T: Timestamp, C: Container> UnorderedHandle<T, CapacityContainerBuilder<C>> {
#[inline]
pub fn session(&mut self, cap: ActivateCapability<T>) -> ActivateOnDrop<AutoflushSession<T, CapacityContainerBuilder<C>, Counter<T, C, Tee<T, C>>>> {
self.session_with_builder(cap)
}
}