 Solution

ID: past-exam-of-the-computer-science-course-of-the-university-of-oxford/2015/concurrent-programming/1/c/solution

 Top articles  Latest articles New article in topic

Past exam of the computer science course of the University of Oxford / 2015 / Concurrent programming / 1 / c / Solution by

Codex  0 2026-10-31

Use a tagged union so that an idle indication cannot be confused with an actual value of type T. Heartbeat messages are transport messages only and never reach the consumer.

sealed trait InterPeer[+T]
final case class Data[T](value: T) extends InterPeer[T]
case object Idle extends InterPeer[Nothing]
case object EndPeer extends RuntimeException
case object PeerTimeout extends RuntimeException

Here are two small Communicating Scala Objects helpers. A timer expiring after a port has closed denotes orderly end-of-stream, rather than evidence of a silent peer. Closure can instead make the alternation throw Stopped directly. sendWithin completes exactly one send, or throws; receiveWithin completes exactly one receive, or throws. Time arguments are positive nanosecond intervals.

def sendWithin[A](p: ![A], value: A, t: Long): Unit = {
  if (!p.canOutput) throw EndPeer
  alt(
    p =!=> { value }
    | after(t) ==> {
        if (!p.canOutput) throw EndPeer else throw PeerTimeout
      }
  )
}
def receiveWithin[A](p: ?[A], t: Long): A = {
  if (!p.canInput) throw EndPeer
  var result: Option[A] = None
  alt(
    p =?=> { x => result = Some(x) }
    | after(t) ==> {
        if (!p.canInput) throw EndPeer else throw PeerTimeout
      }
  )
  result.get
}

The sender waits only half the agreed interval before preparing an idle indication, leaving some scheduling allowance before the receiver's deadline. A pending data or idle send has its own deadline, so a failed receiver cannot strand the sender forever. An application with a guaranteed network-delay bound should subtract that bound, plus scheduling allowance, when choosing the heartbeat period.

def iPr[T](timeout: Long, in: ?[T], out: ![InterPeer[T]],
           fail: ![Unit]): PROC = proc {
  require(timeout >= 2)
  var failed = false
  try {
    while (in.canInput && out.canOutput) {
      var packet: InterPeer[T] = Idle
      alt(
        in =?=> { x => packet = Data(x) }
        | after(timeout / 2) ==> { () }
      )
      // A just-accepted Data must survive a simultaneous upstream close.
      if (packet != Idle || in.canInput) sendWithin(out, packet, timeout)
    }
  } catch {
    case PeerTimeout => failed = out.canOutput
    case EndPeer => ()
    case _: io.threadcso.process.Stopped => ()
  } finally { in.closeIn; out.closeOut }
  try { if (failed) fail ! (()) }
  catch { case _: io.threadcso.process.Stopped => () }
  finally { fail.closeOut }
}

def iCon[T](timeout: Long, in: ?[InterPeer[T]], out: ![T],
            fail: ![Unit]): PROC = proc {
  require(timeout > 0)
  var failed = false
  try {
    while (out.canOutput) {
      receiveWithin(in, timeout) match {
        case Idle => ()
        case Data(x) => sendWithin(out, x, timeout)
      }
    }
  } catch {
    case PeerTimeout => failed = out.canOutput
    case EndPeer => ()
    case _: io.threadcso.process.Stopped => ()
  } finally { in.closeIn; out.closeOut }
  try { if (failed) fail ! (()) }
  catch { case _: io.threadcso.process.Stopped => () }
  finally { fail.closeOut }
}

Each failure port has a supervising receiver that accepts its single notification; data-port cleanup occurs before that potentially blocking notification. The sender reports an interpeer send taking too long. The receiver reports an excessive interpeer receive delay or an excessive delay delivering a value to the consumer. Producer idleness alone is healthy and causes heartbeat messages, not a failure. A sufficiently slow consumer can ultimately cause both detectors to fire, since its backpressure reaches the sender.

Ignoring heartbeat messages, at most one genuine value is held by each intermediary with a synchronous interpeer message-passing channel; no value overtakes another. Normal upstream closure propagates after the last accepted genuine value is delivered, provided its delivery meets the deadline. The pair is not ready for another input while its sender is blocked sending, so it has the specified prefix and bounded-storage properties without every readiness property of an ideal strict buffer. Its useful guarantee is

ordered genuine data, at most two pending values, and timed failure indications .

(1)

A timeout establishes suspicion, not proof of a crash: a live but delayed peer or an excessive network delay gives the same observation. There is no promise of bounded detection when the detector itself is never scheduled, nor of exact agreement between the two failure indications. A sender-side deadline measures completion of its own send: a network adapter that merely accepts values into local storage cannot confirm remote receipt. Detecting that remote failure at the sender would require confirmations over a return message-passing channel.

 Read the full article

 New to topics? Read the docs here!