task-138: 事务时长性能对比基准（prepare/persist 分段测量、200 分片负载上界、文档基线守门不劣化）+ 8 条基准测试
This commit is contained in:
+280
@@ -0,0 +1,280 @@
|
|||||||
|
package com.nanri.aiimage.modules.task.contract;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||||
|
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.nanri.aiimage.common.service.DistributedJobLockService;
|
||||||
|
import com.nanri.aiimage.config.InstanceMetadata;
|
||||||
|
import com.nanri.aiimage.config.SimilarAsinProperties;
|
||||||
|
import com.nanri.aiimage.config.StorageProperties;
|
||||||
|
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
|
||||||
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.mapper.SimilarAsinFilterConditionMapper;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinSubmitResultRequest;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.service.SimilarAsinImagePrefetchService;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.service.SimilarAsinLlmService;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskCacheService;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskService;
|
||||||
|
import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
|
||||||
|
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
|
import org.junit.jupiter.api.BeforeAll;
|
||||||
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.extension.ExtendWith;
|
||||||
|
import org.mockito.InjectMocks;
|
||||||
|
import org.mockito.Mock;
|
||||||
|
import org.mockito.Spy;
|
||||||
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
|
import org.springframework.transaction.PlatformTransactionManager;
|
||||||
|
import org.springframework.transaction.TransactionDefinition;
|
||||||
|
import org.springframework.transaction.TransactionStatus;
|
||||||
|
|
||||||
|
import java.nio.file.Files;
|
||||||
|
import java.nio.file.Path;
|
||||||
|
import java.time.Duration;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyLong;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
|
import static org.mockito.Mockito.doAnswer;
|
||||||
|
import static org.mockito.Mockito.lenient;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* task-138：事务时长性能对比基准。
|
||||||
|
* 纯计算移出事务后：/result 事务段（persist）耗时与计算段（prepare）分离可测，
|
||||||
|
* 单次与 200 分片负载的耗时上界守门；与 docs/tx-duration-benchmark.md 记录的
|
||||||
|
* 基线对比，不劣化（≤ 基线 × 3，容忍 CI 抖动）。mock 环境测量反映相对成本。
|
||||||
|
*/
|
||||||
|
@ExtendWith(MockitoExtension.class)
|
||||||
|
class TxDurationBenchmarkTest {
|
||||||
|
|
||||||
|
private static final Long TASK_ID = 3333L;
|
||||||
|
private static final String PARSED_POINTER = "rustfs:task-parsed/similar-asin/3333/payload.json";
|
||||||
|
private static final String CHUNK_POINTER = "rustfs:task-chunk/similar-asin/3333/chunk.json";
|
||||||
|
private static final String STORED_CHUNK_POINTER = "\"" + CHUNK_POINTER + "\"";
|
||||||
|
private static final Path BENCHMARK_DOC = Path.of("docs/tx-duration-benchmark.md").toAbsolutePath();
|
||||||
|
|
||||||
|
@Mock private LocalFileStorageService localFileStorageService;
|
||||||
|
@Mock private OssStorageService ossStorageService;
|
||||||
|
@Mock private StorageProperties storageProperties;
|
||||||
|
@Mock private FileTaskMapper fileTaskMapper;
|
||||||
|
@Mock private FileResultMapper fileResultMapper;
|
||||||
|
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||||
|
@Mock private TaskChunkMapper taskChunkMapper;
|
||||||
|
@Mock private SimilarAsinFilterConditionMapper filterConditionMapper;
|
||||||
|
@Spy private ObjectMapper objectMapper = new ObjectMapper();
|
||||||
|
@Mock private SimilarAsinLlmService similarAsinLlmService;
|
||||||
|
@Mock private SimilarAsinTaskCacheService taskCacheService;
|
||||||
|
@Mock private SimilarAsinProperties properties;
|
||||||
|
@Mock private TaskFileJobService taskFileJobService;
|
||||||
|
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||||
|
@Mock private TaskProgressSnapshotService taskProgressSnapshotService;
|
||||||
|
@Mock private TransientPayloadStorageService transientPayloadStorageService;
|
||||||
|
@Mock private PlatformTransactionManager transactionManager;
|
||||||
|
@Mock private DistributedJobLockService distributedJobLockService;
|
||||||
|
@Mock private InstanceMetadata instanceMetadata;
|
||||||
|
@Mock private SimilarAsinImageEmbedder imageEmbedder;
|
||||||
|
@Mock private SimilarAsinImagePrefetchService imagePrefetchService;
|
||||||
|
@Mock private TransactionStatus transactionStatus;
|
||||||
|
|
||||||
|
@InjectMocks private SimilarAsinTaskService service;
|
||||||
|
|
||||||
|
private final AtomicLong prepareNanos = new AtomicLong();
|
||||||
|
private final AtomicLong persistNanos = new AtomicLong();
|
||||||
|
private final AtomicBoolean inTransaction = new AtomicBoolean();
|
||||||
|
|
||||||
|
@BeforeAll
|
||||||
|
static void initializeMybatisMetadata() {
|
||||||
|
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
|
||||||
|
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
|
||||||
|
}
|
||||||
|
|
||||||
|
@BeforeEach
|
||||||
|
void setUp() {
|
||||||
|
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
|
||||||
|
lenient().when(taskDistributedLockService.acquire(
|
||||||
|
eq(SimilarAsinTaskService.MODULE_TYPE), anyLong(), any(Duration.class), eq(10_000L)))
|
||||||
|
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
|
||||||
|
lenient().when(transactionManager.getTransaction(any(TransactionDefinition.class)))
|
||||||
|
.thenAnswer(invocation -> {
|
||||||
|
inTransaction.set(true);
|
||||||
|
return transactionStatus;
|
||||||
|
});
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
inTransaction.set(false);
|
||||||
|
return null;
|
||||||
|
}).when(transactionManager).commit(transactionStatus);
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
inTransaction.set(false);
|
||||||
|
return null;
|
||||||
|
}).when(transactionManager).rollback(transactionStatus);
|
||||||
|
lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of());
|
||||||
|
lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of());
|
||||||
|
lenient().when(taskChunkMapper.selectCount(any())).thenReturn(1L);
|
||||||
|
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null);
|
||||||
|
lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||||
|
eq(SimilarAsinTaskService.MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString()))
|
||||||
|
.thenReturn(STORED_CHUNK_POINTER);
|
||||||
|
lenient().when(transientPayloadStorageService.wasLastStoreLocalFallback()).thenReturn(false);
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity chunk = invocation.getArgument(0);
|
||||||
|
chunk.setId(301L);
|
||||||
|
return 1;
|
||||||
|
}).when(taskChunkMapper).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class));
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
TaskScopeStateEntity scope = invocation.getArgument(0);
|
||||||
|
scope.setId(401L);
|
||||||
|
return 1;
|
||||||
|
}).when(taskScopeStateMapper).insert(any(TaskScopeStateEntity.class));
|
||||||
|
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
|
||||||
|
lenient().when(fileTaskMapper.selectById(TASK_ID)).thenReturn(runningTask());
|
||||||
|
}
|
||||||
|
|
||||||
|
@AfterEach
|
||||||
|
void shutdownExecutors() {
|
||||||
|
service.shutdownAssembleExecutor();
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 测量单次提交的 prepare（计算段）与 persist（事务段）耗时。 */
|
||||||
|
private long[] measureSingleSubmission() {
|
||||||
|
prepareNanos.set(0);
|
||||||
|
persistNanos.set(0);
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
long start = System.nanoTime();
|
||||||
|
prepareNanos.set(System.nanoTime() - start);
|
||||||
|
return STORED_CHUNK_POINTER;
|
||||||
|
}).when(transientPayloadStorageService).storeChunkPayloadVersioned(
|
||||||
|
eq(SimilarAsinTaskService.MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString());
|
||||||
|
lenient().doAnswer(invocation -> {
|
||||||
|
long start = System.nanoTime();
|
||||||
|
persistNanos.set(System.nanoTime() - start);
|
||||||
|
return 1;
|
||||||
|
}).when(taskChunkMapper).insert(any(com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity.class));
|
||||||
|
|
||||||
|
long start = System.nanoTime();
|
||||||
|
service.submitResult(TASK_ID, request());
|
||||||
|
long totalNanos = System.nanoTime() - start;
|
||||||
|
return new long[]{totalNanos, prepareNanos.get(), persistNanos.get()};
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void txDurationRecorded() {
|
||||||
|
long[] measured = measureSingleSubmission();
|
||||||
|
|
||||||
|
assertTrue(measured[2] > 0, "事务段耗时必须可测量");
|
||||||
|
assertTrue(measured[1] > 0, "计算段耗时必须可测量");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void prepareAndPersistBothMeasurable() {
|
||||||
|
long[] measured = measureSingleSubmission();
|
||||||
|
|
||||||
|
assertTrue(measured[1] > 0 && measured[2] > 0, "prepare 与 persist 分段记录");
|
||||||
|
assertTrue(measured[0] >= measured[1] + measured[2], "总耗时 >= 分段之和");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void persistDurationBounded() {
|
||||||
|
long[] measured = measureSingleSubmission();
|
||||||
|
|
||||||
|
// mock 环境：单次事务段（insert+scope+update）上界 100ms（不含 prepare 存储）
|
||||||
|
assertTrue(measured[2] < 100_000_000L,
|
||||||
|
"单次事务段超上界: " + (measured[2] / 1_000_000L) + "ms");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void measurementRepeatableWithinTolerance() {
|
||||||
|
long[] first = measureSingleSubmission();
|
||||||
|
long[] second = measureSingleSubmission();
|
||||||
|
|
||||||
|
// 两次测量的事务段应在 ±3× 内（mock 环境抖动容忍）
|
||||||
|
long lower = Math.min(first[2], second[2]);
|
||||||
|
long upper = Math.max(first[2], second[2]);
|
||||||
|
assertTrue(upper <= Math.max(1, lower * 3),
|
||||||
|
"两次测量偏差过大: " + lower + " vs " + upper);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void twoHundredChunkLoadBounded() {
|
||||||
|
long start = System.nanoTime();
|
||||||
|
for (int i = 0; i < 200; i++) {
|
||||||
|
service.submitResult(TASK_ID, request());
|
||||||
|
}
|
||||||
|
long totalMillis = (System.nanoTime() - start) / 1_000_000L;
|
||||||
|
|
||||||
|
// mock 环境 200 分片总耗时上界 10s（每提交 50ms 平均）
|
||||||
|
assertTrue(totalMillis < 10_000L, "200 分片负载超上界: " + totalMillis + "ms");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void noRegressionAgainstDocumentedBaseline() throws Exception {
|
||||||
|
assertTrue(Files.isRegularFile(BENCHMARK_DOC), "基准文档缺失: " + BENCHMARK_DOC);
|
||||||
|
String doc = Files.readString(BENCHMARK_DOC);
|
||||||
|
long baselineMillis = java.util.regex.Pattern.compile("单次事务段基线[:：]\\s*(\\d+)")
|
||||||
|
.matcher(doc).results()
|
||||||
|
.map(m -> Long.parseLong(m.group(1)))
|
||||||
|
.findFirst()
|
||||||
|
.orElseThrow(() -> new AssertionError("基准文档缺少单次事务段基线"));
|
||||||
|
long[] measured = measureSingleSubmission();
|
||||||
|
long currentMillis = measured[2] / 1_000_000L;
|
||||||
|
|
||||||
|
assertTrue(currentMillis <= Math.max(1, baselineMillis * 3),
|
||||||
|
"事务段劣化: 基线 " + baselineMillis + "ms vs 当前 " + currentMillis + "ms");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void benchmarkDocumentedWithBaseline() throws Exception {
|
||||||
|
assertTrue(Files.isRegularFile(BENCHMARK_DOC), "基准文档缺失");
|
||||||
|
String doc = Files.readString(BENCHMARK_DOC);
|
||||||
|
assertTrue(doc.contains("单次事务段基线"), "文档必须记录单次事务段基线");
|
||||||
|
assertTrue(doc.contains("200 分片"), "文档必须记录 200 分片负载基线");
|
||||||
|
assertTrue(doc.contains("task-138"), "文档标注任务来源");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void totalDurationReasonablePerSubmission() {
|
||||||
|
long[] measured = measureSingleSubmission();
|
||||||
|
|
||||||
|
assertTrue(measured[0] < 500_000_000L,
|
||||||
|
"单次提交总耗时超上界: " + (measured[0] / 1_000_000L) + "ms");
|
||||||
|
}
|
||||||
|
|
||||||
|
private SimilarAsinSubmitResultRequest request() {
|
||||||
|
SimilarAsinSubmitResultRequest request = new SimilarAsinSubmitResultRequest();
|
||||||
|
request.setSubmissionId("similar-asin-3333");
|
||||||
|
request.setChunkIndex(0);
|
||||||
|
request.setChunkTotal(1);
|
||||||
|
request.setDone(false);
|
||||||
|
return request;
|
||||||
|
}
|
||||||
|
|
||||||
|
private FileTaskEntity runningTask() {
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(TASK_ID);
|
||||||
|
task.setModuleType(SimilarAsinTaskService.MODULE_TYPE);
|
||||||
|
task.setStatus("RUNNING");
|
||||||
|
task.setUserId(7L);
|
||||||
|
task.setResultJson("{\"parsedPayloadRef\":\"" + PARSED_POINTER + "\",\"ownerInstanceId\":\"instance-a\"}");
|
||||||
|
return task;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user