How thread lock should be handled to keep other threads waiting until downloading a file and then allow all threads to read the file in one go

I am using an ExecutorService fixedThreadPool() to run a TASK.

A TASK here is defined as downloading a file from a specific URL and saving it to the database if it doesn't exist or else read the file from the database only. So it's more like a reader-writer problem where any of the thread of executor thread pool can act as a writer for once and others will be a reader for the subsequent request.

I am using Semaphore to perform this but the issue with this approach is subsequent read requests are happening sequentially.

If 4 TASKs are intended to hit the same URL I needed the synchronization till the file is downloaded and the semaphore is released i.e. out of 4 threads anyone can acquire the lock and rest 3 are waiting. After the download completes all the remaining 3 threads should simultaneously read the downloaded file. But this last step is happening sequentially in my case which will have an impact on project performance as well.

Having said the above use case, the following is my sample code:

Following Runnable is passed to ExecutorService to execute the task on the SharedObject class.

 class DownloadRunnable(SharedObjectMT sharedObject, String url) implement Runnable {
    void run() {
        sharedObject.loadFile(url);
    }
 }

class SharedObjectMT {
    // This Hashmap acts ConcurrentHashMap with URL and semaphore mapping. So
        // multiple threads requesting for the same URL will only be synchronized on their
        // corresponding semaphore. And threads requesting for different URLs 
        // will run concurrently.

    private static HashMap<String, Semaphore> syncMap = new HashMap<>();
    .....
    void loadFile(String url) {
        
        // Let all threads enter sequentially and try to assign a Semaphore for their url in the 
        // hashmap. If the url has never been requested, then only a Semaphore(say S1) will be 
        // assigned to that url. And for all the other threads with *same request url*, this 
        // Semaphore(S1) will be used to handle concurrency.

        synchronized(syncMap) {
             if(syncMap
 == null) {
                syncMap
 = new Semaphore(1);
            }
        }
        
        Semaphore semaphore = syncMap
;

        synchronized(semaphore) {
            ............
            ............
            semaphore.acquire();
            String filePath = findInDatabase(url);
            if(filePath != null) {
                semaphore.release(); // no need to hold semaphore since file already downloaded.
                printStatus("Already downloaded file from url = "+url);
            } else {
                // This DownloadThread is actually a mock of my real project where a third-party 
                // library uses a thread to download the file.

                DownloadThread(() -> {
                    printStatus("Download completed for url= "+ url +". Releasing semaphore.");
                    semaphore.release();
                }).start();
                .............
                .............
            }
        }
    }
}

I know that a single Semaphore can't help me out. Maybe we can use 1 more Semaphore to distinguish between read and write lock or any other locking mechanism. So need some help on what to use for this type of one-time synchronization.

>Note: Please ignore if you find any syntax error in the above code since the actual project is in Kotlin but this is a basic Java multithreading problem so I posted it as a Java code.

I am not sure about Kotlin, but I can demonstrate in Java:

import java.io.IOException;
import java.util.HashMap;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class DownloadOrRead {
        
    //Utility method, which just generates a random String instance...
    private static String randomString(final int length) {
        String alphabet = "abcdefghijklmnopqrstuvwxyz";
        alphabet += alphabet.toUpperCase();
        alphabet += "0123456789";
        final int alphabetSize = alphabet.length();
        final char[] chars = new char[length];
        final Random rand = new Random();
        for (int i = 0; i < chars.length; ++i)
            chars[i] = alphabet.charAt(rand.nextInt(alphabetSize));
        return String.valueOf(chars);
    }
    
    public static class DownLoadCallable implements Callable<String> {
        private final String url;
        
        public DownLoadCallable(final String url) {
            this.url = Objects.requireNonNull(url);
        }
        
        @Override
        public String call() throws IOException, InterruptedException {
            
            /*Utilize url property here to download the file...
            In our case, just simulate a download delay supposedly...*/
            Thread.sleep(5000L + (long) (Math.random() * 10000L));
            
            //Return the file's local path...
            return randomString(20); //In our case, a random String of 20 characters.
        }
    }
    
    //This is the method you are looking for:
    public static String loadPath(final ExecutorService executorService, //Can be shared between calls of loadPath...
                                  final HashMap<String, Future<String>> urlToFuture, //MUST be shared between calls of loadPath!
                                  final String url) //The URL. Can be the same as a URL in a previous call of loadPath.
            throws InterruptedException, ExecutionException {
        final Future<String> future;
        synchronized (urlToFuture) {
            if (!urlToFuture.containsKey(url)) //If nowhere to be seen...
                urlToFuture.put(url, executorService.submit(new DownLoadCallable(url))); //Create a Future...
            future = urlToFuture.get(url); //Obtain the Future (new or old).
        }
        return future.get(); //Outside the synchronized block!
    }
    
    public static void main(final String[] args) {
        
        System.out.println("Creating ExecutorService...");
        final ExecutorService executorService = Executors.newFixedThreadPool(10);
        
        System.out.println("Creating shared map...");
        final HashMap<String, Future<String>> urlToFuture = new HashMap<>();
        
        System.out.println("Creating random URLs...");
        final String[] urls = new String[]{randomString(10), randomString(20), randomString(15)};
        
        try {
            System.out.println("Downloading files sequencially...");
            final Random rand = new Random();
            for (int i = 0; i < 100; ++i)
                System.out.println(loadPath(executorService, urlToFuture, urls[rand.nextInt(urls.length)]));
            
            executorService.shutdown();
            executorService.awaitTermination(10, TimeUnit.MINUTES);
        }
        catch (final InterruptedException | ExecutionException x) {
            System.err.println(x);
        }
    }
}

The whole idea is to submit Callables to the ExecutorService which handle the downloading. We also utilize the Futures returned by the submit method to get the desired result path/file/anything. Just call get on the desired Future object and there you are. The only thing you need to synchronize on is the Map of URLs to Futures.

You will notice, when running this test program, that the first file is blocking until downloaded, then subsequent calls for the same URL are finished immediately (because the URL is already downloaded) and we only block for each new URL (which is not downloaded yet). In this case I am using only 3 random URLs with each URL needing 5 to 15 seconds to complete, which gives us about 15 to 45 seconds uptime, because we download them sequencially.

That concludes the loadPath method. But in the sample code above, the files are downloaded sequencially. If you need multiple Threads also for downloading you may call loadPath from many Threads (without the need of further synchronization somewhere else than the shared Map).

As one can read in this answer here, seems like invoking the get method of the same Future after the operation is complete, will always yield the same object or throw the same Exception if failed. This is something we use to our advantage on this provided code in this post.

Edit 1:

Or even better, as pointed out by @drekbour in the comments, make use of computeIfAbsent and a ConcurrentHashMap for the job, like so:

import java.util.Objects;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class DownloadOrRead1 {
    
    //Utility method, which just generates a random String instance...
    private static String randomString(final int length) {
        String alphabet = "abcdefghijklmnopqrstuvwxyz";
        alphabet += alphabet.toUpperCase();
        alphabet += "0123456789";
        final int alphabetSize = alphabet.length();
        final char[] chars = new char[length];
        final Random rand = new Random();
        for (int i = 0; i < chars.length; ++i)
            chars[i] = alphabet.charAt(rand.nextInt(alphabetSize));
        return String.valueOf(chars);
    }
    
    public static class DownLoadCallable implements Callable<String> {
        private final String url;
        
        public DownLoadCallable(final String url) {
            this.url = Objects.requireNonNull(url);
        }
        
        @Override
        public String call() throws InterruptedException {
            
            System.out.println("Downloading " + url + "...");
            
            /*Utilize url property here to download the file...
            In our case, just simulate a download delay supposedly...*/
            Thread.sleep(5000L + (long) (Math.random() * 10000L));
            
            System.out.println("Downloaded " + url + '.');
            
            //Return the file's local path...
            return randomString(20); //In our case, a random String of 20 characters.
        }
    }
    
    //This is the method you are looking for:
    public static String loadPath(final ExecutorService executorService, //Can be shared between calls of loadPath...
                                  final ConcurrentHashMap<String, Future<String>> urlToFuture, //MUST be shared between calls of loadPath!
                                  final String url) //The URL. Can be the same as a URL in a previous call of loadPath.
            throws InterruptedException, ExecutionException {
        return urlToFuture.computeIfAbsent(url, url2 -> executorService.submit(new DownLoadCallable(url2))).get();
    }
    
    public static void main(final String[] args) {
        
        System.out.println("Creating ExecutorService...");
        final ExecutorService executorService = Executors.newFixedThreadPool(10);
        
        System.out.println("Creating shared Map...");
        final ConcurrentHashMap<String, Future<String>> urlToFuture = new ConcurrentHashMap<>();
        
        System.out.println("Creating random URLs...");
        final String[] urls = new String[]{randomString(10), randomString(10), randomString(10)};
        
        try {
            System.out.println("Downloading files sequencially...");
            final Random rand = new Random();
            for (int i = 0; i < 100; ++i) {
                final String url = urls[rand.nextInt(urls.length)];
                System.out.println("Path for " + url + ": " + loadPath(executorService, urlToFuture, url));
            }
            
            executorService.shutdown();
            executorService.awaitTermination(10, TimeUnit.MINUTES);
        }
        catch (final InterruptedException | ExecutionException x) {
            System.err.println(x);
        }
    }
}