Skip to main content
Added clarification on the changes needed in `Progress` and/or the download threads.
Source Link

As you have some mProgress.wait() and plenty of 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.

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().

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.

As you have some mProgress.wait() and plenty of 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.

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 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().

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.

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().

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.

Source Link

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 plenty of 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 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();
    }
}

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.