pub fn publish_batch(&mut self, data: DataflowEvents) {
        match data {
            // DataflowEvents::Differential(data) => {
            //     for (_t, _wid, x) in data.drain(..) {
            //         match &x {
            //             Merge(e) if e.complete.is_some() => {
            //                 let op_addr = self.op_id_to_op_addr.get(&e.operator).expect("op id in addr not found");
            //                 // we need `ch_source` here, since we're not interested in the operator id
            //                 // provided by the `Batch` event, but where the batch event's records flow to
            //                 // (e.g., a join operator).
            //                 let ch_id = self.op_addr_to_ch_source.get(&op_addr).expect("op addr in chs not found");

            //                 let counter = self.channel_records.entry(*ch_id).or_insert(0);
            //                 *counter += e.complete.unwrap();
            //             }
            //             _ => {}
            //         }
            //     }
            // }
            DataflowEvents::Timely(data) => {
                for (t, wid, x) in data.drain(..) {
                    self.overall_messages += 1;

                    match &x {
                        TimelyEvent::Operates(_) | TimelyEvent::Channels(_) |
                        TimelyEvent::Messages(_) | TimelyEvent::Schedule(_) |
                        TimelyEvent::Text(_) => {
                            // println!("{:?}, {:?}: {:?}", std::time::SystemTime::now(), t, x);
                        },
                        _ => {}
                    }

                    match &x {
                        Text(_) => self.tick_epoch(),
                        Operates(e) => {
                            self.pag_messages += 1;
                            self.fuel -= 1;
                            self.seq_no += 1;

                            // all operates events should happen in the initialization epoch,
                            // i.e., before any Text event epoch markers have been observed
                            assert!(self.next_cap.first == 1 && self.curr_cap == Default::default());

                            self.op_id_to_op_addr.insert(e.id, *e.addr.last().expect("addr empty"));

                            self.buffer.push((self.curr_cap.first, self.seq_no, None, (Default::default(), wid, x)));
                        }
                        Channels(e) => {
                            let ids = self.op_addr_to_ch_target.entry(e.target.0).or_insert(Vec::new());
                            ids.push(e.id);
                            self.op_addr_to_ch_source.insert(e.source.0, e.id);
                        }
                        Schedule(e) => {
                            self.pag_messages += 1;
                            // extend buffer size by 1 to avoid breaking up repositioning of
                            // schedule start events and consequent data messages
                            if self.fuel > 1 || e.start_stop == StartStop::Stop {
                                self.fuel -= 1;
                            }
                            self.seq_no += 1;

                            if self.tick_sys {
                                self.advance_cap(&t);
                            }

                            // fetch length
                            let length = if e.start_stop == StartStop::Stop {
                                let op_addr = self.op_id_to_op_addr.get(&e.id).expect("op id not found");

                                // TODO: refactor
                                if let Some(ch_ids) = self.op_addr_to_ch_target.get(&op_addr) {
                                    let mut len = 0;
                                    for ch_id in ch_ids.iter() {
                                        if let Some(l) = self.channel_records.remove(ch_id) {
                                            len += l;
                                        }
                                    }
                                    if len > 0 {
                                        Some(len)
                                    } else {
                                        None
                                    }
                                } else {
                                    // For Inputs, we won't find a corresponding channel
                                    // (by definition, no channels end at an input)
                                    None
                                }
                            } else {
                                None
                            };

                            self.buffer.push((self.curr_cap.first, self.seq_no, length, (t, wid, x)));
                        }
                        // Remote progress events
                        Progress(e) if e.is_send || e.source != wid => {
                            self.pag_messages += 1;
                            self.fuel -= 1;
                            self.seq_no += 1;

                            if self.tick_sys {
                                self.advance_cap(&t);
                            }

                            self.buffer.push((self.curr_cap.first, self.seq_no, None, (t, wid, x)));
                        }
                        // Data receive events
                        Messages(e) if e.is_send == false => {
                            assert!(e.target == wid);
                            // A. update record counter
                            let counter = self.channel_records.entry(e.channel).or_insert(0);
                            *counter += e.length;

                            // B. if remote message: add to pag events
                            if e.source != e.target {
                                self.pag_messages += 1;
                                self.seq_no += 1;

                                self.buffer.push((self.curr_cap.first, self.seq_no, Some(e.length), (t, wid, x)));

                                // let (last_epoch, last_seq, last_length, (last_t, last_wid, last_x)) = self.buffer.pop()
                                    // .expect("non-empty buffer required");

                                // assert!(if let Schedule(e) = &last_x {e.start_stop == StartStop::Start}
                                        // else {false});

                                // Reposition received remote data message:
                                // 1. push the data message with previous' seq_no
                                // self.buffer.push((self.curr_cap.first, last_seq, Some(e.length), (t, wid, x)));
                                // 2. push the schedule event
                                // self.buffer.push((last_epoch, self.seq_no, last_length, (last_t, last_wid, last_x)));
                            }
                        }
                        // remote data send events
                        Messages(e) if e.source != e.target => {
                            assert!(e.source == wid);

                            self.pag_messages += 1;
                            self.fuel -= 1;
                            self.seq_no += 1;
                            if self.tick_sys {
                                self.advance_cap(&t);
                            }
                            self.buffer.push((self.curr_cap.first, self.seq_no, Some(e.length), (t, wid, x)));
                        }
                        _ => {}
                    }

                    if self.fuel == 0 {
                        self.flush_buffer();
                    }
                }
            }
        }
    }
/// Publishes a batch of logged events and advances the capability.
Publishes a batch of logged events and advances the capability.
Publishes a batch of logged events and advances the capability .
pub fn publish_batch(&mut self, data: DataflowEvents) {
        match data {
            // DataflowEvents::Differential(data) => {
            //     for (_t, _wid, x) in data.drain(..) {
            //         match &x {
            //             Merge(e) if e.complete.is_some() => {
            //                 let op_addr = self.op_id_to_op_addr.get(&e.operator).expect("op id in addr not found");
            //                 // we need `ch_source` here, since we're not interested in the operator id
            //                 // provided by the `Batch` event, but where the batch event's records flow to
            //                 // (e.g., a join operator).
            //                 let ch_id = self.op_addr_to_ch_source.get(&op_addr).expect("op addr in chs not found");
            //                 let counter = self.channel_records.entry(*ch_id).or_insert(0);
            //                 *counter += e.complete.unwrap();
            //             }
            //             _ => {}
            //         }
            //     }
            // }
            DataflowEvents::Timely(data) => {
                for (t, wid, x) in data.drain(..) {
                    self.overall_messages += 1;
                    match &x {
                        TimelyEvent::Operates(_) | TimelyEvent::Channels(_) |
                        TimelyEvent::Messages(_) | TimelyEvent::Schedule(_) |
                        TimelyEvent::Text(_) => {
                            // println!("{:?}, {:?}: {:?}", std::time::SystemTime::now(), t, x);
                        },
                        _ => {}
                    }
                    match &x {
                        Text(_) => self.tick_epoch(),
                        Operates(e) => {
                            self.pag_messages += 1;
                            self.fuel -= 1;
                            self.seq_no += 1;
                            // all operates events should happen in the initialization epoch,
                            // i.e., before any Text event epoch markers have been observed
                            assert!(self.next_cap.first == 1 && self.curr_cap == Default::default());
                            self.op_id_to_op_addr.insert(e.id, *e.addr.last().expect("addr empty"));
                            self.buffer.push((self.curr_cap.first, self.seq_no, None, (Default::default(), wid, x)));
                        }
                        Channels(e) => {
                            let ids = self.op_addr_to_ch_target.entry(e.target.0).or_insert(Vec::new());
                            ids.push(e.id);
                            self.op_addr_to_ch_source.insert(e.source.0, e.id);
                        }
                        Schedule(e) => {
                            self.pag_messages += 1;
                            // extend buffer size by 1 to avoid breaking up repositioning of
                            // schedule start events and consequent data messages
                            if self.fuel > 1 || e.start_stop == StartStop::Stop {
                                self.fuel -= 1;
                            }
                            self.seq_no += 1;
                            if self.tick_sys {
                                self.advance_cap(&t);
                            }
                            // fetch length
                            let length = if e.start_stop == StartStop::Stop {
                                let op_addr = self.op_id_to_op_addr.get(&e.id).expect("op id not found");
                                // TODO: refactor
                                if let Some(ch_ids) = self.op_addr_to_ch_target.get(&op_addr) {
                                    let mut len = 0;
                                    for ch_id in ch_ids.iter() {
                                        if let Some(l) = self.channel_records.remove(ch_id) {
                                            len += l;
                                        }
                                    }
                                    if len > 0 {
                                        Some(len)
                                    } else {
                                        None
                                    }
                                } else {
                                    // For Inputs, we won't find a corresponding channel
                                    // (by definition, no channels end at an input)
                                    None
                                }
                            } else {
                                None
                            };
                            self.buffer.push((self.curr_cap.first, self.seq_no, length, (t, wid, x)));
                        }
                        // Remote progress events
                        Progress(e) if e.is_send || e.source != wid => {
                            self.pag_messages += 1;
                            self.fuel -= 1;
                            self.seq_no += 1;
                            if self.tick_sys {
                                self.advance_cap(&t);
                            }
                            self.buffer.push((self.curr_cap.first, self.seq_no, None, (t, wid, x)));
                        }
                        // Data receive events
                        Messages(e) if e.is_send == false => {
                            assert!(e.target == wid);
                            // A. update record counter
                            let counter = self.channel_records.entry(e.channel).or_insert(0);
                            *counter += e.length;
                            // B. if remote message: add to pag events
                            if e.source != e.target {
                                self.pag_messages += 1;
                                self.seq_no += 1;
                                self.buffer.push((self.curr_cap.first, self.seq_no, Some(e.length), (t, wid, x)));
                                // let (last_epoch, last_seq, last_length, (last_t, last_wid, last_x)) = self.buffer.pop()
                                    // .expect("non-empty buffer required");
                                // assert!(if let Schedule(e) = &last_x {e.start_stop == StartStop::Start}
                                        // else {false});
                                // Reposition received remote data message:
                                // 1. push the data message with previous' seq_no
                                // self.buffer.push((self.curr_cap.first, last_seq, Some(e.length), (t, wid, x)));
                                // 2. push the schedule event
                                // self.buffer.push((last_epoch, self.seq_no, last_length, (last_t, last_wid, last_x)));
                            }
                        }
                        // remote data send events
                        Messages(e) if e.source != e.target => {
                            assert!(e.source == wid);
                            self.pag_messages += 1;
                            self.fuel -= 1;
                            self.seq_no += 1;
                            if self.tick_sys {
                                self.advance_cap(&t);
                            }
                            self.buffer.push((self.curr_cap.first, self.seq_no, Some(e.length), (t, wid, x)));
                        }
                        _ => {}
                    }
                    if self.fuel == 0 {
                        self.flush_buffer();
                    }
                }
            }
        }
    }
pub fn publish_batch ( & mut self , data : DataflowEvents ) { match data { // DataflowEvents::Differential(data) => { //     for (_t, _wid, x) in data.drain(..) { //         match &x { //             Merge(e) if e.complete.is_some() => { //                 let op_addr = self.op_id_to_op_addr.get(&e.operator).expect("op id in addr not found"); //                 // we need `ch_source` here, since we're not interested in the operator id //                 // provided by the `Batch` event, but where the batch event's records flow to //                 // (e.g., a join operator). //                 let ch_id = self.op_addr_to_ch_source.get(&op_addr).expect("op addr in chs not found"); //                 let counter = self.channel_records.entry(*ch_id).or_insert(0); //                 *counter += e.complete.unwrap(); //             } //             _ => {} //         } //     } // } DataflowEvents :: Timely ( data ) => { for ( t , wid , x ) in data . drain ( .. ) { self . overall_messages += 1 ; match & x { TimelyEvent :: Operates ( _ ) | TimelyEvent :: Channels ( _ ) | TimelyEvent :: Messages ( _ ) | TimelyEvent :: Schedule ( _ ) | TimelyEvent :: Text ( _ ) => { // println!("{:?}, {:?}: {:?}", std::time::SystemTime::now(), t, x); } , _ => { } } match & x { Text ( _ ) => self . tick_epoch ( ) , Operates ( e ) => { self . pag_messages += 1 ; self . fuel -= 1 ; self . seq_no += 1 ; // all operates events should happen in the initialization epoch, // i.e., before any Text event epoch markers have been observed assert ! ( self next_cap first 1 self curr_cap Default default ( ) ) ; self . op_id_to_op_addr . insert ( e . id , * e . addr . last ( ) . expect ( " " ) ) ; self . buffer . push ( ( self . curr_cap . first , self . seq_no , None , ( Default :: default ( ) , wid , x ) ) ) ; } Channels ( e ) => { let ids = self . op_addr_to_ch_target . entry ( e . target . 0 ) . or_insert ( Vec :: new ( ) ) ; ids . push ( e . id ) ; self . op_addr_to_ch_source . insert ( e . source . 0 , e . id ) ; } Schedule ( e ) => { self . pag_messages += 1 ; // extend buffer size by 1 to avoid breaking up repositioning of // schedule start events and consequent data messages if self . fuel > 1 || e . start_stop == StartStop :: Stop { self . fuel -= 1 ; } self . seq_no += 1 ; if self . tick_sys { self . advance_cap ( & t ) ; } // fetch length let length = if e . start_stop == StartStop :: Stop { let op_addr = self . op_id_to_op_addr . get ( & e . id ) . expect ( " " ) ; // TODO: refactor if let Some ( ch_ids ) = self . op_addr_to_ch_target . get ( & op_addr ) { let mut len = 0 ; for ch_id in ch_ids . iter ( ) { if let Some ( l ) = self . channel_records . remove ( ch_id ) { len += l ; } } if len > 0 { Some ( len ) } else { None } } else { // For Inputs, we won't find a corresponding channel // (by definition, no channels end at an input) None } } else { None } ; self . buffer . push ( ( self . curr_cap . first , self . seq_no , length , ( t , wid , x ) ) ) ; } // Remote progress events Progress ( e ) if e . is_send || e . source != wid => { self . pag_messages += 1 ; self . fuel -= 1 ; self . seq_no += 1 ; if self . tick_sys { self . advance_cap ( & t ) ; } self . buffer . push ( ( self . curr_cap . first , self . seq_no , None , ( t , wid , x ) ) ) ; } // Data receive events Messages ( e ) if e . is_send == false => { assert ! ( e target wid ) ; // A. update record counter let counter = self . channel_records . entry ( e . channel ) . or_insert ( 0 ) ; * counter += e . length ; // B. if remote message: add to pag events if e . source != e . target { self . pag_messages += 1 ; self . seq_no += 1 ; self . buffer . push ( ( self . curr_cap . first , self . seq_no , Some ( e . length ) , ( t , wid , x ) ) ) ; // let (last_epoch, last_seq, last_length, (last_t, last_wid, last_x)) = self.buffer.pop() // .expect("non-empty buffer required"); // assert!(if let Schedule(e) = &last_x {e.start_stop == StartStop::Start} // else {false}); // Reposition received remote data message: // 1. push the data message with previous' seq_no // self.buffer.push((self.curr_cap.first, last_seq, Some(e.length), (t, wid, x))); // 2. push the schedule event // self.buffer.push((last_epoch, self.seq_no, last_length, (last_t, last_wid, last_x))); } } // remote data send events Messages ( e ) if e . source != e . target => { assert ! ( e source wid ) ; self . pag_messages += 1 ; self . fuel -= 1 ; self . seq_no += 1 ; if self . tick_sys { self . advance_cap ( & t ) ; } self . buffer . push ( ( self . curr_cap . first , self . seq_no , Some ( e . length ) , ( t , wid , x ) ) ) ; } _ => { } } if self . fuel == 0 { self . flush_buffer ( ) ; } } } } }
Publishes a batch of logged events and advances the capability.
Publishes a batch of logged events and advances the capability .
&mut self
data
DataflowEvents
&mut self
data
DataflowEvents