count mode for parallel

Author: sebastianhaeni

src/main/java/ch/sebastianhaeni/pancake/IterativeSolver.java

@@ -12,10 +12,10 @@
 public final class IterativeSolver {
 
  // Change mode here!
- private static final Mode CURRENT_MODE = Mode.COUNT;
+ private static final Mode CURRENT_MODE = Mode.SOLVE;
 
  // Start with alternating sequence
- private static final int[] INITIAL_STATE = Generator.alternate(16);
+ private static final int[] INITIAL_STATE = Generator.alternate(14);
 
  // Start with random sequence
  // private static final int[] INITIAL_STATE = Generator.random(25));

src/main/java/ch/sebastianhaeni/pancake/ParallelSolver.java

@@ -11,7 +11,7 @@ public final class ParallelSolver {
  private static final Mode CURRENT_MODE = Mode.COUNT;
 
  // Start with alternating sequence
- private static final int[] INITIAL_STATE = Generator.alternate(10);
+ private static final int[] INITIAL_STATE = Generator.alternate(16);
 
  // Start with random sequence
  // private static final int[] INITIAL_STATE = Generator.random(25));

src/main/java/ch/sebastianhaeni/pancake/dto/Tags.java

@@ -24,7 +24,7 @@ public enum Tags {
  /**
  * The controller commands the worker to split it's work and send back half of it.
  */
- SPLIT;
+ SPLIT, GATHER;
 
  /**
  * Create an integer for MPI.

src/main/java/ch/sebastianhaeni/pancake/processor/Controller.java

@@ -3,7 +3,6 @@
 import ch.sebastianhaeni.pancake.dto.Tags;
 import ch.sebastianhaeni.pancake.dto.WorkPacket;
 import ch.sebastianhaeni.pancake.model.Node;
-import ch.sebastianhaeni.pancake.util.IntListener;
 import ch.sebastianhaeni.pancake.util.Partition;
 import ch.sebastianhaeni.pancake.util.Status;
 import mpi.MPI;
@@ -17,16 +16,16 @@ public abstract class Controller implements IProcessor {
 
  private static final int INITIAL_WORK_DEPTH = 1000;
 
- private final LinkedBlockingQueue<Integer> idleWorkers = new LinkedBlockingQueue<>(MPI.COMM_WORLD.Size() - 1);
+ final LinkedBlockingQueue<Integer> idleWorkers = new LinkedBlockingQueue<>(MPI.COMM_WORLD.Size() - 1);
  final Stack<Node> stack = new Stack<>();
- private final int[] workers;
- private final int workerCount;
+ final int[] workers;
+ final int workerCount;
  final int[] initialState;
  final Status status = new Status();
 
  private int candidateBound;
- private int bound = -1;
- private int lastIncrease = -1;
+ int bound = -1;
+ int lastIncrease = -1;
 
  Controller(int[] initialState, int workerCount) {
  this.initialState = initialState;
@@ -46,33 +45,9 @@ public void run() {
 
  abstract void work();
 
- private void initializeListeners() {
- for (int worker : workers) {
- (new Thread(new IntListener(Tags.IDLE, this::handleIdle, status, worker))).start();
- }
- }
-
- private void handleIdle(int source, int result) {
- if (idleWorkers.contains(source)) {
- return;
- }
- idleWorkers.add(source);
-
- if (result > bound) {
- bound = result;
- }
-
- if (idleWorkers.size() == workerCount) {
- idleWorkers.clear();
-
- if (lastIncrease == bound) {
- return;
- }
- lastIncrease = bound;
- (new Thread(this::solve)).start();
- }
- }
+ abstract void handleIdle(int source, int[] result);
 
+ abstract void initializeListeners();
 
  void clearListeners() {
  Object[] packetBuf = new Object[]{new WorkPacket(0, 0)};

src/main/java/ch/sebastianhaeni/pancake/processor/CountController.java

@@ -1,12 +1,17 @@
 package ch.sebastianhaeni.pancake.processor;
 
+import ch.sebastianhaeni.pancake.dto.Tags;
+import ch.sebastianhaeni.pancake.util.IntListener;
 import mpi.MPI;
 
 import static ch.sebastianhaeni.pancake.ParallelSolver.CONTROLLER_RANK;
+import static ch.sebastianhaeni.pancake.ParallelSolver.EMPTY_BUFFER;
 import static ch.sebastianhaeni.pancake.util.Output.showCount;
 
 public class CountController extends Controller {
 
+ private boolean foundSolution;
+
  public CountController(int[] initialState, int workerCount) {
  super(initialState, workerCount);
  }
@@ -34,6 +39,57 @@ void work() {
  finishCount(countBuf[0], end - start);
  }
 
+ @Override
+ void initializeListeners() {
+ for (int worker : workers) {
+ (new Thread(new IntListener(Tags.IDLE, this::handleIdle, status, worker, 2))).start();
+ }
+ }
+
+ @Override
+ void handleIdle(int source, int[] result) {
+
+ if (result[1] > 0 && !foundSolution) {
+ // found a solution, we won't increase the bound anymore
+ foundSolution = true;
+ System.out.println("Found the first solution. Not increasing bound anymore.");
+ }
+
+ if (idleWorkers.contains(source)) {
+ return;
+ }
+ idleWorkers.add(source);
+
+ if (result[0] > bound) {
+ bound = result[0];
+ }
+
+ if (idleWorkers.size() == workerCount) {
+ idleWorkers.clear();
+
+ if (foundSolution) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ if (idleWorkers.size() != workerCount) {
+ return;
+ }
+ for (int worker : workers) {
+ MPI.COMM_WORLD.Isend(EMPTY_BUFFER, 0, 0, MPI.INT, worker, Tags.GATHER.tag());
+ }
+ return;
+ }
+
+ if (lastIncrease == bound) {
+ return;
+ }
+ lastIncrease = bound;
+ (new Thread(this::solve)).start();
+ }
+ }
+
  private void finishCount(int count, long millis) {
  showCount(initialState, count, millis);
  clearListeners();

src/main/java/ch/sebastianhaeni/pancake/processor/CountWorker.java

@@ -1,5 +1,7 @@
 package ch.sebastianhaeni.pancake.processor;
 
+import ch.sebastianhaeni.pancake.dto.Tags;
+import ch.sebastianhaeni.pancake.util.IntListener;
 import mpi.MPI;
 
 import static ch.sebastianhaeni.pancake.ParallelSolver.CONTROLLER_RANK;
@@ -10,6 +12,13 @@ public class CountWorker extends Worker {
 
  @Override
  void work() {
+ (new Thread(new IntListener(Tags.GATHER, (source, result) -> {
+ status.done();
+ int[] countResult = new int[1];
+ countResult[0] = count;
+ MPI.COMM_WORLD.Reduce(countResult, 0, new int[1], 0, 1, MPI.INT, MPI.SUM, CONTROLLER_RANK);
+ }, status, CONTROLLER_RANK, 1))).start();
+
  while (!stack.isEmpty()) {
  count();
  }
@@ -37,7 +46,7 @@ private void count() {
  stack.pop();
  } else if (stack.peek().getChildren().empty()) {
  if (stack.peek().getDepth() == 0) {
- requestWork();
+ requestWork(count);
  } else {
  stack.pop();
  }
@@ -54,7 +63,7 @@ private void count() {
  }
 
  if (stack.isEmpty() && !status.isDone()) {
- requestWork();
+ requestWork(count);
  }
  }
 }

src/main/java/ch/sebastianhaeni/pancake/processor/SolveController.java

@@ -2,6 +2,7 @@
 
 import ch.sebastianhaeni.pancake.dto.Tags;
 import ch.sebastianhaeni.pancake.model.Node;
+import ch.sebastianhaeni.pancake.util.IntListener;
 import mpi.MPI;
 
 import java.util.Stack;
@@ -37,6 +38,35 @@ void work() {
  finishSolve(solution[0], end - start);
  }
 
+ @Override
+ void initializeListeners() {
+ for (int worker : workers) {
+ (new Thread(new IntListener(Tags.IDLE, this::handleIdle, status, worker, 1))).start();
+ }
+ }
+
+ @Override
+ void handleIdle(int source, int[] result) {
+ if (idleWorkers.contains(source)) {
+ return;
+ }
+ idleWorkers.add(source);
+
+ if (result[0] > bound) {
+ bound = result[0];
+ }
+
+ if (idleWorkers.size() == workerCount) {
+ idleWorkers.clear();
+
+ if (lastIncrease == bound) {
+ return;
+ }
+ lastIncrease = bound;
+ (new Thread(this::solve)).start();
+ }
+ }
+
  private void finishSolve(Stack<Node> solution, long millis) {
  showSolution(solution, millis);
  clearListeners();

src/main/java/ch/sebastianhaeni/pancake/processor/SolveWorker.java

@@ -36,7 +36,7 @@ private void solve() {
  stack.pop();
  } else if (stack.peek().getChildren().empty()) {
  if (stack.peek().getDepth() == 0) {
- requestWork();
+ requestWork(0);
  } else {
  stack.pop();
  }
@@ -53,7 +53,7 @@ private void solve() {
  }
 
  if (stack.isEmpty() && !status.isDone()) {
- requestWork();
+ requestWork(0);
  }
  }
 

src/main/java/ch/sebastianhaeni/pancake/processor/Worker.java

@@ -47,19 +47,19 @@ private void listenToKill() {
  (new Thread(new IntListener(Tags.KILL, (source, result) -> {
  status.done();
  MPI.COMM_WORLD.Send(EMPTY_BUFFER, 0, 0, MPI.INT, CONTROLLER_RANK, Tags.IDLE.tag());
- }, status))).start();
+ }, status, MPI.ANY_SOURCE, 1))).start();
  }
 
-
  void listenToSplit() {
  splitCommand = MPI.COMM_WORLD.Irecv(EMPTY_BUFFER, 0, 0, MPI.INT, MPI.ANY_SOURCE, Tags.SPLIT.tag());
  }
 
- void requestWork() {
- int[] boundBuf = new int[1];
+ void requestWork(int data) {
+ int[] boundBuf = new int[2];
  boundBuf[0] = candidateBound;
+ boundBuf[1] = data;
 
- MPI.COMM_WORLD.Isend(boundBuf, 0, 1, MPI.INT, CONTROLLER_RANK, Tags.IDLE.tag());
+ MPI.COMM_WORLD.Isend(boundBuf, 0, 2, MPI.INT, CONTROLLER_RANK, Tags.IDLE.tag());
  MPI.COMM_WORLD.Isend(EMPTY_BUFFER, 0, 0, MPI.INT, splitDestination, Tags.SPLIT.tag());
 
  waitForWork();

src/main/java/ch/sebastianhaeni/pancake/util/IntListener.java

@@ -7,34 +7,32 @@
 
 public class IntListener implements Runnable {
  private final Tags tag;
- private final BiConsumer<Integer, Integer> consumer;
+ private final BiConsumer<Integer, int[]> consumer;
  private final Status status;
  private final int source;
+ private final int dataCount;
 
- public IntListener(Tags tag, BiConsumer<Integer, Integer> consumer, Status status) {
- this(tag, consumer, status, MPI.ANY_SOURCE);
- }
-
- public IntListener(Tags tag, BiConsumer<Integer, Integer> consumer, Status status, int source) {
+ public IntListener(Tags tag, BiConsumer<Integer, int[]> consumer, Status status, int source, int dataCount) {
  this.tag = tag;
  this.consumer = consumer;
  this.status = status;
  this.source = source;
+ this.dataCount = dataCount;
  }
 
  @Override
  public void run() {
- int[] result = new int[1];
- mpi.Status response = MPI.COMM_WORLD.Recv(result, 0, 1, MPI.INT, source, tag.tag());
+ int[] result = new int[dataCount];
+ mpi.Status response = MPI.COMM_WORLD.Recv(result, 0, dataCount, MPI.INT, source, tag.tag());
 
  if (status.isDone()) {
  return;
  }
 
- consumer.accept(response.source, result[0]);
+ consumer.accept(response.source, result);
 
  if (!status.isDone()) {
- (new Thread(new IntListener(tag, consumer, status, source))).start();
+ (new Thread(new IntListener(tag, consumer, status, source, dataCount))).start();
  }
  }