stevedlawrence commented on code in PR #1717:
URL: https://github.com/apache/daffodil/pull/1717#discussion_r3830239952
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/DPath.scala:
##########
@@ -252,6 +254,24 @@ final class RuntimeExpressionDPath[T <: AnyRef](
whereBlockedInfo.block(ve.qname, ve.context, 0, ve)
case noLength: InfosetLengthUnknownException =>
whereBlockedInfo.block(noLength.diElement, noLength.erd, 0, noLength)
+ // Register a targeted wake-up alongside the periodic-sweep blocking
+ // above: once this element's length becomes computable (see
+ // CaptureEndOf{Content,Value}LengthUnparser), this suspension is
+ // retried directly instead of waiting for SuspensionTracker's next
+ // sweep. registerLengthStateWaiter (not registerWaiter directly)
+ // also deregisters from whichever LengthState this suspension was
+ // previously registered against, in case an earlier retry blocked
+ // on a different element's length - see Suspension.scala.
+ noLength match {
+ case _: InfosetContentLengthUnknownException =>
+
whereBlockedInfo.registerLengthStateWaiter(noLength.diElement.contentLength)
+ case _: InfosetValueLengthUnknownException =>
+
whereBlockedInfo.registerLengthStateWaiter(noLength.diElement.valueLength)
+ }
Review Comment:
I think if we make `lengthState` a val in InfosetLengthUnknownException we
can avoid the match/case and simplify this to:
```scala
whereBlockedInfo.registerLengthStateWaiter(noLength.lengthState)
```
I don't think there's any harm in making lengthState accessible and it
simplifies our code.
##########
daffodil-core/src/main/scala/org/apache/daffodil/lib/util/MStack.scala:
##########
@@ -167,9 +167,9 @@ private[util] final class MStackOfAnyRef private ()
extends MStack[AnyRef]((n: Int) => new Array[AnyRef](n),
null.asInstanceOf[AnyRef])
object MStackOfAnyRef {
- def apply() = {
+ def apply(initialSize: Int = 32) = {
Review Comment:
Should the other MStacks, e.g. `MStackOfBoolean`, `MStackofInt` have this
initialize size field in the apply method too. I guess we don't currently use
them, but they should be available.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/SuspendableExpression.scala:
##########
@@ -32,12 +32,35 @@ import
org.apache.daffodil.runtime1.processors.unparsers.UState
* dfdl:setVariable expressions (which variables are in-turn used by
* dfdl:outputValueCalc.
*/
+object SuspendableExpression {
+
+ /**
+ * A plain value/variable read can resolve as soon as the referenced
+ * value is known, regardless of whether anything has been written yet
+ * (the forward-reference case this speeds up during build). A
Review Comment:
I dont' think I understand what "(the forward-reference case this speeds up
during build)" means
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/infoset/InfosetImpl.scala:
##########
@@ -572,6 +573,44 @@ sealed abstract class LengthState(ie: DIElement) {
var maybeEndPos0bInBits: MaybeULong = MaybeULong.Nope
var maybeComputedLengthInBits: MaybeULong = MaybeULong.Nope
+ /**
+ * Suspensions blocked on this element's length being unknown
+ * (dfdl:contentLength/dfdl:valueLength), registered via registerWaiter
+ * and retried directly via notifyWaiters once this length becomes
+ * computable, instead of waiting for SuspensionTracker's next periodic
+ * sweep. Purely an optimization alongside that sweep, not a
+ * replacement: a suspension still sits in SuspensionTracker's own
+ * queues too, so requireFinal()'s deadlock detection is unaffected.
+ *
+ * Deduplicated by reference identity - a suspension that re-blocks
+ * here across multiple sweeps re-registers each time, and without
+ * dedup this list would grow without bound.
+ */
+ private var waiters: List[Suspension] = Nil
Review Comment:
Can we rename 'waiters'? Maybe `lengthSuspensions` to make it clear this is
only expected to contain suspensions specifically waiting for this elements
length? I imagine if we ever want additional kinds of waiters we'll want
separate variables so that we can notify them at the right time--different
types of suspensions likely will have different notification triggers.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/DPath.scala:
##########
@@ -252,6 +254,24 @@ final class RuntimeExpressionDPath[T <: AnyRef](
whereBlockedInfo.block(ve.qname, ve.context, 0, ve)
case noLength: InfosetLengthUnknownException =>
whereBlockedInfo.block(noLength.diElement, noLength.erd, 0, noLength)
+ // Register a targeted wake-up alongside the periodic-sweep blocking
+ // above: once this element's length becomes computable (see
Review Comment:
I'm not sure what "periodic-sweep blocking above" is referencing.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -28,7 +28,59 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
private val suspensionsYoung = new Queue[Suspension]
private val suspensionsOld = new Queue[Suspension]
- def suspensions: Seq[Suspension] = suspensionsYoung.toSeq ++
suspensionsOld.toSeq
+ /**
+ * Suspensions unable to make progress until something external
+ * changes (a targeted wake-up firing, or real bytes getting written);
+ * moved out of suspensionsYoung/suspensionsOld so the periodic sweep
+ * (evalSuspensionsThrottled) doesn't keep re-visiting them every tick -
+ * even a cheap per-item skip costs real time at scale.
+ *
+ * Only drained by foldParkedIntoOld, from the two "must attempt
+ * everything" methods below (evalSuspensionsUnthrottled, requireFinal);
+ * the targeted wake-up itself (LengthState.notifyWaiters) retries a
+ * suspension directly regardless of which bucket it's in, so parking
+ * never delays that path.
+ */
+ private val suspensionsParked = new Queue[Suspension]
Review Comment:
Suggests this just wants to be a mutable.HashSet. Suspensions are added to
this when blocked by length state, and when a LengthState changes we can remove
it from the HashSet and add it to the young queue. Depending on how good our
lenght state logic can be to know if it will be blocked or not it could
potentially get blocked again and end up readded, but hopefully not.
This ensures we only ever evaluate this parked suspensions when we really do
think they have a chance of getting evaluated.
##########
daffodil-core/src/main/scala/org/apache/daffodil/lib/util/MStack.scala:
##########
@@ -190,9 +190,14 @@ protected abstract class MStack[@specialized T]
private[util] (
private var index = 0
private var table: Array[T] = null
- def init(): Unit = {
+ /**
+ * initialSize lets a caller that knows its element count up front (e.g.
+ * cloning another MStack of known depth, see UState.cloneForSuspension)
+ * avoid the default 32-slot allocation when that's more than needed.
+ */
Review Comment:
I believe the init function is not a function that users should call. So
this user-related documentation probably doesn't belong here. I would suggest
whatever documentation we do have for initialSize wants to be on the actual
user visible apply methods. I'd also suggest not provided examples referencing
other code--that code could change and then this documentation is wrong and
potentially confusing. If we do want to provide examples, we can include
example usage in this comment. That said, initialSize is pretty self
explanatory, so I'm not sure an example or even documentation provides much
value.
Also, I wonder if we can get rid of the init function entirely, e.g. just do
this in the constructor:
```scala
private var table: Array[T] = arrayAllocator(initialSize)
```
Currently all the MStack* constructors are priate so you have to use the
object apply method, and I think all the objects immediately call init after
allocating an MStack instance. So the init function doesn't seem to really do
anything special. Maybe we should just get rid of it if possible and simplify
the code.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -228,9 +277,51 @@ trait Suspension extends Serializable {
final def isMakingProgress = isMakingProgress_
+ /**
+ * True exactly when this suspension is blocked on
+ * InfosetLengthUnknownException with a targeted wake-up already
+ * registered against the relevant LengthState (set only from
+ * DPath.scala's InfosetLengthUnknownException catch clause, the one
+ * place that calls markWaitingOnLengthState). SuspensionTracker's
+ * periodic sweep uses this to skip re-running doTask until that
+ * wake-up fires. Not inferred from block()'s exc type
+ * (SuspendableOperation.scala also passes a RetryableException that
+ * could structurally match without a wake-up registered) - only the
+ * actual registration site sets this, so the two can't drift apart.
+ */
+ private var isWaitingOnLengthState_ : Boolean = false
+
+ final def markWaitingOnLengthState(): Unit = {
+ isWaitingOnLengthState_ = true
+ }
+
+ final def isWaitingOnLengthState: Boolean = isWaitingOnLengthState_
+
+ /**
+ * Which LengthState (if any) this suspension is registered with as a
+ * waiter. A suspension's dependency can shift between retries (e.g. a
+ * length expression whose which-element branch changes), so it must be
+ * deregistered from the old LengthState before registering a new one,
+ * or the old one's notifyWaiters() would retry it pointlessly.
+ */
+ private var maybeRegisteredLengthState: Maybe[LengthState] = Nope
+
+ final def registerLengthStateWaiter(ls: LengthState): Unit = {
+ if (maybeRegisteredLengthState.isDefined &&
(maybeRegisteredLengthState.get ne ls))
+ maybeRegisteredLengthState.get.removeWaiter(this)
+ maybeRegisteredLengthState = One(ls)
+ ls.registerWaiter(this)
+ }
+
final def block(nodeOrVar: AnyRef, info: AnyRef, index: Long, exc: AnyRef):
Unit = {
Logger.log.debug(s"blocking ${this} due to ${exc}")
+ // Reset unconditionally; DPath.scala's InfosetLengthUnknownException
+ // handling re-sets this immediately via markWaitingOnLengthState when
+ // that's the actual blocking reason, so it never drifts out of sync
+ // with whether a real wake-up is registered.
+ isWaitingOnLengthState_ = false
Review Comment:
I think this goes away if we switch to the idea where parked suspensions are
never evaluated until we determine they should be unblocked.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -46,16 +98,67 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
* evaluate are moved to the old suspensions list. If we evaluate old
* suspensions, we attempt to evaluate them first, with the hope that their
* resolution might make the young suspensions more likely to evaluate.
+ *
+ * skipLengthStateWaiters = true here: a suspension with
+ * isWaitingOnLengthState true has a targeted wake-up already
+ * registered (fired from CaptureEndOf{Content,Value}LengthUnparsers
+ * once its length becomes computable) and can't progress until that
+ * fires - retrying it on the blind periodic schedule first is pure
+ * wasted DPath re-evaluation.
*/
- def evalSuspensions(): Unit = {
+ def evalSuspensions(): Unit =
+ evalSuspensionsThrottled(filterToBuildResolvable = false,
skipLengthStateWaiters = true)
+
+ /**
+ * A discard-sink sweep variant: same throttled cadence as
+ * evalSuspensions, but passes filterToBuildResolvable=true to
+ * evalSuspensionQueue. A suspension whose canResolveWithoutWriting is
+ * false can never be satisfied by a discard-sink traversal no matter
+ * how many retries, so it's skipped-and-requeued instead of really
+ * attempted - unless it's already isWaitingOnLengthState, in which
+ * case it's parked instead (parking only ever follows one of the real
+ * sweep's (evalSuspensions) own unfiltered attempts having set that
+ * flag; this filtered sweep never sets it itself). Either way the
+ * suspension stays pending for that real sweep's later, unfiltered
+ * attempts once real bytes exist for it to depend on.
+ *
+ * Eliminates the wasted doTask cost of this discard-sink sweep for
+ * these suspensions; doesn't eliminate the smaller per-tick
+ * dequeue/requeue cost for suspensions with no targeted wake-up at all
+ * (e.g. padding/target-length SuspendableOperations), which must stay
+ * on the skip-and-requeue path so the real sweep still finds them.
+ *
+ * skipLengthStateWaiters is left false here: canResolveWithoutWriting
Review Comment:
It feels like skipLengthStateWaiters can go away if we move to a deisgn
where we never evaluate LengthState waiters until they are notified, and when
they are notified we move them to young since they are likely to succeed. With
that design, we *always* skip length state waiters until something moves them
to the young queue.
##########
daffodil-core/src/main/scala/org/apache/daffodil/unparsers/runtime1/ElementUnparser.scala:
##########
@@ -360,13 +360,29 @@ class ElementOVCSpecifiedLengthUnparser(
private def suspendableExpression =
new ElementOVCSpecifiedLengthUnparserSuspendableExpression(this, expr)
+ // Negation of SuspendableExpression.canResolveWithoutWriting, computed
+ // once here since expr is already known at unparser-construction time.
+ // True when this OVC's expression can never succeed without a real
+ // written DOS bit position (e.g. references dfdl:valueLength/
+ // contentLength) - the first attempt is then guaranteed to block, no
+ // matter how many times retried.
+ private val requiresWriteTimeValue: Boolean =
+ !SuspendableExpression.canResolveWithoutWriting(expr)
+
Assert.invariant(context.dpathElementCompileInfo.isOutputValueCalc)
override def runContentUnparser(state: UState): Unit = {
computeTargetLength(
state
) // must happen before run() so that we can take advantage of knowing the
length
- suspendableExpression.run(state) // run the expression. It might or might
not have a value.
+ // Skip the guaranteed-to-fail doTask attempt when requiresWriteTimeValue
Review Comment:
Is this true that an OVC with a content/valueLength function is guaranteed
to fail? It's only guaranteed to fail if content/valueLentgh parameter is a
forward lookahead. Referencing the length of a previous element could succeed
without needing to suspend.
I don't think we have the logic to know at compile time is a path expression
is a forward lookahead or not.
Granted, most OVC's that reference the length of something almost always do
a forward reference (usually because the OVC is an element that contains the
length of some future thing). So since this is by far the most common case, if
this "suspend without attempt" thing makes a noticable performance difference
then it's probably worth it keeping. But we should update this comment to make
it clear it's not guaranteed, it's just super common so is an optimization for
the common case.The uncommon case will takea small hit by suspending
unnecessariyl, but that's the tradeoff we're willing to make.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/DPath.scala:
##########
@@ -252,6 +254,24 @@ final class RuntimeExpressionDPath[T <: AnyRef](
whereBlockedInfo.block(ve.qname, ve.context, 0, ve)
case noLength: InfosetLengthUnknownException =>
whereBlockedInfo.block(noLength.diElement, noLength.erd, 0, noLength)
+ // Register a targeted wake-up alongside the periodic-sweep blocking
+ // above: once this element's length becomes computable (see
+ // CaptureEndOf{Content,Value}LengthUnparser), this suspension is
Review Comment:
I think we might have to be careful about where we call notifyWaiters--I
don't think CaptureEndOf*LengthUnparser is the only place where the length of
an element can be finally resolved. In fact, I think it's not uncommon for
those unparsers to not be able to actually capture the actual length and
instead just capture relative bit positions of where the DOS started, which
often isn't enough for
I'm wondering if instead of notifying when an unparser ends, we should
instead notify when something calls `setRel/AbsBitPosition` on the LengthState,
since 1. that is the LengthState that supension is waiting for and 2. the
LengthState knows when it has been calculated.
The one thing I'm not sure about is if the length gets recalculated during
normal buffer resolution or when something (i.e. the suspension) asks for the
length. If it'sonly recalculated when a suspension asks for it, then we're in a
catch-22--we won't notify the suspension until the length is recalculated,
butwe won't recalculate the length until the suspension is run.
I *think* if we notify when things like setAbs/Rel are called, and maybe
some ofther functions, it might be sufficient to ensure the suspensions are
always triggered correctly.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/unparsers/UState.scala:
##########
@@ -553,8 +553,10 @@ final class UStateMain private (
// MStack, since the escape scheme cache logic requires an MStack. We
// reallyjust need the top for cloning for suspensions, but that
// requires changes to how the escape schema cache is accessed, which
- // isn't a trivial change.
- val esClone = new MStackOfMaybe[EscapeSchemeUnparserHelper]()
+ // isn't a trivial change. Sized to the source's actual depth
+ // (instead of MStack's default 32) since nothing ever pushes onto
Review Comment:
We don't need to document the default size of an MStack here. That's an
implementation detail of MStack that doesn't affect this. Same with below
comment.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/infoset/InfosetImpl.scala:
##########
@@ -572,6 +573,44 @@ sealed abstract class LengthState(ie: DIElement) {
var maybeEndPos0bInBits: MaybeULong = MaybeULong.Nope
var maybeComputedLengthInBits: MaybeULong = MaybeULong.Nope
+ /**
+ * Suspensions blocked on this element's length being unknown
+ * (dfdl:contentLength/dfdl:valueLength), registered via registerWaiter
+ * and retried directly via notifyWaiters once this length becomes
+ * computable, instead of waiting for SuspensionTracker's next periodic
+ * sweep. Purely an optimization alongside that sweep, not a
+ * replacement: a suspension still sits in SuspensionTracker's own
+ * queues too, so requireFinal()'s deadlock detection is unaffected.
+ *
+ * Deduplicated by reference identity - a suspension that re-blocks
+ * here across multiple sweeps re-registers each time, and without
+ * dedup this list would grow without bound.
+ */
+ private var waiters: List[Suspension] = Nil
Review Comment:
Should we just make this a mutabla.HashSet? We get duplication detection for
free. It also means the removeWaiting function will be constant time and will
never copy the full list which filterNot will do.
That said, it's possible using a List is still a win if this is expected to
always be a very small list of things, which I imagine it is in most cases. The
overhead of calculating hashes might actually be a bit slowerthan just looking
through a list of a couple things. The List scales very poorly, but it feels
very unlikely to have a bunch of things suspending on the same elements length.
If we do stick with a list, it might be worth documenting that this is
always expected to be very small and so is faster than a HashSet (with testing
to confirm)
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/infoset/InfosetImpl.scala:
##########
@@ -572,6 +573,44 @@ sealed abstract class LengthState(ie: DIElement) {
var maybeEndPos0bInBits: MaybeULong = MaybeULong.Nope
var maybeComputedLengthInBits: MaybeULong = MaybeULong.Nope
+ /**
+ * Suspensions blocked on this element's length being unknown
+ * (dfdl:contentLength/dfdl:valueLength), registered via registerWaiter
+ * and retried directly via notifyWaiters once this length becomes
+ * computable, instead of waiting for SuspensionTracker's next periodic
+ * sweep. Purely an optimization alongside that sweep, not a
+ * replacement: a suspension still sits in SuspensionTracker's own
+ * queues too, so requireFinal()'s deadlock detection is unaffected.
Review Comment:
> a suspension still sits in SuspensionTracker's own queues too, so
requireFinal()'s deadlock detection is unaffected.
1. This comment feels like it describing implementation details about
another part of code. I would suggest we don't do that. If we change that
implementation we also have to remember to update this. I would suggest this
simple says this is the list of suspensions waiting for the length of this
element to be resolved. We then leave it up to the suspension tracker to figure
out what how to handle those.
2. This feels a bit wasteful to have these suspension in the same queues,
aren't they always going to fail until the length is calculated? What if we
keep them completely separate, and when isFinal is true we can copy any
remaining length suspensions to the main queues and then let requireFinal due
it's last pass/deadlock detection. Note that in order for this to work best, we
probably need to ensure that we are notifying these suspensionsa the right
times, since other wise they won't get resolved until the very end.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -104,6 +137,20 @@ trait Suspension extends Serializable {
}
}
+ /**
+ * Public entry point to prepareToSuspend (private below), for callers
+ * that already know - from static information doTask can't see - that
+ * the first attempt is certain to block (e.g. an OVC referencing
+ * dfdl:valueLength/dfdl:contentLength from a discard-sink traversal
Review Comment:
What does a "discard-sink traversal that never writes real bytes" mean?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -104,6 +137,20 @@ trait Suspension extends Serializable {
}
}
+ /**
+ * Public entry point to prepareToSuspend (private below), for callers
+ * that already know - from static information doTask can't see - that
+ * the first attempt is certain to block (e.g. an OVC referencing
+ * dfdl:valueLength/dfdl:contentLength from a discard-sink traversal
+ * that never writes real bytes). Only the wasted doTask attempt is
+ * skipped; prepareToSuspend still runs at this exact call site, so any
+ * caller-specific state-patching (e.g. a cloneForSuspension override)
+ * happens exactly as it would after a real blocked attempt - the
+ * suspension's freeze point isn't deferred to wherever the caller's
+ * result eventually gets used.
+ */
+ final def suspendWithoutAttempting(ustate: UState): Unit =
prepareToSuspend(ustate)
Review Comment:
I'm curious how much this really makes a differences. My assumption was also
that the doTask part of suspension like OVC is not particularly expensive, but
it's everything else related to suspensions (copying the UState, adding to
suspension tracker, evaluating in the suspension tracker, etc.) is the
expensive part.
This change doesn't seem unreasonable,but I'm also curoius if it was found
to make a differences. This PR has kindof three unrelated optimizations (MStack
size limit, parked suspensions, skip first doTask), so I'm curious which one
made a diferent, or if they each just make a little difference that adds up.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -70,26 +85,44 @@ trait Suspension extends Serializable {
protected def doTask(ustate: UState): Unit
+ /**
+ * Guards against runSuspension re-entering for this same instance
+ * while already on-stack. Expected to never trip today (suspend()
+ * clones the ustate at block time, so runSuspension only touches an
+ * isolated per-suspension clone) - exists because LengthState's
+ * targeted wake-up (notifyWaiters) gives a second path into
+ * runSuspension alongside SuspensionTracker's periodic sweeps; if
+ * those two ever nest, this fails loudly instead of silently
+ * corrupting output.
+ */
Review Comment:
What if instead of notifyWaiters actually running those targeted
suspensions, it notifies the suspension tracker to move a suspension from
parked to the head of the young queue, and then we just let the suspension
tracker evaluate things like normal whenever it happens to get triggered next?
It does mean the targeted suspension might get triggered a little later than
normal, but I assume the biggest benefit of this PR is avoiding running
suspensions that we know won't succeed, which this still achieves, just with a
slight delay. And then we can get rid of the isRunning check since there is
still only one entrypoint into running suspensions.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -51,6 +52,20 @@ trait Suspension extends Serializable {
*/
val isReadOnly = false
+ /**
+ * True if this suspension's task might succeed without any bytes having
+ * been written yet (e.g. a plain value/variable read); false if it
+ * needs write-time-only information (e.g. a real DOS bit position, as
+ * dfdl:valueLength/dfdl:contentLength and most padding/fill/alignment
+ * operations do). Distinct from isReadOnly above: those length
+ * operations don't write DOS bytes (so they're isReadOnly) but DO read
+ * write-time-only bit-position state, so they're NOT
+ * canResolveWithoutWriting. Defaults to false; used by
+ * SuspensionTracker.evalBuildResolvableSuspensions to skip retrying a
+ * suspension from a discard-sink traversal that can never satisfy it.
Review Comment:
"discard-sink" is used a lot in this PR and I'm not sure what it means in
this context, can you clarify?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -28,7 +28,59 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
private val suspensionsYoung = new Queue[Suspension]
private val suspensionsOld = new Queue[Suspension]
- def suspensions: Seq[Suspension] = suspensionsYoung.toSeq ++
suspensionsOld.toSeq
+ /**
+ * Suspensions unable to make progress until something external
+ * changes (a targeted wake-up firing, or real bytes getting written);
+ * moved out of suspensionsYoung/suspensionsOld so the periodic sweep
+ * (evalSuspensionsThrottled) doesn't keep re-visiting them every tick -
+ * even a cheap per-item skip costs real time at scale.
+ *
+ * Only drained by foldParkedIntoOld, from the two "must attempt
+ * everything" methods below (evalSuspensionsUnthrottled, requireFinal);
+ * the targeted wake-up itself (LengthState.notifyWaiters) retries a
+ * suspension directly regardless of which bucket it's in, so parking
+ * never delays that path.
+ */
+ private val suspensionsParked = new Queue[Suspension]
+
+ /**
+ * Every still-tracked, not-yet-done suspension across all buckets.
+ * Must include suspensionsParked, or a suspension moving into that
+ * bucket would look like (incorrectly) resolved progress to a caller
+ * comparing this count before/after evalSuspensionsUnthrottled().
+ */
+ def suspensions: Seq[Suspension] =
+ suspensionsYoung.toSeq ++ suspensionsOld.toSeq ++ suspensionsParked.toSeq
+
+ /**
+ * Count of suspensions currently parked (suspensionsParked above) -
+ * unable to progress until the real bytes their wake-up depends on
+ * actually get written (i.e. by a non-discard-sink sweep). Not a good
+ * backlog-sized throttle on its own (see pendingCount below): a
+ * suspension is only classified here once it's actually re-retried
+ * and re-blocks on InfosetLengthUnknownException. A discard-sink sweep
+ * (evalBuildResolvableSuspensions) never performs that retry - it
+ * always skip-and-requeues instead - so nothing parks purely from a
+ * discard-sink sweep before a real sweep (evalSuspensions) has run at
+ * least once.
+ */
+ def parkedCount: Int = suspensionsParked.length
Review Comment:
Never used, suggest we remove it
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/DPath.scala:
##########
@@ -252,6 +254,24 @@ final class RuntimeExpressionDPath[T <: AnyRef](
whereBlockedInfo.block(ve.qname, ve.context, 0, ve)
case noLength: InfosetLengthUnknownException =>
whereBlockedInfo.block(noLength.diElement, noLength.erd, 0, noLength)
+ // Register a targeted wake-up alongside the periodic-sweep blocking
+ // above: once this element's length becomes computable (see
+ // CaptureEndOf{Content,Value}LengthUnparser), this suspension is
+ // retried directly instead of waiting for SuspensionTracker's next
+ // sweep. registerLengthStateWaiter (not registerWaiter directly)
+ // also deregisters from whichever LengthState this suspension was
+ // previously registered against, in case an earlier retry blocked
+ // on a different element's length - see Suspension.scala.
+ noLength match {
+ case _: InfosetContentLengthUnknownException =>
+
whereBlockedInfo.registerLengthStateWaiter(noLength.diElement.contentLength)
+ case _: InfosetValueLengthUnknownException =>
+
whereBlockedInfo.registerLengthStateWaiter(noLength.diElement.valueLength)
+ }
+ // Tell SuspensionTracker's throttled sweep it can skip this
+ // suspension until the wake-up above fires (see
+ // Suspension.isWaitingOnLengthState).
+ whereBlockedInfo.markWaitingOnLengthState()
Review Comment:
Do we need this function, can the logic done is this function be moved to
resgisterLengthStateWaiter? Seems like you can never have one with out the
other so they could be combined.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -28,7 +28,59 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
private val suspensionsYoung = new Queue[Suspension]
private val suspensionsOld = new Queue[Suspension]
- def suspensions: Seq[Suspension] = suspensionsYoung.toSeq ++
suspensionsOld.toSeq
+ /**
+ * Suspensions unable to make progress until something external
+ * changes (a targeted wake-up firing, or real bytes getting written);
+ * moved out of suspensionsYoung/suspensionsOld so the periodic sweep
+ * (evalSuspensionsThrottled) doesn't keep re-visiting them every tick -
+ * even a cheap per-item skip costs real time at scale.
+ *
+ * Only drained by foldParkedIntoOld, from the two "must attempt
+ * everything" methods below (evalSuspensionsUnthrottled, requireFinal);
+ * the targeted wake-up itself (LengthState.notifyWaiters) retries a
+ * suspension directly regardless of which bucket it's in, so parking
+ * never delays that path.
+ */
+ private val suspensionsParked = new Queue[Suspension]
+
+ /**
+ * Every still-tracked, not-yet-done suspension across all buckets.
+ * Must include suspensionsParked, or a suspension moving into that
+ * bucket would look like (incorrectly) resolved progress to a caller
+ * comparing this count before/after evalSuspensionsUnthrottled().
+ */
Review Comment:
We should add a comment that this should only ever be used for debugging
purposes. Combining these queues is a faiarly expensive operation that
internally we should never use in a hot path.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -175,8 +222,10 @@ trait Suspension extends Serializable {
//
// clone the ustate for use when evaluating the expression
//
- // TODO: Performance - copying this whole state, just for OVC is painful.
- // Some sort of copy-on-write scheme would be better.
+ // cloneForSuspension (UState.scala) is a targeted partial clone
+ // (shallow VariableMap copy, stack tops only), not a full deep copy;
+ // its escapeSchemeEVCache/delimiterStack clones are sized to actual
+ // depth instead of MStack's default 32 slots.
Review Comment:
I think this TODO is still needed. Although the cloneForSuspensions is
somewhat targeted and isn't a full deep clone, it does still copy a lot of
things that might never change. I've always wonder if these copies were a
sizable part of suspension overhead, if so copy-on-write implementation could
potentially help with that. Even fixing things so we don't copy the entire
escapeScheme stacks could be an improvement.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -228,9 +277,51 @@ trait Suspension extends Serializable {
final def isMakingProgress = isMakingProgress_
+ /**
+ * True exactly when this suspension is blocked on
+ * InfosetLengthUnknownException with a targeted wake-up already
+ * registered against the relevant LengthState (set only from
+ * DPath.scala's InfosetLengthUnknownException catch clause, the one
+ * place that calls markWaitingOnLengthState). SuspensionTracker's
+ * periodic sweep uses this to skip re-running doTask until that
+ * wake-up fires. Not inferred from block()'s exc type
+ * (SuspendableOperation.scala also passes a RetryableException that
+ * could structurally match without a wake-up registered) - only the
+ * actual registration site sets this, so the two can't drift apart.
+ */
+ private var isWaitingOnLengthState_ : Boolean = false
+
+ final def markWaitingOnLengthState(): Unit = {
+ isWaitingOnLengthState_ = true
+ }
+
+ final def isWaitingOnLengthState: Boolean = isWaitingOnLengthState_
+
+ /**
+ * Which LengthState (if any) this suspension is registered with as a
+ * waiter. A suspension's dependency can shift between retries (e.g. a
+ * length expression whose which-element branch changes), so it must be
+ * deregistered from the old LengthState before registering a new one,
+ * or the old one's notifyWaiters() would retry it pointlessly.
+ */
+ private var maybeRegisteredLengthState: Maybe[LengthState] = Nope
+
+ final def registerLengthStateWaiter(ls: LengthState): Unit = {
+ if (maybeRegisteredLengthState.isDefined &&
(maybeRegisteredLengthState.get ne ls))
+ maybeRegisteredLengthState.get.removeWaiter(this)
+ maybeRegisteredLengthState = One(ls)
+ ls.registerWaiter(this)
Review Comment:
This is a bit confusing to me, since this implies a suspension was waiting
for a previous state, but then somehow now depends on a different state? I'm
not sure how that can ever happen since the suspension should still be blocked
on the previous length state.
Feels like this should be something
```scala
Assert.invariant(!maybeRegisteredLengthState.isDefined)
```
Feels like the suspension tracker needs a function that liked
```scala
def moveFromParked(s: Suspension): Unit = {
Assert.invariant(s.maybeRegisterLengthStateWaiter.isDefined)
s.maybeRegisterLengthStateWaiter.get.removeSuspension(this)
s.maybeRegisterLenghtStateWaiter = Nope
suspensionsParked.remove(s)
suspensionsYoung.push(s)
}
```
So the idea is if something is parked then it must reference a LengthState
in the and the LengthState must references it. Something eventually notifies
the tracker that a parked suspension is unblocked so it unregisteres it (sets
the internal state to Nope and tells the LenghtState to no longer wait on it)
and then moves it to young to be evaluated later.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -46,16 +98,67 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
* evaluate are moved to the old suspensions list. If we evaluate old
* suspensions, we attempt to evaluate them first, with the hope that their
* resolution might make the young suspensions more likely to evaluate.
+ *
+ * skipLengthStateWaiters = true here: a suspension with
+ * isWaitingOnLengthState true has a targeted wake-up already
+ * registered (fired from CaptureEndOf{Content,Value}LengthUnparsers
+ * once its length becomes computable) and can't progress until that
+ * fires - retrying it on the blind periodic schedule first is pure
+ * wasted DPath re-evaluation.
*/
- def evalSuspensions(): Unit = {
+ def evalSuspensions(): Unit =
+ evalSuspensionsThrottled(filterToBuildResolvable = false,
skipLengthStateWaiters = true)
+
+ /**
+ * A discard-sink sweep variant: same throttled cadence as
+ * evalSuspensions, but passes filterToBuildResolvable=true to
+ * evalSuspensionQueue. A suspension whose canResolveWithoutWriting is
+ * false can never be satisfied by a discard-sink traversal no matter
+ * how many retries, so it's skipped-and-requeued instead of really
+ * attempted - unless it's already isWaitingOnLengthState, in which
+ * case it's parked instead (parking only ever follows one of the real
+ * sweep's (evalSuspensions) own unfiltered attempts having set that
+ * flag; this filtered sweep never sets it itself). Either way the
+ * suspension stays pending for that real sweep's later, unfiltered
+ * attempts once real bytes exist for it to depend on.
+ *
+ * Eliminates the wasted doTask cost of this discard-sink sweep for
+ * these suspensions; doesn't eliminate the smaller per-tick
+ * dequeue/requeue cost for suspensions with no targeted wake-up at all
+ * (e.g. padding/target-length SuspendableOperations), which must stay
+ * on the skip-and-requeue path so the real sweep still finds them.
+ *
+ * skipLengthStateWaiters is left false here: canResolveWithoutWriting
+ * already excludes every length-state-blocked suspension from this
+ * sweep's retries, and more completely (it also covers non-length
+ * forward references), so a second filter would be redundant.
+ */
+ def evalBuildResolvableSuspensions(): Unit =
Review Comment:
This is never used, and it's purpose isn't entirely clear, presumably for a
future change. Can this and things related to filterToBuildResolvable be moved
to a separate PR so we can review all the related stuff in one PR?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -28,7 +28,59 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
private val suspensionsYoung = new Queue[Suspension]
private val suspensionsOld = new Queue[Suspension]
- def suspensions: Seq[Suspension] = suspensionsYoung.toSeq ++
suspensionsOld.toSeq
+ /**
+ * Suspensions unable to make progress until something external
+ * changes (a targeted wake-up firing, or real bytes getting written);
+ * moved out of suspensionsYoung/suspensionsOld so the periodic sweep
+ * (evalSuspensionsThrottled) doesn't keep re-visiting them every tick -
+ * even a cheap per-item skip costs real time at scale.
+ *
+ * Only drained by foldParkedIntoOld, from the two "must attempt
+ * everything" methods below (evalSuspensionsUnthrottled, requireFinal);
+ * the targeted wake-up itself (LengthState.notifyWaiters) retries a
+ * suspension directly regardless of which bucket it's in, so parking
+ * never delays that path.
+ */
+ private val suspensionsParked = new Queue[Suspension]
+
+ /**
+ * Every still-tracked, not-yet-done suspension across all buckets.
+ * Must include suspensionsParked, or a suspension moving into that
+ * bucket would look like (incorrectly) resolved progress to a caller
+ * comparing this count before/after evalSuspensionsUnthrottled().
+ */
+ def suspensions: Seq[Suspension] =
+ suspensionsYoung.toSeq ++ suspensionsOld.toSeq ++ suspensionsParked.toSeq
+
+ /**
+ * Count of suspensions currently parked (suspensionsParked above) -
+ * unable to progress until the real bytes their wake-up depends on
+ * actually get written (i.e. by a non-discard-sink sweep). Not a good
+ * backlog-sized throttle on its own (see pendingCount below): a
+ * suspension is only classified here once it's actually re-retried
+ * and re-blocks on InfosetLengthUnknownException. A discard-sink sweep
+ * (evalBuildResolvableSuspensions) never performs that retry - it
+ * always skip-and-requeues instead - so nothing parks purely from a
+ * discard-sink sweep before a real sweep (evalSuspensions) has run at
+ * least once.
+ */
+ def parkedCount: Int = suspensionsParked.length
+
+ /**
+ * Total not-yet-done suspensions across all three buckets, without
+ * allocating (unlike `suspensions` above - not safe to call once per
+ * node). Unlike parkedCount, grows the moment a suspension is created
+ * (trackSuspension), with no dependency on it having been retried yet
+ * - intended as a throttle signal for pacing a discard-sink traversal
+ * against the pending backlog. Still distinguishes the two workload
+ * shapes correctly: a canResolveWithoutWriting=true suspension
+ * resolves within a few ticks of its sibling being added to the tree
+ * (stays small/transient), while a length-dependent one (never
+ * resolvable without real bytes actually being written) accumulates
+ * here unboundedly.
+ */
+ def pendingCount: Int =
+ suspensionsYoung.length + suspensionsOld.length + suspensionsParked.length
Review Comment:
Never used, suggest we remove
--
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]
