To obtain good performance in such programs, it is important to schedule computation and distribute data across the underlying processors and memory modules, so that most cache misses are satisfied from local rather than remote memory. The primary goal of the programming support, therefore, is to enable user control over data placement and computation scheduling.

Cache behavior continues to be the largest single factor affecting performance, and programs with good cache behavior usually have little need for explicit data placement. In programs with high cache misses, if the misses correspond to true data communication between processors, then data placement is unlikely to help. In these cases, it may be necessary to redesign the algorithm to reduce inter-processor communication. Figure 12-2, shows this scenario.

If the misses are to data referenced primarily by a single processor, then data placement may be able to convert remote references to local references, thereby reducing the latency of the miss. The possible options for data placement are automatic page migration or explicit data distribution, either regular or reshaped. The differences between these choices are shown in Figure 12-2.

Automatic page migration requires no user intervention and is based on the run-time cache miss behavior of the program. It can therefore adjust to dynamic changes in the reference patterns. However, the page migration heuristics are deliberately conservative, and may be slow to react to changes in the references patterns. They are also limited to performing page-level allocation of data.

Regular data distribution (performing just page-level placement of the array) is also limited to page-level allocation, but is useful when the page migration heuristics are slow and the desired distribution is known to the programmer.

Finally, reshaped data distribution changes the layout of the array, thereby overcoming the page-level allocation constraints; however, it is useful only if a data structure has the same (static) distribution for the duration of the program. Given these differences, it may be necessary to use each of these options for different data structures in the same program.

Figure 12-2. Cache Behavior and Solutions

The regular data distribution directives try to achieve the desired distribution solely by influencing the mapping of virtual addresses to physical pages without affecting the layout of the data structure. Because the granularity of data allocation is a physical page (at least 16 KB), the achieved distribution is limited by the underlying page granularity. However, the advantages are that regular data distribution directives can be added to an existing program without any restrictions, and can be used for affinity scheduling (see “Affinity Scheduling”).

Similar to regular data distribution, the #pragma distribute_reshape directive specifies the desired distribution of an array. In addition, however, the #pragma distribute_reshape directive declares that the program makes no assumptions about the storage layout of that array. The compiler performs aggressive optimizations for reshaped arrays that violate standard layout assumptions but guarantee the desired data distribution for that array.

The compiler transforms a reshaped array into a pointer to a "processor array." The processor array has one element per processor, with the element pointing to the portion of the array local to the corresponding processor.

Figure 12-3, shows the effect of the #pragma distribute_reshape directive with a block distribution on a one-dimensional array. N is the size of the array dimension, P is the number of processors, and B is the block size on each processor.

Figure 12-3. Implementation of block Distribution

With this implementation, an array reference A[i] is transformed into a two-dimensional reference A[i/B] [i%B], where B, the size of each block, is equal to N/P, rounded up to the nearest integer value (ceiling(N/P)). A[i/B], therefore, points to a processor's local portion of the array, and A[i/B][i%B] refers to a specific element within the local processor's portion.

A cyclic distribution with a chunk size of 1 is implemented as shown in Figure 12-4.

Figure 12-4. Implementation of cyclic(1) Distribution

An array reference, A[i], is transformed to A[i%P] [i/P], where P is the number of threads in that distributed dimension.

Finally, a cyclic distribution with a chunk size that is either a constant greater than 1 or a run-time value (also called block-cyclic) is implemented as shown in Figure 12-5.

Figure 12-5. Implementation of block-cyclic Distribution

An array reference, A[i], is transformed to the three-dimensional reference A[(i/k)%P] [i/(P*k)] [i%k], where P is the total number of threads in that dimension, and k is the chunk size.

The compiler tries to optimize these divide and modulo operations out of inner loops through aggressive loop transformations such as blocking and peeling.

In summary, consider the differences between regular and reshaped data distribution. The advantage of regular distributions is that they do not impose any restrictions on the distributed arrays and can be freely applied in existing codes. Furthermore, they work well for distributions where page granularity is not a problem. For example, consider a block distribution of the columns of a two-dimensional array of size A[r][c] (row-major layout) and distribution [block][*]. If the size of each processor's portion, ceiling(r/P) ×c×element_size, is significantly greater than the page size (16KB on the Origin2000 server), then regular data distribution should be effective in placing the data in the desired fashion.

However, regular data distribution is limited by page-granularity considerations. For instance, consider a [block][block] distribution of a two-dimensional array where the size of a row is much smaller than a page. Each physical page is likely to contain data belonging to multiple processors, making the data-distribution quite ineffective. (Data distribution may still be desirable from affinity-scheduling considerations.)

Reshaped data distribution addresses the problems of regular distributions by changing the layout of the array in memory so as to guarantee the desired distribution. However, because the array no longer conforms to standard storage layout, there are restrictions on the usage of reshaped arrays.

Given both types of data distribution, you can choose between the two based on the characteristics of the particular array in an application.

For a given data structure in the program, you can choose a data distribution option based on the following criteria:

For example, consider the following code:

double A[m][n];
#pragma distribute A[*][block]

In this example, each processor's portion is approximately (m*n/P) elements (8*(m*n/P) bytes), where P is the number of processors. Due to the row-major layout of the array, however, each contiguous piece is only 8*(n/P) bytes. If n is 1000,000 then each contiguous piece is likely to exceed a page and regular distribution is sufficient. If instead n is small, say 10,000, then distribute_reshape is required to obtain the desired distribution.

In contrast, consider the following distribution:

#pragma distribute A[block][*] 

In this example, the size of each processor's portion is a single contiguous piece of (m*n)/P elements (8*(m*n)/P bytes). So if m is 100, for instance, regular distribution may be sufficient even if n is only 10,000.

As this example illustrates, distributing the outer dimensions of an array increases the size of an individual processor's portion (favoring regular distribution), whereas distributing the inner dimensions is more likely to require reshaped distribution.

Finally, the IRIX operating system on Origin2000 follows a default "first-touch" page-allocation policy; that is, each page is allocated from the local memory of the processor that incurs a page-fault on that page. Therefore, in programs where the array is initialized (and consequently first referenced) in parallel, even a regular distribution directive may not be necessary, because the underlying pages are allocated from the desired memory location automatically due to the first-touch policy.

For irregular data structures, you can explicitly place data in the physical memory of a particular processor using the #pragma page_place directive.

By default, the compiler assumes that a distributed array is not dynamically redistributed, and directly schedules a parallel loop for the specified data affinity. In contrast, a redistributed array can have multiple possible distributions, and data affinity for a redistributed array must be implemented in the run-time system based on the particular distribution.

However, the compiler does not know whether or not an array is redistributed, because the array may be redistributed in another function (possibly even in another file). Therefore, you must explicitly specify the #pragma dynamic declaration for redistributed arrays. This directive is required only in those functions that contain a pfor loop with data affinity for that array. This informs the compiler that the array can be dynamically redistributed. Data affinity for such arrays is implemented through a run-time lookup.

This section discusses how the nest clause interacts with the affinity clause when the program has reshaped arrays.

When you combine a nest clause and an affinity clause, the default scheduling is simple, except when the program has reshaped arrays and is compiled -O3. In that case, the default is to use data affinity scheduling for the most frequently accessed reshaped array in the loop (chosen heuristically by the compiler). To obtain simple scheduling even at -O3, you can explicitly specify the schedtype on the parallel loop.

The following example illustrates a nested pfor with an affinity clause:

#pfor nest(i, j) affinity(i, j) = data(A[i][j])
for (i = 2; i < n; i++)
for (j = 2; j < m; j++)
A[i][j] = A[i][j] + i * j;

Details about the following multiprocessing environment variables can be found on the pe_environ(5) man page.

Useful compile-time options include:

-MP:dsm={on, off} (default on)
	All the data-distribution and scheduling features described in this chapter are enabled by default under -mp compilation. To disable all the DSM-specific directives (for example, distribution and affinity scheduling), compile with -MP:dsm=off. Note: Under -mp compilation, the compiler silently generates some book-keeping information under the directory rii_files. This information is used to implement data distribution directives, as well as perform consistency checks of these directives across multiple source files. To disable the processing of the data distribution directives and not generate the rii_files, compile with the -MP:dsm=off option.
-MP:clone={on, off} (default on)
	The compiler automatically clones procedures that are called with reshaped arrays as parameters for the incoming distribution. However, if you have explicitly specified the distribution on all relevant formal parameters, then you can disable auto-cloning with -MP:clone=off. The consistency checking of the distribution between actual and formal parameters is not affected by this flag, and is always enabled.
-MP:check_reshape={on, off} (default off)
	Enables generation of the run-time consistency checks across procedure boundaries when passing reshaped arrays (or portions thereof) as parameters.

The example below distributes sequentially the rows of a matrix. Such a distribution places data effectively only if the size of an individual row (n) exceeds that of a page.

double A[n][n];

/* Distribute columns in cyclic fashion */
#pragma distribute A [cyclic(1)][*]

/* Perform Gaussian elimination across rows
The affinity clause distributes the loop iterations based
on the row distribution of A */
for (i = 0; i< n; i++)
#pragma pfor affinity(j) = data(A[i][j])
for (j = i+1; j < n; j++)
... reduce row j by row i ...  

If the rows are smaller than a page, then it may be beneficial to reshape the array. This is easily specified by changing the keyword from distribute to distribute_reshape.

In addition to overcoming size constraints as shown above, the #pragma distribute_reshape directive is useful when the desired distribution is contrary to the layout of the array. For instance, suppose you want to distribute the columns of a two-dimensional matrix. In the following example, the #pragma distribute_reshape directive overcomes the storage layout constraints to provide the desired distribution:

double A[n][n];

/* Distribute columns in block fashion */
#pragma distribute_reshape A [*][block]


double sum[n];
#pragma distribute sum[block]

/* Perform sum-reduction on the elements of each column */
#pragma pfor local(j) affinity(i) = data(A[j][i])
for (i = 0; i< n; i++)
for (j = 0; j < n; j++)
        sum[i] = sum[i] + A[j][i];

This example demonstrates regular data distribution and data affinity. This example, run on a four-processor Origin2000 server, uses simple block scheduling. Processor 0 will calculate the results of the first 25,000 elements of A, processor 1 will calculate the second 25,000 elements of A, and so on. Arrays B and C are initialized using one processor; therefore, all of the memory pages are touched by the master processor (processor 0) and are placed in processor 0's local memory.

Using data distribution changes the placement of memory pages for arrays A, B, and C to match the data reference pattern. Thus, the code runs 33% faster on a four-processor Origin2000 than it would run using SMP directives without data distribution.

Without Data Distribution

double a[1000000], b[1000000];
double c[1000000];
int i;

#pragma parallel shared(a, b, c) local(i)
#pragma pfor
for (i = 0; i < 100000; i++)
a[i] = b[i] + c[i];

With Data Distribution

double a[1000000], b[1000000];
double c[1000000];
int i;
#pragma distribute a[block], b[block], c[block]

#pragma parallel shared(a, b, c) local(i)
#pragma pfor affinity(i) = data(a[i])
for (i = 0; i < 100000; i++)
         a[i] = b[i] + c[i];

A distributed array can be passed as a parameter to a subroutine that has a matching declaration on the formal parameter:

double A [m][n];
#pragma distribute_reshape A [block][*]
foo (m, n, A);

foo (int p, int q, double A[p][q]){
#pragma distribute_reshape A[block][*]
#pragma pfor affinity (i) = data (A[i][j])
for (i = 0; i < P; i++)
...
}

Because the array is reshaped, it is required that the #pragma distribute_reshape directive in the caller and the callee match exactly. Furthermore, all calls to function foo() must pass in an array with the exact same distribution.

If the array was only distributed (that is, not reshaped) in the above example, then the subroutine foo() can be called from different places with different incoming distributions. In this case, you can omit the distribution directive on the formal parameter, thereby ensuring that any data affinity within the loop is based on the distribution (at run time) of the incoming actual parameter.

double A[m][n], B[p][q];
double A [block][*];
double B [cyclic(1)][*];
foo (m, n, A);
foo (p, q, B);
...
foo (int s, int t, double X[s][t]) {
#pragma pfor affinity (i) = data (X[i+2][j])
for (i = ...)
}

This example shows how an array is redistributed at run time:

bar(int n, double X[n][n]) {
...
#pragma redistribute X [*][cyclic(<expr>)]
...
}

foo() {
double LocalArray [1000][1000];
#pragma distribute LocalArray [*][block]
/* the call to bar() may redistribute LocalArray */
#pragma dynamic LocalArray
...
bar(1000, LocalArray);
/* The distribution for the following pfor */
is not known statically */
#pragma pfor affinity (i) = data (LocalArray[i][j])
...
}

The next example illustrates a situation where the #pragma dynamic directive can be optimized away. The main routine contains a local array A that is both distributed and dynamically redistributed. This array is passed as a parameter to foo() before being redistributed, and to bar() after being (possibly) redistributed. The incoming distribution for foo() is statically known; you can specify a #pragma distribute directive on the formal parameter, thereby obtaining more efficient static scheduling for the affinity pfor. The subroutine bar(), however, can be called with multiple distributions, requiring run-time scheduling of the affinity pfor.

main () {
double A[m][n];
#pragma distribute A [block][*]
#pragma dynamic A
foo (A);
if (...) {
#pragma redistribute A [cyclic(x)][*]
}
bar (A);
}

void foo (double A[m][n]) {
/* Incoming distribution is known to the user */
#pragma distribute A[block][*]
#pragma pfor affinity (i) = data (A[i][j])
     ...
}

void bar (double A[m][n]) {
/* Incoming distribution is not known statically */
#pragma dynamic A
#pragma pfor affinity (i) = data (A[i][j])
     ...
}

The example below consists of a large array that is conceptually partitioned into unequal portions, one for each processor. This array is indexed through an index array idx, which stores the starting index value and the size of each processor's portion.

double A[N];
/* idx ---> index array containing start index into A [idx[p][0]]
and size [idx[p][1]] for each processor */
int idx [P][2];
#pragma page_place A[idx[0][0], idx[0][1]*8, 0)
#pragma page_place A[idx[1][0], idx[1][1]*8, 1)
#pragma page_place A[idx[2][0], idx[2][1]*8, 2)
...
#pragma pfor affinity (i) = thread(i)
for (i = 0; i < P-1; i++)
     ... process elements on processor i ...
     ... A[idx[i][0]] to A[idx[i][0]+idx[i][1]] ...