Past exam of the computer science course of the University of Oxford / 2015 / Concurrent programming / 1 / d / Solution 2026-10-31
View more
A channel multiplexer adds a stream identity; a channel demultiplexer removes it and selects the corresponding output. Add an explicit per-stream end marker, since closing the shared message-passing channel cannot identify which producer has finished. End(i) is represented by None; Some(x) represents a genuine value, including a null value if the application permits one.
final case class MUX[T](stream: Int, item: Option[T])
The following Communicating Scala Objects implementation uses a synchronous one-to-one mid. As in part (a), a short timer detects closure even when no communication is enabled. Only the channel multiplexer owns the input ports, and only the channel demultiplexer owns the output ports. The channel multiplexer takes at most one input value before its corresponding mid write; it never accepts a second input while that write is pending.
def iMux[T](ins: Seq[?[T]], mid: ![MUX[T]]): PROC = proc {
  val ended = Array.fill(ins.length)(false)
  val poll = 1000000L
  try {
    while (mid.canOutput && ended.contains(false)) {
      for (i <- ins.indices if !ended(i) && !ins(i).canInput) {
        mid ! MUX[T](i, None)
        ended(i) = true
      }
      if (ended.contains(false)) {
        alt(
          (| (for (i <- ins.indices) yield
            (!ended(i) && ins(i)) =?=> { x => mid ! MUX(i, Some(x)) }))
          | after(poll) ==> { () }
        )
      }
    }
  } catch { case _: io.threadcso.process.Stopped => () }
  finally { ins.foreach(_.closeIn); mid.closeOut }
}

def iDemux[T](mid: ?[MUX[T]], outs: Seq[![T]]): PROC = proc {
  val ended = Array.fill(outs.length)(false)
  val poll = 1000000L
  try {
    while (mid.canInput && ended.contains(false) &&
           outs.indices.forall(i => ended(i) || outs(i).canOutput)) {
      alt(
        mid =?=> { packet =>
          val i = packet.stream
          require(i >= 0 && i < outs.length && !ended(i))
          packet.item match {
            case Some(x) => outs(i) ! x
            case None => ended(i) = true; outs(i).closeOut
          }
        }
        | after(poll) ==> { () }
      )
    }
  } catch { case _: io.threadcso.process.Stopped => () }
  finally { mid.closeIn; outs.foreach(_.closeOut) }
}
For each stream, the output sequence remains a prefix of its accepted input sequence. One value can be pending in the channel demultiplexer, and one in the channel multiplexer, possibly from the same stream. Hence its discrepancy is at most two:
(1)
There is a shared total of two pending values, so these processes do not supply the independent readiness of a parallel collection of capacity-two strict buffers. An unready consumer blocks the channel demultiplexer's ordinary send; the channel multiplexer can accept one further value and then blocks too. This is head-of-line blocking.
The alternation may choose an input unfairly; replacing its selection policy by round-robin selection helps among simultaneously ready producers. Even fair input choice cannot compensate for a consumer that blocks the entire channel demultiplexer. Closed-input scanning does allow a producer's end marker eventually to reach its consumer, but only after the shared transport and any earlier delivery become serviceable. Its own accepted values precede its end marker, so that consumer drains cleanly.
Consumer-driven closure is more restrictive. With no return message-passing channel, the sender cannot promptly learn which producer should stop. This implementation deliberately shuts down the whole multiplexed channel when it detects any unfinished consumer closing, thereby cancelling the other pairs too. If the channel demultiplexer is already blocked on some other consumer, even that detection waits for the blocked send. Thus producer-driven ends can be represented separately but need not be delivered promptly; consumer-driven cancellation is global and need not be detected promptly. Merely adding an end marker does not solve either delay.
Read the full article
Past exam of the computer science course of the University of Oxford / 2015 / Concurrent programming / 1 / e / Solution 2026-10-31
View more
Use a return control channel to carry both cancellation and one-slot flow control. The latter is what prevents a silent consumer from obstructing other streams. Keep the preceding MUX type and add
sealed trait CTRL
final case class Grant(stream: Int) extends CTRL
final case class Cancel(stream: Int) extends CTRL
Here is a precise event-loop specification of the modified generators. choose means a fair Communicating Scala Objects guarded alternation over the listed input events, output events and a short closure-poll timer; every selected transition below is finite and contains no further blocking communication. An output transition commits its state change only after its selected send succeeds. The notation is operational pseudocode, with the requested signatures; it does not introduce an assumed nonblocking send primitive.
def iMux[T](ins: Seq[?[T]], ctrl: ?[CTRL], mid: ![MUX[T]]): PROC
  for each stream i:
    credit[i] = true; inputOpen[i] = true; terminal[i] = false
    pending[i] = None                     // Option[MUX[T]]
  until every terminal[i]:
    poll closures:
      if !ins[i].canInput: inputOpen[i] = false
      if !terminal[i] && !inputOpen[i] && pending[i].isEmpty:
        pending[i] = Some(MUX(i, None))   // end follows earlier data
    choose:
      ctrl ? Grant(i):
        if !terminal[i]: credit[i] = true
      ctrl ? Cancel(i):
        terminal[i] = true; pending[i] = None; ins[i].closeIn
      ins[i] ? x, enabled when inputOpen[i] && !terminal[i]
                              && credit[i] && pending[i].isEmpty:
        credit[i] = false; pending[i] = Some(MUX(i, Some(x)))
      mid ! pending[i].get, enabled when pending[i].nonEmpty:
        if pending[i].get.item.isEmpty: terminal[i] = true
        pending[i] = None
      poll timer: no action
  finally: close every ins[i] input; close mid output and ctrl input

def iDemux[T](ctrl: ![CTRL], mid: ?[MUX[T]], outs: Seq[![T]]): PROC
  for each stream i:
    slot[i] = None; ended[i] = false; cancelled[i] = false
  replies = empty FIFO queue of CTRL
  until every (cancelled[i] || (ended[i] && slot[i].isEmpty)):
    poll closures:
      if !cancelled[i] && !ended[i] && !outs[i].canOutput:
        cancelled[i] = true; slot[i] = None
        delete any queued Grant(i); append Cancel(i) to replies
      if !cancelled[i] && ended[i] && slot[i].nonEmpty
                                      && !outs[i].canOutput:
        cancelled[i] = true; slot[i] = None
      if ended[i] && slot[i].isEmpty: outs[i].closeOut
    choose:
      mid ? MUX(i, Some(x)):
        if !cancelled[i]: assert slot[i].isEmpty; slot[i] = Some(x)
      mid ? MUX(i, None):
        ended[i] = true
      outs[i] ! slot[i].get, enabled when !cancelled[i] && slot[i].nonEmpty:
        slot[i] = None
        if !ended[i]: append Grant(i) to replies
      ctrl ! replies.front, enabled when replies.nonEmpty:
        remove replies.front
      poll timer: no action
  finally: close mid input, every output, and ctrl output
There is one receive event on mid, with the two packet forms dispatched in its body, and one send event on mid, offering a fairly chosen pending stream. There is likewise one send event on ctrl. If mid becomes exhausted, the channel demultiplexer marks any remaining streams ended, discards queued replies, and drains or cancels their slots; the sender has then finished and needs no more grants. If the control input closes, the channel multiplexer enters cleanup. If the receiver finds its control output closed, it discards queued replies and disables further control sends, but continues receiving end markers and draining its slots; sender completion must not discard already accepted values. A channel closing during an individual guarded operation is handled locally: a failed outs(i) send cancels that stream and queues Cancel(i) unless it had already ended; exhaustion of a shared port uses the cleanup rules just stated. Ports that closed while their data events were disabled are found by the poll. These rules specify closure handling as well as normal transitions.
The credit starts at one and is consumed when an input is accepted. It is returned only after the corresponding output finishes. Thus the one permitted outstanding value for stream is either pending at the sender or stored at the receiver, never both. An end marker needs no credit and follows that stream's last data packet on the ordered shared message-passing channel. It can arrive while that value is still waiting for its consumer; the receiver then delays closing the output until its slot empties. The invariant is
(1)
As in a strict buffer, cancellation can discard the last outstanding value. A cancelled receiver ignores a data packet already in flight; a later grant cannot resurrect a cancelled sender. No selected data delivery blocks the event loop, and control replies are queued rather than sent inside a receive handler. In particular, there is no cycle in which the sender blocks sending data while the receiver blocks sending cancellation. Closing the control output at final receiver termination also releases the sender if a last cancellation was still queued.
Under fair event selection and eventual scheduling, an individual cancellation reaches its producer without waiting for any unrelated consumer. Producer closure similarly delivers its own end marker without waiting for another stream to drain. This gives finite flow control equivalent in ordering and termination to capacity-one strict buffers, with finite protocol delays. A physical shared message-passing channel necessarily serializes transfers, so instantaneous readiness and absolute real-time delay bounds are not identical to physically separate buffers.
Read the full article