Frohe Weihnachten PKS WiSe 16/17 - Hier der einzige Lösungsvorschlag seit 6 Jahren AJP und PKS

ForkJoinHistogramService.java

package de.uniba.wiai.dsg.pks.assignment1.histogram.threaded.forkjoin;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

import de.uniba.wiai.dsg.pks.assignment1.histogram.Histogram;
import de.uniba.wiai.dsg.pks.assignment1.histogram.HistogramService;
import de.uniba.wiai.dsg.pks.assignment1.histogram.HistogramServiceException;
import net.jcip.annotations.Immutable;
import net.jcip.annotations.ThreadSafe;

@Immutable
public class ForkJoinHistogramService implements HistogramService {

    @Immutable
    private static final class ImmutableHistogram {

        public static final int ALPHABET_SIZE = 26;

        public static final ImmutableHistogram EMPTY = new ImmutableHistogram(0, 0, 0, new long[ALPHABET_SIZE]);
        public static final ImmutableHistogram EMPTY_DIRECTORY = new ImmutableHistogram(0, 0, 1, new long[ALPHABET_SIZE]);
        public static final ImmutableHistogram EMPTY_FILE = new ImmutableHistogram(0, 1, 0, new long[ALPHABET_SIZE]);

        public static ImmutableHistogram fromFile(Path file) {
            try {
                List<String> lines = Files.readAllLines(file, StandardCharsets.UTF_8);
                return new ImmutableHistogram(lines.size(), 1, 0, linesToDistribution(lines));
            } catch (IOException e) {
                // in case of an error, we simply measure only that a file was found
                return EMPTY_FILE;
            }
        }

        private static long[] linesToDistribution(List<String> lines) {
            long[] distribution = new long[ALPHABET_SIZE];
            for (String line : lines) {
                for (char c : line.toCharArray()) {
                    if (c >= 'a' && c <= 'z') {
                        distribution[c - 'a']++;
                    } else if (c >= 'A' && c <= 'Z') {
                        distribution[c - 'A']++;
                    }
                }
            }
            return distribution;
        }

        private final long[] distribution = new long[ALPHABET_SIZE];
        private final long lines;
        private final long files;
        private final long directories;

        private ImmutableHistogram(long lines, long files, long directories, long[] distribution) {
            this.lines = lines;
            this.files = files;
            this.directories = directories;
            System.arraycopy(Objects.requireNonNull(distribution), 0, this.distribution, 0, ALPHABET_SIZE);
        }

        public long[] getDistribution() {
            long[] result = new long[ALPHABET_SIZE];
            System.arraycopy(distribution, 0, result, 0, ALPHABET_SIZE);
            return result;
        }

        public long getLines() {
            return lines;
        }

        public long getFiles() {
            return files;
        }

        public long getDirectories() {
            return directories;
        }

        public String toString() {
            return String.format("Histogram with %d directories and %d files that contain %d lines", directories, files, lines);
        }

        public ImmutableHistogram add(ImmutableHistogram otherImmutableHistogram) {
            long[] newDistribution = new long[ALPHABET_SIZE];
            for (int i = 0; i < ALPHABET_SIZE; i++) {
                newDistribution[i] = this.distribution[i] + otherImmutableHistogram.distribution[i];
            }
            return new ImmutableHistogram(this.lines + otherImmutableHistogram.lines,
                    this.files + otherImmutableHistogram.files,
                    this.directories + otherImmutableHistogram.directories,
                    newDistribution);
        }

        public Histogram toHistogram() {
            // note: the int variables in the constructor of Histogram had to be changed to long for that to work
            return new Histogram(getDistribution(), lines, files, directories);
        }
    }

    @ThreadSafe
    private static final class ConsoleOutputConsumer implements Runnable {

        public static final String POISON_PILL = "DONE";

        private final BlockingQueue<String> blockingQueue;

        private ConsoleOutputConsumer(BlockingQueue<String> blockingQueue) {
            this.blockingQueue = Objects.requireNonNull(blockingQueue);
        }

        @Override
        public void run() {
            while (true) {
                final String element = takeWithGuardedWait();
                if (POISON_PILL.equals(element)) {
                    return;
                }
                System.out.println(element);
            }
        }

        private String takeWithGuardedWait() {
            while (true) {
                try {
                    return blockingQueue.take();
                } catch (InterruptedException ignored) {
                }
            }
        }
    }

    @ThreadSafe
    private static final class FileTreeHistogramForkJoinTask extends RecursiveTask<ImmutableHistogram> {

        private final Path currentPath;
        private final FileExtension fileExtension;
        private final BlockingQueue<String> blockingQueue;

        private FileTreeHistogramForkJoinTask(Path currentPath, FileExtension fileExtension, BlockingQueue<String> blockingQueue) {
            this.currentPath = Objects.requireNonNull(currentPath);
            this.fileExtension = Objects.requireNonNull(fileExtension);
            this.blockingQueue = Objects.requireNonNull(blockingQueue);
        }

        @Override
        protected ImmutableHistogram compute() {
            // base cases: files
            // recursive case: directories

            // base case
            if (fileExtension.isFileWithExtension(currentPath)) {
                return ImmutableHistogram.fromFile(currentPath);
            } else if (Files.isRegularFile(currentPath)) {
                return ImmutableHistogram.EMPTY;
            }

            // split and fork
            List<FileTreeHistogramForkJoinTask> tasks = new ArrayList<>();
            try (final DirectoryStream<Path> paths = Files.newDirectoryStream(currentPath)) {
                for (Path path : paths) {
                    if (Files.isDirectory(path) || fileExtension.isFileWithExtension(path)) {
                        final FileTreeHistogramForkJoinTask task = new FileTreeHistogramForkJoinTask(path, fileExtension, blockingQueue);
                        task.fork();
                        tasks.add(task);
                    }
                }
            } catch (IOException e) {
                // we explicitly return upon failure with an empty result instead of an exception
                return print(ImmutableHistogram.EMPTY_DIRECTORY);
            }

            // join and aggregate
            ImmutableHistogram result = ImmutableHistogram.EMPTY_DIRECTORY; // count root folder as well
            for (FileTreeHistogramForkJoinTask task : tasks) {
                result = result.add(task.join());
            }

            return print(result);
        }

        private ImmutableHistogram print(ImmutableHistogram result) {
            final String output = String.join(" ", currentPath.toAbsolutePath().toString(), result.toString());
            while (true) {
                try {
                    this.blockingQueue.put(output);
                    break;
                } catch (InterruptedException ignored) {
                }
            }
            return result;
        }
    }

    @Immutable
    private static class FileExtension {

        private final String fileExtension;

        private FileExtension(String fileExtension) {
            this.fileExtension = Objects.requireNonNull(fileExtension);
        }

        public boolean isFileWithExtension(Path path) { // can also be used as a predicate
            return Files.isRegularFile(path) && path.toString().endsWith(fileExtension);
        }
    }

    public ForkJoinHistogramService() {
        // REQUIRED FOR GRADING - DO NOT REMOVE DEFAULT CONSTRUCTOR
        // but you can add code below
    }

    @Override
    public Histogram calculateHistogram(String rootDirectory, String fileExtension)
            throws HistogramServiceException {
        Objects.requireNonNull(rootDirectory);
        Objects.requireNonNull(fileExtension);

        if(fileExtension.isEmpty()) {
            throw new HistogramServiceException("fileExtension is empty");
        }

        if(!Files.isDirectory(Paths.get(rootDirectory))) {
            throw new HistogramServiceException("rootDirectory is no directory");
        }

        ExecutorService executorService = Executors.newSingleThreadExecutor();
        try {
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(1000);
            executorService.execute(new ConsoleOutputConsumer(blockingQueue));
            try {
                ForkJoinPool forkJoinPool = new ForkJoinPool();
                try {
                    final Path currentPath = Paths.get(rootDirectory);
                    final FileTreeHistogramForkJoinTask task = new FileTreeHistogramForkJoinTask(currentPath, new FileExtension(fileExtension), blockingQueue);
                    return forkJoinPool.invoke(task).toHistogram();
                } finally {
                    forkJoinPool.shutdown(); // we explicitly do not wait to until it is terminated so that the code is faster
                }
            } finally {
                while (true) { // send poison pill through guarded wait
                    try {
                        blockingQueue.put(ConsoleOutputConsumer.POISON_PILL);
                        break;
                    } catch (InterruptedException e) {
                    }
                }
            }
        } finally {
            executorService.shutdown(); // we explicitly do not wait to until it is terminated so that the code is faster
        }
    }

    @Override
    public String toString() {
        return "ForkJoinHistogramService";
    }
}