dbsp_nexmark: Fix event rate.

blp · blp · commit 77879667ecb1 · 2024-08-14T15:37:24.000-07:00
Nexmark query q15 ran very slowly as the number of events increased past
about 10M, causing runs of 100M to take multiple minutes when most of the
other queries ran much faster.  We traced this back to the
`first_event_rate` option, which controlled the number of events emitted
for each simulated second, with a default value of 10,000,000.  Because
q15 aggregates by day, this meant that all 100M generated events would be
in the same aggregation group.

We looked at what the Flink generator does.  It also has a
`first_event_rate` setting, which defaults to 10,000.  If we interpret
this as a rate per second (as Feldera interpreted this setting), then
this would still put all of the events in one day because 100,000,000 /
10,000 = 10,000, which is less than the 86,400 seconds in one day.
However, despite the name, the Flink setting is the interval between
events in microseconds, which means that its default was actually 100
events per second, or 8.64 million events/day, which is significantly less
than the 100 million events in the simulation.

This commit changes the Feldera Nexmark generator to default to the same
inter-event interval.  I thought that it was too confusing to retain this
setting under the same name, so I changed its name to `event_interval`,
denominated in milliseconds.

This speeds up q15 for me from about 180s to about 36s, a 5x speedup.

Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/benchmark/run-nexmark.sh b/benchmark/run-nexmark.sh
@@ -462,7 +462,6 @@ case $runner:$language in
 	esac
 	for query in $queries; do
 	    run_log $CARGO bench --bench nexmark -- \
-		--first-event-rate=10000000 \
 		--max-events=$events \
 		--cpu-cores $cores \
 		--num-event-generators $cores \
diff --git a/crates/nexmark/src/config.rs b/crates/nexmark/src/config.rs
@@ -91,9 +91,9 @@ pub struct GeneratorOptions {
     #[clap(long, default_value = "46", env = "NEXMARK_BID_PROPORTION")]
     pub bid_proportion: usize,
 
-    /// Initial overall event rate (per second).
-    #[clap(long, default_value = "10000000", env = "NEXMARK_FIRST_EVENT_RATE")]
-    pub first_event_rate: usize,
+    /// Time between events, in milliseconds.
+    #[clap(long, default_value = "10", env = "NEXMARK_EVENT_INTERVAL")]
+    pub event_interval: usize,
 
     /// Ratio of bids to 'hot' auctions compared to all other auctions.
     #[clap(long, default_value = "2", env = "NEXMARK_HOT_AUCTION_RATIO")]
@@ -170,7 +170,7 @@ impl Default for GeneratorOptions {
             avg_bid_byte_size: 100,
             avg_person_byte_size: 200,
             bid_proportion: 46,
-            first_event_rate: 10_000_000,
+            event_interval: 10,
             hot_auction_ratio: 2,
             hot_bidders_ratio: 4,
             hot_sellers_ratio: 4,
diff --git a/crates/nexmark/src/generator/config.rs b/crates/nexmark/src/generator/config.rs
@@ -35,11 +35,6 @@ pub struct Config {
     /// generating different event numbers (using `first_event_number +
     /// events_count_so_far*num_generators`, for example).
     pub first_event_number: usize,
-
-    /// Delay between events, in microseconds. If the array has more than one
-    /// entry then the rate is changed every {@link #stepLengthSec}, and wraps
-    /// around.
-    pub inter_event_delay_us: [f64; 1],
 }
 
 /// Implementation of config methods based on the Java implementation at
@@ -51,16 +46,6 @@ impl Config {
         first_event_id: u64,
         first_event_number: usize,
     ) -> Config {
-        // The inter_event_delay is calculated as the number (or fraction) of
-        // micro seconds that pass between each event. Unlike the Java
-        // implementation, this is not dependent on the number of generators
-        // because each generator here returns interleaved events. For example,
-        // with 3 generators, the first generator emits events based on the
-        // event numbers 0, 3 and 6 etc., where as the Java implementation uses
-        // 0, 1 and 2 locally for each generator and so adds a factor of
-        // num_generators.
-        let inter_event_delay = 1_000_000.0 / (options.first_event_rate as f64);
-
         // Original Java implementation says:
         // "Scale maximum down to avoid overflow in getEstimatedSizeBytes."
         // but including to ensure similar behavior.
@@ -84,7 +69,6 @@ impl Config {
             first_event_id,
             max_events,
             first_event_number,
-            inter_event_delay_us: [inter_event_delay],
         }
     }
 
@@ -115,7 +99,7 @@ impl Config {
     // What timestamp should the event with `eventNumber` have for this
     // generator?
     pub fn timestamp_for_event(&self, event_number: u64) -> u64 {
-        self.base_time + (self.inter_event_delay_us[0] * event_number as f64) as u64 / 1000
+        self.base_time + self.options.event_interval as u64 * event_number
     }
 }
 
@@ -264,13 +248,11 @@ pub mod tests {
         assert_eq!(config.next_adjusted_event_number(num_events), expected);
     }
 
-    // With the default first event rate of 10_000_000 events per second there
-    // is 1_000_000 µs/s / 10_000_000 events/s = 0.1µs / event, so the timestamp
-    // should increase by 0.1µs, or 0.0001ms for each event.
+    // The default event interval is 10 ms.
     #[rstest]
-    #[case(10_000, 1)]
-    #[case(20_000, 2)]
-    #[case(50_000, 5)]
+    #[case(1, 10)]
+    #[case(2, 20)]
+    #[case(5, 50)]
     fn test_timestamp_for_event_single_generator(#[case] event_number: u64, #[case] expected: u64) {
         assert_eq!(
             Config::default().timestamp_for_event(event_number),
diff --git a/crates/nexmark/src/generator/mod.rs b/crates/nexmark/src/generator/mod.rs
@@ -295,7 +295,7 @@ pub mod tests {
                 "got: {:?}, want: Event::NewAuction(_)",
                 next_event.event
             );
-            assert_eq!(next_event.event_timestamp, event_num / 10);
+            assert_eq!(next_event.event_timestamp, event_num * 10);
         }
 
         // And the rest of the events in the first epoch are bids.
diff --git a/crates/nexmark/src/lib.rs b/crates/nexmark/src/lib.rs
@@ -297,7 +297,6 @@ pub mod tests {
     #[test]
     fn test_source_with_multiple_generators() {
         let options = GeneratorOptions {
-            first_event_rate: 1_000_000,
             num_event_generators: 3,
             max_events: 10,
             ..GeneratorOptions::default()
diff --git a/scripts/bench.bash b/scripts/bench.bash
@@ -21,18 +21,16 @@ rm -rf ${RESULTS_DIR}
 mkdir -p ${RESULTS_DIR}
 
 # Run nexmark benchmark
-EVENT_RATE=10000000
 MAX_EVENTS=100000000
 GENERATORS=8
 CORES=6
 if [ "$SMOKE" != "" ]; then
-  EVENT_RATE=10000000
   MAX_EVENTS=1000000
 fi
 
 FILES=( "q0" "q1" "q2" "q3" "q4" "q5" "q6" "q7" "q8" "q9" "q12" "q13" "q14" "q15" "q16" "q17" "q18" "q19" "q20" "q21" "q22" )
 for FILE in "${FILES[@]}"
-  do cargo bench --bench nexmark -- --first-event-rate=${EVENT_RATE} --max-events=${MAX_EVENTS} --cpu-cores ${CORES}  --num-event-generators ${GENERATORS} --source-buffer-size 10000 --input-batch-size 40000 --csv ${NEXMARK_CSV_FILE} --query $FILE
+  do cargo bench --bench nexmark -- --max-events=${MAX_EVENTS} --cpu-cores ${CORES}  --num-event-generators ${GENERATORS} --source-buffer-size 10000 --input-batch-size 40000 --csv ${NEXMARK_CSV_FILE} --query $FILE
 done
 mkdir -p ${NEXMARK_RESULTS_DIR}
 mv crates/nexmark/${NEXMARK_CSV_FILE} $NEXMARK_RESULTS_DIR
@@ -90,14 +88,12 @@ fi
 #mv crates/dbsp/${LDBC_CSV_FILE} ${LDBC_RESULTS_DIR}
 
 # Run nexmark benchmark with persistence
-EVENT_RATE=5000000
 MAX_EVENTS=3000000
 CORES=1
 if [ "$SMOKE" != "" ]; then
-  EVENT_RATE=5000000
   MAX_EVENTS=100000
 fi
-cargo bench --bench nexmark -- --first-event-rate=${EVENT_RATE} --max-events=${MAX_EVENTS} --cpu-cores ${CORES} --num-event-generators 6 --source-buffer-size 10000 --input-batch-size 40000 --csv ${NEXMARK_DRAM_CSV_FILE}
+cargo bench --bench nexmark -- --max-events=${MAX_EVENTS} --cpu-cores ${CORES} --num-event-generators 6 --source-buffer-size 10000 --input-batch-size 40000 --csv ${NEXMARK_DRAM_CSV_FILE}
 mv crates/nexmark/${NEXMARK_DRAM_CSV_FILE} $NEXMARK_RESULTS_DIR 
-#cargo bench --bench nexmark --features persistence -- --first-event-rate=${EVENT_RATE} --max-events=${MAX_EVENTS} --cpu-cores ${CORES} --num-event-generators 6 --source-buffer-size 10000 --input-batch-size 40000 --csv ${NEXMARK_PERSISTENCE_CSV_FILE}
+#cargo bench --bench nexmark --features persistence -- --max-events=${MAX_EVENTS} --cpu-cores ${CORES} --num-event-generators 6 --source-buffer-size 10000 --input-batch-size 40000 --csv ${NEXMARK_PERSISTENCE_CSV_FILE}
 #mv crates/nexmark/${NEXMARK_PERSISTENCE_CSV_FILE} $NEXMARK_RESULTS_DIR 

Original file line number	Diff line number	Diff line change
`@@ -295,7 +295,7 @@ pub mod tests {`
`295`	`295`	`"got: {:?}, want: Event::NewAuction(_)",`
`296`	`296`	`next_event.event`
`297`	`297`	`);`
`298`		`- assert_eq!(next_event.event_timestamp, event_num / 10);`
	`298`	`+ assert_eq!(next_event.event_timestamp, event_num * 10);`
`299`	`299`	`}`
`300`	`300`
`301`	`301`	`// And the rest of the events in the first epoch are bids.`