fix(core): 修复 workload 缓冲/传感器索引的溢出与并发问题 (#534)

* fix(core): make workload buffer/sensor indexing overflow- and concurrency-safe

- generateOneRow used Math.abs(stepOffset*(deviceIndex+1)) % bufferSize;
  Math.abs(Long.MIN_VALUE) is negative, yielding a negative buffer index.
  Use Math.floorMod, which is always in [0, bufferSize).
- SingletonWorkDataWorkLoad cast curLoop before the modulo ((int)curLoop % n)
  so a curLoop past Integer.MAX_VALUE produced a wrong/negative device id;
  cast after the modulo instead.
- Derive the sensor cursor from the same reserved curLoop so the
  (device, sensor) pairing is atomic instead of using a separate counter
  that a concurrent caller could advance between reads.
Covered by WorkloadIndexTest and SingletonWorkDataWorkLoadConcurrencyTest.

* ci: trigger CI

* test(core): make SingletonWorkDataWorkLoadConcurrencyTest order-independent

GenerateDataWorkLoad.workloadValues is a static final array built once by
initWorkloadValues(), which returns null when config.hasWrite() (the first
field of OPERATION_PROPORTION) is 0. If a preceding test leaves
OPERATION_PROPORTION on a read-only value (e.g. OperationControllerTest
ends on "0:1:...") and the workload class is first loaded during this
test, workloadValues stays null and getOneBatch() NPEs on every thread.

Force a write workload in setUp (restored in tearDown) so the array is
populated whenever this test triggers the class load, regardless of order.

---------

Co-authored-by: spricoder <maybell_uteka@mail.com>

Author: SpriCoder

core/src/main/java/cn/edu/tsinghua/iot/benchmark/workload/GenerateDataWorkLoad.java

@@ -69,7 +69,9 @@ public long getBatchNumber() {
   protected List<Object> generateOneRow(int deviceIndex, int colIndex, long stepOffset)
       throws WorkloadException {
     List<Object> values = new ArrayList<>(config.getSENSOR_NUMBER());
-    int index = (int) (Math.abs(stepOffset * (deviceIndex + 1)) % config.getWORKLOAD_BUFFER_SIZE());
+    int index =
+        (int)
+            Math.floorMod(stepOffset * (deviceIndex + 1), (long) config.getWORKLOAD_BUFFER_SIZE());
     if (colIndex == -1) {
       for (int i = 0; i < config.getSENSOR_NUMBER(); i++) {
         values.add(workloadValues[i][index]);

core/src/main/java/cn/edu/tsinghua/iot/benchmark/workload/SingletonWorkDataWorkLoad.java

@@ -33,21 +33,28 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
 public class SingletonWorkDataWorkLoad extends GenerateDataWorkLoad {
   private static final Logger LOGGER = LoggerFactory.getLogger(SingletonWorkDataWorkLoad.class);
   private static final List<Sensor> SENSORS = Collections.synchronizedList(config.getSENSORS());
   private static SingletonWorkDataWorkLoad singletonWorkDataWorkLoad = null;
-  private static final AtomicInteger sensorIndex = new AtomicInteger();
   private final AtomicLong insertLoop = new AtomicLong(0);
+
+  /**
+   * The value insertLoop starts at; the sensor cursor is derived as (curLoop - insertStartIndex).
+   */
+  private final long insertStartIndex;
+
   private static final List<Integer> deviceIds = MetaUtil.sortDeviceId();
 
   private SingletonWorkDataWorkLoad() {
     if (config.isIS_OUT_OF_ORDER()) {
       long startIndex = (long) (config.getLOOP() * config.getOUT_OF_ORDER_RATIO());
       this.insertLoop.set(startIndex);
+      this.insertStartIndex = startIndex;
+    } else {
+      this.insertStartIndex = 0L;
     }
   }
 
@@ -79,15 +86,20 @@ public IBatch getOneBatch() throws WorkloadException {
       if (config.isIS_SENSOR_TS_ALIGNMENT()) {
         sensors = SENSORS;
       } else {
-        int sensorId = sensorIndex.getAndIncrement() % config.getSENSOR_NUMBER();
+        // Derive the sensor cursor from the SAME curLoop reserved above so the (device, sensor)
+        // pairing is atomic: a single insertLoop.getAndIncrement() fixes both. A separate counter
+        // could be incremented by a concurrent caller between the two reads, mis-pairing the device
+        // with another caller's sensor (defect #3). Single-threaded this yields the same sequence,
+        // because insertLoop and the old sensor counter advanced in lockstep on this branch.
+        int sensorId = (int) ((curLoop - insertStartIndex) % config.getSENSOR_NUMBER());
         batch.setColIndex(sensorId);
         sensors.add(SENSORS.get(sensorId));
       }
       DeviceSchema deviceSchema =
           new DeviceSchema(
-              MetaUtil.getDeviceId(deviceIds.get((int) curLoop % config.getDEVICE_NUMBER())),
+              MetaUtil.getDeviceId(deviceIds.get((int) (curLoop % config.getDEVICE_NUMBER()))),
               sensors,
-              MetaUtil.getTags(deviceIds.get((int) curLoop % config.getDEVICE_NUMBER())));
+              MetaUtil.getTags(deviceIds.get((int) (curLoop % config.getDEVICE_NUMBER()))));
       // create data of batch
       List<Record> records = new ArrayList<>();
       // Sometimes the current time is used as the timestamp of records, considering the requirement

core/src/test/java/cn/edu/tsinghua/iot/benchmark/workload/SingletonWorkDataWorkLoadConcurrencyTest.java

@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package cn.edu.tsinghua.iot.benchmark.workload;
+
+import cn.edu.tsinghua.iot.benchmark.BenchmarkTestBase;
+import cn.edu.tsinghua.iot.benchmark.conf.Config;
+import cn.edu.tsinghua.iot.benchmark.conf.ConfigDescriptor;
+import cn.edu.tsinghua.iot.benchmark.entity.Batch.IBatch;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Concurrency regression test for defect #3 — {@code SingletonWorkDataWorkLoad} pairing atomicity.
+ *
+ * <p>{@code getOneBatch()} draws the device cursor ({@code insertLoop}) and the sensor cursor from
+ * two <em>independent</em> atomic counters. Each increment is atomic on its own, but the pairing
+ * between them is not: under concurrency thread A can take device cursor {@code L} while thread B
+ * races ahead and takes the sensor cursor that should have paired with {@code L}, so the {@code
+ * (device, sensor)} mapping gets shuffled.
+ *
+ * <p>With {@code DEVICE_NUMBER == SENSOR_NUMBER} the correct mapping is a clean bijection: device
+ * {@code p} is always paired with the same sensor (regardless of the counter's starting offset). So
+ * the number of distinct {@code (device, colIndex)} pairs must equal the number of distinct devices
+ * seen. Under the bug a device gets paired with several sensors and the pair count exceeds the
+ * device count. This holds single-threaded for both the buggy and fixed code; only concurrency
+ * breaks it, which is exactly what this test exercises.
+ */
+public class SingletonWorkDataWorkLoadConcurrencyTest extends BenchmarkTestBase {
+
+  private static final Config config = ConfigDescriptor.getInstance().getConfig();
+
+  /** DEVICE_NUMBER == SENSOR_NUMBER so the correct device→sensor mapping is a bijection. */
+  private static final int SIZE = 200;
+
+  private static final int THREADS = 16;
+  private static final int CALLS_PER_THREAD = 600;
+
+  private int origDeviceNumber;
+  private boolean origSensorTsAlignment;
+  private int origDeviceNumPerWrite;
+  private int origBatchSize;
+  private boolean origOutOfOrder;
+  private int origDataClientNumber;
+  private int origSchemaClientNumber;
+  private String origOperationProportion;
+
+  @Before
+  public void setUp() {
+    origDeviceNumber = config.getDEVICE_NUMBER();
+    origSensorTsAlignment = config.isIS_SENSOR_TS_ALIGNMENT();
+    origDeviceNumPerWrite = config.getDEVICE_NUM_PER_WRITE();
+    origBatchSize = config.getBATCH_SIZE_PER_WRITE();
+    origOutOfOrder = config.isIS_OUT_OF_ORDER();
+    origDataClientNumber = config.getDATA_CLIENT_NUMBER();
+    origSchemaClientNumber = config.getSCHEMA_CLIENT_NUMBER();
+    origOperationProportion = config.getOPERATION_PROPORTION();
+
+    // Write must be enabled BEFORE the workload classes load: GenerateDataWorkLoad builds its
+    // static final workloadValues array once, in initWorkloadValues(), which returns null when
+    // config.hasWrite() (OPERATION_PROPORTION's first field) is 0. getOneBatch() then dereferences
+    // that null array -> NPE. A preceding test (e.g. OperationControllerTest) can leave
+    // OPERATION_PROPORTION on a read-only value, so we force a write workload here rather than
+    // relying on the shared singleton's current state.
+    config.setOPERATION_PROPORTION("1:0:0:0:0:0:0:0:0:0:0:0:0");
+    // SENSOR_NUMBER stays at its default (200) so the pre-populated SENSORS list is consistent.
+    config.setDEVICE_NUMBER(SIZE);
+    // Force the buggy else-branch that pairs a device with a single sensor cursor.
+    config.setIS_SENSOR_TS_ALIGNMENT(false);
+    // One device per batch → one colIndex per batch, the unit we collect.
+    config.setDEVICE_NUM_PER_WRITE(1);
+    // Keep each getOneBatch() cheap; we care about the (device, sensor) pairing, not the records.
+    config.setBATCH_SIZE_PER_WRITE(1);
+    config.setIS_OUT_OF_ORDER(false);
+    // Keep the singletons that DataWorkLoad eagerly builds consistent with the end-to-end test.
+    config.setDATA_CLIENT_NUMBER(1);
+    config.setSCHEMA_CLIENT_NUMBER(1);
+  }
+
+  @After
+  public void tearDown() {
+    config.setDEVICE_NUMBER(origDeviceNumber);
+    config.setIS_SENSOR_TS_ALIGNMENT(origSensorTsAlignment);
+    config.setDEVICE_NUM_PER_WRITE(origDeviceNumPerWrite);
+    config.setBATCH_SIZE_PER_WRITE(origBatchSize);
+    config.setIS_OUT_OF_ORDER(origOutOfOrder);
+    config.setDATA_CLIENT_NUMBER(origDataClientNumber);
+    config.setSCHEMA_CLIENT_NUMBER(origSchemaClientNumber);
+    config.setOPERATION_PROPORTION(origOperationProportion);
+  }
+
+  @Test
+  public void devicePairsWithExactlyOneSensorUnderConcurrency() throws Exception {
+    SingletonWorkDataWorkLoad workload = SingletonWorkDataWorkLoad.getInstance();
+
+    Set<String> distinctDevices = ConcurrentHashMap.newKeySet();
+    Set<String> distinctPairs = ConcurrentHashMap.newKeySet();
+    CopyOnWriteArrayList<Throwable> failures = new CopyOnWriteArrayList<>();
+
+    CyclicBarrier startLine = new CyclicBarrier(THREADS);
+    CountDownLatch done = new CountDownLatch(THREADS);
+    for (int t = 0; t < THREADS; t++) {
+      new Thread(
+              () -> {
+                try {
+                  startLine.await();
+                  for (int i = 0; i < CALLS_PER_THREAD; i++) {
+                    IBatch batch = workload.getOneBatch();
+                    String device = batch.getDeviceSchema().getDevice();
+                    int sensor = batch.getColIndex();
+                    distinctDevices.add(device);
+                    distinctPairs.add(device + "#" + sensor);
+                  }
+                } catch (Throwable e) {
+                  failures.add(e);
+                } finally {
+                  done.countDown();
+                }
+              })
+          .start();
+    }
+    done.await();
+
+    assertTrue("getOneBatch threw under concurrency: " + failures, failures.isEmpty());
+    // Sanity: enough calls actually hit a spread of devices, otherwise the assertion is vacuous.
+    assertTrue(
+        "expected to exercise many devices but only saw " + distinctDevices.size(),
+        distinctDevices.size() > 1);
+    assertEquals(
+        "each device must pair with exactly one sensor; more pairs than devices means the device "
+            + "and sensor cursors were mismatched by a concurrent caller (defect #3)",
+        distinctDevices.size(),
+        distinctPairs.size());
+  }
+}

core/src/test/java/cn/edu/tsinghua/iot/benchmark/workload/WorkloadIndexTest.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package cn.edu.tsinghua.iot.benchmark.workload;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Documents the integer-overflow fixes (issue #13) in workload index calculation: why {@code
+ * Math.floorMod} replaces {@code Math.abs(...) % n} in {@code GenerateDataWorkLoad.generateOneRow},
+ * and why the cast must happen after the modulo in {@code SingletonWorkDataWorkLoad.getOneBatch}.
+ * These assertions capture the JDK behaviour that makes the old expressions produce negative
+ * (out-of-bounds) indices and the new ones safe.
+ */
+public class WorkloadIndexTest {
+
+  @Test
+  public void testFloorModNeverNegativeOnLongOverflow() {
+    int bufferSize = 100;
+    // stepOffset * (deviceIndex + 1) can overflow to Long.MIN_VALUE
+    long overflowed = Long.MIN_VALUE;
+    // old approach was buggy: Math.abs(Long.MIN_VALUE) stays negative -> negative index
+    assertTrue("Math.abs(Long.MIN_VALUE) is still negative", Math.abs(overflowed) < 0);
+    // fixed approach: floorMod with a positive divisor is always in [0, bufferSize)
+    long index = Math.floorMod(overflowed, (long) bufferSize);
+    assertTrue("floorMod index out of range", index >= 0 && index < bufferSize);
+  }
+
+  @Test
+  public void testModuloBeforeCastNeverNegativeDeviceIndex() {
+    int deviceNumber = 50;
+    long curLoop = (long) Integer.MAX_VALUE + 100; // exceeds int range
+    // old approach was buggy: cast happens before modulo -> overflow to negative
+    assertTrue("cast-before-mod overflows to negative", ((int) curLoop) % deviceNumber < 0);
+    // fixed approach: modulo first keeps the index in [0, deviceNumber)
+    int index = (int) (curLoop % deviceNumber);
+    assertTrue("mod-before-cast index out of range", index >= 0 && index < deviceNumber);
+  }
+}