4
\$\begingroup\$

I have a task at hand to build a utility which

  1. Scans through a log file.

  2. Rolls over if a log file is reset.

  3. Scans through each line of the log file.

  4. Each line is sent to an executor service and checks are performed: which include looking for a particular word in the line, if a match is found I forward this line for further processing which includes splitting up the line and forming JSON.

  5. This JSON is sent across to a server using a CloseableHttpCLient with connection keep alive and ServiceUnavailableRetryStrategy patterns.

EntryPoint FileTailReader:(Started from Main)

   public class FileTailReader implements Runnable {

    private final File file;
    private long filePointer;
    private String url;
    private static volatile boolean keepLooping = true; // TODO move to main class
    private static final Logger logger = LogManager.getLogger(Main.class);
    private ExecutorService executor;
    private List<Future<?>> futures;


    public FileTailReader(File file, String url, ExecutorService executor, List<Future<?>> futures) {
        this.file = file;
        this.url = url;
        this.executor = executor;
        this.futures = futures;

    }

    private HttpPost getPost() {
        HttpPost httpPost = new HttpPost(url);
        httpPost.setHeader("Accept", "application/json");
        httpPost.setHeader("Content-type", "application/json");
        return httpPost;
    }

    @Override
    public void run() {
        long updateInterval = 100;
        try {
            ArrayList<String> batchArray = new ArrayList<>();
            HttpPost httpPost = getPost();
            CloseableHttpAsyncClient closeableHttpClient = getCloseableClient();
            Path path = Paths.get(file.toURI());
            BasicFileAttributes basicFileAttributes = Files.readAttributes(path, BasicFileAttributes.class);
            Object fileKey = basicFileAttributes.fileKey();
            String iNode = fileKey.toString();  // iNode is common during file roll
            long startTime = System.nanoTime();
            while (keepLooping) {

                Thread.sleep(updateInterval);
                long len = file.length();

                if (len < filePointer) {

                    // Log must have been rolled
                    // We can spawn a new thread here to read the remaining part of the rolled file.
                    // Compare the iNode of the file in tail with every file in the dir, if a match is found
                    // - we have the rolled file
                    // This scenario will occur only if our reader lags behind the writer - No worry

                    RolledFileReader rolledFileReader = new RolledFileReader(iNode, file, filePointer, executor,
                            closeableHttpClient, httpPost, futures);
                    new Thread(rolledFileReader).start();

                    logger.info("Log file was reset. Restarting logging from start of file.");
                    this.appendMessage("Log file was reset. Restarting logging from start of file.");
                    filePointer = len;
                } else if (len > filePointer) {
                    // File must have had something added to it!
                    RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
                    randomAccessFile.seek(filePointer);
                    FileInputStream fileInputStream = new FileInputStream(randomAccessFile.getFD());
                    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fileInputStream));
                    String bLine;
                    while ((bLine = bufferedReader.readLine()) != null) {
                        // We will use an array to hold 100 lines, so that we can batch process in a
                        // single thread
                        batchArray.add(bLine);
                        switch (batchArray.size()) {

                            case 1000:
                                appendLine((ArrayList<String>) batchArray.clone(), closeableHttpClient, httpPost);
                                batchArray.clear();
                                break;
                        }
                    }

                    if (batchArray.size() > 0) {
                        appendLine((ArrayList<String>) batchArray.clone(), closeableHttpClient, httpPost);
                    }

                    filePointer = randomAccessFile.getFilePointer();
                    randomAccessFile.close();
                    fileInputStream.close();
                    bufferedReader.close();
                   // logger.info("Total time taken: " + ((System.nanoTime() - startTime) / 1e9));

                }

                //boolean allDone = checkIfAllExecuted();
               // logger.info("isAllDone" + allDone + futures.size());

            }
            executor.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
            this.appendMessage("Fatal error reading log file, log tailing has stopped.");
        }
    }

    private void appendMessage(String line) {
        System.out.println(line.trim());
    }

    private void appendLine(ArrayList<String> batchArray, CloseableHttpAsyncClient client, HttpPost httpPost) {
        Future<?> future = executor.submit(new LocalThreadPoolExecutor(batchArray, client, httpPost));
        futures.add(future);

    }

    private boolean checkIfAllExecuted() {
        boolean allDone = true;
        for (Future<?> future : futures) {
            allDone &= future.isDone(); // check if future is done
        }
        return allDone;
    }

    //Reusable connection
    private RequestConfig getConnConfig() {
        return RequestConfig.custom()
                .setConnectionRequestTimeout(5 * 1000)
                .setConnectTimeout(5 * 1000)
                .setSocketTimeout(5 * 1000).build();
    }

    private PoolingNHttpClientConnectionManager getPoolingConnManager() throws IOReactorException {
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
        PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);
        cm.setMaxTotal(1000);
        cm.setDefaultMaxPerRoute(1000);

        return cm;
    }

    private CloseableHttpAsyncClient getCloseableClient() throws IOReactorException {
        CloseableHttpAsyncClient httpAsyncClient = HttpAsyncClientBuilder.create()
                .setDefaultRequestConfig(getConnConfig())
                .setConnectionManager(getPoolingConnManager()).build();

        httpAsyncClient.start();

        return httpAsyncClient;


                /*.setServiceUnavailableRetryStrategy(new ServiceUnavailableRetryStrategy() {
                    @Override
                    public boolean retryRequest(
                            final HttpResponse response, final int executionCount, final HttpContext context) {
                        int statusCode = response.getStatusLine().getStatusCode();
                        return statusCode != HttpURLConnection.HTTP_OK && executionCount < 5;
                    }

                    @Override
                    public long getRetryInterval() {
                        return 0;
                    }
                }).build();*/
    }


}

I am using an implementation of Rabin Karp for string find:

public class RabinKarp {
    private final String pat;      // the pattern  // needed only for Las Vegas
    private long patHash;    // pattern hash value
    private int m;           // pattern length
    private long q;          // a large prime, small enough to avoid long overflow
    private final int R;           // radix
    private long RM;         // R^(M-1) % Q

    /**
     * Preprocesses the pattern string.
     *
     * @param pattern the pattern string
     * @param R       the alphabet size
     */
    public RabinKarp(char[] pattern, int R) {
        this.pat = String.valueOf(pattern);
        this.R = R;
        throw new UnsupportedOperationException("Operation not supported yet");
    }

    /**
     * Preprocesses the pattern string.
     *
     * @param pat the pattern string
     */
    public RabinKarp(String pat) {
        this.pat = pat;      // save pattern (needed only for Las Vegas)
        R = 256;
        m = pat.length();
        q = longRandomPrime();

        // precompute R^(m-1) % q for use in removing leading digit
        RM = 1;
        for (int i = 1; i <= m - 1; i++)
            RM = (R * RM) % q;
        patHash = hash(pat, m);
    }

    // Compute hash for key[0..m-1].
    private long hash(String key, int m) {
        long h = 0;
        for (int j = 0; j < m; j++)
            h = (R * h + key.charAt(j)) % q;
        return h;
    }

    // Las Vegas version: does pat[] match txt[i..i-m+1] ?
    private boolean check(String txt, int i) {
        for (int j = 0; j < m; j++)
            if (pat.charAt(j) != txt.charAt(i + j))
                return false;
        return true;
    }

    // Monte Carlo version: always return true
    // private boolean check(int i) {
    //    return true;
    //}

    /**
     * Returns the index of the first occurrrence of the pattern string
     * in the text string.
     *
     * @param txt the text string
     * @return the index of the first occurrence of the pattern string
     * in the text string; n if no such match
     */
    public int search(String txt) {
        int n = txt.length();
        if (n < m) return n;
        long txtHash = hash(txt, m);

        // check for match at offset 0
        if ((patHash == txtHash) && check(txt, 0))
            return 0;

        // check for hash match; if hash match, check for exact match
        for (int i = m; i < n; i++) {
            // Remove leading digit, add trailing digit, check for match.
            txtHash = (txtHash + q - RM * txt.charAt(i - m) % q) % q;
            txtHash = (txtHash * R + txt.charAt(i)) % q;

            // match
            int offset = i - m + 1;
            if ((patHash == txtHash) && check(txt, offset))
                return offset;
        }

        // no match
        return -1;
    }


    // a random 31-bit prime
    private static long longRandomPrime() {
        BigInteger prime = BigInteger.probablePrime(31, new Random());
        return prime.longValue();
    }
}

Here is my RolledFileReader

public class RolledFileReader implements Runnable {

    private static final Logger logger = LogManager.getLogger(RolledFileReader.class);

    private String iNode;
    private File tailedFile;
    private long filePointer;
    private ExecutorService executor;
    private CloseableHttpAsyncClient client;
    private HttpPost httpPost;
    List<Future<?>> futures;

    public RolledFileReader(String iNode, File tailedFile, long filePointer, ExecutorService executor,
                            CloseableHttpAsyncClient client, HttpPost httpPost, List<Future<?>> futures) {
        this.iNode = iNode;
        this.tailedFile = tailedFile;
        this.filePointer = filePointer;
        this.executor = executor;
        this.client = client;
        this.httpPost = httpPost;
        this.futures = futures;
    }

    @Override
    public void run() {
        try {
            inodeReader();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public void inodeReader() throws Exception {
        String fParent = tailedFile.getParentFile().toString();
        File[] files = new File(fParent).listFiles();
        if (files != null) {
            Arrays.sort(files, Collections.reverseOrder()); // Probability of finding the file at top increases
            for (File file : files) {
                if (file.isFile()) {
                    Path path = Paths.get(file.toURI());
                    BasicFileAttributes basicFileAttributes = Files.readAttributes(path, BasicFileAttributes.class);
                    Object fileKey = basicFileAttributes.fileKey();
                    String matchInode = fileKey.toString();
                    if (matchInode.equalsIgnoreCase(iNode) && file.length() > filePointer) {
                        //We found a match - now process the remaining file - we are in a separate thread
                        readRolledFile(file, filePointer);

                    }
                }
            }

        }
    }


    public void readRolledFile(File rolledFile, long filePointer) throws Exception {
        ArrayList<String> batchArray = new ArrayList<>();
        RandomAccessFile randomAccessFile = new RandomAccessFile(rolledFile, "r");
        randomAccessFile.seek(filePointer);
        FileInputStream fileInputStream = new FileInputStream(randomAccessFile.getFD());
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fileInputStream));
        String bLine;
        while ((bLine = bufferedReader.readLine()) != null) {

            batchArray.add(bLine);
            switch (batchArray.size()) {
                case 1000:
                    executor.execute(new LocalThreadPoolExecutor((ArrayList<String>) batchArray.clone(), client, httpPost));
            }
        }

        if (batchArray.size() > 0) {
            executor.execute(new LocalThreadPoolExecutor((ArrayList<String>) batchArray.clone(), client, httpPost));
        }
    }


}

And my executor service LocalThreadPoolExecutor:

   public class LocalThreadPoolExecutor implements Runnable {
    private static final Logger logger = LogManager.getLogger(Main.class);

    private final ArrayList<String> payload;
    private final CloseableHttpAsyncClient client;
    private final HttpPost httpPost;
    private HttpContext context;
    private final RabinKarp searcher = new RabinKarp("JioEvents");

    public LocalThreadPoolExecutor(ArrayList<String> payload, CloseableHttpAsyncClient client,
                                   HttpPost httpPost) {
        this.payload = payload;
        this.client = client;
        this.httpPost = httpPost;
    }

    @Override
    public void run() {
        try {
            for (String line : payload) {
                int offset = searcher.search(line);
                switch (offset) {
                    case -1:
                        break;
                    default:
                        String zeroIn = line.substring(offset).toLowerCase();
                        String postPayload = processLogs(zeroIn);
                        if (null != postPayload) {
                            postData(postPayload, client, httpPost);
                        }
                }
            }
       // logger.info("Processed a batch of: "+payload.size());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private String processLogs(String line) {
        String[] jsonElements = line.split("\\|");
        switch (jsonElements.length) {
            case 15:
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("customerID", jsonElements[1]);
                jsonObject.put("mobileNumber", jsonElements[2]);
                jsonObject.put("eventID", jsonElements[3]);
                jsonObject.put("eventType", jsonElements[4]);
                jsonObject.put("eventDateTime", jsonElements[5]);
                jsonObject.put("eventResponseCode", jsonElements[6]);
                jsonObject.put("sourceSystem", jsonElements[7]);
                jsonObject.put("clientID", jsonElements[8]);
                jsonObject.put("serverHostName", jsonElements[9]);
                jsonObject.put("serverIPAddress", jsonElements[10]);
                jsonObject.put("serverSessionID", jsonElements[11]);
                jsonObject.put("softwareVersion", jsonElements[12]);
                jsonObject.put("deviceInfo", jsonElements[13]);
                jsonObject.put("userAgent", jsonElements[14]);
                return jsonObject.toString();
        }
        return null;
    }

    private void postData(String data, CloseableHttpAsyncClient client, HttpPost httpPost) throws Exception {

        StringEntity entity = new StringEntity(data);
        httpPost.setEntity(entity);
        Future<HttpResponse> future = client.execute(httpPost, context, null);
     //   HttpResponse response = future.get();
     //   logger.info("Resp is: "+response.getStatusLine().getStatusCode());

    }

}

And finally the Main class:

public class Main {
    private static final Logger logger = LogManager.getLogger(Main.class);
    private static final ExecutorService executor = Executors.newFixedThreadPool(25);
    private static final List<Future<?>> futures = new ArrayList<>();

    private static void usage() {
        System.out.println("Invalid usage");
    }

    public static void main(String[] args) {

        if (args.length < 2) {
            usage();
            System.exit(0);
        }
        String url = args[0];
        String fPath = args[1];

        File log = new File(fPath);
        FileTailReader fileTailReader = new FileTailReader(log, url, executor, futures);

        new Thread(fileTailReader).start(); // Can issue multiple threads with an executor like so, for multiple files


    }

}

The purpose of declaring member variables in Main is that I can later on add ShutdownHooks.

I am interested in knowing how I can make this code faster. Right now I am getting a throughput of 300000 lines per 8876 millis. Which is not going well with my peers.

Edit:

I changed the way RandomAccessFile is reading from the file and I have observed a considerable increase in speed, however I am still looking for fresh pointers to enhance and optimize this utility:

else if (len > filePointer) {
                    // File must have had something added to it!
                    long startTime = System.nanoTime();
                    RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
                    randomAccessFile.seek(filePointer);
                    FileInputStream fileInputStream = new FileInputStream(randomAccessFile.getFD());
                    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fileInputStream));
                    String bLine;
                    logger.info("Pointer: "+filePointer+" fileLength: "+len);
                    while ((bLine = bufferedReader.readLine()) != null) {
                        this.appendLine(bLine, httpclient, httpPost);
                    }
                    logger.info("Total time taken: " + ((System.nanoTime() - startTime) / 1e9));
                    filePointer = randomAccessFile.getFilePointer();
                    logger.info("FilePointer reset to: "+filePointer);
                    randomAccessFile.close();
                    fileInputStream.close();
                    bufferedReader.close();
                }

I also added a bit of batch processing in the above snippet (Code from FileTailReader is edited to demonstrate the same in particular addition of batchArray which is a list) - I see an improvement of 10 seconds. Now the program executes in 21 point some milli seconds.

\$\endgroup\$
3
+100
\$\begingroup\$

To be honest, the code is way too chaotic and very hard to understand. Main object oriented principles are not followed, a lot of bad naming, passing around of variables everywhere. Just following the code flow is very hard.

The main concerns:

SRP

The SRP: Single Responsibility Principle. Your FileTailReader does like everything. It's something like FileTailReader-Runnable-HttpPostCreating-ThreadManaging-HttpConnectionPoolManaging-Thread-Thing. Your types should do one thing.

I'd suggest you first start to refactor all the responsibilities into separate types. I mean, you just provided us a list of those responsilities.

This, and the passing around of the variables is hurting the most (e.g. the List<Future<?>> futures, which is passed from the Main type to everywhere. I do not accept the "Shutdown hook" argument.

Thread management

When I see Threads, I'm usually very suspicious - it's a very icky source of bugs. And when I see Threads starting other Threads, which then start other Threads again... well, actually, I have never seen that. If a bug occurs: Good luck, or better: good night.

First of all, I don't see any reason, why FileTailReader should be a Thread? Shouldn't it be possible to start the app twice with a different File? As long as you don't need to tail dozens of files, I wouldn't implement this type as Runnable.

Why does RolledFileReader need to be a Thread? If the file is "rolled over", the actual thread is done anyway? Even if not, e.g. the new file has to be processed, too, I'd just "read the rest".

Performance

A priori: If you want to improve "performance" (whatever that exactly is!), you have to measure.

HDD

Quote: 300000 lines per 8876

My test file is 100MB's with 1'024'000 lines (100 chars length). I read it completely in 186ms with a BufferedReader and 8192 buffer size. To count the lines, using a LineNumberReader, it takes 281ms. And that's on a ~3 year old fair SSD. I wouldn't care about read speed from the disk, so that's why I would get rid of all the Threads you are using for accessing the file system. As long as you do not read with some stupid buffer sizes, reading will always be faster than writing anyway. I'm also fairly certain, that the removal of those Threads will improve the throughput, because thread management is not free. In general, I usually say: Adding more code and complexity will reduce performance - which of course, is not entirely true, but I try to avoid doing performance-improvements, as long as a certain part of the code is not a problem (which reading from a file is not).

Logging

Configure your logging correspondingly. Check your Logging API's performance guide.

RabinKarp

When I call the search method 10'000 times, I got ~9 millis, using indexOf, I got ~1ms. Again: This type has added complexity, without being a problem and without measurement.

HTTP

Your main bottleneck will be the http processing. If you can avoid it, avoid it. If you can reduce the amount of requests (called remote calls in general), reduce it. It - mainly the latency and the performance of the backend system - is usually the only component, which you can not influence.

Reduce amount of calls

... if possible. In your case: Try not to send one json string per line, try to send as much as you can.

Pool

Use a (Http)ConnectionPool - which you do. Web Servers usually have different configurations about their http pool (or "Connectors"): The amount of threads, which will process requests, the amount of open connections (which will wait until the thread pool has a thread free) and often the amount of http connections open per client ip. This certainly has to match the configuration of your pool. If your app opens too many http connections, they will just wait, memory consumption will rise, memory management will take cpu time, you kill the backend server (yay). I see, your connection pool has a maximum of 1000 connections. That's a lot. Our company website has about 1000 in total, for several different servers, the max connections for our productive server are usually around 100 or 150.

Async of HTTP calls

Since the http requests will be the main bottleneck, I'm not sure if the asynchronous http calls will improve the performance. The asynchronous call will be executed in a separate Thread again, which takes ressources, but your main app will have to wait for the backend anyway, right? Again: Measure.

GZIP

Gzipping may take some cpu time, but transfering bytes is often way more expensive than gzipping Strings. Of course, this has to be measured, too.

JSON

There's different implementations (I can't see which one you are using) which of course do perform differently, too.

Memory / GC

I won't go into much detail, but if the concern is performance, you should look into garbage collection. There's a few common rules to follow, as using fixed heap sizes, or in your case, using a througput collector instead of a low latency collector.

Hope this helps,

slowy

\$\endgroup\$
  • \$\begingroup\$ I take the SRP part, I have created a separate class and delegated the connection creation there. However when you say, that objects should not be passed along, like now I am initing them in the main class. You do not suggest how to share these objects between classes without a compromise on speed. \$\endgroup\$ – User3 Jun 1 '17 at 12:58
  • \$\begingroup\$ I did suggest to remove all Threading, except for the HTTP calls, so this question is obsolete. And even if it were a performance problem - and I'm fairly certain, it is not: You're hunting nanos. While reading from a file is millis, so is doing http requests. Even if you improve the speed up to 99%, it's still ~0% of the overall performance - that's not worth it, from an economic point of view. If you really want improve the performance, change the architecture (do the http server stuff locally). Or buy better hardware. It's cheaper than maintaining non maintainable code. \$\endgroup\$ – slowy Jun 1 '17 at 16:11
2
\$\begingroup\$

Ok, so my 5c is that just looking at the code, I can clearly see that you should rethink what each class should do and re-module the whole solution again. For example, the http communication seems scattered across both FileTailReader and RolledFileReader. After reading the aim of your task, it seems that you have a typical IN - process - OUT scenario. In this scenario all parts should be as much independent as can be, communicating between themselves through interfaces. Such decoupling would at least allow you to test each module separately. In your implementation, I think, that each IN contains some part of OUT which. Another thing - you've added some async behavior to your *Reader classes. While I'm not sure what they actually do (sorry tl;dr) you might benefit from decoupling the file parsing logic from async behaviour. Again, it might at least, allow you to easily run tests. Next thing, you create some object within classes with new(), maybe you should create them externally and pass as a constructor parameter, in other words, do dependency injection. You don't need to use any fancy framework, just create all the components in Main() in proper order and start job. If it makes sense to define an interface for any of the injected classes, consider doing it. Same goes for all hardcoded parameters - put them in a *Config class and just set as default values.

\$\endgroup\$
  • \$\begingroup\$ Thanks for your valuable advise, I have improved my code based on what you mentioned \$\endgroup\$ – User3 Jun 7 '17 at 8:18

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.