Skip to content

Improve unparse suspension resolution and reduce clone overhead - #1717

Closed
olabusayoT wants to merge 7 commits into
apache:mainfrom
olabusayoT:daf-3065-suspension-tracking
Closed

olabusayoT wants to merge 7 commits into
apache:mainfrom
olabusayoT:daf-3065-suspension-tracking

Conversation

@olabusayoT

Copy link
Copy Markdown
Contributor

Suspensions blocked on a not-yet-known DFDL length (dfdl:valueLength, dfdl:contentLength) now register directly on the referenced element's LengthState and get retried the moment that length becomes known, instead of waiting for SuspensionTracker's periodic sweep (LengthState.notifyWaiters, InfosetImpl.scala; registration in DPath.scala's InfosetLengthUnknownException handling).

SuspensionTracker gains a dedicated pendingCount, a separate suspensionsParked queue for suspensions with a registered targeted wake-up (removed from the periodic rotation entirely, pruned once resolved so a suspension that resolves out-of-band doesn't sit retained for the rest of the document), and
evalBuildResolvableSuspensions, a sweep variant for callers that can't write real bytes yet (a discard-sink traversal), gated by Suspension.canResolveWithoutWriting.

Suspension.suspendWithoutAttempting registers and tracks a suspension without ever calling doTask, for callers that already know, from static information doTask itself can't see, that the first attempt is certain to block; used by ElementOVCSpecifiedLengthUnparser when its expression can never resolve without a real written DOS bit position.

cloneForSuspension (UState.scala) sizes its escapeSchemeEVCache/ delimiterStack MStack clones to the source's actual depth instead of MStack's default 32-slot allocation.

DAFFODIL-3065

Suspensions blocked on a not-yet-known DFDL length (dfdl:valueLength,
dfdl:contentLength) now register directly on the referenced element's
LengthState and get retried the moment that length becomes known,
instead of waiting for SuspensionTracker's periodic sweep
(LengthState.notifyWaiters, InfosetImpl.scala; registration in
DPath.scala's InfosetLengthUnknownException handling).

SuspensionTracker gains a dedicated pendingCount, a separate
suspensionsParked queue for suspensions with a registered targeted
wake-up (removed from the periodic rotation entirely, pruned once
resolved so a suspension that resolves out-of-band doesn't sit
retained for the rest of the document), and
evalBuildResolvableSuspensions, a sweep variant for callers that can't
write real bytes yet (a discard-sink traversal), gated by
Suspension.canResolveWithoutWriting.

Suspension.suspendWithoutAttempting registers and tracks a suspension
without ever calling doTask, for callers that already know, from
static information doTask itself can't see, that the first attempt is
certain to block; used by ElementOVCSpecifiedLengthUnparser when its
expression can never resolve without a real written DOS bit position.

cloneForSuspension (UState.scala) sizes its escapeSchemeEVCache/
delimiterStack MStack clones to the source's actual depth instead of
MStack's default 32-slot allocation.

DAFFODIL-3065
@olabusayoT

Copy link
Copy Markdown
Contributor Author

FYI using daffodil performance command

+---------+-----+------------+----------+----------+-----------+---------+----------+------------+-----------+-------------+                                                                                                      
|       N |   T |  main tput |   A tput |   tput Δ |  main lat |   A lat |    lat Δ |  main susp |    A susp |  susp match |                                                                                                      
+---------+-----+------------+----------+----------+-----------+---------+----------+------------+-----------+-------------+                                                                                                      
|       1 |   1 |       1.45 |     1.46 |    +0.3% |    672.44 |  669.81 |    -0.4% |       4000 |      4000 |         yes |                                                                                                      
|       1 |   4 |       1.44 |     1.48 |    +3.0% |    679.15 |  658.51 |    -3.0% |       4000 |      4000 |         yes |                                                                                                      
|       1 |   8 |       1.43 |     1.49 |    +4.0% |    683.19 |  656.02 |    -4.0% |       4000 |      4000 |         yes |                                                                                                      
|    2500 |   1 |      55.16 |    60.04 |    +8.8% |     18.13 |   16.65 |    -8.2% |   10000000 |  10000000 |         yes |                                                                                                      
|    2500 |   4 |     131.90 |   145.82 |   +10.6% |     30.28 |   27.39 |    -9.6% |   10000000 |  10000000 |         yes |                                                                                                      
|    2500 |   8 |     131.73 |   139.70 |    +6.0% |     60.62 |   57.15 |    -5.7% |   10000000 |  10000000 |         yes |                                                                                                      
|    5000 |   1 |      61.54 |    67.66 |    +9.9% |     16.25 |   14.79 |    -9.0% |   20000000 |  20000000 |         yes |                                                                                                      
|    5000 |   4 |     152.76 |   181.06 |   +18.5% |     26.16 |   22.10 |   -15.5% |   20000000 |  20000000 |         yes |                                                                                                      
|    5000 |   8 |     153.04 |   168.74 |   +10.3% |     52.24 |   47.36 |    -9.3% |   20000000 |  20000000 |         yes |                                                                                                      
|   10000 |   4 |     163.49 |   182.48 |   +11.6% |     24.45 |   21.91 |   -10.4% |   40000000 |  40000000 |         yes |                                                                                                      
|   10000 |   8 |     168.06 |   178.36 |    +6.1% |     47.57 |   44.83 |    -5.8% |   40000000 |  40000000 |         yes |                                                                                                      
+---------+-----+------------+----------+----------+-----------+---------+----------+------------+-----------+-------------+


object MStackOfAnyRef {
def apply() = {
def apply(initialSize: Int = 32) = {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

* 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.
*/

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:

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.

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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure what "periodic-sweep blocking above" is referencing.

Comment on lines +265 to +270
noLength match {
case _: InfosetContentLengthUnknownException =>
whereBlockedInfo.registerLengthStateWaiter(noLength.diElement.contentLength)
case _: InfosetValueLengthUnknownException =>
whereBlockedInfo.registerLengthStateWaiter(noLength.diElement.valueLength)
}

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think if we make lengthState a val in InfosetLengthUnknownException we can avoid the match/case and simplify this to:

whereBlockedInfo.registerLengthStateWaiter(noLength.lengthState)

I don't think there's any harm in making lengthState accessible and it simplifies our code.

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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

* discard-sink sweep before a real sweep (evalSuspensions) has run at
* least once.
*/
def parkedCount: Int = suspensionsParked.length

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Never used, suggest we remove it

* here unboundedly.
*/
def pendingCount: Int =
suspensionsYoung.length + suspensionsOld.length + suspensionsParked.length

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Never used, suggest we remove

* (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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unfortunately some test in our current test rig break with the proposed "lengthstate suspensions never evaluate until notified" plan, because of DOS Splitting/merging, so in those situations, the registered wakeup never gets fired. Here's is a more details explanation from claude

Here's what's actually going on in TestOutputValueCalc1's OutputValueCalc_01:

Trace (OutputValueCalc_01: x needs valueLength(y), y needs valueLength(z)):

first-attempt SuspendableExpression(x, valueLength(y)) → isDone=false (y unwritten) 
first-attempt SimpleTypeRetryUnparser(x) → isDone=false (x has no value) 
first-attempt SuspendableExpression(y, valueLength(z)) → isDone=false (z unwritten) 
first-attempt SimpleTypeRetryUnparser(y) → isDone=false (y has no value) 

retry SuspendableExpression(x) isWaitingOnLengthState=false 
registerWaiter valueLen(y) <- SuspendableExpression(x) ← x registers its ONE targeted wake-up, on y's length 
retry-result SuspendableExpression(x) → isDone=false 

retry SimpleTypeRetryUnparser(x) → isDone=false 
retry SuspendableExpression(y) → isDone=true (z is a plain string, already known) 
retry SimpleTypeRetryUnparser(y) → isDone=true (y's bytes now written) 
retry SimpleTypeRetryUnparser(x) → isDone=false (still no value for x) 
retry SimpleTypeRetryUnparser(x) → isDone=false 

retry SuspendableExpression(x) isWaitingOnLengthState=true ← STILL registered, no notify ever fired 
retry-result SuspendableExpression(x) → isDone=true ← succeeds anyway, via a plain re-attempt 

The point: x registers exactly once, against y's ValueLengthState. y's length never triggers that registered wake-up as y's length only becomes computable as a side effect of y's own SimpleTypeRetryUnparser succeeding (a completely different suspension object, blocked on y's value, not on any length at all) plus the surrounding DOS-splitting machinery converging. Grepping the run for the wake-up call (moveFromParkedToYoung, the only path a registered notify can take) and for the notify-check itself found zero hits. x succeeds purely because something keeps giving it a plain, unconditional re-attempt, not because anything ever told it "your wait is over."

So "never evaluate a suspension until it's notified" is unsound as a general design: x's only completion path here is a blind retry with a stale, never-fired registration still sitting on it. A purely event-driven model has no event to catch this, which is exactly why the periodic retry has to stay.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah right, eventually the state of DOSs change, which allows LengthState to be calculated. But the LengthState never knows that the DOS state changed. Maybe an option is to add a member to DOSs to track the LengthState's that have suspensions that depend on that DOS? When the DOS state changes it could notify the LengthStates that they could be resolvable, and then the LengthStates could unpark the suspension?

I also wonder if this approach can be generalized--it sounds like someting we could potentially use for other suspensions. Essentially, we have things that keep track of certain Suspensions because they are waiting for state to change to the point whre those suspensions are likely to be resolvable and then things that notify those suspensions that the calculation could succeed. For example, maybe something like this:

class SuspensionWaiter {

	// list of suspensions that are parked until something notifies this waiter
	val suspensions = mutable.Set[Suspension].empty

	// list of classes that maintain some state that when changed could notify this waiter
        // that their suspensions might now be resolvable
	val notifiers = mutable.Set[SuspensionWaiterNotifier].empty

	def addSuspension(s: Suspension) = {
		suspensionTracker.park(s)
		suspensions.addOne(s)
	}

	def notify(): Unit = {
		suspensions.foreach { s => suspensionTracker.moveFromParkedToYoung(s) }
		suspensions.clear()
		notifiers.foreach {
			_.waiters.remove(this)
		}
	}
}

class SuspensionWaiterNotifier {
	val waiters = mutable.Set[SuspensionWaiter].empty

	def addWaiter(w: SuspensionWaiter) = waiters.addOne(w)

	def notify(): Unit = {
		waiters.foreach(_.notify)
	}
}

And then in LengthState, this might look something like

class LengthState {

	// already existing members
	var maybeStartDataOutputStream: Maybe[DataOutputStream] = Nope
	var maybeEndDataOutputStream: Maybe[DataOutputStream] = Nope

	// new waiter val
	val suspensionWaiter = new SuspensionWaiter {
		override def addSuspension(s: Suspension) {
			super.addSuspension(s)
			maybeStartDataOutputStream.foreach { dos => dos.suspensionWaiterNotifier.addWaiter(this) }
			maybeEndDataOutputStream.foreach { dos => dos.suspensionWaiterNotifier.addWaiter(this) }

		}
	}
}

And when a suspension is blocked on a LengthState, it would do

lengthState.suspensionWaiter.addSuspension(suspension)

When this is called, the suspension gets parked, and the DOSs are told to notify the waiter if something changes.

And then the DataOutputStreams has something like

class DataOutputStreamImplMixin {

	val suspensionWaiterNotifierLengthState = new SuspensionWaiterNotifier()

	// ... when DOS state changes where a length state might be resolvable
	suspensionWaiterNotifierLengthState.notify()
}

That will notify each of the waiters that depend on its state (if any). The waiters will then move their suspensions from parked to young to be evaluated some point soon. And then they'll tell the notifiers to no longer notify this length state waiter since it no longer is waiting on suspensions.

Also note that the LengthState class could also call suspensionWaiter.notify(), for example if setAbsBitPosition is changed--a waiter does not have to be notified by the SuspensionWaiterNotifiers. And some suspensionWaiter might not even have notifiers. Those are just a helpful class to use when other classes maintain state that the waiter classes depend on.

Also, note that the Notifier could override the notify() function to actually examine state and determine if it really should unpark the suspension and clean up the state. The notify() function doesn't necessarily require unparking. But the default, and likely most common implementation, could be to just unpark the suspension, let things run, and if they block again then they just register with suspensionWaiter/Notifiers and repeat the process.

"never evaluate a suspension until it's notified" is unsound as a general design:

Note that we should still unpark all parked suspensions when tracker.requireFinal is called, which should allow them to be resolved. So even if something is never notified (which is probably a bug and we should probably warn) we could still have a backup to ensure things at least succeed.


Also, there are likely other designs. For example, maybe the SuspensionWaiter becomes part of a the Suspension class, and parked Suspension's just keep track of a list of SuspensionNotifiers. Then the both the DOS and LengthState classes would have SuspensionNotifiers which keep track of the Suspensions that depend on them. One downside to this approach is it doesn't allow LenghtStates to overrule a notification from a DOS if it wants, since ultimately the LengthState knows if the change to the DOS state makes a difference to the LengthState or not. If we moved the waiters into Suspensions, a DOS would unpark the and there's nothing the LengthState waiters could do about it. But maybe we don't need that level of granularity for this. It does feel a bit simpler to not really have a separate concept of SuspensionWaiters, and instead there are only things that can notify a Suspension that it could be unparked.

* sweep's retries, and more completely (it also covers non-length
* forward references), so a second filter would be redundant.
*/
def evalBuildResolvableSuspensions(): Unit =

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

…able wake-ups

LengthState.suspensionWaiter.notifySuspensions() is called from all
four position setters (setAbsStartPos0bInBits, setRelStartPos0bInBits,
setAbsEndPos0bInBits, setRelEndPos0bInBits) and from both
absolute-position migrations in recheckStreams, so a suspension
blocked on an unknown position gets a targeted wake-up instead of
only the periodic sweep.

DirectOrBufferedDataOutputStream gains a lazily-allocated
FinishedListener registry, fired once when a DOS transitions to
Finished. FinishedListener is a small callback trait declared in
daffodil-io so that package's dependency stays one-directional;
LengthState implements it rather than daffodil-io referencing the
infoset layer directly. LengthState tracks at most one such DOS at a
time (maybeBlockingDos), for the case in maybeLengthInBits where it's
blocked on a specific DOS in a relative-DOS chain becoming finished,
registering with whichever DOS is currently blocking it and
deregistering from any prior one first.

The register/remove/notify bookkeeping behind this wake-up is a
standalone, reusable org.apache.daffodil.runtime1.processors.SuspensionWaiter
class rather than something specific to LengthState. notifySuspensions()
does nothing when nothing is registered, since that's the overwhelmingly
common case on every call site that triggers it (every variable set,
every position update). LengthState delegates to a SuspensionWaiter via
its suspensionWaiter field, and VariableInstance now has its own:
VariableHasNoValue and VariableSuspended (thrown from
VariableMap1.readVariable) previously registered no targeted wake-up at
all, so a suspension blocked on an undefined or in-process variable only
ever retried on the periodic sweep. VariableInstance.suspensionWaiter is
a @transient var, reinitialized in a custom readObject, rather than a
lazy val: VariableInstance is part of the compiled schema's serialized
state (saved/reloaded via DataProcessor.save) so the field can't survive
a save/reload, but it's also read on every setVariable/setDefaultValue
call, and a lazy val's thread-safe access check would be paid on every
one of those even though a VariableInstance is never shared across
threads.

A SuspensionWaiter's notifySuspensions() hands each registered
suspension to Suspension.moveFromParkedToYoung(), which moves it back
into its own SuspensionTracker's young queue for a real attempt.
Suspension.suspend() stashes that tracker (maybeTracker) the first
time it blocks, which is why UStateMain.suspensionTracker widened from
private to private[processors].

Suspension.isWaitingOnWaiter is derived from maybeRegisteredWaiter.isDefined
instead of a separately tracked boolean. Suspension.maybeRegisterWaiterFor
is a single shared helper used from both DPath.scala's expression
evaluation and SuspendableOperation's retry loop, so padding/target-length
operations and variable reads all get the same targeted wake-up.
ElementOVCSpecifiedLengthUnparser always attempts its expression once
before suspending, since the prior static-shape check couldn't
distinguish a reference guaranteed to block from one whose target had
already finished unparsing.

A DOS-splitting chain's cumulative length can become computable
without any single setter or FinishedListener ever firing, so no
targeted wake-up covers that case. evalParkedSuspensions still gives
every parked suspension a real, unconditional retry on the same
reduced cadence as old suspensions, bounding staleness for that case
instead of deferring it to requireFinal, and skips this work entirely
when suspensionsParked is empty.

Removed evalBuildResolvableSuspensions, evalSuspensionsUnthrottled,
parkedCount, pendingCount, Suspension.canResolveWithoutWriting, and
suspendWithoutAttempting: dead code with zero callers on this branch.

MStack's initialSize constructor parameter now threads through
MStackOfBoolean/MStackOfInt/MStackOfLong the same way it already did
for MStackOf/MStackOfMaybe/MStackOfAnyRef.

DAFFODIL-3065
@olabusayoT
olabusayoT force-pushed the daf-3065-suspension-tracking branch 2 times, most recently from 103dd0d to 7461af0 Compare August 25, 2026 18:59

@stevedlawrence stevedlawrence left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks nice, I think we need additional documentation since it's still not totally clear to me how the park stuff interacts with the young/old queues, and when things are supposed to moved from one to the other. There's also maybe some room for some simplfication, which feels important since I think this is all just going to get even more complex as we deal with suspensions.

* daffodil-io depend only on this trait, not on whoever implements it,
* keeps the dependency one-directional.
*/
trait FinishedListener {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this needs a new name to make it clear this is about a DOS finishing. I think a number of things in Daffodil have a concept of being finishes/finalized, so that trait and allbacks probably want to differentiate that to make it clear what is actually being finished. You could maybe getaway with notifyFinished as the call back name if it also accepts the DOS as a parameter so the callback knows which one was finished if it cares.


// Lazily allocated - most DOS instances never have anyone waiting on
// their isFinished transition, so this stays unallocated for those.
private var maybeFinishedListeners: Maybe[mutable.HashSet[FinishedListener]] = Nope

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm wondering if this wants to be a var immutable Set?

My thinking is that in most cases there will probably zero or just a very small number of listeners, and immutable Sets are optimized in those cases, with special EmptySet, Set1, Set2, Set3, and Set4 classes. And beyond that it is still fairly efficient to avoid copies when adding new elements.

This would clean up the Maybe isEmpty/get stuff and also remove overhead related to hashsets (e.g. hash calculations, array allocations, bucket allocations, etc). I'm not sure it will make much of a difference in practice since these are almost always empty and I imagine a Nope and a Set.Empty are basically the same.

val toNotify = maybeFinishedListeners.get.toArray
maybeFinishedListeners.get.clear()
toNotify.foreach(_.notifyFinished())
}

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One drawback with this notifying only on is finished is suspensions don't necessarily need a DOS to be finished to be able to resolve. In some cases they might just need the starting absolute bit position to allow length to be calculatable. isFinish will work, but it might delay the suspension for much longer.

This is one reason where the suggested approach about making this more specific to suspensions has advantages. This could notify the suspension waiters when maybeAbStartingBitPos0b gets set rather than waiting for it to be finished. The waiter can then examine the state and act accordingly. It's also more generic and could be used in other suspension optimizations that wait on different state. This approach only works for DOSs, we'll need a new mechanism if we want some other kind of listern (e.g. element inf InfosetImpl becomes final).

// top), so there's no need to keep this registration around afterward.
private def notifyFinishedListeners(): Unit = {
if (maybeFinishedListeners.isDefined) {
val toNotify = maybeFinishedListeners.get.toArray

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we just do maybeFinishedListeners.get.foreach(_.notifyFinished()) and avoid the array allocation?

private def notifyFinishedListeners(): Unit = {
if (maybeFinishedListeners.isDefined) {
val toNotify = maybeFinishedListeners.get.toArray
maybeFinishedListeners.get.clear()

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can just set maybeFinishedListerns = Nope and let the hash set be garbage collected, once this is finihsed we'll never need the hashSet again,so we don't really care what its state is. And I imagine the clear() function zero's out the backing array which is going to be a bit slower.

}

private[runtime1] def isRegisteredSuspension(s: Suspension): Boolean =
suspensions.contains(s)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggest we just make these public

// caller of runSuspension.
def notifySuspensions(): Unit = {
if (suspensions.nonEmpty) {
val toMove = suspensions.toArray

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This probably just wants to be

suspensions.foreach { ... }
suspensions.clear()

No need to copy to an array first

// setDefaultValue read this on every call, and a lazy val's
// thread-safe access check would be paid on every one of those even
// though a VariableInstance is never shared across threads.
@transient var suspensionWaiter: SuspensionWaiter = new SuspensionWaiter

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you make this a @transient lazy val then you shouldn't need the readObject. On deserialization a lazy val just becomes unset and so the first time it's referenced a new one will be created.


// Suspensions blocked reading this variable before it had a value
// (VariableHasNoValue/VariableSuspended), given a real retry once it's
// set. A variable is only ever set once, so unlike a length this never

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've avoid mentioning how this differs from the length. Just describe how this specific implemenation works. If we ever change how length works we don't need to update this unrelated comment.

// Always try the expression first: whether the specific referenced
// occurrence (e.g. an already-finished earlier sibling) is already
// resolved is a run-time fact the expression's static shape alone
// can't tell us, even when it references valueLength/contentLength.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure what this comment is trying to say. Sounds like a fancy way to saw what the original comment said. we need to run the expression, it might need to succeed be something it's known yet, which is just standard suspension behavior.

@olabusayoT
olabusayoT force-pushed the daf-3065-suspension-tracking branch 2 times, most recently from 4403f40 to 297084c Compare September 1, 2026 17:25
Suspended unparse operations previously relied entirely on
SuspensionTracker's periodic sweep to retry. This adds
SuspensionWaiter, a register/notify class wired into every common
blocking point so retries are triggered by the event that unblocks
them, plus a shared DataOutputStreamEventListener/Registry replacing
three near-duplicate listener classes.

LengthState.suspensionWaiter registers broadly across every DOS whose
resolution could make its length computable;
PrefixLengthSuspendableOperation, NeedValueAndTargetLengthMixin,
SuppressableSeparatorUnparserSuspendableOperation,
ChoiceUnusedUnparserSuspendableOperation, and
AlignmentFillUnparserSuspendableMixin gain the same targeted
registration, previously absent.

SuspensionTracker's force-retry paths now clear a suspension's
registrations before retrying it, since a reentrant notify from within
a force-retry pass could otherwise hand it back to its tracker twice
while still registered; a reentrant notify mid-pass is now tracked and
honored once the pass settles. UState.suspensionTracker becomes
reachable through a suspension's cloned savedUstate.

DirectOrBufferedDataOutputStream.equals/hashCode's broken "equal if
eq" override, silently relying on Set's reference-equality
short-circuit, is fixed to real identity equality, now load-bearing
for LengthState's new DOS sets.

Suspension.isWaitingOnWaiter is renamed isParked; the periodic sweep
now diverts an isParked suspension immediately rather than on its
next dequeue. Registration/notify storage (SuspensionWaiter,
dosListeners) is lazily allocated and skips a Map for the common
single-registrant case.

VariableInstance's suspensionWaiter and related wiring (added
alongside LengthState's own) are removed here; this branch stays
scoped to LengthState wake-ups. LengthState's lazy-waiter pattern
moves into a new, reusable HasSuspensionWaiter trait.

ElementOVCSpecifiedLengthUnparser.runContentUnparser's comment is
reverted to its pre-existing simple form: the unconditional
suspendableExpression.run(state) call here already matches main, so
the more elaborate comment wasn't describing any actual behavior
change, just explaining a decision that happened to restore parity
with what main already did.

DAFFODIL-3065
@olabusayoT
olabusayoT force-pushed the daf-3065-suspension-tracking branch from 297084c to 63af290 Compare September 1, 2026 22:52
SuspensionWaiter's overflow storage (used once a waiter gains a
second, distinct registrant) was a mutable.LinkedHashMap, paying a
hash-table entry allocation on every registration and again, via
iterator.collect{}.toList, materializing an extra iterator wrapper
and a linked list on every notifySuspensions() call. Allocation
profiling (on a downstream branch's own targeted-wake-up work)
found a waiter that grows past its sole-registrant slot almost never
holds more than two registrants at once, so the hash table's own
per-entry overhead buys nothing here.

Replaces the map with a small mutable SuspensionWaiterRegistration
holder (suspension + cond) stored in a linear-scan
mutable.ArrayBuffer: registerSuspension/removeSuspension use
indexWhere/remove-by-index instead of map put/remove,
notifySuspensions loops directly into a pre-sized buffer instead of
building an iterator-plus-list, and forceRetryAll/clear/
isRegisteredSuspension are adjusted to the same structure. Reentrancy
safety is preserved exactly as before: notifySuspensions fully
computes which registrants are handled before removing any of them
or calling moveFromParkedToYoung, since that call can reenter this
same waiter's register/removeSuspension.

No externally visible behavior change; registerSuspension/
notifySuspensions/forceRetryAll/clear keep their existing signatures.

DAFFODIL-3065
@olabusayoT

olabusayoT commented Sep 3, 2026 •

Copy link
Copy Markdown
Contributor Author

Sorry for the delayed results. WE needed some performance updatesafter profiling

  +----------+---+---------------+---------+---------------------------+                                                                                                                                                            
  | workload | t | branch vs main  | Welch t | significant?              |                                                                                                                                                            
  +----------+---+---------------+---------+---------------------------+                                                                                                                                                            
  | sparse   | 1 | -19.52%       | -19.07  | yes, extremely            |                                                                                                                                                            
  | sparse   | 4 |  -5.65%       |  -3.22  | yes                       |                                                                                                                                                            
  | dense    | 1 |  +1.27%       |   0.80  | no - essentially even     |                                                                                                                                                            
  | dense    | 4 |  -1.05%       |  -0.26  | no - essentially even     |                                                                                                                                                            
  +----------+---+---------------+---------+---------------------------+

  +--------+----+----------------------------+-------------+-------------+---------+---------+                                                                                                                                      
  | workload | t | metric                    |    main     |    brance     |  pct    | Welch t |                                                                                                                                      
  +--------+----+----------------------------+-------------+-------------+---------+---------+                                                                                                                                      
  | sparse | 1  | throughput (files/sec)     |     8.940   |    11.100   | +24.16% |  20.28  |                                                                                                                                      
  | sparse | 1  | latency (ms)               |   111.902   |    90.035   | -19.54% | -19.08  |                                                                                                                                      
  | sparse | 4  | throughput (files/sec)     |    19.525   |    19.964   |  +2.25% |   0.53  |                                                                                                                                      
  | sparse | 4  | latency (ms)               |   203.616   |   192.270   |  -5.57% |  -3.13  |                                                                                                                                      
  | dense  | 1  | throughput (files/sec)     |     0.491   |     0.485   |  -1.15% |  -0.74  |                                                                                                                                      
  | dense  | 1  | latency (ms)               |  2039.470   |  2065.425   |  +1.27% |   0.80  |                                                                                                                                      
  | dense  | 4  | throughput (files/sec)     |     1.060   |     1.039   |  -1.95% |  -0.44  |                                                                                                                                      
  | dense  | 4  | latency (ms)               |  3438.972   |  3550.129   |  +3.23% |   0.53  |                                                                                                                                      
  +--------+----+----------------------------+-------------+-------------+---------+---------+

evalParkedSuspensions copied suspensionsParked into a retry queue
unconditionally, but a suspension resolved out-of-band (not via its own
registered wake-up or this method's own retry) stays a member with
isDone already true; feeding it to evalSuspensionQueue then trips that
method's dequeued-suspension-is-never-done invariant. Prune already-done
members out of suspensionsParked before building the retry queue.

DAFFODIL-3065

@jadams-tresys jadams-tresys left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A few cleanup suggestions

Have you verified any of the performance gains on any of our actual schemas? I know I ran our usual performance tests on some of these changes but wanted to check if any other performance testing had been done outside of fabricated test schemas.

Comment on lines +439 to +441
// maybeRegisterWaiterFor, below, reconciles the registration
// against what this block needs, touching nothing when a re-block
// on the same reason is already correctly registered.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does it make sense to have this comment bet 20+ lines away from where the call actually is? I'd probably just remove this comment.

private var soleSuspension: Suspension = null
private var soleCond: () => Boolean = null
private var overflow: mutable.ArrayBuffer[SuspensionWaiterRegistration] = null

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there any gain from having this overflow logic as opposed to just having the mutable.ArrayBuffer and always calling .tail to get the last entry? We can still initialize it with just 1 entry so I wouldn't think there would be a significant allocation overhead and it would make a lot of this much simpler.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it's a performance gain that keeps us from allocating an array buffer on every single suspension waiter (which are mostly zero registrants), as opposed to only when they are actual registrants.

This came out of a profiling session for allocation hotspots

dp
}

@Test def testManyForceRetryCyclesResolveCorrectly(): Unit = {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I feel like the 2 tests in this file could just be TDML tests or should at least make use of the TestUtils code instead of recreating a bunch of existing functionality

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch! Will do

@stevedlawrence stevedlawrence left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1, it feels like this still adds lots of complication that is hard to follow. But I don't see anything clearly wrong and tests seem to pass, and it does show performance gains. I suspect some day we might want to revisit this and see if there is a simpler design that gives similar results, but seems a resonable first step towards improving suspension performance

* this bookkeeping cost on every push isn't worth it normally; flip to
* true only while profiling to check a use's initialSize choice.
*/
var trackMaxSizeReached: Boolean = false

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Feels like this wants to be a val, this should never change during normal usage. This is purely for developers to determine if certain stacks should have a different default size. I don't thin it's unreasonable to require a recompile to do that kind of testing.

}
setFinishedRegistrations(finishedTargets)
registeredAbsBitPosDoses = absBitPosTargets
}

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this logic to track middle doses is new from previous changes. Did you notice if that made any impact to performance. It adds complexity and now I'm wondering if that added complexity is wroth it. Maybe only the start and end DOSes are all that really matters in most cases and we can just wait for those to be finished.

_suspensionWaiter
}

protected def notifySuspensionWaiterIfAllocated(): Unit = {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggest we remove these IfAllocated suffix. It seems reasonble to make this a no-op if there's no suspension waiter. I don't think callers would care. Just add documentation to the functions

// As written, we have a bunch of suspensions that occur, but have
// specifically known length of zero bits. So nothing being written out.
// In that case, why do we need to split at all?
//

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might want to add a TODO to this comment?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comment predates this branch, all we did want wrap it in a try block.. I can add a todo statement tho

// single Maybe. One registry covers every such fact: a suspension that
// only cares about one of them just re-checks that specific fact when
// notified, rather than needing a separate registry per fact.
final class DosRegistrations(

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Feels odd for the Suspension class have something specific about DOSes. I thought the goal of the SuspensionWaiter logic was so that outside things can track whatever state they care about (whehter it be DOSes, Variables, Nodes, etc.), and when their state changes they notify suspension that they are/might be ready to be reevaluated.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are two different registry classes:

  • DataOutputStreamListenerRegistry[L] (DataOutputStream.scala:68-91) lives on the DOS side. Each DirectOrBufferedDataOutputStream owns one, holding the set of listeners waiting on that one DOS's fact (Set[L]). One DOS can have many listeners. This answers "who do I tell when I change?"
  • DosRegistrations (Suspension.scala:309-328) lives on the Suspension side. It holds the set of DOSes this one suspension has registered itself with (Set[DataOutputStream]). One suspension to many DOSes (ex: a length calc suspensions can be watching both its start and end DOS at once). This answers "what do I need to unregister from when I'm done?"

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note this above is specific to DOSes, variable/nodes use suspension waiters maybeRegisteredWaiters path

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How complex would it be for suspensions to use something similar for DOSes? It feels like Suspensions don't want to have to know details about things, it's the SuspensionWaiters that want to know those details.

Falling back to the periodic sweep for the middle-dos case is a latency
detail, not a correctness one. Since the finished and abs-bit-pos DOS
registrations always target the same set once middleDOS tracking is gone,
they're now built as one set instead of two.

Renamed notifySuspensionWaiterIfAllocated/clearSuspensionWaiterIfAllocated
to notifySuspensionWaiter/clearSuspensionWaiter (documented as no-ops), and Suspension's dosListeners
field to registeredDoses to describe what it holds rather than reading
backwards. Added a tunables parameter to TestUtils.testUnparsing and
rewrote TestOutputValueCalcParkedSuspensionRetry and part of
LengthStateWaiterTest to use it instead of hand-rolled compile/unparse
boilerplate. Changed MStack.trackMaxSizeReached from a runtime-settable
var to a final val, a deliberate compile-time constant rather than a
toggle other code can flip back off.

DAFFODIL-3065
@olabusayoT
olabusayoT force-pushed the daf-3065-suspension-tracking branch from 5dc9308 to 336813a Compare September 21, 2026 21:30
Suspension previously implemented DataOutputStreamEventListener
directly and maintained its own DosRegistrations (a Set[DataOutputStream]
it registered itself with via dos.registerListener/removeListener) for
a DOS's abs-bit-pos/zeroLengthStatus values settling. This mirrors the
pattern already used for the "finished" case instead: DataOutputStream
now exposes a settledWaiter: SuspensionWaiter that suspensions register
against directly through the ordinary registerSuspension/removeSuspension
API, the same way LengthState/VariableInstance already do for their own
values. DataOutputStreamEventListener, notifyKnown, and the registerListener/
removeListener methods are gone.

Suspension's own bookkeeping (renamed DosRegistrations to
AdditionalWaiterRegistrations) now tracks a Set[SuspensionWaiter] rather
than a Set[DataOutputStream], since a suspension can still need more than
one such waiter at once (e.g. both its start and end DOS). clearRegisteredWaiter
now takes the waiter being cleared, since a suspension can be dropped from
either its one primary waiter slot or from this additional set, and needs
to know which one is asking so it updates the right side.

DataOutputStreamListenerRegistry (daffodil-io) is unchanged and still
used for the finished-listener case; only its second use (the settled-
value case) migrated away from it.

DAFFODIL-3065
@olabusayoT

Copy link
Copy Markdown
Contributor Author

No performance gains for the added complication

@olabusayoT olabusayoT closed this Sep 28, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants