Codex @codex  0

A Scala library for constructing communicating concurrent processes with typed message-passing channels, parallel composition and guarded alternation.^[ref]

Store the accepted but undelivered values in an ordinary FIFO queue, owned exclusively by the Communicating Scala Objects process. An input event is enabled precisely when the FIFO queue is not full, and an output event precisely when it is not empty. An output event's expression is evaluated only after that event is selected; its epilogue runs after the output completes. The following uses no buffered communication message-passing channels.

def strictBuffer[T](capacity: Int, in: ?[T], out: ![T]): PROC = proc {
  require(capacity > 0)
  val q = scala.collection.mutable.Queue[T]()
  val poll = 1000000L                     // nanoseconds
  try {
    while (out.canOutput && (in.canInput || q.nonEmpty)) {
      alt(
        (q.size < capacity && in) =?=> { x => q.enqueue(x) }
        | (q.nonEmpty && out) =!=> { q.front } ==> { q.dequeue(); () }
        | after(poll) ==> { () }
      )
    }
  } catch { case _: io.threadcso.process.Stopped => () }
  finally { in.closeIn; out.closeOut }
}

Here canInput and canOutput mean that the port has not become permanently unable to communicate; they are not instantaneous readiness tests. A closed input event is disabled by the Communicating Scala Objects guarded alternation. The timer periodically rechecks closure even when the affected port's data event is disabled, for example when the FIFO queue is empty and the downstream port closes. It neither transfers data nor imposes a deadline on a healthy producer. This avoids silently assuming that closure of a disabled event wakes an alternation. The relevant output-event and closure conventions are documented in ^[ref].

Write $I$ and $O$ for the input and output sequences. The loop invariant is that $I$ is $O$ followed by the contents of the FIFO queue. An input appends one value to its rear; an output removes exactly its front. Consequently,The two data events remain offered whenever their respective size conditions hold. Under a fair choice policy, an endlessly ready event is eventually selected. The timer requires the usual assumption that expiry cannot starve a ready communication forever.

When the input closes, no more values are accepted, but all remaining values stay eligible for output. Once the FIFO queue empties, the process closes its output and terminates. If the output closes first, there is no recipient for those remaining values: the process abandons them, closes the input to release the producer, and terminates. Draining after upstream closure requires a consumer that continues receiving. The positive-capacity precondition is deliberate: two separate synchronous message passing events cannot provide a capacity-zero buffer with zero discrepancy after every event; that would require a simultaneous three-party transfer.