Improve unparse suspension resolution and reduce clone overhead

daffodil-core/src/main/scala/org/apache/daffodil/io/DataOutputStream.scala

@@ -34,6 +34,62 @@ object ZeroLengthStatus {
   object Unknown extends ZeroLengthStatus
 }
 
+/**
+ * Callback for code outside daffodil-io that wants to know, without
+ * polling for it, the moment some fact about a specific DataOutputStream
+ * settles into its final, permanent value: currently its absolute bit
+ * position (some suspensions only need this, not the DOS to be fully
+ * finished, to become resolvable: e.g. alignment fill, or a length
+ * calculation where the other endpoint is already absolute) or its
+ * zeroLengthStatus (e.g. deciding whether to suppress a separator).
+ * Unlike the finished notification below, this one's only implementer
+ * (Suspension) isn't a SuspensionWaiter; each
+ * suspension may be watching a different DOS from its own current
+ * writing context, so it registers itself directly. That's why this
+ * stays a generic listener trait instead of also being collapsed onto
+ * SuspensionWaiter. A single notifyKnown covers every such fact (rather
+ * than a differently-named method per fact) since every implementer to
+ * date reacts the same way regardless of which one changed: worth a
+ * real attempt again.
+ */
+trait DataOutputStreamEventListener {
+  def notifyKnown(dos: DataOutputStream): Unit
+}
+
+/**
+ * Shared bookkeeping for a registry of listeners waiting on some
+ * DataOutputStream fact that settles once and never changes again:
+ * register, remove, and clear-then-notify. notify is supplied
+ * per instance rather than via subclassing, so a use site just
+ * constructs one directly with its own dos and notify callback (e.g.
+ * `new DataOutputStreamListenerRegistry[SuspensionWaiter](this, (w, _) =>
+ * w.notifySuspensions())`) instead of a registry subclass per event.
+ */
+private[io] final class DataOutputStreamListenerRegistry[L](
+  dos: DataOutputStream,
+  notify: (L, DataOutputStream) => Unit
+) {
+  private var listeners: Set[L] = Set.empty
+
+  def register(l: L): Unit = { listeners = listeners + l }
+
+  def remove(l: L): Unit = { listeners = listeners - l }
+
+  // Drops every registered listener without notifying them, for a DOS
+  // being reset for reuse rather than reaching the fact they're waiting on.
+  def clear(): Unit = { listeners = Set.empty }
+
+  // Resets the field to empty BEFORE running any callback, not after:
+  // toNotify aliases the old (immutable, so this is free) Set, so a
+  // re-entrant registration during notification lands in a fresh
+  // `listeners` and survives instead of being wiped out afterward.
+  def clearAndNotifyAll(): Unit = {
+    val toNotify = listeners
+    listeners = Set.empty
+    toNotify.foreach(notify(_, dos))
+  }
+}
+
 /**
  * There is an asymmetry between DataInputStream and DataOutputStream with respect to the
  * positions and limits in the bit stream.
@@ -234,6 +290,22 @@ trait DataOutputStream extends DataStreamCommon {
   def setFinished(finfo: FormatInfo): Unit
   def isFinished: Boolean
 
+  /**
+   * True once this DOS's own position/content can never change again:
+   * either finished, or direct and merged/flushed away without ever
+   * passing through the finished state. Callers that only care whether
+   * it's safe to treat this DOS's data as permanent should use this.
+   */
+  def isFinishedOrDead: Boolean
+
+  /**
+   * Registers/deregisters a callback for one of this DOS's facts
+   * settling into its final value; see DataOutputStreamEventListener
+   * for what a listener does with that notification.
+   */
+  def registerListener(l: DataOutputStreamEventListener): Unit
+  def removeListener(l: DataOutputStreamEventListener): Unit
+
   /**
    * This function deletes any temnporary files that have been generated
    */

daffodil-core/src/main/scala/org/apache/daffodil/io/DataOutputStreamImplMixin.scala

@@ -95,6 +95,7 @@ trait DataOutputStreamImplMixin
           // we know that no actual writes occurred to this DOS, so it is zero length.
           // And now that it is finsihed, that can never change.
           zlStatus_ = Zero
+          notifyListeners()
         } else {
           // do nothing. It stays what it is, Unknown.
         }
@@ -108,9 +109,29 @@ trait DataOutputStreamImplMixin
    * if the amount written was 1 bit or more.
    */
   final protected def setNonZeroLength(): Unit = {
+    val wasUnknown = zlStatus_ eq ZeroLengthStatus.Unknown
     zlStatus_ = ZeroLengthStatus.NonZero
+    // Unknown -> NonZero happens exactly once per DOS; subsequent writes
+    // find it already NonZero.
+    if (wasUnknown) {
+      notifyListeners()
+    }
   }
 
+  private val stateChangeListeners =
+    new DataOutputStreamListenerRegistry[DataOutputStreamEventListener](
+      this,
+      (l, d) => l.notifyKnown(d)
+    )
+
+  def registerListener(l: DataOutputStreamEventListener): Unit =
+    stateChangeListeners.register(l)
+
+  def removeListener(l: DataOutputStreamEventListener): Unit =
+    stateChangeListeners.remove(l)
+
+  private def notifyListeners(): Unit = stateChangeListeners.clearAndNotifyAll()
+
   /**
    * Once we determine what it is, this will hold the absolute bit pos
    * of the first bit of this buffer.
@@ -168,6 +189,10 @@ trait DataOutputStreamImplMixin
     maybeAbsStartingBitPos0b_ = MaybeULong.Nope
     relBitPos0b_ = ULong(0)
     zlStatus_ = ZeroLengthStatus.Unknown
+    // A listener registered against the pre-reset facts would otherwise
+    // fire (or stay silently registered) against facts from this DOS's
+    // next, unrelated lifetime.
+    stateChangeListeners.clear()
   }
 
   def setAbsStartingBitPos0b(newStartingBitPos0b: ULong): Unit = {
@@ -181,6 +206,10 @@ trait DataOutputStreamImplMixin
       this.maybeAbsolutizedRelativeStartingBitPosInBits_.isEmpty
     ) {
       this.maybeAbsStartingBitPos0b_ = mv
+      // maybeAbsBitPos0b was Nope and is now defined for the first (and
+      // only) time. This DOS's absolute position never becomes unknown
+      // again once known, so this is the one moment to notify.
+      notifyListeners()
     } else if (this.maybeAbsStartingBitPos0b_.isDefined) {
       this.maybeAbsolutizedRelativeStartingBitPosInBits_ = this.maybeAbsStartingBitPos0b_
       this.maybeAbsStartingBitPos0b_ = mv
@@ -358,6 +387,7 @@ trait DataOutputStreamImplMixin
 
   @inline private[io] final def isDead = { _dosState =:= Uninitialized }
   @inline override final def isFinished = { _dosState =:= Finished }
+  @inline override final def isFinishedOrDead = { isFinished || isDead }
   // @inline override def setFinished(finfo: FormatInfo) { _dosState = Finished }
   @inline private[io] final def isActive = { _dosState =:= Active }
   @inline private[io] final def isReadOnly = { isFinished && isBuffering }

daffodil-core/src/main/scala/org/apache/daffodil/io/DirectOrBufferedDataOutputStream.scala

@@ -31,6 +31,7 @@ import org.apache.daffodil.lib.util.Maybe
 import org.apache.daffodil.lib.util.Maybe.*
 import org.apache.daffodil.lib.util.MaybeULong
 import org.apache.daffodil.lib.util.Misc
+import org.apache.daffodil.runtime1.processors.SuspensionWaiter
 
 import passera.unsigned.ULong
 
@@ -239,9 +240,12 @@ class DirectOrBufferedDataOutputStream private[io] (
    * Two of these are equal if they are eq.
    * This matters because we compare them to see if we are making forward progress
    */
-  override def equals(other: Any): Boolean = AnyRef.equals(other)
+  override def equals(other: Any): Boolean = other match {
+    case that: AnyRef => this.eq(that)
+    case _ => false
+  }
 
-  override def hashCode(): Int = AnyRef.hashCode()
+  override def hashCode(): Int = System.identityHashCode(this)
 
   override def toString: String = {
     lazy val buf = bufferingJOS.getBuf
@@ -509,6 +513,10 @@ class DirectOrBufferedDataOutputStream private[io] (
         // so now the first one is an EMPTY not necessarily a finished buffered DOS
         first.convertToDirect(directStream) // first is now the direct stream
         directStream.setDOSState(Uninitialized) // old direct stream is now dead
+        // Dead, not finished, but its own position/content is just as
+        // permanent. Anyone waiting on this specific DOS via
+        // registerFinishedListener needs to hear about it too.
+        directStream.notifyFinishedListeners()
         directStream = first // long live the new direct stream!
         Logger.log.debug(s"New direct DOS $directStream")
 
@@ -548,6 +556,7 @@ class DirectOrBufferedDataOutputStream private[io] (
             jos.close()
           }
           directStream.setDOSState(Uninitialized) // not just finished. We're dead now.
+          directStream.notifyFinishedListeners()
         } else {
           // the last stream we merged forward into was not finished.
           Assert.invariant(directStream.isActive)
@@ -576,9 +585,35 @@ class DirectOrBufferedDataOutputStream private[io] (
         val f = _following.get
         f.maybeAbsBitPos0b // requesting this pulls the absolute position info forward.
       }
+
+      notifyFinishedListeners()
     }
   }
 
+  // Registers SuspensionWaiters directly rather than through
+  // DataOutputStreamEventListener: this event only ever has a
+  // SuspensionWaiter as a registrant, so depending on
+  // runtime1.processors.SuspensionWaiter here is an accepted exception.
+  private val finishedListeners = new DataOutputStreamListenerRegistry[SuspensionWaiter](
+    this,
+    (w, _) => w.notifySuspensions()
+  )
+
+  def registerFinishedListener(w: SuspensionWaiter): Unit =
+    finishedListeners.register(w)
+
+  def removeFinishedListener(w: SuspensionWaiter): Unit =
+    finishedListeners.remove(w)
+
+  // setFinished() is one-shot per DOS, so there's no need to keep this
+  // registration around afterward. Finishing/dying is also the only event
+  // that can resolve a still-Unknown zeroLengthStatus to Zero; querying
+  // the getter forces that check rather than duplicating it here.
+  private def notifyFinishedListeners(): Unit = {
+    finishedListeners.clearAndNotifyAll()
+    val _ = zeroLengthStatus
+  }
+
   /**
    * This override implements a critical behavior, which is that when we ask for
    * an absolute bit position, if we have it great. if we don't, we look at the