fix high cpu idle usage in _buffer and _forget (#7904)

GitOrigin-RevId: 822f7d7fae87fca8629169fa1f5c62d0f0720e79

CHANGELOG.md

@@ -5,6 +5,9 @@ All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ## [Unreleased]
 
+### Fixed
+- temporal behaviors in temporal operators (`windowby`, `interval_join`) now consume no CPU when no data passes through them.
+
 ## [0.16.2] - 2024-12-19
 
 ### Added

external/timely-dataflow/communication/src/allocator/counters.rs

@@ -34,7 +34,7 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
         //     if self.count != 0 {
         //         self.events
         //             .borrow_mut()
-        //             .push_back(self.index);
+        //             .push(self.index);
         //         self.count = 0;
         //     }
         // }
@@ -43,10 +43,19 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
         // }
         // TODO: Version above is less chatty, but can be a bit late in
         //       moving information along. Better, but needs cooperation.
-        self.events
-            .borrow_mut()
-            .push(self.index);
 
+        // [Pathway extension]: do not push index if the message is none.
+        // Without it, the program never parks if Variable is used. Then None messages
+        // circulate and in every call to `step_or_park` some operators are activated.
+        // This Pusher is only used by Thread allocator and apparently ignoring None
+        // messages here doesn't block the computation. (The Thread allocator is used
+        // if only one worker is used and always for Pipeline pact).
+        if element.is_some() {
+            self.events
+                .borrow_mut()
+                .push(self.index);
+        }
+
         self.pusher.push(element)
     }
 }