ParallelGeneticMain.java

package kz.moe.classifier;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.reflect.TypeToken;
import kz.moe.classifier.genetic.creature.BasicCreature;
import kz.moe.classifier.genetic.creature.Chromosome;
import kz.moe.classifier.genetic.creature.Gene;
import kz.moe.classifier.genetic.creature.ICreature;
import kz.moe.classifier.genetic.crossover.ICrossover;
import kz.moe.classifier.genetic.crossover.OnePointCrossover;
import kz.moe.classifier.genetic.fitness.IFitness;
import kz.moe.classifier.genetic.fitness.KappaFitness;
import kz.moe.classifier.genetic.mutation.IMutation;
import kz.moe.classifier.genetic.mutation.UniformMutation;
import kz.moe.classifier.genetic.selection.ISelection;
import kz.moe.classifier.genetic.selection.RouletteSelection;
import kz.moe.classifier.utils.ClassifierHelper;
import mpi.MPI;
import org.apache.lucene.document.Document;

import javax.jms.JMSException;
import java.io.IOException;
import java.lang.reflect.Type;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;

import static kz.moe.classifier.utils.GeneticUtils.generateRandomChromosome;

/**
 * User: Sanzhar Aubakirov
 * Date: 1/21/16
 */
public class ParallelGeneticMain {
    public static final int MASTER_RANK = 0;
    public static final int MAXIMUM_MESSAGE_SIZE = 1000;
    public static final Gson GSON = new GsonBuilder().create();

    private final static List<Document> newsAndNotifications = ClassifierHelper.readDocumentsFromFile();
    private static final Type listType = new TypeToken<List<Gene>>() {
    }.getType();
    public static final String EXIT_CODE_FOR_ALL = "EXIT";
    public static int rank;

    private static ICreature maximum = null; // used only by master

    public static void main(String[] args) throws IOException, JMSException {

        MPI.Init(args);
        rank = MPI.COMM_WORLD.Rank();
        int sizeMPJ = MPI.COMM_WORLD.Size();
        final IFitness fitness = new KappaFitness();
        final ISelection selection = new RouletteSelection(sizeMPJ - 1);
        final IMutation mutation = new UniformMutation();
        final ICrossover crossover = new OnePointCrossover();


        if (rank != MASTER_RANK) {
            System.out.println("\tSlave with rank " + rank + " activated");
            doSlaveAction(fitness);
        } else {
            System.out.println("\tMaster activated");
            createFirstGeneration(sizeMPJ - 1, fitness);
            doMasterAction(sizeMPJ - 1, fitness, selection, mutation, crossover);
            MPI.Finalize();
        }

    }

    private static void doMasterAction(int generationSize, IFitness fitness, ISelection selection, IMutation mutation,
                                       ICrossover crossover) {
        boolean solutionFound = false;

        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                System.out.println("\n\t EXITING SEND TO ALL");
                for (int i = 0; i < generationSize; i++) {
                    sendMessageToSlave(i + 1, EXIT_CODE_FOR_ALL);
                }
            }
        });

        while (!solutionFound) {
            final List<ICreature> creatures = new ArrayList<>();
            for (int i = 0; i < generationSize; i++) {
                char[] tmp = new char[MAXIMUM_MESSAGE_SIZE];
                MPI.COMM_WORLD.Recv(tmp, 0, MAXIMUM_MESSAGE_SIZE, MPI.CHAR, MPI.ANY_SOURCE, 0);
                final String[] msgParts = new String(tmp).split("/");
                final String genesJSON = msgParts[0];
                final List<Gene> genes = GSON.fromJson(genesJSON, listType);
                final Double fitnessValue = Double.parseDouble(msgParts[1]);
                final ICreature newCreature = new BasicCreature(new Chromosome(genes), fitness);
                newCreature.setFitnessValue(fitnessValue);
                creatures.add(newCreature);
            }

            for (ICreature creature : creatures) {
                final Double fitnessValue = creature.getFitnessValue();
                if (fitnessValue > 0.95) {
                    final Double check = creature.calculateFitness(newsAndNotifications);
                    if (check >= 0.95) {
                        creature.getChromosome().printGenes();
                        System.out.println(creature.getResult().getReadyToPrintResult().toString());
                        for (int i = 0; i < generationSize; i++) {
                            sendMessageToSlave(i + 1, EXIT_CODE_FOR_ALL);
                        }
                        solutionFound = true;
                    } else {
                        System.out.println("Found solution not reliable");
                        maximum = null;
                    }
                }
            }

            boolean maximumChanged = false;
            for (ICreature creature : creatures) {
                if (maximum == null || maximum.getFitnessValue() < creature.getFitnessValue()) {
                    maximum = creature;
                    maximumChanged = true;
                }
            }
            if (maximumChanged && maximum.getFitnessValue() > 0.7) { // 0.7 here just to save time
                System.out.println("maximum fitness\t" + maximum.getFitnessValue());
                final Double nextFitness = maximum.calculateFitness(newsAndNotifications);
                System.out.println("next fitness\t" + nextFitness);
                System.out.println(maximum.getResult().getReadyToPrintResult().toString());
                maximum.getChromosome().printGenes();
                if (maximum.getFitnessValue() - 0.1 >= nextFitness) {
                    maximum = null;
                }
            }

            final List<ICreature> survived = selection.with(creatures, crossover);
            final List<ICreature> nextGen = new ArrayList<>();
            for (ICreature creature : survived) {
                if (new SecureRandom().nextInt(100) > 70) {
                    nextGen.add(mutation.with(creature));
                } else {
                    nextGen.add(creature);
                }
            }

            // send message to QUEUE for the next iteration
            for (int i = 0; i < nextGen.size(); i++) {
                final String genesJSON = GSON.toJson(nextGen.get(i).getChromosome().getChromosomeArray());
                sendMessageToSlave(i + 1, genesJSON + "/");
            }
        }
    }

    private static void doSlaveAction(IFitness fitness) throws JMSException {
        while (true) {
            final char[] tmp = new char[MAXIMUM_MESSAGE_SIZE];
            MPI.COMM_WORLD.Recv(tmp, 0, MAXIMUM_MESSAGE_SIZE, MPI.CHAR, MASTER_RANK, 0);
            final String genesJSON = new String(tmp).split("/")[0];
            if (genesJSON.equals(EXIT_CODE_FOR_ALL)) {
                MPI.Finalize();
                return;
            }
            final List<Gene> genes = GSON.fromJson(genesJSON, listType);
            final ICreature creature = new BasicCreature(new Chromosome(genes), fitness);
            final Random forSeed = new Random();
            final long seed = forSeed.nextLong();
            final Double calculatedFitness = creature.calculateFitness(getRandomSublistOfDocuments(seed));
            System.out.println("rank" + rank +
                                       "\tseed\t" + seed +
                                       "\tgenes\t" + genesJSON +
                                       "\tfitness\t" + calculatedFitness);
            sendCalculatedFitnessToMaster(genesJSON, calculatedFitness);
        }
    }

    /**
     * Send calculated fitness to the master
     */
    private static void sendCalculatedFitnessToMaster(String genesJSON, Double calculatedFitness) {
        final int tag = 0;
        final int offset = 0;
        final String queueMessage = genesJSON + "/" + calculatedFitness;
        MPI.COMM_WORLD.Send(queueMessage.toCharArray(), offset, queueMessage.length(), MPI.CHAR, MASTER_RANK, tag);
    }

    private static void sendMessageToSlave(int rank, String genesJSON) {
        final int offset = 0;
        MPI.COMM_WORLD.Send(genesJSON.toCharArray(), offset, genesJSON.length(), MPI.CHAR, rank, 0);
    }

    private static List<Document> getRandomSublistOfDocuments(final long seed) {
        final List<Document> copy = new ArrayList<>(newsAndNotifications);
        final Random rnd = new Random(seed);
        Collections.shuffle(copy, rnd);
        return copy.subList(0, 2000);
    }

    private static void createFirstGeneration(int generatonSize, IFitness fitness) {
        final List<ICreature> creatures = new ArrayList<>();
        for (int i = 0; i < generatonSize; i++) {
            Chromosome e = generateRandomChromosome();
            creatures.add(new BasicCreature(e, fitness));
        }
        for (int i = 0; i < creatures.size(); i++) {
            final String genesJSON = GSON.toJson(creatures.get(i).getChromosome().getChromosomeArray());
            sendMessageToSlave(i + 1, genesJSON + "/");
        }
    }

}