Remove GuardedMessageEvent

Author: frankmcsherry

timely/src/dataflow/operators/generic/builder_rc.rs

@@ -21,7 +21,6 @@ use crate::dataflow::operators::generic::handles::{InputHandleCore, new_input_ha
 use crate::dataflow::operators::generic::operator_info::OperatorInfo;
 use crate::dataflow::operators::generic::builder_raw::OperatorShape;
 use crate::progress::operate::PortConnectivity;
-use crate::logging::TimelyLogger as Logger;
 
 use super::builder_raw::OperatorBuilder as OperatorBuilderRaw;
 
@@ -35,22 +34,19 @@ pub struct OperatorBuilder<G: Scope> {
     /// For each input, a shared list of summaries to each output.
     summaries: Vec<Rc<RefCell<PortConnectivity<<G::Timestamp as Timestamp>::Summary>>>>,
     produced: Vec<Rc<RefCell<ChangeBatch<G::Timestamp>>>>,
-    logging: Option<Logger>,
 }
 
 impl<G: Scope> OperatorBuilder<G> {
 
     /// Allocates a new generic operator builder from its containing scope.
     pub fn new(name: String, scope: G) -> Self {
-        let logging = scope.logging();
         OperatorBuilder {
             builder: OperatorBuilderRaw::new(name, scope),
             frontier: Vec::new(),
             consumed: Vec::new(),
             internal: Rc::new(RefCell::new(Vec::new())),
             summaries: Vec::new(),
             produced: Vec::new(),
-            logging,
         }
     }
 
@@ -90,7 +86,7 @@ impl<G: Scope> OperatorBuilder<G> {
         let shared_summary = Rc::new(RefCell::new(connection.into_iter().collect()));
         self.summaries.push(Rc::clone(&shared_summary));
 
-        new_input_handle(input, Rc::clone(&self.internal), shared_summary, self.logging.clone())
+        new_input_handle(input, Rc::clone(&self.internal), shared_summary)
     }
 
     /// Adds a new output to a generic operator builder, returning the `Push` implementor to use.

timely/src/dataflow/operators/generic/handles.rs

@@ -17,7 +17,6 @@ use crate::dataflow::channels::Message;
 use crate::communication::{Push, Pull};
 use crate::{Container, Data};
 use crate::container::{ContainerBuilder, CapacityContainerBuilder};
-use crate::logging::TimelyLogger as Logger;
 
 use crate::dataflow::operators::InputCapability;
 use crate::dataflow::operators::capability::CapabilityTrait;
@@ -31,7 +30,6 @@ pub struct InputHandleCore<T: Timestamp, C: Container, P: Pull<Message<T, C>>> {
     /// Each timestamp received through this input may only produce output timestamps
     /// greater or equal to the input timestamp subjected to at least one of these summaries.
     summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,
-    logging: Option<Logger>,
 }
 
 /// Handle to an operator's input stream, specialized to vectors.
@@ -82,13 +80,9 @@ impl<T: Timestamp, C: Container, P: Pull<Message<T, C>>> InputHandleCore<T, C, P
     /// ```
     #[inline]
     pub fn for_each<F: FnMut(InputCapability<T>, &mut C)>(&mut self, mut logic: F) {
-        let mut logging = self.logging.take();
         while let Some((cap, data)) = self.next() {
-            logging.as_mut().map(|l| l.log(crate::logging::GuardedMessageEvent { is_start: true }));
             logic(cap, data);
-            logging.as_mut().map(|l| l.log(crate::logging::GuardedMessageEvent { is_start: false }));
         }
-        self.logging = logging;
     }
 
 }
@@ -150,13 +144,11 @@ pub fn new_input_handle<T: Timestamp, C: Container, P: Pull<Message<T, C>>>(
     pull_counter: PullCounter<T, C, P>,
     internal: Rc<RefCell<Vec<Rc<RefCell<ChangeBatch<T>>>>>>,
     summaries: Rc<RefCell<PortConnectivity<T::Summary>>>,
-    logging: Option<Logger>
 ) -> InputHandleCore<T, C, P> {
     InputHandleCore {
         pull_counter,
         internal,
         summaries,
-        logging,
     }
 }