Eliaaazzz commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3819854589
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark
handling. Elements pop
+ // in timestamp order, so the first selected one is the
earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer
above the
+ // watermark, must not consume the single key slot, or the
bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark
is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a
cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// hasBuildableDataLocked reports whether a key that isn't in progress heads
its
+// heap with data, or with a timer the watermark has reached. Callers hold
ss.mu.
+func (ss *stageState) hasBuildableDataLocked(watermark mtime.Time) bool {
+ for k, dnt := range ss.pendingByKeys {
Review Comment:
It stops at the first key it could build, so the full pass over
pendingByKeys only happens when the answer is no, which is exactly the case
that previously rescheduled empty bundles without bound; one scan to park the
stage replaces that loop. On the productive path it usually stops at the first
key that heads with data. The bundleReady call site is behind em.sawResidual,
so pipelines without a self checkpointing source never evaluate it, and the
builder call site is short circuited by timerCleared.
The cost that remains is one scan per unproductive build on a stateful stage
with many pending keys headed by future timers. An index of buildable keys
would remove it, but buildability depends on the current watermark, since a key
headed by a future timer becomes buildable when the watermark moves, so the
index needs its own bookkeeping on every watermark advance. I would leave that
until a profile shows this scan, and can file it as a follow up if you prefer.
--
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]
