Kniha An%20Introduction%20to%20Distributed%20Processing

hannels. (A connected directed graph is a directed graph whose underlying
undirected graph is connected.) For a task t, we let In
t
⊆ D
T
denote the set of edges directed
towards t and Out
t
⊆ D
T
the set of edges directed away from t. Channels in In
t
are those on
which t receives messages and channels in Out
t
are those on which t sends messages. We
also let n
t
= |In
t
|, that is, n
t
denotes the number of channels on which t may receive
messages.
A task t is a reactive (or message-driven) entity, in the sense that normally it only performs
computation (including the sending of messages to other tasks) as a response to the receipt
of a message from another task. An exception to this rule is that at least one task must be
allowed to send messages out "spontaneously" (i.e., not as a response to a message
receipt) to other tasks at the beginning of its execution, inasmuch as otherwise the assumed
message-driven character of the tasks would imply that every task would idle indefinitely and
no computation would take place at all. Also, a task may initially perform computation for
initialization purposes.
Algorithm Task_t, given next, describes the overall behavior of a generic task t. Although in
this algorithm we (for ease of notation) let tasks compute and then send messages out, no
such precedence is in fact needed, as computing and sending messages out may constitute
intermingled portions of a task's actions.
Algorithm Task_t:

Do some computation;
send one message on each channel of a (possibly empty) subset of
Out
t
;
repeat
receive message on c
1
∈ In
t
and B
1
→
Do some computation;
send one message on each channel of a (possibly empty)
subset of Out
t

or…
or
receive message on c
nt
∈ In
t
and B
nt
→
Do some computation;
send one message on each channel of a (possibly empty)
subset of Out
t

until global termination is known to t.


There are many important observations to be made in connection with Algorithm Task_t. The
first important observation is in connection with how the computation begins and ends for
task t. As we remarked earlier, task t begins by doing some computation and by sending
messages to none or more of the tasks to which it is connected in G
T
by an edge directed
away from it (messages are sent by means of the operation send). Then t iterates until a
global termination condition is known to it, at which time its computation ends. At each
iteration, t does some computation and may send messages. The issue of global termination
will be thoroughly discussed in Section 6.2 in a generic setting, and before that in various
other chapters it will come up in more particular contexts. For now it suffices to notice that t
acquires the information that it may terminate its local computation by means of messages
received during its iterations. If designed correctly, what this information signals to t is that
no message will ever reach it again, and then it may exit the repeat…until loop.
The second important observation is on the construction of the repeat…until loop and on
the semantics associated with it. Each iteration of this loop contains n
t
guarded commands
grouped together by or connectives. A guarded command is usually denoted by
guard → command,
where, in our present context, guard is a condition of the form
receive message on c
k
∈ In
t
and B
k

for some Boolean condition B
k
, where 1 ≤ k ≤ n
t
. The receive appearing in the description
of the guard is an operation for a task to receive messages. The guard is said to be ready
when there is a message available for immediate reception on channel c
k
and furthermore
the condition B
k
is true. This condition may depend on the message that is available for
reception, so that a guard may be ready or not, for the same channel, depending on what is
at the channel to be received. The overall semantics of the repeat…until loop is then the
following. At each iteration, execute the command of exactly one guarded command whose
guard is ready. If no guard is ready, then the task is suspended until one is. If more than one
guard is ready, then one of them is selected arbitrarily. As the reader will verify by our many
distributed algorithm examples along the book, this possibility of nondeterministically
selecting guarded commands for execution provides great design flexibility.
Our final important remark in connection with Algorithm Task_t is on the semantics
associated with the receive and send operations. Although as we have remarked the use of
a receive in a guard is to be interpreted as an indication that a message is available for
immediate receipt by the task on the channel specified, when used in other contexts this
operation in general has a blocking nature. A blocking receive has the effect of suspending
the task until a message arrives on the channel specified, unless a message is already there
to be received, in which case the reception takes place and the task resumes its execution
immediately.
The send operation too has a semantics of its own, and in general may be blocking or
nonblocking. If it is blocking, then the task is suspended until the message can be delivered
directly to the receiving task, unless the receiving task happens to be already suspended for
message reception on the corresponding channel when the send is executed. A blocking
send and a blocking receive constitute what is known as task rendez-vous, which is a
mechanism for task synchronization. If the send operation has a nonblocking nature, then
the task transmits the message and immediately resumes its execution. This nonblocking
version of send requires buffering for the messages that have been sent but not yet
received, that is, messages that are in transit on the channel. Blocking and nonblocking
send operations are also sometimes referred to as synchronous and asynchronous,
respectively, to emphasize the synchronizing effect they have in the former case. We refrain
from using this terminology, however, because in this book the words synchronous and
asynchronous will have other meanings throughout (cf. Section 2.1). When used, as in
Algorithm Task-t, to transmit messages to more than one task, the send operation is
assumed to be able to do all such transmissions in parallel.
The relation of blocking and nonblocking send operations with message buffering
requirements raises important questions related to the design of distributed algorithms. If, on
the one hand, a blocking send requires no message buffering (as the message is passed
directly between the synchronized tasks), on the other hand a nonblocking send requires the
ability of a channel to buffer an unbounded number of messages. The former scenario poses
great difficulties to the program designer, as communication deadlocks occur with great ease
when the programming is done with the use of blocking operations only. For this reason,
however unreal the requirement of infinitely many buffers may seem, it is customary to start
the design of a distributed algorithm by assuming nonblocking operations, and then at a later
stage performing changes to yield a program that makes use of the operations provided by
the language at hand, possibly of a blocking nature or of a nature that lies somewhere in
between the two extremes of blocking and nonblocking send operations.
The use of nonblocking send operations does in general allow the correctness of distributed
algorithms to be shown more easily, as well as their properties. We then henceforth assume
that, in Algorithm Task_t, send operations have a nonblocking nature. Because Algorithm
Task_t is a template for all the algorithms appearing in the book, the assumption of
nonblocking send operations holds throughout. Another important aspect affecting the
design of distributed algorithms is whether the channels in D
T
deliver messages in the FIFO
order or not. Although as we remarked in Section 1.3 this property may at times be essential,
we make no assumptions now, and leave its treatment to be done on a case-by-case basis.
We do make the point, however, that in the guards of Algorithm Task_t at most one
message can be available for immediate reception on a FIFO channel, even if other
messages have already arrived on that same channel (the available message is the one to
have arrived first and not yet received). If the channel is not FIFO, then any message that
has arrived can be regarded as being available for immediate reception.

1.5 Handling infinite-capacity channels
As we saw in Section 1.4, the blocking or nonblocking nature of the send operations is
closely related to the channels ability to buffer messages. Specifically, blocking operations
require no buffering at all, while nonblocking operations may require an infinite amount of
buffers. Between the two extremes, we say that a channel has capacity k ≥ 0 if the number
of messages it can buffer before either a message is received by the receiving task or the
sending task is suspended upon attempting a transmission is k. The case of k = 0
corresponds to a blocking send, and the case in which k → ∞ corresponds to a nonblocking
send.
Although Algorithm Task_t of Section 1.4 is written under the assumption of infinite-capacity
channels, such an assumption is unreasonable, and must be dealt with somewhere along
the programming process. This is in general achieved along two main steps. First, for each
channel c a nonnegative integer b(c) must be determined that reflects the number of buffers
actually needed by channel c. This number must be selected carefully, as an improper
choice may introduce communication deadlocks in the program. Such a deadlock is
represented by a directed cycle of tasks, all of which are suspended to send a message on
the channel on the cycle, which cannot be done because all channels have been assigned
insufficient storage space. Secondly, once the b(c)'s have been determined, Algorithm
Task_t must be changed so that it now employs send operations that can deal with the new
channel capacities. Depending on the programming language at hand, this can be achieved
rather easily. For example, if the programming language offers channels with zero capacity,
then each channel c may be replaced with a serial arrangement of b(c) relay tasks
alternating with b(c) + 1 zero-capacity channels. Each relay task has one input channel and
one output channel, and has the sole function of sending on its output channel whatever it
receives on its input channel. It has, in addition, a storage capacity of exactly one message,
so the entire arrangement can be viewed as a b(c)-capacity channel.
The real problem is of course to determine values for the b(c)'s in such a way that no new
deadlock is introduced in the distributed algorithm (put more optimistically, the task is to
ensure the deadlock-freedom of an originally deadlock-free program). In the remainder of
this section, we describe solutions to this problem which are based on the availability of a
bound r(c), provided for each channel c, on the number of messages that may require
buffering in c when c has infinite capacity. This number r(c) is the largest number of
messages that will ever be in transit on c when the receiving task of c is itself attempting a
message transmission, so the messages in transit have to be buffered.
Although determining the r(c)'s can be very simple for some distributed algorithms (cf.
Sections 5.4 and 8.5), for many others such bounds are either unknown, or known
imprecisely, or simply do not exist. In such cases, the value of r(c) should be set to a "large"
positive integer M for all channels c whose bounds cannot be determined precisely. Just how
large this M has to be, and what the limitations of this approach are, we discuss later in this
section.
If the value of r(c) is known precisely for all c ∈ D
T
, then obviously the strategy of assigning
b(c) = r(c) buffers to every channel c guarantees the introduction of no additional deadlock,
as every message ever to be in transit when its destination is engaged in a message
transmission will be buffered (there may be more messages in transit, but only when their
destination is not engaged in a message transmission, and will therefore be ready for
reception within a finite amount of time). The interesting question here is, however, whether
it can still be guaranteed that no new deadlock will be introduced if b(c) < r(c) for some
channels c. This would be an important strategy to deal with the cases in which r(c) = M for
some c ∈ D
T
, and to allow (potentially) substantial space savings in the process of buffer
assignment. Theorem 1.1 given next concerns this issue.
Theorem 1.1
Suppose that the distributed algorithm given by Algorithm Task_t for all t ∈ N
T
is deadlock-
free. Suppose in addition that G
T
contains no directed cycle on which every channel c is
such that either b(c) < r(c) or r(c) = M. Then the distributed algorithm obtained by replacing
each infinite-capacity channel c with a b(c)-capacity channel is deadlock-free.
Proof: A necessary condition for a deadlock to arise is that a directed cycle exists in G
T

whose tasks are all suspended on an attempt to send messages on the channels on that
cycle. By the hypotheses, however, every directed cycle in G
T
has at least one channel c for
which b(c) = r(c) < M, so at least the tasks t that have such channels in Out
t
are never
indefinitely suspended upon attempting to send messages on them.
The converse of Theorem 1.1 is also often true, but not in general. Specifically, there may be
cases in which r(c) = M for all the channels c of a directed cycle, and yet the resulting
algorithm is deadlock-free, as M may be a true upper bound for c (albeit unknown). So
setting b(c) = r(c) for this channel does not necessarily mean providing it with insufficient
buffering space.
As long as we comply with the sufficient condition given by Theorem 1.1, it is then possible
to assign to some channels c fewer buffers than r(c) and still guarantee that the resulting
distributed algorithm is deadlock-free if it was deadlock-free to begin with. In the remainder
of this section, we discuss two criteria whereby these channels may be selected. Both
criteria lead to intractable optimization problems (i.e., NP-hard problems), so heuristics need
to be devised to approximate solutions to them (some are provided in the literature).
The first criterion attempts to save as much buffering space as possible. It is called the
space-optimal criterion, and is based on a choice of M such that

where C
+
is the set of channels for which a precise upper bound is not known. This criterion
requires a subset of channels C ⊆ D
T
to be determined such that every directed cycle in G
T

has at least one channel in C, and such that

is minimum over all such subsets (clearly, C and C
+
are then disjoint, given the value of M,
unless C
+
contains the channels of an entire directed cycle from G
T
). Then the strategy is to
set

which ensures that at least one channel c from every directed cycle in G
T
is assigned b(c) =
r(c) buffers (Figure 1.1). By Theorem 1.1, this strategy then produces a deadlock-free result
if no directed cycle in G
T
has all of its channels in the set C
+
. That this strategy employs the
minimum number of buffers comes from the optimal determination of the set C.
The space-optimal approach to buffer assignment has the drawback that the concurrency in
intertask communication may be too low, inasmuch as many channels in D
T
may be
allocated zero buffers. Extreme situations can happen, as for example the assignment of
zero buffers to all the channels of a long directed path in G
T
. A scenario might then happen
in which all tasks in this path (except the last one) would be suspended to communicate with
its successor on the path, and this would only take place for one pair of tasks at a time.
When at least one channel c has insufficient buffers (i.e., b(c) < r(c)) or is such that r(c) = M,
a measure of concurrency that attempts to capture the effect we just described is to take the
minimum, over all directed paths in G
T
whose channels c all have b(c) < r(c) or r(c) = M, of
the ratio

where L is the number of channels on the path. Clearly, this measure can be no less than
1/|N
T
| and no more than 1/2, as long as the assignment of buffers conforms to the
hypotheses of Theorem 1.1. The value of 1/2, in particular, can only be achieved if no
directed path with more than one channel exists comprising channels c such that b(c) < r(c)
or r(c) = M only.
Another criterion for buffer assignment to channels is then the concurrency-optimal criterion,
which also seeks to save buffering space, but not to the point

Figure 1.1: A graph G
T
is shown in part (a). In the graphs of parts (b) through (d),
circular nodes are the nodes of G
T
, while square nodes represent buffers assigned to
the corresponding channel in G
T
. If r(c) = 1 for all c ∈ {c
1
, c
2
, c
3
, c
4
}, then parts (b)
through (d) represent three distinct buffer assignments, all of which deadlock-free. Part
(b) shows the strategy of setting b(c) =r(c) for all c ∈{c
1
, c
2
,c
3
, c
4
}. Parts (c) and (d)
represent, respectively, the results of the space-optimal and the concurrency-optimal
strategies.
that the concurrency as we defined might be compromised. This criterion looks for buffer
assignments that yield a level of concurrency equal to 1/2, and for this reason does not allow
any directed path with more than one channel to have all of its channels assigned insufficient
buffers. This alone is, however, insufficient for the value of 1/2 to be attained, as for such it is
also necessary that no directed path with more than one channel contain channels c with r(c)
= M only. Like the space-optimal criterion, the concurrency-optimal criterion utilizes a value
of M such that

This criterion requires a subset of channels C ⊆ D
T
to be found such that no directed path
with more than one channel exists in G
T
comprising channels from C only, and such that

is maximum over all such subsets (clearly, C
+
⊆ C, given the value of M, unless C
+
contains
the channels of an entire directed path from G
T
with more than one channel). The strategy is
then to set

thereby ensuring that at least one channel c in every directed path with more than one
channel in G
T
is assigned b(c) = r(c) buffers, and that, as a consequence, at least one
channel c from every directed cycle in G
T
is assigned b(c) = r(c) buffers as well (Figure 1.1).
By Theorem 1.1, this strategy then produces a deadlock-free result if no directed cycle in G
T

has all of its channels in the set C
+
. The strategy also provides concurrency equal to 1/2 by
our definition, as long as C
+
does not contain all the channels of any directed path in G
T
with
more than one channel. Given this constraint that optimal concurrency must be achieved (if
possible), then the strategy employs the minimum number of buffers, as the set C is
optimally determined.

1.6 Processor allocation
When we discussed the routing of messages among processors in Section 1.3 we saw that
addressing a message at the task level requires knowledge by the processor running the
task originating the message of the processor on which the destination task runs. This
information is provided by what is known as an allocation function, which is a mapping of the
form

where N
T
and N
P
are, as we recall, the node sets of graphs G
T
(introduced in Section 1.4)
and G
P
(introduced in Section 1.3), respectively. The function A is such that A(t) = p if and
only if task t runs on processor p.
For many of the systems reviewed in Section 1.1 the allocation function is given naturally by
how the various tasks in N
T
are distributed throughout the system, as for example computer
networks and networks of workstations. However, for multiprocessors and also for networks
of workstations when viewed as parallel processing systems, the function A has to be
determined during what is called the processor allocation step of program design. In these
cases, G
T
should be viewed not simply as the task graph introduced earlier, but rather as an
enlargement of that graph to accommodate the relay tasks discussed in Section 1.5 (or any
other tasks with similar functions—cf. Exercise 4).
The determination of the allocation function A is based on a series of attributes associated
with both G
T
and G
P
. Among the attributes associated with G
P
is its routing function, which,
as we remarked in section 1.3, can be described by the mapping

For all p,q ∈ N
P
,R(p,q) is the set of links on the route from processor p to processor q,
possibly distinct from R(q,p) and such that R(p, p) = . Additional attributes of G
P
are the
relative processor speed (in instructions per unit time) of p ∈ N
P
, s
p
, and the relative link
capacity (in bits per unit time) of (p,q) ∈ E
P
, c(
p
,
q
) (the same in both directions). These
numbers are such that th