Reduction

Reduction operation is widely used to gather values of some variable stored on different threads.

import java.io.IOException;
import org.pcj.*;
@RegisterStorage(PcjReduction.Shared.class)
public class PcjReduction implements StartPoint {
@Storage(PcjExample.class)
enum Shared { a }
public long a;
public static void main(String[] args) throws IOException {
String nodesFile = "nodes.txt";
PCJ.executionBuilder (PcjExample.class)
.addNodes(new File("nodes.txt"))
.start();
}
@Override
public void main() throws Throwable {
a = PCJ.myId() + 10; // set value of a
long sum = 0;
if (PCJ.myId() == 0) {
sum = PCJ.reduce(Lomg::sum, Shared.a);
}
PCJ.barrier();
System.out.println(PCJ.myId()+ " "+ sum);
}
}

In the presented example the values are communicated to the thread with the id 0. Than reduction operation such as summation is performed. Than value of the variable a (Shared variable) stored at the thread p is communicated to the thread 0 and added to the local variable sum.

import java.io.IOException;
import org.pcj.*;
@RegisterStorage(PcjReductionGet.Shared.class)
public class PcjReductionGet implements StartPoint {
@Storage(PcjReductionGet.class)
enum Shared { a }
long a;
public static void main(String[] args) throws IOException {
String nodesFile = "nodes.txt";
PCJ.executionBuilder (PcjExample.class)
.addNodes(new File("nodes.txt"))
.start();
}
@Override
public void main() throws Throwable {
a = PCJ.myId() + 10; // set value of a
long sum = 0;
PCJ.barrier();
if (PCJ.myId() == 0) {
for (int p = 1; p < PCJ.threadCount(); p++) {
sum = sum + (long) PCJ.get(p, Shared.a);
}
}
PCJ.barrier();
System.out.println(PCJ.myId()+ " "+ sum);
}
}

The presented algorithm of the reduction is based on the synchronous communication PCJ.get(). The summation is performed at thread 0 as data is arrived.

The asynchronous version requires additional storage at the thread 0. The al[]variable stores values of a variable communicated in asynchronous way:

import java.io.IOException;
import org.pcj.*;
@RegisterStorage(PcjReductionGet.Shared.class)
public class PcjReductionGet implements StartPoint {
@Storage(PcjReductionGet.class)
enum Shared { a }
long a;
public static void main(String[] args) throws IOException {
String nodesFile = "nodes.txt";
PCJ.executionBuilder (PcjExample.class)
.addNodes(new File("nodes.txt"))
.start();
}
@Override
public void main() throws Throwable {
PcjFuture aL[] = new PcjFuture[PCJ.threadCount()];
PCJ.barrier();
long sum;
if (PCJ.myId() == 0) { // Asynchronous communication
for (int p = 0; p < PCJ.threadCount(); p++) {
aL[p] = PCJ.asyncGet(p, Shared.a);
}
}
PCJ.barrier(); // Synchronization
sum = 0;
if (PCJ.myId() == 0) { // Sumation of local values
for (int p = 0; p < PCJ.threadCount(); p++) {
sum = sum + (long) aL[p].get();
}
}
System.out.println(PCJ.myId() + " " + sum);
}
}