Add build/write-prefetch unparse path with coroutine-driven write dispatch

daffodil-core/src/main/scala/org/apache/daffodil/core/runtime1/ElementBaseRuntime1Mixin.scala

@@ -27,6 +27,7 @@ import org.apache.daffodil.lib.schema.annotation.props.gen.LengthKind
 import org.apache.daffodil.lib.schema.annotation.props.gen.Representation.Text
 import org.apache.daffodil.lib.util.Delay
 import org.apache.daffodil.lib.util.Maybe
+import org.apache.daffodil.runtime1.dpath.SuspendableExpression
 import org.apache.daffodil.runtime1.dsom.DPathElementCompileInfo
 import org.apache.daffodil.runtime1.processors.ElementRuntimeData
 import org.apache.daffodil.runtime1.processors.RuntimeData
@@ -134,6 +135,19 @@ trait ElementBaseRuntime1Mixin { self: ElementBase =>
     isReferenced || mightHaveSuspensions
   }
 
+  /**
+   * True if the schema has at least one dfdl:outputValueCalc element whose
+   * expression can resolve without writing (schema-wide; consult only via
+   * schemaSet.root). Gates useBuildWritePrefetch: if false, every OVC needs an
+   * actual written byte position, so racing build ahead is never beneficial.
+   */
+  final lazy val hasAnyPrefetchBeneficialOVC: Boolean =
+    schemaSet.allSchemaComponents.exists {
+      case e: ElementBase if e.isOutputValueCalc =>
+        SuspendableExpression.canResolveWithoutWriting(e.ovcCompiledExpression)
+      case _ => false
+    }
+
   final override lazy val dpathCompileInfo = dpathElementCompileInfo
 
   /**

daffodil-core/src/main/scala/org/apache/daffodil/core/runtime1/SchemaSetRuntime1Mixin.scala

@@ -91,7 +91,8 @@ trait SchemaSetRuntime1Mixin {
         root.elementRuntimeData,
         variableMap,
         allLayers,
-        layerRuntimeCompiler
+        layerRuntimeCompiler,
+        root.hasAnyPrefetchBeneficialOVC
       )
     if (root.numComponents > root.numUniqueComponents)
       Logger.log.debug(

daffodil-core/src/main/scala/org/apache/daffodil/core/util/TestUtils.scala

@@ -43,6 +43,7 @@ import org.apache.daffodil.lib.iapi.*
 import org.apache.daffodil.lib.util.*
 import org.apache.daffodil.lib.xml.*
 import org.apache.daffodil.runtime1.iapi.DFDL
+import org.apache.daffodil.runtime1.infoset.InfosetInputter
 import org.apache.daffodil.runtime1.infoset.ScalaXMLInfosetInputter
 import org.apache.daffodil.runtime1.infoset.ScalaXMLInfosetOutputter
 import org.apache.daffodil.runtime1.processors.DataProcessor
@@ -118,9 +119,11 @@ object TestUtils {
     testSchema: scala.xml.Elem,
     infosetXML: Node,
     unparseTo: String,
-    areTracing: Boolean = false
+    areTracing: Boolean = false,
+    tunables: Map[String, String] = Map.empty
   ): java.util.List[api.Diagnostic] = {
-    val compiler = Compiler().withTunable("allowExternalPathExpressions", "true")
+    val compiler =
+      Compiler().withTunable("allowExternalPathExpressions", "true").withTunables(tunables)
     val pf = compiler.compileNode(testSchema)
     if (pf.isError) throwDiagnostics(pf.getDiagnostics)
     var u = saveAndReload(pf.onPath("/").asInstanceOf[DataProcessor])
@@ -198,6 +201,32 @@ object TestUtils {
     p
   }
 
+  /**
+   * Compiles testSchema with the given tunables and returns the resulting
+   * DataProcessor, with no saveAndReload round-trip (unlike compileSchema)
+   * since some callers build test-only state directly off the live object.
+   */
+  def compileForUnparse(
+    testSchema: Node,
+    tunables: Map[String, String] = Map.empty
+  ): DataProcessor = {
+    val pf = Compiler().withTunables(tunables).compileNode(testSchema)
+    if (pf.isError) throwDiagnostics(pf.getDiagnostics)
+    val dp = pf.onPath("/").asInstanceOf[DataProcessor]
+    if (dp.isError) throwDiagnostics(dp.getDiagnostics)
+    dp
+  }
+
+  /**
+   * Builds a fresh InfosetInputter walking infosetXML against dp, already
+   * initialized (root TRD pushed) the same way a real unparse would.
+   */
+  def newInitializedInputter(infosetXML: Node, dp: DataProcessor): InfosetInputter = {
+    val inputter = new InfosetInputter(new ScalaXMLInfosetInputter(infosetXML))
+    inputter.initialize(dp.ssrd.elementRuntimeData, dp.tunables)
+    inputter
+  }
+
   private def runSchemaOnRBC(
     testSchema: Node,
     data: ReadableByteChannel,

daffodil-core/src/main/scala/org/apache/daffodil/lib/util/MStack.scala

@@ -24,6 +24,12 @@ import Maybe.*
 object MStack {
   final case class Mark(val v: Int) extends AnyVal
   val nullMark = Mark(0)
+
+  /**
+   * Off by default: growing past initialSize isn't itself wrong, so paying
+   * this bookkeeping cost on every push isn't worth it normally.
+   */
+  final val trackMaxSizeReached: Boolean = false
 }
 
 /**
@@ -33,35 +39,25 @@ object MStack {
  * catches improper initialization. These were not initializing properly,
  * so the idiom evolved to use the scala initializers.
  */
-final class MStackOfBoolean private ()
-  extends MStack[Boolean]((n: Int) => new Array[Boolean](n), false)
+final class MStackOfBoolean private (initialSize: Int)
+  extends MStack[Boolean]((n: Int) => new Array[Boolean](n), false, initialSize)
 
 object MStackOfBoolean {
-  def apply() = {
-    val stk = new MStackOfBoolean()
-    stk.init()
-    stk
-  }
+  def apply(initialSize: Int = 32) = new MStackOfBoolean(initialSize)
 }
 
-final class MStackOfInt extends MStack[Int]((n: Int) => new Array[Int](n), 0)
+final class MStackOfInt(initialSize: Int)
+  extends MStack[Int]((n: Int) => new Array[Int](n), 0, initialSize)
 
 object MStackOfInt {
-  def apply() = {
-    val stk = new MStackOfInt()
-    stk.init()
-    stk
-  }
+  def apply(initialSize: Int = 32) = new MStackOfInt(initialSize)
 }
 
-final class MStackOfLong extends MStack[Long]((n: Int) => new Array[Long](n), 0L)
+final class MStackOfLong(initialSize: Int)
+  extends MStack[Long]((n: Int) => new Array[Long](n), 0L, initialSize)
 
 object MStackOfLong {
-  def apply() = {
-    val stk = new MStackOfLong()
-    stk.init()
-    stk
-  }
+  def apply(initialSize: Int = 32) = new MStackOfLong(initialSize)
 }
 
 /**
@@ -75,11 +71,11 @@ object MStackOfLong {
  * So we use an Array[AnyRef] as the representation here, and we
  * convert null to Nope, and an actual object reference to One(x)
  */
-final class MStackOfMaybe[T <: AnyRef] {
+final class MStackOfMaybe[T <: AnyRef](initialSize: Int = 32) {
 
   override def toString = delegate.toString
 
-  private val delegate = new MStackOf[T]
+  private val delegate = new MStackOf[T](initialSize)
   private val nullT = null.asInstanceOf[T]
 
   def copyFrom(other: MStackOfMaybe[T]) = delegate.copyFrom(other.delegate)
@@ -120,6 +116,7 @@ final class MStackOfMaybe[T <: AnyRef] {
   def toListMaybe = delegate.toList.map { (x: AnyRef) =>
     Maybe(x) // Scala compiler bug without this cast
   }
+  def maxSizeReached = delegate.maxSizeReached
 }
 
 /**
@@ -135,15 +132,15 @@ final class MStackOfMaybe[T <: AnyRef] {
  * an object reference or null, and call Maybe(thing) explicitly outside the
  * iteration. Maybe(null) is Nope, and Maybe(thing) is One(thing) if thing is not null.
  */
-final class MStackOf[T <: AnyRef] extends Serializable {
+final class MStackOf[T <: AnyRef](initialSize: Int = 32) extends Serializable {
 
   override def toString = delegate.toString
 
   def copyFrom(other: MStackOf[T]) = delegate.copyFrom(other.delegate)
 
   @inline final def length = delegate.length
 
-  private val delegate = MStackOfAnyRef()
+  private val delegate = MStackOfAnyRef(initialSize)
 
   @inline final def mark = delegate.mark
   @inline final def reset(m: MStack.Mark) = delegate.reset(m)
@@ -156,22 +153,23 @@ final class MStackOf[T <: AnyRef] extends Serializable {
   @inline final def isEmpty = delegate.isEmpty
   def clear() = delegate.clear()
   def toList = delegate.toList
+  def maxSizeReached = delegate.maxSizeReached
 
   def iterator = delegate.iterator.asInstanceOf[ResettableIterator[T]]
 
   lazy val iter = delegate.iter.asInstanceOf[ResettableIterator[T]]
 
 }
 
-private[util] final class MStackOfAnyRef private ()
-  extends MStack[AnyRef]((n: Int) => new Array[AnyRef](n), null.asInstanceOf[AnyRef])
+private[util] final class MStackOfAnyRef private (initialSize: Int)
+  extends MStack[AnyRef](
+    (n: Int) => new Array[AnyRef](n),
+    null.asInstanceOf[AnyRef],
+    initialSize
+  )
 
 object MStackOfAnyRef {
-  def apply() = {
-    val stk = new MStackOfAnyRef()
-    stk.init()
-    stk
-  }
+  def apply(initialSize: Int = 32) = new MStackOfAnyRef(initialSize)
 }
 
 /**
@@ -184,16 +182,23 @@ object MStackOfAnyRef {
  */
 protected abstract class MStack[@specialized T] private[util] (
   arrayAllocator: (Int) => Array[T],
-  nullValue: T
+  nullValue: T,
+  initialSize: Int = 32
 ) {
 
   private var index = 0
-  private var table: Array[T] = null
+  private var table: Array[T] = arrayAllocator(initialSize)
 
-  def init(): Unit = {
-    index = 0
-    table = arrayAllocator(32)
-  }
+  private var maxSizeReached_ = 0
+
+  /**
+   * The largest this stack's length has ever grown to, across its
+   * whole lifetime (not just its current length; pops don't reduce
+   * this). Diagnostic only: useful for profiling to check whether a
+   * particular use's initialSize is well-chosen, not for any runtime
+   * decision. Always 0 unless MStack.trackMaxSizeReached is enabled.
+   */
+  final def maxSizeReached: Int = maxSizeReached_
 
   def copyFrom(other: MStack[T]): Unit = {
     this.index = other.index
@@ -212,6 +217,9 @@ protected abstract class MStack[@specialized T] private[util] (
       }
 
     }
+    if (MStack.trackMaxSizeReached && other.maxSizeReached_ > this.maxSizeReached_) {
+      this.maxSizeReached_ = other.maxSizeReached_
+    }
   }
   // private var currentIteratorIndex = -1
 
@@ -254,6 +262,9 @@ protected abstract class MStack[@specialized T] private[util] (
     if (index == table.length) table = growArray(table)
     table(index) = x
     index += 1
+    if (MStack.trackMaxSizeReached && index > maxSizeReached_) {
+      maxSizeReached_ = index
+    }
   }
 
   /**

daffodil-core/src/main/scala/org/apache/daffodil/runtime1/dpath/SuspendableExpression.scala

@@ -32,12 +32,25 @@ import org.apache.daffodil.runtime1.processors.unparsers.UState
  * dfdl:setVariable expressions (which variables are in-turn used by
  * dfdl:outputValueCalc.
  */
+object SuspendableExpression {
+
+  /** True unless expr calls valueLength/contentLength (which needs a
+   * written DOS position, not merely a known value); other reads
+   * resolve once the value is known. A static property of expr alone;
+   * it doesn't know whether the referenced position is already written. */
+  def canResolveWithoutWriting(expr: CompiledExpression[AnyRef]): Boolean =
+    expr.valueReferencedElementInfos.isEmpty && expr.contentReferencedElementInfos.isEmpty
+}
+
 trait SuspendableExpression extends Suspension {
 
   override val isReadOnly = true
 
   protected def expr: CompiledExpression[AnyRef]
 
+  override def canResolveWithoutWriting: Boolean =
+    SuspendableExpression.canResolveWithoutWriting(expr)
+
   override def toString =
     "SuspendableExpression(" + rd.diagnosticDebugName + ", expr=" + expr.prettyExpr + ")"
 

daffodil-core/src/main/scala/org/apache/daffodil/runtime1/infoset/InfosetImpl.scala

@@ -176,13 +176,13 @@ sealed trait DINode {
   private var _isFinal: Boolean = false
 
   /**
-  * Use to mark a node as final, indicating that its value will not change or have
-  * any children added to it. Setting an element as final does not preclude it from
-  * being discarded by backtracking, i.e. it is only locally final, but might still
-  * be inside an enclosing PoU.
-  *
-  * This cannot be called if an element is already marked as final to help ensure
-  * correct use.
+   * Use to mark a node as final, indicating that its value will not change or have
+   * any children added to it. Setting an element as final does not preclude it from
+   * being discarded by backtracking, i.e. it is only locally final, but might still
+   * be inside an enclosing PoU.
+   *
+   * This cannot be called if an element is already marked as final to help ensure
+   * correct use.
    */
   def setFinal(): Unit = {
     Assert.invariant(!_isFinal)
@@ -1334,6 +1334,11 @@ final class DIArray(
 
   final def freeChildIfNoLongerNeeded(index: Int, doFree: Boolean): Unit = {
     val node = _contents(index)
+    // Under build/write-prefetch, both BuildState and write's writeContent
+    // dispatch can reach this slot; write may have already freed (nulled) it
+    // by the time build's independent call arrives here (build's doFree is
+    // always false, so this would only mark wouldHaveBeenFreed, moot here).
+    if (node == null) return
     if (!node.erd.dpathElementCompileInfo.isReferencedByExpressions) {
       if (doFree) {
         // set to null so that the garbage collector can free this node
@@ -1825,6 +1830,9 @@ sealed class DIComplex(override val erd: ElementRuntimeData)
 
   def freeChildIfNoLongerNeeded(index: Int, doFree: Boolean): Unit = {
     val node = child(index)
+    // Under build/write-prefetch, write may have already freed (nulled)
+    // this slot by the time build's independent call arrives here.
+    if (node == null) return
     if (!node.erd.dpathElementCompileInfo.isReferencedByExpressions) {
       if (doFree) {
         // set to null so that the garbage collector can free this node