differential_dataflow/
logging.rs

1//! Loggers and logging events for differential dataflow.
2
3use columnar::Columnar;
4use serde::{Deserialize, Serialize};
5
6/// Container builder for differential log events.
7pub type DifferentialEventBuilder = timely::container::CapacityContainerBuilder<Vec<(std::time::Duration, DifferentialEvent)>>;
8
9/// Logger for differential dataflow events.
10pub type Logger = ::timely::logging_core::TypedLogger<DifferentialEventBuilder, DifferentialEvent>;
11
12/// Enables logging of differential dataflow events.
13pub fn enable<W>(worker: &mut timely::worker::Worker, writer: W) -> Option<Box<dyn std::any::Any+'static>>
14where
15    W: std::io::Write + 'static,
16{
17    worker.log_register().and_then(|mut log_register| {
18        let writer = ::timely::dataflow::operators::capture::EventWriter::new(writer);
19        let mut logger = ::timely::logging::BatchLogger::new(writer);
20        log_register.insert::<DifferentialEventBuilder,_>("differential/arrange", move |time, data| logger.publish_batch(time, data))
21    })
22}
23
24/// Possible different differential events.
25///
26/// The spine reports batches through `BatchEvent`, `MergeEvent`, and `DropEvent`, which balance:
27/// a consumer that adds one batch per `BatchEvent`, and removes one per `MergeEvent` with `complete`
28/// set and per `DropEvent`, counts the batches the trace holds. Batches without updates are counted,
29/// and report a length of zero.
30#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
31pub enum DifferentialEvent {
32    /// Batch creation.
33    Batch(BatchEvent),
34    /// Merge start and stop events.
35    Merge(MergeEvent),
36    /// Batch dropped when trace dropped.
37    Drop(DropEvent),
38    /// A merge failed to complete in time.
39    MergeShortfall(MergeShortfall),
40    /// Trace sharing event.
41    TraceShare(TraceShare),
42    /// Batcher size event
43    Batcher(BatcherEvent),
44}
45
46/// A batch introduced to a trace.
47///
48/// See [`DifferentialEvent`] for how batch, merge, and drop events balance.
49#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
50pub struct BatchEvent {
51    /// Operator identifier.
52    pub operator: usize,
53    /// Number of updates in the batch.
54    pub length: usize,
55}
56
57impl From<BatchEvent> for DifferentialEvent { fn from(e: BatchEvent) -> Self { DifferentialEvent::Batch(e) } }
58
59
60/// Either the start or end of a merge event.
61#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
62pub struct BatcherEvent {
63    /// Operator identifier.
64    pub operator: usize,
65    /// Change in records.
66    pub records_diff: isize,
67    /// Change in used size.
68    pub size_diff: isize,
69    /// Change in capacity.
70    pub capacity_diff: isize,
71    /// Change in number of allocations.
72    pub allocations_diff: isize,
73}
74
75impl From<BatcherEvent> for DifferentialEvent { fn from(e: BatcherEvent) -> Self { DifferentialEvent::Batcher(e) } }
76
77/// A batch dropped by a trace.
78///
79/// See [`DifferentialEvent`] for how batch, merge, and drop events balance.
80#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
81pub struct DropEvent {
82    /// Operator identifier.
83    pub operator: usize,
84    /// Number of updates in the batch.
85    pub length: usize,
86}
87
88impl From<DropEvent> for DifferentialEvent { fn from(e: DropEvent) -> Self { DifferentialEvent::Drop(e) } }
89
90/// Either the start or end of a merge of two batches into one.
91///
92/// See [`DifferentialEvent`] for how batch, merge, and drop events balance.
93#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
94pub struct MergeEvent {
95    /// Operator identifier.
96    pub operator: usize,
97    /// Which order of magnitude.
98    pub scale: usize,
99    /// Length of first trace.
100    pub length1: usize,
101    /// Length of second trace.
102    pub length2: usize,
103    /// None implies a start; otherwise the number of updates in the merged batch.
104    pub complete: Option<usize>,
105}
106
107impl From<MergeEvent> for DifferentialEvent { fn from(e: MergeEvent) -> Self { DifferentialEvent::Merge(e) } }
108
109/// A merge failed to complete in time.
110#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
111pub struct MergeShortfall {
112    /// Operator identifier.
113    pub operator: usize,
114    /// Which order of magnitude.
115    pub scale: usize,
116    /// By how much were we short.
117    pub shortfall: usize,
118}
119
120impl From<MergeShortfall> for DifferentialEvent { fn from(e: MergeShortfall) -> Self { DifferentialEvent::MergeShortfall(e) } }
121
122/// Either the start or end of a merge event.
123#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Serialize, Deserialize, Columnar)]
124pub struct TraceShare {
125    /// Operator identifier.
126    pub operator: usize,
127    /// Change in number of shares.
128    pub diff: isize,
129}
130
131impl From<TraceShare> for DifferentialEvent { fn from(e: TraceShare) -> Self { DifferentialEvent::TraceShare(e) } }
differential_dataflow/logging.rs

differential_dataflow/
logging.rs
