File downloader using Java, multithreading and HTTP Range request

Question

I'm making a file downloader using Java. The project is on GitHub; I have included the most relevant excerpts here.

I separate the file into parts and download them using HTTP Range request, with each part being handled by a thread. All of the threads share an object, which is used to keep track of the progress and print the progress bar. The client is something like this:

public class DownloadThread implements Runnable {

    // Constructors and helper methods go here
    // ...

    public void downloadToFile(HttpURLConnection conn) throws IOException {
        //...

        while (mDownloadedSize < contentLength) {
            /*
             * Code to get the data from the input stream goes here.
             */

            // After getting the data, the thread updates the progress and
            // notifies the other threads.
            synchronized (mProgress) {
                // mProgress is the object used to update the progress
                // result is the number of bytes read from the input stream
                mProgress.downloadedCount += result;

                // ...
                mProgress.updateProgressBar();

                mProgress.notifyAll();
            }
        }
    }

    // ...
}

This is the main download class, which is used to manage the smaller download thread objects.

public class Download implements Runnable {

    // Other functions go here.
    // ...

    @Override
    public void run() {
        try {
            // Get the file name and create the URL object
            String fileName = new File(mUrl).getName();
            URL url = new URL(mUrl);

            // Check the validity of the URL
            HttpResult result = checkURLValidity(url);
            long contentSize = result.contentLength;
            int responseCode = result.responseCode;

            if (contentSize == -1 || responseCode != 200) {
                String errMessage = "Error while checking URL validity!";
                errMessage += "\nResponse code: " + responseCode;
                errMessage += "\nContent size: " + contentSize;
                throw new RuntimeException(errMessage);
            }

            // Notify the progress object of the result of the check
            synchronized (mProgress) {
                mProgress.mURLVerifyResult.contentLength = contentSize;
                mProgress.mURLVerifyResult.responseCode = responseCode;
                mProgress.notifyAll();
            }

            // Start threads to download.
            ArrayList<DownloadThread> downloadParts;

            mProgress.startDownloadTimeStamp = Instant.now();

            try {
                downloadParts = startDownloadThreads(url, contentSize,
                        mPartsCount, mProgress);
            } catch (RuntimeException ex) {
                throw ex;
            }

            // Wait for the threads to finish downloading
            for (int i = 0; i < downloadParts.size(); i++) {
                DownloadThread currentThread = downloadParts.get(i);
                currentThread.joinThread();
                if (currentThread.getDownloadedSize() != currentThread.getPartSize()) {
                    throw new RuntimeException("Download incompleted at part "
                            + (i + 1) + ": " + currentThread.getDownloadedSize());
                }
            }

            // Notify that all parts have finished downloading
            synchronized (mProgress) {
                mProgress.downloadFinished = true;
                mProgress.notifyAll();
            }

            // Join the mPartsCount together
            joinDownloadedParts(fileName, downloadParts);

            // Delete part files
            try {
                for (int i = 0; i < downloadParts.size(); i++) {
                    String partName = "." + fileName + ".part" + (i + 1);
                    Path filePath = Paths.get(partName);
                    Files.deleteIfExists(filePath);
                }
            } catch (IOException ex) {
                // If failed to delete then just ignore the exception.
                // What can we do?
            }

            // Notify that all parts have finished joining.
            synchronized (mProgress) {
                mProgress.joinPartsFinished = true;
                mProgress.notifyAll();
            }

        } catch (RuntimeException | InterruptedException | IOException ex) {
            // If an exception is thrown, put it in the progress object and
            // notify the other threads.
            synchronized (mProgress) {
                mProgress.ex = ex;
                mProgress.notifyAll();
            }
        }
    }
}

The object used to keep track of the progress (mProgress) also has a member ex, which is used to store the exception thrown by any of the thread.

This is the main class:

public class Main {

    public static String mURL;

    private static final String PROGRAM_DIR = System.getenv("HOME")
            + "/.QTDownloader";
    private static final String DOWNLOADED_LIST_FILENAME = PROGRAM_DIR
            + "/.filelist.csv";

    /**
     *
     * @param args Array of arguments.
     * @throws java.lang.InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        if (args.length == 0) {
            printUsage(args);
            System.exit(0);
        }

        // Read the arguments
        HashMap<String, String> userOptions = new HashMap<>();
        try {
            userOptions = readArgumentOptions(args);
        } catch (RuntimeException ex) {
            printErrorMessage(ex);
        }

        mURL = args[args.length - 1]; // The url is the last argument.
        int partsCount = 8;           // Number of parts to divide to download.

        // Get the list of downloaded files
        HashMap<String, DownloadSession> downloadSessionList = null;
        try {
            downloadSessionList = getListOfDownloadedFiles();
        } catch (IOException ex) {
            printErrorMessage(ex);
        }

        // Check if the file has been downloaded or not
        String fileName = new File(mURL).getName();
        DownloadSession currentDownloadSession;
        currentDownloadSession = checkIfFileWasDownloaded(downloadSessionList, fileName, mURL);

        // If the file was downloaded before.
        boolean downloaded = currentDownloadSession.alreadyDownloaded;

        // If the last attempt to download the file was interrupted and
        // the user chose to resume downloading.
        userOptions.put("resume", currentDownloadSession.resumeDownload ? "y" : "n");

        // If the user chooses to cancel downloading, exit the program
        if (currentDownloadSession.cancelDownload) {
            return;
        }

        System.out.print("\n");

        // If failed to read from the download list file
        // create a new hashmap for the download sessions list
        if (downloadSessionList == null) {
            downloadSessionList = new HashMap<>();
        }

        if (downloaded) {
            currentDownloadSession = downloadSessionList.get(mURL);
            currentDownloadSession.setDownloadSize(-1);
        } else {
            currentDownloadSession = new DownloadSession(fileName, mURL, -1);
            downloadSessionList.put(mURL, currentDownloadSession);
        }

        try {
            writeInfo(downloadSessionList);
        } catch (IOException ex) {
            printErrorMessage(ex);
        }

        // Create a Progress object to keep track of the download
        Progress progress = new Progress();

        // Start new download with the given URL
        DateFormat dateFormat = new SimpleDateFormat("dd-MM-yyyy HH:mm:ss");
        Date date = new Date();
        System.out.println("--- " + dateFormat.format(date) + " ---\n");
        System.out.println("Downloading from: " + mURL);

        Download newDownload = new Download(mURL, partsCount, progress, userOptions);

        // Start the download.
        Instant start = Instant.now();
        newDownload.startThread();

        // Verify URL
        System.out.println("Sending HTTP request...");
        synchronized (progress) {
            // Wait until verification of the URL succeeds or an exception is thrown.
            while (progress.mURLVerifyResult.responseCode == 0
                    && progress.ex == null) {
                progress.wait();
            }

            if (progress.ex == null) {
                // If no exception was thrown, URL verification succeeds.
                System.out.println("Response code: "
                        + progress.mURLVerifyResult.responseCode);
                System.out.println("Fize size: "
                        + Utility.readableFileSize(progress.mURLVerifyResult.contentLength));
            } else {
                // Else print the error message and exit.
                printErrorMessage(progress.ex);
            }
        }

        System.out.println();

        // Wait for the download to finish
        Instant downloadFinish = null;

        synchronized (progress) {
            // Wait until the download finishes or an exception is thrown.
            while (!progress.downloadFinished && progress.ex == null) {
                progress.wait();
            }

            if (progress.ex == null) {
                // If no exception was thrown. the file was downloaded successfully.
                downloadFinish = Instant.now();
                double downloadTime = ((double) (Duration.between(start,
                        downloadFinish).toMillis())) / 1000;

                System.out.println("\n\nTotal download time: " + downloadTime);
            } else {
                // Else print the error message and exit.
                printErrorMessage(progress.ex);
            }
        }

        // Wait for the parts to finish joining.
        Instant joinFinishedTime;

        synchronized (progress) {
            // Wait until all parts finish joining or an exception is thrown.
            while (!progress.joinPartsFinished && progress.ex == null) {
                progress.wait();
            }

            if (progress.ex == null) {
                // If no exception is thrown, parts joining succeeds.
                joinFinishedTime = Instant.now();
                double joinTime = ((double) (Duration.between(downloadFinish,
                        joinFinishedTime).toMillis())) / 1000;

                System.out.println("Total join time: " + joinTime);
            } else {
                // Else print the error message and exit.
                printErrorMessage(progress.ex);
            }
        }

        // Wait for the main download thread to end.
        try {
            newDownload.joinThread();
        } catch (InterruptedException ex) {
            printErrorMessage(ex);
        }

        // Save the download to the downloaded file list
        currentDownloadSession.setDownloadSize(progress.downloadedCount);
        try {
            writeInfo(downloadSessionList);
        } catch (IOException ex) {
            printErrorMessage(ex);
        }

        // Print the current time
        date = new Date();
        System.out.println("Finished downloading!");
        System.out.println("\n--- " + dateFormat.format(date) + " ---");
    }

    // Other functions go here.
}

As you can see, each time after the thread reads from the input stream (or download from the source), it updates the number of downloaded bytes count in the keeping track of the progress object, then notifies the other threads.

I suspect that this does slow down the download a little bit (just a little bit, since the download speed is still good enough compare to wget and chrome). Is this design flawed, and can it be improved in any way?

Answer 1

Naming convention, access level and exception handling

Java's naming convention do not follow the hungarian notation and thus do not include a leading m for member variable as in mProgress which therefore should probably get renamed to progress. You should also make use of encapsulation and avoid accessing fields directly:

while (progress.mURLVerifyResult.responseCode == 0
                && progress.ex == null) {

If you insist in accessing fields rather then getter/setter methods internally, then at least declare these fields as package-private instead of public:

public class Progress {
    public HttpResult mURLVerifyResult;
    public Exception ex;
    public boolean downloadFinished;
    public boolean joinPartsFinished;

    public long downloadedCount;
    public long time;
    public long sizeChange;
    public long percentageCount;

    public Instant startDownloadTimeStamp;
    ...
}

In additon to that, specifying percentageCount as long is probably a waste of memory as under normal circumstances this should never exceed 100, therefore int or short are probably enough.

Instead of ignoring a caught exception:

catch (IOException ex) {
    // If failed to delete then just ignore the exception.
    // What can we do?
}

you should at least log the exception or write it to the error stream. If you log it on warn-, debug- or trace-level is up to you, but at least log it somewhere.

Something like this however, does not make any sense:

try {
    downloadParts = startDownloadThreads(url, contentSize, mPartsCount, mProgress);
} catch (RuntimeException ex) {
    throw ex;
}

Either do something with the exception or don't catch the runtime exception at all at this place.

Don't repeat yourself (DRY) - Refactor synchronized progress logic

As the main method contains the synchronized progress logic three times, this could be refactored to something like:

private static void synchronizeProgress(Progress progress, boolean condition, ProgressHandler handler) {
    synchronized (progress) {
        // Wait until all parts finish joining or an exception is thrown.
        while (condition && progress.ex == null) {
            progress.wait();
        }

        if (progress.ex == null) {
            handler.handle(progress.mURLVerifyResult);
        } else {
            // Else print the error message and exit.
            printErrorMessage(progress.ex);
        }
    }
}

Where the ProgressHandler is a simple interface

public interface ProgressHandler {
    void handle(HttpResult verifyResult);
}

In public static void main(...) the current sychronized(progress) blocks could now be refactored to

synchronizeProgress(progress, 
                    progress.mURLVerifyResult.responseCode == 0, 
                    (verifyResult) -> {
    // If no exception was thrown, URL verification succeeds.
    System.out.println("Response code: "+ verifyResult.responseCode);
    System.out.println("Fize size: "
                       + Utility.readableFileSize(verifyResult.contentLength));
});
...
synchronizeProgress(progress,
                    !progress.downloadFinished,
                    (verifyResult) -> {
    // If no exception was thrown. the file was downloaded successfully.
    downloadFinish = Instant.now();
    double downloadTime = ((double) (Duration.between(start,
                    downloadFinish).toMillis())) / 1000;

    System.out.println("\n\nTotal download time: " + downloadTime);
});
...
synchronizeProgress(progress,
                    !progress.joinPartsFinished,
                    (verifyResult) -> {
    // If no exception is thrown, parts joining succeeds.
    Instant joinFinishedTime = Instant.now();
    double joinTime = ((double) (Duration.between(downloadFinish,
                    joinFinishedTime).toMillis())) / 1000;

    System.out.println("Total join time: " + joinTime);
});

Logical Issues

You set an synchronization point while waiting for the validation of the URI to finish. This does not make sense to me as all the download threads will use the same URI and thus need to be verified only once before passing the URI to the Download object.

You defined Download as a runnable but yet do not really make use of the thread as you have only one URL to download defined. Also, Progress is only able to handle one download URL at a time. Maybe the Progress class can therefore be managed by the Download class directly.

Consider replacing wait and notifyAll with Java concurrent utils

As you have some mProgress.wait() and mProgress.notifyAll() invocations throughout your code, which are inherited from Object, this should probably get replaced with one of the Java concurrent utility classes like Lock, Condition, CountDownLatch or CyclicBarrier.

The JavaDoc for CountDownLatch f.e. states:

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

A CountDownLatch is initialized with a given count. The await methods block until the current count reaches zero due to invocations of the countDown() method, after which all waiting threads are released and any subsequent invocations of await return immediately. This is a one-shot phenomenon -- the count cannot be reset. If you need a version that resets the count, consider using a CyclicBarrier.

As the main thread is the only class waiting for finish of download threads and the cyclic barrier is able to execute a further runnable once all or enough threads have signaled their readyness (I'll explain in a minute), I recommend therefore using a cyclic barrier. This can also be resetted and therefore reused. It is cruicial to understand that Java will continue every thread that is waiting at a barrier point after all or enough threads invoked await().

You therefore can create a barrier point in the main method like this:

public static void main(String ... args) {
    DownloadPartMerger merger = new DownloadPartMerger();
    // download threads + the main thread which is also waiting for the synch
    CyclicBarrier barrier = new CyclicBarrier(downloadThreads + 1, merger);
    ...
    Progress progress = new Progress(barrier);
    ...
    downloadParts(barrier, ...);
    ...
    Result result = merger.getResult();
}

Where the former synchronizeProgress method now equals something like:

public static void downloadParts(CyclicBarrier barrier, ...) {
    try {
        barrier.await();
        // all threads signaled their achievement of the barrier point
        handler.handle(...);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (BrokenBarrierException e) {
        e.printStackTrace();
    }
}

The Progress class has to either provide a getBarrier() method the download threads could use to signal their await() or the progress class should implement a wait() method which invokes the await() signal.

If needed, the barrier can be resetted to the initial state via invoking barrier.reset(); The cyclic barrier takes an optional runnable as second argument which is executed by the last thread that invoked the await() method.

The API documentation therefore states:

A CyclicBarrier supports an optional Runnable command that is run once per barrier point, after the last thread in the party arrives, but before any threads are released. This barrier action is useful for updating shared-state before any of the parties continue.

A barrier point is reached if enough thread signaled an barrier.await(); The optional thread is useful to merge the downloaded parts to a single downloaded object. In the sample code above this is represented by the DownloadPartMerger object which is a Runnable and should do the work which is currently done in your third synchronization point (Wait for the parts to finish joining).

Further considerations

Currently you spawn 8 download threads per file and iterate over the files sequentially and block on joining the threads in the download thread. This will logically wait for the first part to finish before the next part is waited for to finish. In reality even if you send request 1 before request 2 it might be possible for response 2 being received before response 1.

While the data is still available within the threads, you could work with Future and Callable and also with Java's Execution framework to write the data as fast as possible to the backing part files and stop the worker threads.

Other than that, you could also define a byte array which holds the returned content length initially with null values. Upon reception of a byte segment, you could copy the bytes using System.arraycopy(receivedPartBytes, 0, downloadedContent, startPos, receivedPartBytes.length); from the part-byte array to the byte array of the full result (downloadedContent in the sample) directly, though I'm not sure if invoking arraycopy is thread-safe.