A directed graph needs an explicit termination argument. After assigning its parent in a rooted tree, a vertex must still receive later traffic: having no new graph neighbours to discover is not proof that other senders have finished. Use a discovery wave together with duplicate-suppressed announcements of activated vertices. Both kinds of message identify their immediate sender, who has already activated.
sealed trait Msg { def from: NodeId }
final case class Discover(from: NodeId) extends Msg
final case class Active(from: NodeId, origin: NodeId) extends MsgLet
adj = neighbours.distinct. The actual communication objects must expose both input and output ports: the paper's Seq[![Msg]] annotation supplies only output capabilities, although the following sentence also requires receiving. Set up one shared-writer, single-reader buffered message-passing channel per vertex, publish its output in toNode, and keep its input in fromNode. Only node i reads fromNode(i). These are the two ports of the same underlying mailbox, not additional graph edges. The question's prohibition on buffered message-passing channels applies to Question 1(a), not to this question.Take , reliable delivery, eventual scheduling, finite
adj, and mailbox capacity . Every node has at most distinct incoming neighbours, including itself if there is a self-loop. The bound below ensures that no send blocks merely because a mailbox's receiver is not currently reading. Do not close these shared mailboxes on local process termination: late, redundant traffic is harmless and bounded. Their creator can release them after the parallel composition has joined. For example, the concrete setup can useval mailbox = Vector.fill(N)(N2NBuf[Msg](B, 0, 1))
val toNode: Seq[![Msg]] = mailbox
val fromNode: Seq[?[Msg]] = mailboxHere the zero writer count makes the shared output independent of individual writer departures; processes do not call its
closeOut during the search.def span: PROC = proc {
val adj = neighbours.distinct
var started = false
val known = scala.collection.mutable.Set[NodeId]()
def announce(origin: NodeId): Unit = {
if (!known(origin)) {
known += origin
for (w <- adj) toNode(w) ! Active(me, origin)
}
}
def activate(sender: NodeId): Unit = {
started = true
parent = if (me == 0) 0 else sender
for (w <- adj if me == 0 || w != parent)
toNode(w) ! Discover(me)
announce(me)
}
if (me == 0) activate(0)
while (known.size < N) {
val message = fromNode(me) ? ()
if (!started) activate(message.from)
message match {
case Discover(_) => ()
case Active(_, origin) => announce(origin)
}
}
}An announcement may be a vertex's first incoming message. It then serves as a discovery message too: its immediate sender is already active, and the receiving vertex adopts that sender as its parent in a rooted tree before forwarding anything. This preserves the required first-message rule. In particular, an inactive vertex never forwards announcements while remaining inactive, which could invalidate the parent-order argument.
For each nonroot vertex, its parent in a rooted tree activated strictly earlier. Repeated parent-following therefore cannot make a graph cycle and eventually reaches the unique initially active vertex, zero. Each parent assignment uses an actual incoming directed edge. Strong connectivity ensures eventual activation of every vertex: if an activated set were proper, a directed path from zero to an inactive vertex would have a first edge leaving that set. Its sender sends a discovery message along that edge; excluding its own parent does not exclude an inactive neighbour.
Each activation sends at most one
Discover on each outgoing directed edge. For each of the origins, each vertex forwards Active at most once on each outgoing directed edge. Therefore a mailbox receives at most messages over the whole run, proving the chosen nonblocking storage bound. Each origin's announcement eventually reaches every vertex, since every first receipt forwards it on all outgoing directed edges. Thus every known set reaches size and every process terminates, even if redundant messages remain unread:No timeout is used to conclude that the distributed work has finished. This uses larger finite mailboxes to make the termination proof simple, rather than assuming that simultaneous synchronous broadcasts cannot deadlock.Use one logical traversal token carrying the discovered vertex set and the active LIFO stack of vertices. The current vertex either discovers its first still-unvisited outgoing graph neighbour, or returns to its preceding stack entry. The former records both the child's parent in a rooted tree and the parent's child. Unlike concurrent flooding discovery, this visits one entire branch before the next, giving an actual depth-first search.
The important directed-graph detail is that a child's parent need not be an outgoing graph neighbour. Return the token by routing it through the original directed graph; a routing hop does not count as a traversal visit. The following duplicate-suppressed flooding route requires no previously known topology and sends only to immediate outgoing graph neighbours.
final case class Token(visited: Set[NodeId], stack: List[NodeId], serial: Int)
sealed trait Msg
final case class Route(serial: Int, target: NodeId, token: Token) extends Msg
case object Finished extends MsgHere
stack.head is the current vertex. Use the same paired toNode and fromNode ports as in (i), now with the new Msg type. Set mailbox capacity to , normalize neighbours with distinct, and again leave the mailboxes open until all processes have joined. The following helpers are local functions inside each node's generated process.def span: PROC = proc {
val adj = neighbours.distinct
val routed = scala.collection.mutable.Set[Int]()
var entered = false
var done = false
def flood(message: Msg): Unit =
for (w <- adj) toNode(w) ! message
def move(target: NodeId, state: Token): Unit = {
val s = state.serial + 1
routed += s
flood(Route(s, target, state.copy(serial = s)))
}
def step(t: Token): Unit = {
assert(t.stack.head == me)
if (!entered) {
entered = true
parent = if (t.stack.tail.isEmpty) me else t.stack.tail.head
children = Nil
}
adj.find(w => !t.visited(w)) match {
case Some(w) =>
children = children :+ w
move(w, Token(t.visited + w, w :: t.stack, t.serial))
case None =>
t.stack.tail match {
case p :: rest => move(p, t.copy(stack = p :: rest))
case Nil => flood(Finished); done = true
}
}
}
if (me == 0) step(Token(Set(0), List(0), 0))
while (!done) {
fromNode(me) ? () match {
case r @ Route(s, target, t) =>
if (!routed(s)) {
routed += s
if (target == me) step(t) else flood(r)
}
case Finished => flood(Finished); done = true
}
}
}Every route number is globally unique: there is one logical token, and its serial increases at every move. The sender inserts that serial in its own
routed set before flooding; this stops a returning copy from circulating again. Intermediate vertices forward the first copy of an envelope and ignore later copies. Its target consumes exactly one copy without forwarding it. Because the directed graph is strongly connected, a simple directed path from sender to target exists; its intermediate vertices eventually relay the envelope. Thus each move eventually finishes.A routed return can pass through a vertex not yet discovered by the depth-first search. That vertex only forwards the envelope: it does not run
step, change entered, or assign a parent. This distinction is essential. For the directed cycle , returning from two to one uses , while the logical traversal stack changes from to .
A depth-first search return follows a routed directed path when the child has no reverse directed edge
. Details. 
The discovered vertex set only grows. Each newly discovered vertex is pushed once and later popped once; an exhausted vertex can never acquire a new unvisited neighbour as this set grows. Consequently there are exactly downward moves and return moves. When the root exhausts its outgoing graph neighbours, every vertex has been visited, by the standard reachability argument for depth-first search. Each nonroot parent was recorded along the actual downward directed edge, and
children records exactly those assignments in discovery order.There are route envelopes, each forwarded at most once per outgoing directed edge, and one final
Finished flood. Each node forwards Finished before exiting; strong connectivity therefore delivers it everywhere. There are at most total messages into any mailbox, below the proposed capacity. Even a mailbox whose process has already exited cannot block later sends. All processes terminate withThis deliberately trades communication efficiency for a simple valid solution on a genuinely directed graph. A direct parent acknowledgement would instead require an additional bidirectional-edge assumption, which the question does not supply.Use the recorded child list as the depth-first traversal order. A leaf of a tree has no children, so it contributes its own identity once. Closure cancels the remaining traversal; a normal completion closes the output so that a consumer can detect end-of-fringe.
def dfFringe(node: Array[Node], root: NodeId, fringe: ![NodeId]): PROC = proc {
def walk(v: NodeId): Unit = {
val cs = node(v).children
if (cs.isEmpty) fringe ! v
else cs.foreach(walk)
}
try { walk(root) }
catch { case _: io.threadcso.process.Stopped => () }
finally { fringe.closeOut }
}A completed finite spanning tree has no recursive child graph cycle, so this visits a finite number of vertices. A blocked leaf send is released if its output closes. A singleton root is itself a leaf of a tree and is output once. With sufficient call-stack space,
Run two
dfFringe producers on synchronous one-to-one message-passing channels, and compare one received leaf from each. Catch closure separately on each input: otherwise the first closure could hide an extra value in the other fringe. An extra value or unequal pair is enough to stop immediately.def sameFringe(node1: Array[Node], root1: NodeId,
node2: Array[Node], root2: NodeId): Boolean = {
val a = OneOne[NodeId]
val b = OneOne[NodeId]
var equal = true
def next(p: ?[NodeId]): Option[NodeId] =
try { Some(p ? ()) }
catch { case _: io.threadcso.process.Stopped => None }
val compare = proc {
var done = false
try {
while (!done) {
val x = next(a)
val y = next(b)
(x, y) match {
case (None, None) => done = true
case (Some(u), Some(v)) if u == v => ()
case _ => equal = false; done = true
}
}
} finally { a.closeIn; b.closeIn }
}
(dfFringe(node1, root1, a) || dfFringe(node2, root2, b) || compare)()
equal
}After successful comparisons, the first identities in the two leaf sequences agree. The first differing pair disproves equality of sequences; simultaneous exhaustion proves it; exhaustion on exactly one side disproves it. Therefore,Closing both inputs on every comparator exit releases either producer if it is waiting in an output. The closure handler in
dfFringe then returns normally, and the parallel composition joins before equal is read. For finite trees and eventual scheduling, the comparator's next read either obtains a leaf or observes completion, so there is no deadlock caused by reading the two inputs in a fixed order.Only a constant amount of leaf data is stored. After an early mismatch, each producer may already have traversed toward its next leaf and be blocked offering it; cancellation prevents unbounded speculation or deliberate construction of the remaining fringe. Occasionally a short unequal fringe is already completely traversed when its final distinguishing value is compared. The streaming guarantee is that neither complete fringe needs to be materialized in advance, and work stops as soon as inequality is established.
Articles by others on the same topic
There are currently no matching articles.