1
\$\begingroup\$

This is my first program in akka so I wanted to know if the program is efficient and is using the advantages of actor model.


The program's purpose is to scan a given directory for any files and print the number of lines in each file.

  1. The main Application class will create the actor system and send a Scan message to a FileScanner actor.
  2. The FileScanner actor will scan the given directory, and for each file it will create a new FileParser actor and send a Parse message. Also, all the fileparser actors are passed the same Aggregator actor Reference.
  3. The FileParser actor will parse the given file, and for each line it will send a Line message to the Aggregator Actor.
  4. The Aggregator actor will maintain a count of the number of lines for each file in an instance hashmap and will print the line count for each "End" message it receives. Once all files are processed, it will shutdown the actor system.

A few points which I need clarity on:

  1. There is a separate FileParser actor for each file. Is this fine? What is the benefit of using a router which routes to FileParser actor? Will its use only help in controlling the number of fileparser actors and also how load is distributed among these actors?

  2. There is a single Aggregator actor which counts the number of lines for each file. It's using an instance HashMap and I hope this is fine. Or will a separate aggregator actor for each file improve performance?

  3. Also, I am passing the number of files to Aggregator actor while it's created so that I can shutdown the actor system once all files are processed. If I have a separate Aggregator for each file, I'm not sure how to shutdown.

  4. Each file is only sequentially processed i.e a FileParser actor is reading the file sequentially and and then invoking aggregator for each line. Is this fine or can it be improved.


Application

/**
 * The Application program bootstraps the actorsystem for parsing files in a
 * given directory and finding their linecount
 * 
 * @author 
 * @version 1.0
 */
public class Application {
    public void start(String directoryPath) {
        ActorSystem actorSystem = ActorSystem.create("logProcessor");
        ActorRef fileScanner = actorSystem.actorOf(
                Props.create(FileScanner.class), "fileScanner");
        fileScanner.tell(new Scan(directoryPath), ActorRef.noSender());
    }

    public static void main(String[] args) {

        if (args.length < 1) {
            System.out
                    .println("Usage: java -jar log-process-1.0-SNAPSHOT.jar <directorypath>");
            System.exit(0);
        }
        String path = args[0];
        Application application = new Application();
        application.start(path);
    }
}

FileScanner

/**
 * The FileScanner program scans for files in a given directory
 * 
 * @author 
 * @version 1.0
 */
public class FileScanner extends UntypedActor {

    public FileScanner() {
    }

    /**
     * Invoked by the Actor System to scan a given directory
     * 
     * @param message
     *            The message to process
     */
    public void onReceive(Object message) {
        ActorRef parser;
        if (message instanceof Scan) {
            Scan scan = (Scan) message;
            System.out.println("Scan directory: " + scan.getDirectory());

            // Only top level files in the directory are read.No recursion is
            // done
            File directory = new File(scan.getDirectory());
            // Incase of large number of files,we need to optimize below call.
            File[] files = directory.listFiles();
            // Required to shutdown actorsystem after all files are processed
            int numberOfFiles = 0;

            /*
             * To only count the files and ignore any folders
             */
            for (File file : files) {
                if (file.isFile())
                    numberOfFiles++;
            }
            ActorRef aggregator = getContext()
                    .actorOf(Props.create(Aggregator.class, numberOfFiles),
                            "aggregator");
            File file;
            for (int i = 0; i < files.length; i++) {
                file = files[i];
                if (!file.isFile())
                    continue;
                System.out.println(file.getName());

                /*
                 * Use a unique identifier(counter) for actor names as file
                 * names can have special characters(ex:readme (copy).md) and
                 * hence cannot be directly used as actor names
                 * 
                 * Docs:Actor paths MUST: not start with `$`, // include only
                 * ASCII letters and can only contain these special //
                 * characters: -_.*$+:@&=,!~';.
                 */
                parser = getContext().actorOf(
                        Props.create(FileParser.class, aggregator),
                        "parser-" + i);
                parser.tell(new Parse(file.getAbsolutePath()), getSelf());
            }
        } else {
            unhandled(message);
        }
    }
}

FileParser

/**
 * The FileScanner program scans for files in a given directory
 * 
 * @author 
 * @version 1.0
 */
public class FileParser extends UntypedActor {

    /**
     * An aggregator actor reference to send file events to.
     */
    private ActorRef aggregator;

    public FileParser(ActorRef aggregator) {
        this.aggregator = aggregator;
    }

    /**
     * Invoked by the mailbox when it receives a thread timeslice and a message
     * is available to it from FileScanner.It reads only text files and any
     * other files are not handled
     * 
     * @param message
     *            The message to process
     */
    public void onReceive(Object message) {
        if (message instanceof Parse) {
            Parse parseMessage = (Parse) message;
            System.out.println("Parse File : " + parseMessage.getFilePath());

            Path file = Paths.get(parseMessage.getFilePath());
            try {
                BufferedReader reader = Files.newBufferedReader(file,
                        StandardCharsets.UTF_8);
                String line = null;
                Line lineMessage = new Line(parseMessage.getFilePath());
                aggregator.tell(new Start(parseMessage.getFilePath()),
                        getSelf());
                while ((line = reader.readLine()) != null) {
                    aggregator.tell(lineMessage, getSelf());
                }
                aggregator.tell(new End(parseMessage.getFilePath()), getSelf());
            } catch (IOException e) {
                e.printStackTrace();
            }
        } else {
            unhandled(message);
        }
    }

}

Aggregator

/**
 * The Aggregator program counts the number of lines for each file.
 * 
 * @author 
 * @version 1.0
 */
public class Aggregator extends UntypedActor {
    /**
     * Number of files that are processed.
     */
    private int numberOfFiles;
    /**
     * Number of lines per file
     */
    private HashMap<String, Integer> lineCount;
    /**
     * A running count of processed files
     */
    private int count;

    public Aggregator(int numberOfFiles) {
        this.numberOfFiles = numberOfFiles;
        lineCount = new HashMap<>();
    }

    /**
     * Invoked by the mailbox when it receives a thread timeslice and a file
     * event(start,line or end) is available from FileParser
     * 
     * @param message
     *            The message to process
     */
    public void onReceive(Object message) {
        if (message instanceof End) {
            End end = (End) message;
            count++;// we can use bigint incase of long files.
            System.out.println("Line count of file : " + end.getFilePath()
                    + " is :" + lineCount.get(end.getFilePath()));
            if (count >= numberOfFiles) {
                getContext().system().shutdown();
            }
        } else if (message instanceof Line) {
            Line line = (Line) message;
            String path = line.getFilePath();
            if (lineCount.containsKey(path)) {
                lineCount.put(path, lineCount.get(path) + 1);
            } else {
                lineCount.put(path, 1);
            }
        } else if (message instanceof Start) {

        } else {
            unhandled(message);
        }
    }

}
\$\endgroup\$
  • \$\begingroup\$ Re routers: at your volumes (I doubt a folder will ahve more than some thouands files) not much difference. You can control the load balancing strategy of routers. Some are out of the box (round robin, lowest mailbox, etc), but you can even write your own strategy. \$\endgroup\$ – Diego Martinoia May 3 '17 at 13:36
1
\$\begingroup\$

I haven't used akka before, so I can't help much with those bits. For performance things, it is best to actually run the code and profile it. It's possible that it would be more efficient to just message Aggregator once per file after counting the lines for a specific file in FileParser. (This would also simplify the logic in Aggregator significantly.)

However, there are some Java things that can be improved.


Aggregator reports the counts to stdout. That might be fine for an example like this, but in a real application the business logic of how results are collected should not be tied up with how the results are displayed to a user. There should be a method that returns the results from Aggregator.


File.listFiles() has an overload that takes a FileFilter. This will allow you to reject non-files when the first list is collected. This saves a full iteration over all of the directory contents to get the number of files. And means that the code doesn't need to perform the same check again when getting the FileParser.


if (lineCount.containsKey(path)) {
    lineCount.put(path, lineCount.get(path) + 1);
} else {
    lineCount.put(path, 1);
}

If you are using Java8, this can be replaced with:

lineCount.merge(path, 1, (oldV, newV) -> oldV + newV);

You have to read documentation of Aggregator.count to find out what it is counting. Making the name better can resolve this.

\$\endgroup\$
  • \$\begingroup\$ Yes that's a good point.,but I had to message once per line as this was needed for the task that I had to demonstrate. \$\endgroup\$ – whokares Nov 23 '16 at 14:48

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.