Combining static and dynamic scheduling on distributed-memory multiprocessors.
Loops are a large source of parallelism for many numerical applications. An important issue in the parallel execution of loops is how to schedule them so that the workload is well balanced among the processors. Most existing loop scheduling algorithms were designed for shared-memory multiprocessors, with uniform memory access costs. These approaches are not suitable for distributed-memory multiprocessors where data locality is a major concern and communication costs are high. This paper presents a new scheduling algorithm in which data locality is taken into account. Our approach combines both worlds, static and dynamic scheduling, in a two-level (overlapped) fashion. This way data locality is considered and communication costs are limited. The performance of the new algorithm is evaluated on a CM-5 message-passing distributed-memory multiprocessor.
Introduction
Loops exhibit most of the parallelism present in numerical
programs. Therefore, distributing the workload of a loop
evenly among the processors is a critical factor in the efficient
execution of this kind of parallel programs. A loop
scheduling strategy assigns iterations to the processors of the
machine in such a way that all of them finish their work-load
at more or less the same time. A simple strategy is
static scheduling, which determines the assignment of iterations
to processors at compile-time. But in many situations,
the workload of the iterations is unpredictable at compile-
time. Dynamic scheduling strategies have been developed
to handle such situations, solving the assignment problem
at run-time.
When considering cross-iteration data dependencies we
can classify the loops into three major types, doserial, doall
and doacross [Pol88]. In this paper we are primarily concerned
with doall or parallel loops, in which there is no data
dependence between any pair of their iterations. Otherwise,
Published in Proc. ACM Int'l Conf. on Supercomputing,
Manchester, UK, July 11-15, 1994
the loop is taken as serial. From the point of view of the
scheduling a loop is uniform if the execution times of different
iterations are the same (e.g. matrix multiplication),
semi-uniform, if they depend on the index of the loop (e.g.
adjoint convolution), or non-uniform, if they depend on the
data (e.g. transitive closure).
We can always obtain an optimal (or near optimal) static
schedule for uniform loops. If we have a parallel machine
with P processors, we must simply partition the set of iterations
of the loop into P chunks of dP=Ne iterations each,
where N is the length of the loop. Now we assign each of
the chunks to each of the processors. We can still obtain an
(near) optimal static schedule for semi-uniform loops. For
example, consider the parallelized adjoint convolution algorithm
which exhibits a triangular iteration space,
We can establish a (near) optimal static schedule using a
kind of cyclic loop distribution scheme. We assign the iteration
rP (i from 1 to P and .) to processor
or to processor
Clearly, it is not possible to define an optimal static
schedule for a non-uniform loop because the data is known
only at run-time. These loops are frequent in many applications
such as image processing, sparse matrix computation
and partial differential equations, among others. Most
of the dynamic scheduling strategies used for balancing the
workload of a non-uniform parallel loop were designed for
shared-memory machines, with uniform memory access cost
(UMA). Therefore, the data locality is not taken into con-
sideration. Nevertheless, in a distributed-memory machine
the data locality is a major concern and must be optimized.
This paper concentrates on non-uniform parallel loops
and discusses an efficient solution for scheduling this kind of
loops on distributed-memory parallel machines. Our strategy
combines features from static and dynamic scheduling
techniques and succeeds in keeping the cost associated to remote
memory accesses low. Section 2 describes some scheduling
strategies proposed in the literature, specially for distributed-memory
machines. Our approach for performing loop
scheduling is explained in Section 3. Section 4 shows the
experimental results obtained on the CM-5 and, finally, the
paper is concluded in Section 5.
Related work
We can find some static scheduling solutions proposed in the
literature. In [PKP89] an optimal compile-time scheduling
strategy is proposed for perfectly-nested parallel loops with
constant bounds. Sarkar and Hennessy [SH86] proposed a
scheduling method based on splitting a program into sequential
tasks. In [BB91] the scheduling is solved as a linear optimization
problem. Tawbi and Feautrier [TF92] tested some
heuristics to solve the processor allocation and scheduling
problem. The application of symbolic analysis to derive an
optimal static scheduling can be found in [HP93].
For non-uniform problems it is necessary to balance the
load dynamically, at run-time. In self-scheduling (SS) [TY86],
the simplest strategy, each processor repeatedly selects and
executes one iteration until all iterations are executed. An
experimental comparison of self-scheduling and static pre-scheduling
algorithms can be found in [Zha91]. (Fixed-size)
chunk scheduling [KW85] reduces the high synchronization
overhead found in SS by scheduling chunks of more than one
iteration as units. GSS [PK87] schedules large chunks at the
beginning of the loop (low synchronization overhead) and
small ones toward the end of the loop, trying to balance the
workload. In Factoring scheduling [HSF92] the allocation of
iterations to processors proceeds in phases. In each phase,
a part of the remaining iterations is divided equally among
the available processors. Trapezoid scheduling [TN93] is a
variation of GSS where the size of the successive chunks
decreases linearly instead of exponentially. In [Luc92] a tapering
scheduling is proposed, similar to GSS and Trapezoid
scheduling but with a different function for the decreasing
size of the chunks.
All of above dynamic scheduling were designed for shared-memory
parallel machines, where an efficient method of synchronization
and data sharing is found. For distributed-memory
parallel machines, there is a big difference between
the access time to the local memory and to the remote mem-
ories. Then, the synchronization overhead and the data-sharing
cost is high in comparison with access cost of local
data. A much better scheduling performance is obtained if
the locality of the data is exploited. It was not until very
recently that some dynamic scheduling strategies which exploit
the data locality have appeared in the literature.
In [ML94], the affinity dynamic scheduling (ADS) is introduced
for NUMA machines. This algorithm, as GSS or
Trapezoid scheduling, assigns large chunks of iterations at
the start of loop execution and uses a deterministic assignment
policy to ensure that a chunk is always assigned to
the same processor. This way, data locality is exploited,
but it only works in the case that the same data is reused
several times. That happens, for example, when the parallel
loop is surrounded by a sequential one, and then the
same chunk is executed repeatedly. A locality-based dynamic
scheduling (LDS), designed for NUMA machines, is
described in [LTSS93]. LDS considers the data space is partitioned
throughout the processors. It computes the sizes of
the chunks using a GSS-like scheme. Each chunk is assigned
to a processor by demand and it includes those iterations
whose data is stored in the local memory of the processor
assigned.
In the case of message-passing machines, Rudolph and
Polychronopoulos [RP89] implemented and evaluated static,
self-, chunk and guided-self scheduling in the iPSC/2 hyper-cube
multicomputer. It was the first attempt to evaluate
some shared-memory dynamic scheduling algorithms on a
distributed-memory multiprocessor.
do foreverf /* executed by all processors */
Static scheduling level();
Dynamic scheduling level();g

Figure

1: Hybrid scheduling (HS) algorithm
In distributed-memory machines purely dynamic scheduling
strategies are poor performers due to their relatively
high synchronization overhead. On the other hand, static
scheduling schemes are not suitable for non-uniform prob-
lems. In both situations, a hybrid scheme can perform bet-
ter, as it can exploit data locality and exhibit relatively low
synchronization overhead, and it can balance the workload
dynamically at the same time.
An example of such an approach is presented in [LSL92],
where the two-phase Safe Self-Scheduling (SSS) is described.
In the first phase, a subset of the iterations of the loop is
distributed uniformly among the processors. The second
phase of SSS is activated at run-time when a processor becomes
idle. A chunk of not yet executed iterations is chosen
and assigned dynamically to each requesting processor.
This strategy considers a centralized queue of iterations and
that all processors can access all data. Distributed Safe Self-Scheduling
(DSSS), an adaptation of SSS to message-passing
machines is discussed in [SLL93]. The data is partitioned
into small blocks of the same size and distributed with partial
redundancy among all the processors. Now DSSS proceeds
as SSS but the chunks of iterations are assigned to
the processors that have the corresponding data. DSSS is
generalized in [LS93].
The dynamic phase in DSSS is fired when a processor
becomes idle. This processor is in charge of the scheduling
duties and distributes by demand the rest of the workload
among all the other processors. Inherently this dynamic
redistribution of load may imply a low performance.
This paper describes a new approach to hybrid schedul-
ing. We propose a two-level scheme in which both phases
(static and dynamic) overlap. ADS uses a similar two-level
scheme, where the exploitation of data-locality is a major
concern and the load-balancing is accomplished by demand
(when a processor becomes idle). In our case, data-locality
is considered, and we try to predict in advance the work-load
unbalances in order to obtain a better load-balance
and to reduce the communication overhead. While the processors
are executing their statically distributed workload,
a dynamic redistribution of the rest of the workload is in
action. Hence the communication overhead associated with
the dynamic level of the scheduling is hidden by the local
computations. Our experiments show that with such a two-level
strategy we can obtain a good performance.
3 Scheduling scheme
The parallel loop we consider in the rest of the paper has
the following simple structure,
loop body(I) (1)
The hybrid scheduling (HS) we propose is accomplished
in two levels, as shown in Figure 1.
Static level: First, we statically partition and evenly distribute
the iteration space of the parallel loop among the
processors of the machine. If we denote by P the number of
processors, each processor will execute a total of
iterations of the parallel loop (1),
local loop body(i)
(note that we use lowercase indices to represent local variables
and uppercase for the global ones). If we denote by
ff the numeric address of each processor,
then the relation between the local and global indices, I
depends on the static distribution scheme we have
used. For example, in the case of a block distributed loop,
we have
The form of the local loop body depends on the data
distribution scheme chosen. If we choose to partition the
data throughout the local memories of all the processors,
then local loop body(i) is identical to loop body(i). If
we prefer to replicate all the data on all local memories
then local loop body(i) is equal to loop body(I), where
Other combinations can be found if a partial
replication of the data is considered.
Dynamic level: The second level requires splitting each
local iteration space in a set of chunks. The sizes of the
chunks may be constant or variable. Indeed, we may use one
of the dynamic scheduling algorithms found in the literature
in order to accomplish the partitioning of the local iteration
spaces. For example, if the GSS algorithm is considered,
the set of n local iterations would be partitioned into a total
of S iteration groups (chunks), where
e is the
size of chunk number i, S. Therefore, the local
loop is transformed into
upper-bound
lower-bound, upper-bound
local-loop-body(i)
where chunk size(c) computes the size of chunk number c.
All the actions associated with the dynamic level of HS
are carried out at the boundary between the execution of
one chunk and the following one. This way, we can establish
the hybrid degree between the two extremes, static and dy-
namic, by choosing the size (and the number) of the chunks.
When a certain unbalance of the workload is detected the
dynamic level of HS becomes activated and tries to reduce
the unbalance below a given threshold. During this redistribution
the local computations are still going on.
Specifically, just after the execution of each chunk of iterations
and before starting with the next one, all processors
carry out some actions in order to arrange themselves from
the fastest to the slowest one (that is, from the most lightly
loaded to the most heavily loaded processor) at that point
of execution. Then, a redistribution of load (chunks of it-
erations) is executed, if it is necessary. This redistribution
process overlaps the execution of the local chunks.
A critical point in HS is how to arrange efficiently such
classification of the processors, in order to determine which
processors send load and which receive it. We solve this
Workers:
local queue is not empty )f
Send a load msg to the scheduler;
Remove a chunk from my local queue;
Execute the chunk removed;g
else if ( FirstTime == YES )f
Send a load needed msg to the scheduler;
Scheduler:
local queue is not empty )f
Increment my workload counter;
Remove a chunk from my local queue;
Execute the chunk removed;g
else if ( FirstTime == YES )f
Select mhlw, the most heavily loaded worker;
Send a load request msg to mhlw;
if ( all workers are dead
&& my local queue is empty
&& no messages traveling ) exit();
load msgs has been received )

Figure

2: Static scheduling level
problem assigning the classification work to one of the pro-
cessors. All the processors are workers and one of them
is, in addition, in charge of the dynamic scheduling duties.
Therefore, all of the workers must send messages periodically
to the scheduler with the information needed in order
to establish their classification at the right times (at chunk
boundaries).
We have implemented HS in a message-passing distributed-memory
machine using a simple protocol of seven different
types of messages. The meaning of these messages is
described in Table 1. The rest of this section is dedicated to
a detailed description of both levels in HS.
3.1 Static scheduling level

Figure

2 shows a detailed description of the static level in
HS from both points of view, workers and scheduler (which
is also a worker). Under this level, each worker sends one
(empty) load message to the scheduler just before executing
one of its local chunks. Therefore, the time elapsed between
two consecutive load messages is equal to the time elapsed
between the beginning of the execution of one local chunk
and the next one. The scheduler carries out the classification
of all the workers just after the execution of each of its local
chunks (which is coded in the WorkloadRedistribution()
function in Figure 2).
When a worker becomes idle it sends a load needed message
to the scheduler, requesting a non-executed chunk from
other worker. The search of this worker is accomplished by
the scheduler during its dynamic level, unless the requesting
worker is the scheduler itself. Note that, in this case, the load
request message is sent only if the selected worker (mhlw) has
a minimum number of non-executed chunks, parameterized
by the constant TransferLimit. With this mechanism the
Type Meaning
Load A worker sends it to the scheduler before the execution of each local chunk
Load needed A worker sends it to the scheduler when it becomes idle
Load migrated A worker sends it to a requesting processor with its next non-executed local chunk
Request failed A worker sends it to a requesting processor when its local queue is almost empty
Data returned It is used to return the remotely processed data block to its original place
Load request The scheduler requests to a worker to send its next non-executed chunk to a processor
Load finished There are no more chunks to be execute remotely

Table

1: Messages used in the HS implementation
Scheduler:
Read all load msgs received and
update workload counters;
Classify the corresponding workers
according to their workload;
while ( there are workers in the classification list )f
Remove mllw, the most lightly loaded worker
in the classification list;
Choose mhlw, the most heavily loaded worker,
from the classification list;
Send a load migrated msg to mllw with
my next non-executed local chunk;
else
Send a load request msg to mhlw with
mllw as the requesting processor;g
if ( no message has been sent )f
if ( mllw has sent a load needed msg )f
Send a load finished msg to mllw;
Reduce number of workers alive;
if ( all workers are dead
&& my local queue is empty
&& no messages traveling ) exit();

Figure

3: Workload redistribution
migration of the remaining chunks is avoided as the communication
latency would probably eliminate the benefits
associated with balancing this remaining load (this is specially
true if we use algorithms like GSS to partition the
local loops, as the last chunks are very small).
3.2 Workload redistribution
Periodically, the scheduler tries to evenly redistribute the
load among all the workers. After the execution of each of
the local chunks, the scheduler reads all the load messages
received and by counting them it can determine the relative
local "speed" of all the workers. Note that this is a rather
"instantaneous" measure, because the scheduler counts only
the load messages received during the execution of its last
local chunk. Based on this information, the scheduler classifies
the workers that have sent load messages in order of
decreasing speed. Once the classification is completed, the
scheduler orders the heavy-loaded processors to send chunks
to the light-loaded ones (see Figure 3). This load redistribution
is accomplished while the the rest of the workers are
executing their local chunks.
A worker do not send a load message before starting to
execute a remote chunk (see next subsection). This way, a
fast worker will receive a number of successive remote chunks
and the frequency of arrival of its load messages will be reduced
as a consequence. Thus, the scheduler will eventually
classify this worker as slow and will not send more remote
load to it until it becomes fast at a later stage (if that hap-
pens).
With this mechanism, the scheduler tries to periodically
delay the fastest processors and to alleviate the load of the
slowest ones. There is a tradeoff between the accuracy of
the measures of the relative speeds and the communication
cost associated with that measure. The more accuracy we
achieve the more sensitive the scheduler is to changes in the
local loads and thus the better is the balancing of the load,
because the response to these changes is faster and more
precise. But in order to obtain high accuracy the workers
must send load messages with a high frequency (by reducing
the size of the chunks). That can increase the communication
overhead to intolerable levels. The tradeoff between
classification accuracy and associated communication cost is
established by choosing the size of the chunks of iterations.
In order to prevent thrashing of remote chunks, we incorporate
two simple mechanisms. First, a remote chunk is
always executed in the destination processor (redistribution
of remote chunks to several workers is prohibited). Second,
the workload counters used to select the slow workers (load[]
vector in Figure 3) are modified during the load redistribution
process. If a chosen mhlw sends one of its non-executed
local chunks, then its workload counter is decremented ac-
cordingly. This way a single slow (heavy loaded) worker
is prevented from sending too many chunks with the risk
of having that processor lose all of its local chunks. Likewise
the workload counter of the corresponding destination
worker is incremented for each remote chunk assigned, trying
to prevent bulk arrivals.
3.3 Dynamic processor coordination
We hide (overlap) most of the communication latency of
redistributing load by executing local computations at the
same time. The processors continue to execute local chunks
while the transferred chunks are traveling through the interconnection
network. Between each local chunk and the
following one, the workers check the arrival of a control mes-
sage, with, possibly, a remote chunk to be executed. If it
is not arrived, the workers immediately begin the execution
of the following local chunk. Otherwise, they execute the
remote chunk and, finally, return the processed remote data
to the owner. This last action is only necessary if we do not
want to modify the original (static) data distribution. In
some cases this modification is irrelevant, for example, when
Workers:
while ( there are pending messages )f
switch ( type of the message received )f
case load finished msg:
if ( there are no pending messages ) exit();
break;
case load request msg:
Send a load migrated msg to the requesting
processor with my next non-executed chunk;
else
Send a request failed msg to
the requesting processor;
break;
case load migrated msg:
Execute migrated chunk;
Send a data returned msg with the migrated data
to its owner (if necessary);
if ( no load finished msg received )
else if ( no messages traveling ) exit();
break;
case request failed msg:
if (no load finished msg received )
else if ( no messages traveling ) exit();
break;
case data returned msg:
Store the remotely processed data block
in its original location;
else if ( no messages traveling ) exit();
break;gg

Figure

4: Dynamic scheduling level (workers)
we want to execute the same parallel loop several times, one
after another. In this case, we can delay the reconstruction
of the original data distribution until the last time the
parallel loop is executed.
Due to the communication latency, we lose part of the
load balance because the decision of redistributing load and
the redistribution itself are carried out at different times.
For example, it is possible for a fast processor to enter a
heavily loaded part of its local computation before it receives
a previously transferred remote chunk. In this case,
we further increase the load of a, currently, slow processor.
Nevertheless, the increase of the error in balancing the load
is overcome by the benefit of reducing the communication
overhead. The net result is beneficial, as our experiments
show.
During the workload redistribution phase explained in
the above section the scheduler determines the set of workers
which must send chunks and the set which must receive
them. Now we need to establish a dynamic processor co-ordination
in order to efficiently transfer the chunks among
the selected workers. During this coordination all workers
check different kinds of messages and execute some actions
according to the control scheme shown in figures 4 and 5).
One can argue that if the number of workers is very large
then the scheduler can become a bottleneck. With a large
number of workers, the scheduler would probably execute
just the first of its local chunks and then redistribute the
rest of them among the workers. But that depends on the
Scheduler:
while ( there are pending messages )f
switch ( type of the message received )f
case load needed msg:
Select mhlw, the most heavily loaded worker;
Send a load migrated msg to the requesting
processor with my next non-executed chunk;
else Send a load request msg to mhlw;
Send a load finished msg to
the requesting processor;
Reduce number of workers alive;
if ( all workers are dead
&& my local queue is empty
&& no messages traveling ) exit();
break;
case load migrated msg:
Execute migrated chunk;
Send a data returned msg with the migrated data
to its owner (if necessary);
break;
case request failed msg:
break;
case data returned msg:
Store the remotely processed data block
in its original location;
else if ( no messages traveling ) exit();
break;gg

Figure

5: Dynamic scheduling level (scheduler)
static load initially assigned to the scheduler in relation with
the loads of the workers. Indeed our experiments show that,
in general, the efficiency of the parallel loop is lower if the
scheduler is one of the initially lightly loaded processors.
Moreover the scheduler is in charge of the periodic classification
of the workers, core of the workload redistribution
process. This may introduce another potencial performance
bottleneck. In this paper we show results using just one
scheduler but the possibility of having several schedulers will
be considered in the future.
4 Experimental evaluation
We have implemented HS on the CM-5 using the CMMD
message-passing library [Thi91]. In addition to HS and in
order to make comparisons, the fully static and fully dynamic
schedulings were implemented as well. We then measured
the performance of all the scheduling algorithms using
two benchmark programs.
4.1 Benchmark programs
We have chosen matrix multiplication (MM) and (a simpli-
fied) transitive closure (TC) as our benchmark programs.
MM is a typical example of a very regular (uniform) prob-
lem, and TC represents an irregular (non uniform) one,
where the workload of the parallel loop depends heavily on
the data content.
ffl Matrix Multiplication: This program has three perfectly
nested loops, where the innermost one contains
a reduction operation. The other two loops are fully
parallel. In our experiments we have parallelized the
program at the outermost loop.
The regularity of this program will be useful in evaluating
the effectiveness of HS as compared to (optimal)
static scheduling.
Transitive closure: The TC program is a 3-depth nested
loop as well. The outermost loop is serial and the the
next one is parallel. For this reason the first loop was
dropped in our experiments (that is, just one iteration
of the outermost loop was considered), as follows,
The sequential innermost loop may or may not be exe-
cuted, and hence the computation is non-uniform, depending
on the input data. This program will allow us
to evaluate the performance of HS in a fully dynamic
environment.
4.2 Scheduling algorithms
We have implemented the following scheduling algorithms:
static scheduling (Static), HS with data partitioned through-out
the local memories (HSp), HS with data replicated in all
the local memories (HSr) and dynamic scheduling with data
partitioned throughout the local memories, considering the
scheduler as a worker too (DynamicP) or in charge of the
scheduling duties only (DynamicPS).
Static scheduling consists in a block-wise static distribution
of the iterations of the parallelized loop. The matrices
are partitioned and block distributed as well. In the case
of MM, matrices A and C are block partitioned by rows
(and the partitions stored in the respective local matrices a
and c) and matrix B is replicated. The local code for each
processor is as follows,
For the TC, matrix A is block partitioned by rows as
well (and stored in local matrix a) and the first row of A is
replicated (and renamed as A1[]). The local TC is now as
Due to the fact that the execution time of the loop body of
the parallelized loop in TC is very small, we have introduced
another loop surrounding it (granularity control loop), with
a number of LoopBodySize iterations. This loop permits
parameterizing the loop body size and studying its effect in
our experiments.
In order to obtain an initially poor workload balance for
static scheduling in the TC benchmark, we have taken an
input matrix A with 1's in the first half of its rows, and 0's in
the rest of them. With this data, static scheduling performs
around 50% efficiency. This way we can clearly evaluate
the benefits of including dynamic actions in the scheduling
strategy for this irregular problem.
We have considered the same data distribution scheme
for the static level in HSp. With HSp and HSr we can analyze
the global impact of data migration on the performance
of HS, as the partitioning scheme may imply a great amount
of data migration. In any case, the scheme chosen to partition
the local loops is GSS(0) for the scheduler and GSS(2)
for the rest of the workers. With GSS(2) we obtain a reasonably
small communication overhead due to the relatively
reduced number of messages sent to the scheduler by the
workers (as the chunks presents bigger sizes). On the other
hand, using GSS(0) for the scheduler we improve its "sensi-
tiveness" to dynamic workload changes and hence obtain a
reasonable load balancing. We have selected the value 2 (1)
for the TransferLimit parameter for the workers (sched-
uler). The same algorithm, GSS(0), was chosen for the fully
dynamic schedulings.
In all the cases where data was migrated we have always
considered that it returns to the original owner just after
the execution of the remote chunk.
4.3 Results and interpretation

Figure

6 (a) displays the efficiency versus the number of processors
for MM in the case of two different matrix sizes, comparing
the performance of the three scheduling algorithms,
static, hybrid and dynamic. In the computation of the efficiency
we have executed the same scheduling algorithm on
one processor. We can see that the best performer is static
scheduling and the worst performer is the dynamic one. HS
performs quite similar to static scheduling, especially in the
case of sufficiently large matrices. This behavior is explained
by the fact that all the workers process (almost) the same
workload (because the loop was evenly distributed across
the local memories at compile-time). All the workers send
load messages more or less at the same time to the sched-
uler. If the execution time of the chunks is sufficiently large
then the messages arrive to the scheduler with a difference
in time (due to different communication paths taken) that
is smaller than the execution time of a single chunk. This
way, the workload counters contain the same number (or at
most with a difference of one unit) for all the processors.
Thus the scheduler never decides to migrate any of the local
chunks and HS works similar to the static strategy (with
some overhead associated with the control messages). More-
over, a possible load unbalance may appear only during the
MM (100*100)0.90.70.50.3
Number of processors
Efficiency
(a)
Number of processors0.90.70.50.3
MM (400*400)
HSr
Number of processors
Efficiency
(b)
HSr
MM (400*400)
Number of processors0.990.970.950.93
Efficiency

Figure

Efficiency for matrix multiplication
last local chunk for some of the processors. In this case the
TransferLimit parameter prevents the migration of the
remaining chunks. Besides this explains that both versions,
HSp and HSr, perform roughly in a similar way, as shown
in

Figure

6 (b).
For the TC program, HSp performs best for large loop
bodies (see Figure 7). We have experimented with two matrix
sizes and three values for the upper bound of the granularity
control loop (LoopBodySize), 10, 100 and 1000. If the
dimension of the input matrix and/or the size of the parallelized
loop body is sufficiently large the efficiency of HSp
rises to around 75 to 80%. Instead the best purely dynamic
scheduling is always below the static one. These results
shows the advantages of considering data locality (static
level in HS) and the overlapping between the chunk migrations
and the local computations (dynamic level in HS). If
the data or the size of the loop body are small the chunk
migration costs are only partially hidden by the local computation
time. In this case the performance of HS is very poor
and becomes worse as the number of processors increases.
Note that for a large input matrix the efficiency of HSp remains
around 75% for any number of processors considered.
Therefore the use of just one scheduler is sufficient in order
to balance the load among a few dozens of processors.
The input matrix chosen (half 1's and half 0's) splits
the processors into two classes, heavily loaded and lightly
loaded processors. With the initial static loop distribution,
half of the processors are in charge of all of the computations
and the other half are idle. We have experimented the
influence of the initial workload assigned to the scheduler on
the performance of HS. The results are shown in Figure 8,
where the data partitioned and replicated versions of HS are
considered too. On the one hand, it is important to point
out that, for large loop bodies, the efficiency of HS is not
greatly affected by the data distribution scheme chosen. On
the other hand, the efficiency is noticeably worse if a lightly
loaded (initially idle) scheduler is chosen. This is reasonable
because a scheduler of this kind goes into the workload redistribution
level very frequently. It thus sends many messages
to the heavily loaded processors ordering them to migrate
many of their chunks. Hence the total communication cost
is high. This behavior is prevented if the scheduler has a
heavy workload, but at expense of a probably worse work-load
balance. Indeed, we have found many cases where the
load is balanced worse if the scheduler is a heavily loaded
worker than if it is not. Nevertheless the efficiency of the
former case is better due to its limited communication cost.
Number of processors0.50.30.12 4 8
Efficiency Static
Number of processors
Efficiency
Number of processors0.60.40.2
Efficiency
Number of processors
Efficiency
Number of processors
Efficiency
Number of processors
Efficiency

Figure

7: Efficiency for transitive closure
Conclusions
We have studied the problem of loop scheduling on distributed-memory
multiprocessors. If the loop is regular a compile-time
(static) scheduling is a (near) optimal solution, where
there is no need to migrate load among the processors. If
the loop is irregular (the execution time of the loop body
depends on the data, for example) we will thus need to consider
runtime (dynamic) techniques in order to manage load
imbalances well. Most dynamic scheduling algorithms do
their work well but in a shared-memory environment. If the
memory is distributed we will have to take data locality into
account if we want to keep communication overhead under
certain limit.
We believe that the best solution would be to combine
static and dynamic strategies in a loop scheduling algorithm
if we want to obtain high performance on a distributed-memory
multiprocessor. We have described such an algo-
rithm, Hybrid Scheduling (HS), in which the static and dynamic
nature is presented in two levels. This conception
permits overlapping the local computations with the migration
of part of the workload. This way a large part of the
high communication overhead associated with migrations
does not contribute to the completion time. In order to
orchestrate the dynamic workload balance one of the workers
is, at the same time, in charge of the dynamic scheduling
duties.
We have evaluated the performance of HS on a CM-5.
For a regular program (matrix multiplication), HS performs
near static scheduling, depending on data size. In this case,
HS does not carry out any load migration and the overhead
associated with the dynamic level is very low. For an irregular
program (transitive closure), HS performs best if the
loop body is sufficiently large. For many data sizes the load
is not very well balanced but the communication overhead
associated with migrations is low. Our results indicate that
the combination of both effects is in general beneficial for a
machine as the CM-5.
An important result is that the performance of HS is
not very "sensitive" to the data distribution scheme chosen
for the benchmarks tested. This is important in many
applications in which it is not easy to find the best data dis-
tribution. Further investigations include the in depth study
of the influence of the data distribution chosen at the static
level on the performance. Moreover we are working on a
multi-scheduler extension of HS and in obtaining more experimental
results in a different kind of distributed-memory
machine.

Acknowledgements

This research has benefited significantly from discussions
with and suggestions from Constantine D. Polychronopou-
los. Most of this material and all the experimental results
were obtained during a research stay in the Center for Supercomputing
Research and Development, University of Illinois
at Urbana-Chaimpaign. We would like to thank David
Padua and his compiler group for the support they gave us.
This research was supported in part by the CICYT under
grant TIC92-0942-C03-03 and the Xunta de Galicia under
grant XUGA20604A90



--R

"A Scheduling Algorithm for Parallelizable Dependent Tasks"
"Symbolic Analysis: A Basis for Parallelization, Optimization and Scheduling of Programs"
"Factoring, a Method for Scheduling Parallel Loops"
"Allocating Inde- pending Subtasks on Parallel Processors"
"Self-Scheduling on Distributed-Memory Machines"
"Scheduling Parallel Loops with Variable Length Iteration Execution Times on Parallel Comput- ers"
"Locality and Loop Scheduling on NUMA Multi- processors"
"A Dynamic Scheduling Method for Irregular Parallel Programs"
"Using Processor Affinity in Loop Scheduling on Shared-Memory Multiprocessors"
"Guided Self-Scheduling: A Practical Scheduling Scheme for Parallel Supercomputers"
"Utilizing Multidimensional Loop Parallelism on Large-Scale Parallel Processor Systems"
Parallel Programming and Compilers.
"An Efficient Message-Passing Scheduler Based on Guided Self-Scheduling"
"Compile-Time Partitioning and Scheduling of Parallel Programs"
"Schedul- ing Non-Uniform Parallel Loops on Distributed Memory Machines"
"Processor Allocation and Loop Scheduling on Multiprocessor Com- puters"


"Processor Self-Scheduling for Multiple Nested Parallel Loops"
"Dynamic and Static Load Scheduling Performance on NUMA Shared Memory Multi- processors"
--TR
Allocating independent subtasks on parallel processors
Compile-time partitioning and scheduling of parallel programs
Guided self-scheduling: A practical scheduling scheme for parallel supercomputers
Utilizing Multidimensional Loop Parallelism on Large Scale Parallel Processor Systems
Dynamic and static load scheduling performance on a NUMA shared memory multiprocessor
Factoring: a method for scheduling parallel loops
A dynamic scheduling method for irregular parallel programs
Processor allocation and loop scheduling on multiprocessor computers
Self-scheduling on distributed-memory machines
An efficient message-passing scheduler based on guided self scheduling
Parallel Programming and Compilers
Trapezoid Self-Scheduling
Using Processor Affinity in Loop Scheduling on Shared-Memory Multiprocessors
Symbolic Analysis

--CTR
Babak Hamidzadeh , Lau Ying Kit , David J. Lilja, Dynamic Task Scheduling Using Online Optimization, IEEE Transactions on Parallel and Distributed Systems, v.11 n.11, p.1151-1163, November 2000
distributed-memory multiprocessors;dynamic and static scheduling;loop scheduling;load balancing;message-passing