nzw921rx commented on PR #12418:
URL: https://github.com/apache/seatunnel/pull/12418#issuecomment-5776652268
> Thanks both for keeping this moving.
>
> @Vivek1106-04, the full sweep is useful and it actually lines up with
Issue 1 from my earlier review: `triggerBodyMicros` (0 vs 500) barely moves the
numbers at any pipeline count, which is exactly what you'd expect if the
measured task almost never has to wait behind the periodic body. The knee at
500 pipelines looks like thread/CPU contention from ~1000 live threads in one
JVM rather than scheduling-queue pressure, so I'd read this sweep as "per-pool
wake-up latency under near-zero load," not yet as a load-bearing comparison for
#12165 - which is really the same thing @nzw921rx is getting at below.
>
> @nzw921rx, I think you're right that the harness measures thread wake-up
delay rather than queueing under the engine's real trigger pattern, and that
this limits how far the numbers can be trusted as a stand-in for the actual
scheduling model comparison - that's the same gap I flagged as Issue 1
(non-blocking) in my review.
>
> On routing through the real path: there's already a testing hook at that
boundary -
`CheckpointCoordinator#scheduleTriggerPendingCheckpoint(CheckpointType, long)`
(`CheckpointCoordinator.java:547-553`) is `@VisibleForTesting protected`. The
catch is that it schedules `tryTriggerPendingCheckpoint(checkpointType)`, which
on the happy path calls `createPendingCheckpoint(...)` and
`startTriggerPendingCheckpoint(...)` (`CheckpointCoordinator.java:816-824`) -
i.e. it builds a real `PendingCheckpoint`, writes to checkpoint storage, and
sends real RPC operations out to task locations. Driving the benchmark through
that method means constructing a real (or fully mocked, as
`CheckpointCoordinatorTest` does with Mockito) `CheckpointManager`, execution
plan and task locations, so the sample would then include storage and RPC noise
on top of scheduling delay, which is a different and heavier thing to measure
than what this PR set out to isolate.
>
> A cheaper middle ground that gets at the same question without that
machinery: my Issue 1 write-up already proposed arming both a trigger-like task
and a watchdog-like task per pool with a duty cycle long enough that both of a
pool's 2 workers can be occupied, so the measured task genuinely has to queue
behind them - closer to how `tryTriggerPendingCheckpoint`'s own re-scheduling
(`scheduleTriggerPendingCheckpoint(checkpointType, 500L)` when
`pendingCounter.get() > 0`, line ~816) can pile a second timer onto the same
pool. That would show whether pool width matters under real contention, still
as a standalone harness.
>
> Could you say which of these you're looking for: exercising
`tryTriggerPendingCheckpoint`'s full side effects (storage + RPC, higher
fidelity but a much bigger, more intrusive change), or a queuing-faithful
standalone harness along those lines? Either is a reasonable ask, but they're
different asks with different cost, and it'd help @Vivek1106-04 to know which
one unblocks this PR versus which is really scoped to the #12165 discussion.
@DanielLeens Thank you for your reply
Real RPC/IO can cause distortion in test scheduling, but existing custom
thread pools cannot test the engine's true methods. I provide the following
ideas:
Add new construction methods to eliminate existing RPC behaviors and mockups
such as IMAP
Design:
# Checkpoint Scheduling Benchmark Design
## Goal
Measure the scheduling delay of the real `CheckpointCoordinator` path
without starting SeaTunnel or Hazelcast nodes.
The benchmark must not override `tryTriggerPendingCheckpoint`.
## Measurement Boundary
```text
Start:
scheduleTriggerPendingCheckpoint()
Measured:
scheduler queueing
→ scheduler wake-up
→ shared dispatch, when applicable
→ tryTriggerPendingCheckpoint()
→ state checks
→ coordinator lock
→ submission of createPendingCheckpoint to executorService
End:
executorService.execute() receives the checkpoint creation task
```
The remaining checkpoint path continues outside the measured region:
```text
CheckpointID allocation
→ PendingCheckpoint creation
→ barrier RPC
→ task ACK
→ checkpoint storage
→ IMap state update
→ checkpoint completion
```
## CheckpointCoordinator Constructor
The production constructor delegates to a testing constructor that accepts
the real scheduler.
```java
public CheckpointCoordinator(
CheckpointManager manager,
CheckpointStorage checkpointStorage,
CheckpointConfig checkpointConfig,
long jobId,
CheckpointPlan plan,
CheckpointIDCounter checkpointIdCounter,
PipelineState pipelineState,
ExecutorService executorService,
IMap<Object, Object> runningJobStateIMap,
boolean isStartWithSavePoint,
CheckpointMonitorService checkpointMonitorService) {
this(
manager,
checkpointStorage,
checkpointConfig,
jobId,
plan,
checkpointIdCounter,
pipelineState,
executorService,
runningJobStateIMap,
isStartWithSavePoint,
checkpointMonitorService,
manager.leaseCheckpointScheduler(plan.getPipelineId()));
}
```
```java
@VisibleForTesting
protected CheckpointCoordinator(
CheckpointManager manager,
CheckpointStorage checkpointStorage,
CheckpointConfig checkpointConfig,
long jobId,
CheckpointPlan plan,
CheckpointIDCounter checkpointIdCounter,
PipelineState pipelineState,
ExecutorService executorService,
IMap<Object, Object> runningJobStateIMap,
boolean isStartWithSavePoint,
CheckpointMonitorService checkpointMonitorService,
PipelineCheckpointScheduler scheduler) {
// Keep the existing initialization unchanged.
this.executorService = executorService;
this.checkpointManager = manager;
this.checkpointStorage = checkpointStorage;
this.checkpointIdCounter = checkpointIdCounter;
this.runningJobStateIMap = runningJobStateIMap;
this.checkpointMonitorService = checkpointMonitorService;
this.scheduler = scheduler;
// Initialize the remaining existing fields normally.
}
```
On `dev`, the same testing constructor accepts the real per-pipeline
`ScheduledExecutorService`. The two revisions are compiled separately by the
benchmark workflow.
## Scheduling Probe
Use the constructor-injected `ExecutorService` as the measurement probe.
```java
final class ProbingExecutorService extends AbstractExecutorService {
private final ExecutorService delegate;
private final AtomicReference<CountDownLatch> nextProbe =
new AtomicReference<>();
ProbingExecutorService(ExecutorService delegate) {
this.delegate = delegate;
}
void arm(CountDownLatch probe) {
nextProbe.set(probe);
}
@Override
public void execute(Runnable command) {
CountDownLatch probe = nextProbe.getAndSet(null);
if (probe != null) {
probe.countDown();
}
delegate.execute(command);
}
// Delegate shutdown and lifecycle methods to delegate.
}
```
The probe is armed only after coordinator initialization and task readiness
have completed.
## Benchmark Coordinator
The benchmark subclass only exposes the protected scheduling method.
```java
final class BenchmarkCheckpointCoordinator
extends CheckpointCoordinator {
void scheduleForBenchmark() {
scheduleTriggerPendingCheckpoint(
CheckpointType.CHECKPOINT_TYPE,
0L);
}
}
```
It must not override:
```java
tryTriggerPendingCheckpoint(...)
```
Therefore all existing checks, locking and pending-checkpoint behavior
remain active.
## Fixture
The Fixture uses real scheduling implementations and local in-memory
replacements for external services.
```text
Scheduler real
CheckpointCoordinator real
tryTriggerPendingCheckpoint real
ExecutorService real delegate with a submission probe
CheckpointIDCounter in-memory AtomicLong
IMap ConcurrentHashMap-backed adapter
CheckpointStorage in-memory implementation
RPC successful local fake
Task ACK successful local callback
SeaTunnel nodes not started
Hazelcast cluster not started
```
The memory-backed `IMap` must support the operations used by
`CheckpointCoordinator`:
```java
get()
set()
compute()
remove()
```
Example:
```java
ConcurrentMap<Object, Object> state =
new ConcurrentHashMap<>();
IMap<Object, Object> imap = mock(IMap.class);
when(imap.get(any()))
.thenAnswer(invocation ->
state.get(invocation.getArgument(0)));
doAnswer(invocation -> {
state.put(
invocation.getArgument(0),
invocation.getArgument(1));
return null;
})
.when(imap)
.set(any(), any());
when(imap.compute(any(), any()))
.thenAnswer(invocation -> {
Object key = invocation.getArgument(0);
BiFunction<Object, Object, Object> function =
invocation.getArgument(1);
return state.compute(key, function);
});
when(imap.remove(any()))
.thenAnswer(invocation ->
state.remove(invocation.getArgument(0)));
```
The RPC fake returns a completed `InvocationFuture`. After a barrier is
sent, it feeds successful `TaskAcknowledgeOperation` instances back to the
coordinator so that `pendingCounter` returns to zero.
The Fixture must report all tasks as `READY_START` and wait for
initialization to finish before arming the scheduling probe.
## Benchmark Invocation
```java
CountDownLatch triggerSubmitted = new CountDownLatch(1);
probingExecutor.arm(triggerSubmitted);
coordinator.scheduleForBenchmark();
triggerSubmitted.await();
```
After the measured invocation, wait for the mocked ACK and checkpoint
completion outside the measured region:
```java
fixture.awaitCheckpointCompleted();
```
This prevents one sample from leaking pending checkpoint state into the next
sample.
## Comparison
The existing workflow compiles and runs each revision separately:
```text
A1 = dev
B1 = candidate PR
B2 = candidate PR
A2 = dev
```
`dev` uses its real per-pipeline scheduler. The candidate PR uses its real
shared scheduler. Both revisions keep the same Fixture behavior, workload,
parameters and measurement boundary.
@Vivek1106-04 We look forward to your reply
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
