stevedlawrence commented on code in PR #1717:
URL: https://github.com/apache/daffodil/pull/1717#discussion_r3862624332
##########
daffodil-core/src/main/scala/org/apache/daffodil/io/DirectOrBufferedDataOutputStream.scala:
##########
@@ -576,6 +577,35 @@ class DirectOrBufferedDataOutputStream private[io] (
val f = _following.get
f.maybeAbsBitPos0b // requesting this pulls the absolute position info
forward.
}
+
+ notifyFinishedListeners()
+ }
+ }
+
+ // 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
+
+ def registerFinishedListener(fl: FinishedListener): Unit = {
+ if (maybeFinishedListeners.isEmpty) {
+ maybeFinishedListeners = One(mutable.HashSet.empty)
+ }
+ maybeFinishedListeners.get.add(fl)
+ }
+
+ def removeFinishedListener(fl: FinishedListener): Unit = {
+ if (maybeFinishedListeners.isDefined) {
+ maybeFinishedListeners.get.remove(fl)
+ }
+ }
+
+ // setFinished() is one-shot per DOS (Assert.usage(!isFinished) at its
+ // top), so there's no need to keep this registration around afterward.
+ private def notifyFinishedListeners(): Unit = {
+ if (maybeFinishedListeners.isDefined) {
+ val toNotify = maybeFinishedListeners.get.toArray
+ maybeFinishedListeners.get.clear()
Review Comment:
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.
##########
daffodil-core/src/main/scala/org/apache/daffodil/io/DirectOrBufferedDataOutputStream.scala:
##########
@@ -576,6 +577,35 @@ class DirectOrBufferedDataOutputStream private[io] (
val f = _following.get
f.maybeAbsBitPos0b // requesting this pulls the absolute position info
forward.
}
+
+ notifyFinishedListeners()
+ }
+ }
+
+ // 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
+
+ def registerFinishedListener(fl: FinishedListener): Unit = {
+ if (maybeFinishedListeners.isEmpty) {
+ maybeFinishedListeners = One(mutable.HashSet.empty)
+ }
+ maybeFinishedListeners.get.add(fl)
+ }
+
+ def removeFinishedListener(fl: FinishedListener): Unit = {
+ if (maybeFinishedListeners.isDefined) {
+ maybeFinishedListeners.get.remove(fl)
+ }
+ }
+
+ // setFinished() is one-shot per DOS (Assert.usage(!isFinished) at its
+ // top), so there's no need to keep this registration around afterward.
+ private def notifyFinishedListeners(): Unit = {
+ if (maybeFinishedListeners.isDefined) {
+ val toNotify = maybeFinishedListeners.get.toArray
Review Comment:
Can we just do `maybeFinishedListeners.get.foreach(_.notifyFinished())` and
avoid the array allocation?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspendableOperation.scala:
##########
@@ -79,6 +79,10 @@ trait SuspendableOperation extends Suspension {
if (ustate.currentInfosetNodeMaybe.isDefined)
ustate.currentInfosetNodeMaybe.get
else "No Node"
block(nodeOpt, ustate.getDataOutputStream, 0, e)
+ // Registers the same targeted wake-up DPath.scala's expression
+ // evaluation does, if e has a known SuspensionWaiter to
+ // register with; a no-op for any other RetryableException.
+ maybeRegisterWaiterFor(e)
Review Comment:
Why do we do this both here and in DPath.scala? Wouldn't DPath.scala have
already registered this suspension? Feels like DPath is a better place for that
since it knows all about the diferent kinds of exceptions that suspensions
might hit. I'm not sure we want the Suspension class to have to know about
those exceptions.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -228,9 +253,53 @@ trait Suspension extends Serializable {
final def isMakingProgress = isMakingProgress_
+ // Which SuspensionWaiter (if any) this suspension is registered with.
+ // block() unconditionally deregisters this before any retry, so
+ // registerWaiter never finds one already set.
+ private var maybeRegisteredWaiter: Maybe[SuspensionWaiter] = Nope
+
+ /**
+ * True exactly when a targeted wake-up is registered against some
+ * SuspensionWaiter (set only from registerWaiter). SuspensionTracker's
+ * periodic sweep uses this to skip re-running doTask until that
+ * wake-up fires.
Review Comment:
Aren't suspensions that have an associated waiter in the Parked queue, which
is never evaluated? I thought SuspensionWaiters just move things from Parked to
Young when they are ready to be tried again?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -94,23 +152,38 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
}
/**
- * Attempt to evaluate suspensions on the provie queue. Keep repeating the
- * evaluates as long as some progress is being made. Suspensions that
- * evaluate sucessfully are removed from the queue. Once suspensions make no
- * further progress and are all blocked, we return. Blocked suspensions put
- * back on the same queue.
+ * Repeatedly attempts suspensions on queue until no progress is made;
+ * still-blocked ones go back on the queue. A suspension with
+ * isWaitingOnWaiter true is parked instead of really attempted: a
+ * guaranteed external wake-up already exists for it.
*/
- private def evalSuspensionQueue(queue: Queue[Suspension]): Unit = {
+ private def evalSuspensionQueue(
+ queue: Queue[Suspension],
+ skipWaiters: Boolean = false
Review Comment:
Is it possible to avoid the skipWaiter? I think the implication is that
parked suspensions could end up in the young or old queues which feels
incorrect to me. Feels like parked suspensions never want to be mied into those
queus. We can move things around between queues, but being in multiple feels
like it could lead to complications.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -94,23 +152,38 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
}
/**
- * Attempt to evaluate suspensions on the provie queue. Keep repeating the
- * evaluates as long as some progress is being made. Suspensions that
- * evaluate sucessfully are removed from the queue. Once suspensions make no
- * further progress and are all blocked, we return. Blocked suspensions put
- * back on the same queue.
+ * Repeatedly attempts suspensions on queue until no progress is made;
+ * still-blocked ones go back on the queue. A suspension with
+ * isWaitingOnWaiter true is parked instead of really attempted: a
+ * guaranteed external wake-up already exists for it.
*/
- private def evalSuspensionQueue(queue: Queue[Suspension]): Unit = {
+ private def evalSuspensionQueue(
+ queue: Queue[Suspension],
+ skipWaiters: Boolean = false
+ ): Unit = {
var countOfNotMakingProgress = 0
while (!queue.isEmpty && countOfNotMakingProgress < queue.length) {
val s = queue.dequeue()
- suspensionStatRuns += 1
- s.runSuspension()
- if (!s.isDone) queue.enqueue(s)
- if (s.isDone || s.isMakingProgress) {
+ if (s.isDone) {
+ // resolved out of band; queue got smaller for free, so this
Review Comment:
How can it get resolved out of band? Isn't the only place we eval
suspensions in this function? And even if a suspension gets notified because
another suspension was run, that just moves the suspension from to young, it
doesn't evaluate the suspnension.
##########
daffodil-core/src/main/scala/org/apache/daffodil/lib/util/MStack.scala:
##########
@@ -33,35 +33,25 @@ object MStack {
* catches improper initialization. These were not initializing properly,
* so the idiom evolved to use the scala initializers.
*/
-final class MStackOfBoolean private ()
Review Comment:
Thinking about these MStack changes, we might want to add some diagnostic
ability to get an idea of how big our various MStacks are likely to grow. If a
certain MStack almost always grows above a certain size, we might want to
consider a larger initial size to avoid extra allocations and copies.
##########
daffodil-core/src/main/scala/org/apache/daffodil/io/DirectOrBufferedDataOutputStream.scala:
##########
@@ -576,6 +577,35 @@ class DirectOrBufferedDataOutputStream private[io] (
val f = _following.get
f.maybeAbsBitPos0b // requesting this pulls the absolute position info
forward.
}
+
+ notifyFinishedListeners()
+ }
+ }
+
+ // 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
+
+ def registerFinishedListener(fl: FinishedListener): Unit = {
+ if (maybeFinishedListeners.isEmpty) {
+ maybeFinishedListeners = One(mutable.HashSet.empty)
+ }
+ maybeFinishedListeners.get.add(fl)
+ }
+
+ def removeFinishedListener(fl: FinishedListener): Unit = {
+ if (maybeFinishedListeners.isDefined) {
+ maybeFinishedListeners.get.remove(fl)
+ }
+ }
+
+ // setFinished() is one-shot per DOS (Assert.usage(!isFinished) at its
+ // top), so there's no need to keep this registration around afterward.
+ private def notifyFinishedListeners(): Unit = {
+ if (maybeFinishedListeners.isDefined) {
+ val toNotify = maybeFinishedListeners.get.toArray
+ maybeFinishedListeners.get.clear()
+ toNotify.foreach(_.notifyFinished())
}
Review Comment:
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).
##########
daffodil-core/src/main/scala/org/apache/daffodil/io/DirectOrBufferedDataOutputStream.scala:
##########
@@ -576,6 +577,35 @@ class DirectOrBufferedDataOutputStream private[io] (
val f = _following.get
f.maybeAbsBitPos0b // requesting this pulls the absolute position info
forward.
}
+
+ notifyFinishedListeners()
+ }
+ }
+
+ // 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
Review Comment:
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.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/unparsers/UState.scala:
##########
@@ -707,7 +712,10 @@ final class UStateMain private (
* All the other clones used for outputValueCalc, those never
* need to add any.
*/
- private val suspensionTracker =
+ // private[processors], not private: Suspension.suspend stashes this so
+ // notifyWaiters can hand a suspension back to its tracker instead of
+ // running it directly.
+ private[processors] val suspensionTracker =
Review Comment:
Suggest we just make thsi public. I think we are moving to a world where
lots of things might want to interact with the suspension tracker. And
"processors" already contains pretty much all our runtime things, so I don't
think limiting to that does much.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/infoset/InfosetImpl.scala:
##########
@@ -533,7 +536,7 @@ sealed trait DITerm {
* are defined, then that start pos is a relative start pos, and we'll need to
compute the
* absolute start pos once we find out the absolute start pos of the data
output stream.
*/
-sealed abstract class LengthState(ie: DIElement) {
+sealed abstract class LengthState(ie: DIElement) extends FinishedListener {
Review Comment:
Does it simplify things if we register the SuspensionWaiter with DOSs
instead of FinishedListeners? It seems the FinishedListener just delegates to
the suspension waiter so calling the SuspensionWater directly just removes that
level of indirection and simplifies APIs a bit.
And FinishedListeners are only used in the context of SuspensionWaiters, I
don't know of anything else that needs DOS notifications?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/DPath.scala:
##########
@@ -250,8 +250,15 @@ final class RuntimeExpressionDPath[T <: AnyRef](
}
case ve: VariableException =>
whereBlockedInfo.block(ve.qname, ve.context, 0, ve)
+ // Registers a targeted wake-up for when the variable is set,
+ // alongside the periodic-sweep blocking above.
+ whereBlockedInfo.maybeRegisterWaiterFor(ve)
Review Comment:
Can we split variable waiters into a separate PR? This PR is already complex
enough.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/DPathRuntime.scala:
##########
@@ -205,7 +206,11 @@ case class VRef(vrd: VariableRuntimeData, context:
ThrowsSDE) extends RecipeOp {
override def run(dstate: DState): Unit = {
if (dstate.parseOrUnparseState.isEmpty)
- throw new VariableHasNoValue(vrd.globalQName, vrd)
+ // No live parse/unparse state to look up the real VariableInstance
+ // in, so there's nothing a targeted wake-up could ever register
+ // against anyway; a fresh, disconnected instance is a harmless
+ // placeholder here.
Review Comment:
I'm not sure I understand why a disconnected instance is harmless, and it
feels questionable to me? When this exception is handled, we'll end up
registering a Suspension with this new `VariableInstance`, but nothing,
including the VariableMap, knows about this instance. So that means when
something does VariableMap.setVariable, this VariableInstance will never get
changed.
That said, I'm not even sure when parseOrUnparseState would be empty. Is
that just during schema compilation, in which case we don't won't be doing
suspensions anyway? And that's why a dummy value is safe? If that's the case, I
think this comment needs to be expanded to make that more clear.
##########
daffodil-core/src/main/scala/org/apache/daffodil/io/DataOutputStream.scala:
##########
@@ -34,6 +34,16 @@ object ZeroLengthStatus {
object Unknown extends ZeroLengthStatus
}
+/**
+ * Callback for code outside daffodil-io that wants to know when a
+ * DataOutputStream becomes finished, without polling isFinished. Letting
+ * daffodil-io depend only on this trait, not on whoever implements it,
+ * keeps the dependency one-directional.
+ */
+trait FinishedListener {
Review Comment:
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.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/infoset/InfosetImpl.scala:
##########
@@ -572,12 +575,59 @@ sealed abstract class LengthState(ie: DIElement) {
var maybeEndPos0bInBits: MaybeULong = MaybeULong.Nope
var maybeComputedLengthInBits: MaybeULong = MaybeULong.Nope
+ /**
+ * Suspensions blocked specifically on this element's length, registered
+ * via suspensionWaiter and given a real retry once the length becomes
+ * computable. A separate SuspensionWaiter per waiter kind (were one
+ * ever added) so each can be notified on its own trigger.
+ */
+ val suspensionWaiter: SuspensionWaiter = new SuspensionWaiter {
+ // A notify (from a position setter, recheckStreams, or a DOS
+ // finishing) is only ever a hint that the length might be known now,
+ // never a guarantee - re-verify before waking anything up.
+ override def notifySuspensions(): Unit = {
+ if (maybeLengthInBits().isDefined) super.notifySuspensions()
+ }
+ }
+
+ // private[infoset], not private, so tests in this package can assert on it
directly.
+ private[infoset] def isRegisteredWaiter(s: Suspension): Boolean =
+ suspensionWaiter.isRegisteredSuspension(s)
+
+ override def notifyFinished(): Unit = suspensionWaiter.notifySuspensions()
+
+ // The one DirectOrBufferedDataOutputStream (if any) that maybeLengthInBits
+ // is currently waiting to finish - discovered while computing it, not a
+ // fixed reference like maybeStartDataOutputStream/maybeEndDataOutputStream.
+ private var maybeBlockingDos: Maybe[DirectOrBufferedDataOutputStream] = Nope
Review Comment:
It'm wondering if this is too pessimistic? It assumes there is a single DOS
that must become final before this can make progress, but I'm not sure that's
always the case. maybeLengthInBits has a number of different ways it can
calculate the length, and it uses different methods depending on the the states
of the start/end DOS.
So for example, the first time maybeLengthInBits runs it might drop into the
the logic where it needs a middle DOS to be final. But then if it were run
again at some point later, maybe the absolute positions of the start and end
DOSs might have been resolved and so we could calculate the length without
needing that middle DOS that we are blocking on?
Instead of blocking and waiting for a single DOS, what if instead we waiting
on a bunch of DOSs, including both startDOS and endDOSs as well as any middle
DOS that fail. Any of those can notify the listeners (and do so on any state
change rather than just final), and when a DOS state changes we can try to
recalculate the length.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -186,7 +189,29 @@ trait Suspension extends Serializable {
savedUstate_ = cloneUState
- ustate.asInstanceOf[UStateMain].addSuspension(this)
+ val mainUState = ustate.asInstanceOf[UStateMain]
+ maybeTracker = One(mainUState.suspensionTracker)
Review Comment:
Instead of copying the tracker out of the ustate, thoughts on changing
UStateForSuspension so that it also has a reference of the tracker? That way
anything that's doing anything with UState (which is most things, including
Suspensions) always has access to the tracker, even if it's UStateForSuspension
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -228,9 +253,53 @@ trait Suspension extends Serializable {
final def isMakingProgress = isMakingProgress_
+ // Which SuspensionWaiter (if any) this suspension is registered with.
+ // block() unconditionally deregisters this before any retry, so
+ // registerWaiter never finds one already set.
+ private var maybeRegisteredWaiter: Maybe[SuspensionWaiter] = Nope
Review Comment:
Why do we need to keep track of the registered waiter? Isn't the waiter the
thing that keeps track of the suspension and tells the tracker to move the
suspension to young, at which point the waiter can can stop waiting on the
suspension? Might siplify things if the Suspension don't need to care about the
waiters. For example, now we don't have to deregsiter the suspension from the
waiter in `block()` and make sure things stay in sync.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/Suspension.scala:
##########
@@ -228,9 +253,53 @@ trait Suspension extends Serializable {
final def isMakingProgress = isMakingProgress_
+ // Which SuspensionWaiter (if any) this suspension is registered with.
+ // block() unconditionally deregisters this before any retry, so
+ // registerWaiter never finds one already set.
+ private var maybeRegisteredWaiter: Maybe[SuspensionWaiter] = Nope
+
+ /**
+ * True exactly when a targeted wake-up is registered against some
+ * SuspensionWaiter (set only from registerWaiter). SuspensionTracker's
+ * periodic sweep uses this to skip re-running doTask until that
+ * wake-up fires.
+ */
+ final def isWaitingOnWaiter: Boolean = maybeRegisteredWaiter.isDefined
+
+ final def registerWaiter(w: SuspensionWaiter): Unit = {
+ Assert.invariant(maybeRegisteredWaiter.isEmpty)
+ maybeRegisteredWaiter = One(w)
+ w.registerSuspension(this)
+ }
+
+ /**
+ * Registers a targeted wake-up when exc (the reason this suspension
+ * just blocked, via block() immediately beforehand) has a known
+ * SuspensionWaiter to register with - a no-op otherwise. Shared by
+ * DPath.scala's expression evaluation and SuspendableOperation's
+ * retry loop.
+ */
+ final def maybeRegisterWaiterFor(exc: AnyRef): Unit = exc match {
+ case noLength: InfosetLengthUnknownException =>
+ registerWaiter(noLength.lengthState.suspensionWaiter)
+ case noVar: VariableException =>
+ registerWaiter(noVar.variableInstance.suspensionWaiter)
+ case _ => // no targeted wake-up available for any other blocking reason
+ }
Review Comment:
I'm not sure the Suspension class wants to have to know about all the
different kinds of exceptions. That feels like it wants to be handled in DPath,
which I think you already do?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -65,17 +82,58 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
}
}
- /**
- * Evaluates all suspensions until either they are all evaluated or a
- * deadlock is detected. This moves all young suspensions to the old queue,
- * and evaluates all old suspensions. If the old queue is non-empty, that
- * means some suspensions are blocked, likely due to a circular deadlock, and
- * we output diagnostics.
- */
- def requireFinal(): Unit = {
+ // Some suspensions only ever resolve through a real, unconditional
+ // retry rather than their own registered wake-up actually firing (a
+ // length that only becomes computable through the DOS-splitting
+ // machinery's cumulative progress, not one identifiable event). This
+ // bounds how long such a suspension waits to requireFinal.
Review Comment:
Is it possible to just make our suspension notifications more careful about
how they notify and make sure they are waiting on everything that could allow
them to resolve. Or is it really not possible in all cases.
It feels like periodically evaluating parked suspensions is not that much
different from just putting things in old. I guess the main difference is that
we can bump things to Young, which we can't currently do with old?
And maybe it just requires slightly more complex logic in the how we track
the things that notified the waiters? For example, each DOS has a list of
things it will notify. Maybe as DOS become final and foldedinto other DOS they
copy the list finished listeners.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -65,17 +82,58 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
}
}
- /**
- * Evaluates all suspensions until either they are all evaluated or a
- * deadlock is detected. This moves all young suspensions to the old queue,
- * and evaluates all old suspensions. If the old queue is non-empty, that
- * means some suspensions are blocked, likely due to a circular deadlock, and
- * we output diagnostics.
- */
- def requireFinal(): Unit = {
+ // Some suspensions only ever resolve through a real, unconditional
+ // retry rather than their own registered wake-up actually firing (a
+ // length that only becomes computable through the DOS-splitting
+ // machinery's cumulative progress, not one identifiable event). This
+ // bounds how long such a suspension waits to requireFinal.
+ private def evalParkedSuspensions(): Unit = {
+ if (suspensionsParked.isEmpty) return
+ val toRetry = new Queue[Suspension]
+ toRetry ++= suspensionsParked
+ suspensionsParked.clear()
+ evalSuspensionQueue(toRetry)
+ // Re-park only what's still waiting on a registered waiter; anything
+ // else left over blocked on something unrelated, so it belongs back
+ // in the normal rotation to get real retries again, not skip-parked
+ // forever.
+ toRetry.foreach { s =>
+ if (s.isWaitingOnWaiter) suspensionsParked.add(s) else
suspensionsOld.enqueue(s)
Review Comment:
I see, so this is why suspensions need to know whatis waiting on them? Since
we periodically evaluate the parked once, and we essentilly need be able to
know which ones succeeded and which ones didn't and are still waiting?
Feels like things are simplified quite a bit if we can be more confident
that parked suspensions will eventually get notified.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionWaiter.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.daffodil.runtime1.processors
+
+import scala.collection.mutable
+
+/**
+ * Tracks suspensions parked waiting on some state change that might make
+ * them resolvable (a LengthState's length becoming known, a variable
+ * being set, etc.), and moves them back to young once notified.
+ *
+ * A call to notifySuspensions() is only ever a hint that a retry might
+ * now succeed, never a guarantee - the state that changed might not be
+ * the thing this particular waiter's suspensions actually need. Override
+ * notifySuspensions() to re-verify before calling super.notifySuspensions()
+ * wherever that distinction matters; the default just unparks
+ * unconditionally, which is correct when the caller already knows the
+ * state is fully resolved (e.g. a variable that was just set).
+ */
+class SuspensionWaiter {
+
+ private val suspensions: mutable.HashSet[Suspension] = mutable.HashSet.empty
Review Comment:
Similar to another comment, we might want to consider a immtuable var Set, I
doubt the suspension list ever gets very large.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/VariableMap1.scala:
##########
@@ -103,6 +103,27 @@ class VariableInstance private (val rd:
VariableRuntimeData) extends Serializabl
// either the defaultValue expression or by an external binding
var firstInstanceInitialValue: DataValuePrimitiveNullable = DataValue.NoValue
+ // 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
Review Comment:
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.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -94,23 +152,38 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
}
/**
- * Attempt to evaluate suspensions on the provie queue. Keep repeating the
- * evaluates as long as some progress is being made. Suspensions that
- * evaluate sucessfully are removed from the queue. Once suspensions make no
- * further progress and are all blocked, we return. Blocked suspensions put
- * back on the same queue.
+ * Repeatedly attempts suspensions on queue until no progress is made;
+ * still-blocked ones go back on the queue. A suspension with
+ * isWaitingOnWaiter true is parked instead of really attempted: a
+ * guaranteed external wake-up already exists for it.
*/
- private def evalSuspensionQueue(queue: Queue[Suspension]): Unit = {
+ private def evalSuspensionQueue(
+ queue: Queue[Suspension],
+ skipWaiters: Boolean = false
+ ): Unit = {
var countOfNotMakingProgress = 0
while (!queue.isEmpty && countOfNotMakingProgress < queue.length) {
val s = queue.dequeue()
- suspensionStatRuns += 1
- s.runSuspension()
- if (!s.isDone) queue.enqueue(s)
- if (s.isDone || s.isMakingProgress) {
+ if (s.isDone) {
+ // resolved out of band; queue got smaller for free, so this
+ // counts as progress the same as a successful run below.
+ countOfNotMakingProgress = 0
+ } else if (skipWaiters && s.isWaitingOnWaiter) {
Review Comment:
I see, another reason why suspensions need to know if they are parked or
not, because the suspension tracker must be the thing that moves them to the
parked set after evaluating it. Can this be renamed isParked instead of
isWaitingOnWaiter?
So just to confirm, the flow is something like
1. We evaluate a Suspension from this queue
2. That fails, something detects that failure and wants to park it, so it
sets up a SuspensionWaiter, which sets isParked = true, but does not actually
add it to the parked set
3. We return back to here, this detects isParked is true and instead of
requeing it adds it to the suspensionsParked
I think we need a big comment somewhere that explains how all this parked
stuff interacts with the young and old queues. There's a lot of different
interactions and it's not totally clear exactly how it's expected to work.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionTracker.scala:
##########
@@ -42,20 +59,20 @@ class SuspensionTracker(suspensionWaitYoung: Int,
suspensionWaitOld: Int) {
/**
* Attempts to evaluate suspensions. Old suspensions are evaluated less
- * frequently than young suspensions. Any young suspensions that fail to
- * 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.
+ * frequently than young suspensions. A suspension with isWaitingOnWaiter
+ * true has a targeted wake-up already registered, so it's parked here
+ * instead of wasting a retry on it.
*/
- def evalSuspensions(): Unit = {
+ def evalSuspensions(): Unit = evalSuspensionsThrottled()
+
+ private def evalSuspensionsThrottled(): Unit = {
if (count % suspensionWaitOld == 0) {
- evalSuspensionQueue(suspensionsOld)
+ evalSuspensionQueue(suspensionsOld, skipWaiters = true)
+ evalParkedSuspensions()
Review Comment:
Why are we evaluated parked suspensions here? Don't we know parked
suspensions are only evaluated when they are notified, and if it's parked
doesn't that imply nothing has notified it?
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionWaiter.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.daffodil.runtime1.processors
+
+import scala.collection.mutable
+
+/**
+ * Tracks suspensions parked waiting on some state change that might make
+ * them resolvable (a LengthState's length becoming known, a variable
+ * being set, etc.), and moves them back to young once notified.
+ *
+ * A call to notifySuspensions() is only ever a hint that a retry might
+ * now succeed, never a guarantee - the state that changed might not be
+ * the thing this particular waiter's suspensions actually need. Override
+ * notifySuspensions() to re-verify before calling super.notifySuspensions()
+ * wherever that distinction matters; the default just unparks
+ * unconditionally, which is correct when the caller already knows the
+ * state is fully resolved (e.g. a variable that was just set).
+ */
+class SuspensionWaiter {
+
+ private val suspensions: mutable.HashSet[Suspension] = mutable.HashSet.empty
+
+ private[runtime1] def registerSuspension(s: Suspension): Unit = {
+ suspensions.add(s)
+ }
+
+ private[runtime1] def removeSuspension(s: Suspension): Unit = {
+ suspensions.remove(s)
+ }
+
+ private[runtime1] def isRegisteredSuspension(s: Suspension): Boolean =
+ suspensions.contains(s)
Review Comment:
Suggest we just make these public
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/VariableMap1.scala:
##########
@@ -103,6 +103,27 @@ class VariableInstance private (val rd:
VariableRuntimeData) extends Serializabl
// either the defaultValue expression or by an external binding
var firstInstanceInitialValue: DataValuePrimitiveNullable = DataValue.NoValue
+ // 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
+ // needs a re-verify override - by the time notifySuspensions() is
+ // called below, the value truly is available.
+ //
+ // transient: VariableInstance is part of the compiled schema's
+ // serialized state (saved/reloaded via DataProcessor.save), but a
+ // waiter only ever holds transient in-progress suspensions, never
+ // anything meaningful to persist across a save/reload. A plain var
+ // reinitialized in readObject rather than a lazy val: setVariable and
+ // 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
Review Comment:
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.
##########
daffodil-core/src/main/scala/org/apache/daffodil/runtime1/processors/SuspensionWaiter.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.daffodil.runtime1.processors
+
+import scala.collection.mutable
+
+/**
+ * Tracks suspensions parked waiting on some state change that might make
+ * them resolvable (a LengthState's length becoming known, a variable
+ * being set, etc.), and moves them back to young once notified.
+ *
+ * A call to notifySuspensions() is only ever a hint that a retry might
+ * now succeed, never a guarantee - the state that changed might not be
+ * the thing this particular waiter's suspensions actually need. Override
+ * notifySuspensions() to re-verify before calling super.notifySuspensions()
+ * wherever that distinction matters; the default just unparks
+ * unconditionally, which is correct when the caller already knows the
+ * state is fully resolved (e.g. a variable that was just set).
+ */
+class SuspensionWaiter {
+
+ private val suspensions: mutable.HashSet[Suspension] = mutable.HashSet.empty
+
+ private[runtime1] def registerSuspension(s: Suspension): Unit = {
+ suspensions.add(s)
+ }
+
+ private[runtime1] def removeSuspension(s: Suspension): Unit = {
+ suspensions.remove(s)
+ }
+
+ private[runtime1] def isRegisteredSuspension(s: Suspension): Boolean =
+ suspensions.contains(s)
+
+ def isEmpty: Boolean = suspensions.isEmpty
+
+ def clear(): Unit = suspensions.clear()
+
+ // Hands every waiter to its tracker (Suspension.moveFromParkedToYoung)
+ // for a real attempt on its next sweep; SuspensionTracker is the only
+ // caller of runSuspension.
+ def notifySuspensions(): Unit = {
+ if (suspensions.nonEmpty) {
+ val toMove = suspensions.toArray
Review Comment:
This probably just wants to be
```scala
suspensions.foreach { ... }
suspensions.clear()
```
No need to copy to an array first
##########
daffodil-core/src/main/scala/org/apache/daffodil/unparsers/runtime1/ElementUnparser.scala:
##########
@@ -366,7 +366,11 @@ class ElementOVCSpecifiedLengthUnparser(
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.
+ // 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.
Review Comment:
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.
--
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]
