stevedlawrence commented on code in PR #1717:
URL: https://github.com/apache/daffodil/pull/1717#discussion_r3843490248
##########
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:
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:
```scala
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
```scala
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
```scala
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
```scala
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.
--
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]
