Design an image downloading library.

package com.androidinterview.imagedownloader;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;

// Files on disk, keyed by URL, evicted least recently used first. This is the
// whole persistence story for an image downloader, because the thing being
// cached is already a file and turning it into rows would only add a layer.
public final class ImageDiskCache {

    private final File directory;
    private final File temporaries;
    private final long maxBytes;
    // File name to its size, in access order, so eviction takes the front of
    // the iterator and never sorts. The size is recorded at insert rather than
    // read back from the file later, so the budget cannot drift when a file is
    // replaced or deleted underneath us.
    private final LinkedHashMap<String, Long> entries = new LinkedHashMap<>(16, 0.75f, true);
    private long bytes;

    public ImageDiskCache(File directory, long maxBytes) {
        this.directory = directory;
        this.temporaries = new File(directory, "tmp");
        this.maxBytes = maxBytes;
        temporaries.mkdirs();
        File[] existing = directory.listFiles(File::isFile);
        if (existing != null) {
            // A restart has to learn what is already on disk, or the budget
            // starts at zero underneath a full cache. Last modified is write
            // time, so recency is exact within a session and approximate
            // across a restart. DiskLruCache persists a journal to close that
            // gap, and that is the price of doing better.
            Arrays.sort(existing, Comparator.comparingLong(File::lastModified));
            for (File file : existing) {
                entries.put(file.getName(), file.length());
                bytes += file.length();
            }
        }
    }

    public synchronized File get(String url) {
        String name = nameFor(url);
        Long size = entries.get(name);
        if (size == null) {
            return null;
        }
        File file = new File(directory, name);
        if (!file.exists()) {
            // Deleted from outside, a user clearing storage. Forget it now,
            // or the budget keeps counting bytes that are no longer there.
            entries.remove(name);
            bytes -= size;
            return null;
        }
        return file;
    }

    // One temporary file per URL, at a name derived from the URL. That is what
    // makes resume work. A killed process leaves the part file behind, the next
    // request for the same URL finds it, and its length is the offset to ask
    // the server to continue from. There is no separate state table because
    // the file's own length is the state.
    public File temporaryFileFor(String url) {
        return new File(temporaries, nameFor(url) + ".part");
    }

    // The temporary file is written first and moved into place last. A process
    // killed mid download then leaves a part file in the temp directory rather
    // than a half written image at a cache path, which a later hit would
    // happily serve as though it were whole.
    public synchronized File commit(String url, File temporary) throws IOException {
        File target = new File(directory, nameFor(url));
        long size = temporary.length();
        if (!temporary.renameTo(target)) {
            throw new IOException("Could not commit " + temporary);
        }
        // A re-download of a URL already in the cache replaces the file, so
        // the old size comes off before the new one goes on. Adding without
        // subtracting inflates the total on every refresh of the same avatar
        // until the cache evicts everything else to satisfy a budget it is no
        // longer measuring.
        Long previous = entries.put(target.getName(), size);
        bytes += size - (previous == null ? 0 : previous);
        trim();
        return target;
    }

    private void trim() {
        Iterator<Map.Entry<String, Long>> oldestFirst = entries.entrySet().iterator();
        while (bytes > maxBytes && oldestFirst.hasNext()) {
            Map.Entry<String, Long> victim = oldestFirst.next();
            bytes -= victim.getValue();
            // This can delete a file a caller is decoding right now, because
            // get hands out a path with no lease on it. The real fix is a
            // reference count or handing back an open stream instead of a
            // path, which is what DiskLruCache's Snapshot is.
            new File(directory, victim.getKey()).delete();
            oldestFirst.remove();
        }
    }

    // A URL is not a legal file name, so the key is a digest of it. It also
    // keeps a query string full of signed parameters out of the file system.
    private static String nameFor(String url) {
        return UUID.nameUUIDFromBytes(url.getBytes(StandardCharsets.UTF_8)).toString();
    }
}

package com.androidinterview.imagedownloader;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.function.Consumer;

// Cache first, then one download per URL however many callers asked for it,
// with a cap on how many connections any single host sees at once.
public final class ImageDownloader {

    // Streams a response body into the destination file from resumeFrom, so
    // nothing large is ever held in memory. The contract is the Range dance
    // the file downloader answer writes out in full. Send Range from
    // resumeFrom, append on a 206, and on a 200 truncate the file to zero and
    // write from the start, because the server sent the whole thing again.
    // Whatever networking library the app has already chosen implements this.
    public interface Fetcher {
        void fetchTo(String url, File destination, long resumeFrom) throws IOException;
    }

    public interface Callback {
        void onReady(File file);

        void onFailed(Exception cause);
    }

    private final ImageDiskCache cache;
    private final Fetcher fetcher;
    private final int perHostLimit;
    private final ExecutorService pool;
    // Callbacks are posted here rather than run on the pool thread, because
    // the caller is usually about to touch a view. On Android this is a
    // Handler on the main looper wrapped as an Executor.
    private final Executor callbackExecutor;
    // One permit set per host, never pruned. An app talks to a handful of
    // image hosts, so the map stays small enough not to bother bounding.
    private final Map<String, Semaphore> hostLimits = new ConcurrentHashMap<>();
    // URL to everyone waiting on it. This map is the deduplication layer, and
    // it is the only state the downloader keeps beyond the cache.
    private final Map<String, List<Callback>> waiting = new HashMap<>();

    public ImageDownloader(
            ImageDiskCache cache, Fetcher fetcher, int concurrency, int perHostLimit, Executor callbackExecutor) {
        this.cache = cache;
        this.fetcher = fetcher;
        this.perHostLimit = perHostLimit;
        this.callbackExecutor = callbackExecutor;
        // Downloads are I O bound, so the pool can be larger than the core
        // count. It still needs a cap, because an unbounded pool turns a feed
        // full of thumbnails into a few hundred open sockets.
        this.pool = Executors.newFixedThreadPool(concurrency);
    }

    public void request(String url, Callback callback) {
        File hit = cache.get(url);
        if (hit != null) {
            callbackExecutor.execute(() -> callback.onReady(hit));
            return;
        }
        synchronized (waiting) {
            List<Callback> already = waiting.get(url);
            if (already != null) {
                // Ten rows asking for the same avatar is one download and ten
                // callbacks. Joining the download that is already running,
                // rather than starting a second one, is the whole point.
                already.add(callback);
                return;
            }
            List<Callback> first = new ArrayList<>();
            first.add(callback);
            waiting.put(url, first);
        }
        pool.execute(() -> download(url));
    }

    private void download(String url) {
        Semaphore host = hostLimits.computeIfAbsent(hostOf(url), absent -> new Semaphore(perHostLimit));
        // A global cap is not enough. What gets you throttled is simultaneous
        // connections to one host, so the permit is per host and the surplus
        // workers wait here rather than at somebody else's server.
        host.acquireUninterruptibly();
        try {
            // The part file for this URL is the resume point. Whatever a
            // previous attempt managed to write is still there, and its length
            // is where the next request asks the server to continue.
            File temporary = cache.temporaryFileFor(url);
            fetcher.fetchTo(url, temporary, temporary.length());
            File file = cache.commit(url, temporary);
            deliver(url, callback -> callback.onReady(file));
        } catch (Exception failure) {
            // The part file is deliberately left in place. It is what the next
            // attempt resumes from.
            deliver(url, callback -> callback.onFailed(failure));
        } finally {
            host.release();
        }
    }

    // The waiting list is taken out of the map before it is walked, so a
    // callback that immediately asks for the same URL again starts a fresh
    // download instead of joining one that has already finished.
    private void deliver(String url, Consumer<Callback> outcome) {
        List<Callback> callbacks;
        synchronized (waiting) {
            callbacks = waiting.remove(url);
        }
        if (callbacks != null) {
            callbackExecutor.execute(() -> callbacks.forEach(outcome));
        }
    }

    private static String hostOf(String url) {
        String host = URI.create(url).getHost();
        return host == null ? "" : host;
    }
}

package com.androidinterview.imagedownloader

import java.io.File
import java.util.UUID

// Files on disk, keyed by URL, evicted least recently used first. This is the
// whole persistence story for an image downloader, because the thing being
// cached is already a file and turning it into rows would only add a layer.
class ImageDiskCache(private val directory: File, private val maxBytes: Long) {

    private val temporaries = File(directory, "tmp").apply { mkdirs() }

    // File name to its size, in access order, so eviction takes the front of
    // the iterator and never sorts. The size is recorded at insert rather than
    // read back from the file later, so the budget cannot drift when a file is
    // replaced or deleted underneath us.
    private val entries = LinkedHashMap<String, Long>(16, 0.75f, true)
    private var bytes = 0L

    init {
        // A restart has to learn what is already on disk, or the budget starts
        // at zero underneath a full cache. Last modified is write time, so
        // recency is exact within a session and approximate across a restart.
        // DiskLruCache persists a journal to close that gap, and that is the
        // price of doing better.
        directory.listFiles(File::isFile).orEmpty().sortedBy(File::lastModified).forEach {
            entries[it.name] = it.length()
            bytes += it.length()
        }
    }

    @Synchronized
    operator fun get(url: String): File? {
        val name = nameFor(url)
        val size = entries[name] ?: return null
        val file = File(directory, name)
        if (file.exists()) return file
        // Deleted from outside, a user clearing storage. Forget it now, or the
        // budget keeps counting bytes that are no longer there.
        entries.remove(name)
        bytes -= size
        return null
    }

    // One temporary file per URL, at a name derived from the URL. That is what
    // makes resume work. A killed process leaves the part file behind, the next
    // request for the same URL finds it, and its length is the offset to ask
    // the server to continue from. There is no separate state table because
    // the file's own length is the state.
    fun temporaryFileFor(url: String) = File(temporaries, nameFor(url) + ".part")

    // The temporary file is written first and moved into place last. A process
    // killed mid download then leaves a part file in the temp directory rather
    // than a half written image at a cache path, which a later hit would
    // happily serve as though it were whole.
    @Synchronized
    fun commit(url: String, temporary: File): File {
        val target = File(directory, nameFor(url))
        val size = temporary.length()
        check(temporary.renameTo(target)) { "Could not commit $temporary" }
        // A re-download of a URL already in the cache replaces the file, so
        // the old size comes off before the new one goes on. Adding without
        // subtracting inflates the total on every refresh of the same avatar
        // until the cache evicts everything else to satisfy a budget it is no
        // longer measuring.
        bytes += size - (entries.put(target.name, size) ?: 0L)
        trim()
        return target
    }

    private fun trim() {
        val oldestFirst = entries.entries.iterator()
        while (bytes > maxBytes && oldestFirst.hasNext()) {
            val victim = oldestFirst.next()
            bytes -= victim.value
            // This can delete a file a caller is decoding right now, because
            // get hands out a path with no lease on it. The real fix is a
            // reference count or handing back an open stream instead of a
            // path, which is what DiskLruCache's Snapshot is.
            File(directory, victim.key).delete()
            oldestFirst.remove()
        }
    }

    // A URL is not a legal file name, so the key is a digest of it. It also
    // keeps a query string full of signed parameters out of the file system.
    private fun nameFor(url: String) = UUID.nameUUIDFromBytes(url.toByteArray()).toString()
}

package com.androidinterview.imagedownloader

import java.io.File
import java.net.URI
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.Executor
import java.util.concurrent.Executors
import java.util.concurrent.Semaphore

// Streams a response body into the destination file from resumeFrom, so
// nothing large is held in memory. The contract is the Range dance the file
// downloader answer writes out in full. Send Range from resumeFrom, append on
// a 206, and on a 200 truncate the file to zero and write from the start,
// because the server sent the whole thing again. Whatever networking library
// the app already chose implements this.
fun interface Fetcher {
    fun fetchTo(url: String, destination: File, resumeFrom: Long)
}

// Cache first, then one download per URL however many callers asked for it,
// with a cap on how many connections any single host sees at once.
class ImageDownloader(
    private val cache: ImageDiskCache,
    private val fetcher: Fetcher,
    concurrency: Int,
    private val perHostLimit: Int,
    // Callbacks are posted here rather than run on the pool thread, because
    // the caller is usually about to touch a view. On Android this is a
    // Handler on the main looper wrapped as an Executor.
    private val callbackExecutor: Executor,
) {
    // Downloads are I O bound, so the pool can be larger than the core count.
    // It still needs a cap, because an unbounded pool turns a feed full of
    // thumbnails into a few hundred open sockets.
    private val pool = Executors.newFixedThreadPool(concurrency)

    // One permit set per host, never pruned. An app talks to a handful of
    // image hosts, so the map stays small enough not to bother bounding.
    private val hostLimits = ConcurrentHashMap<String, Semaphore>()

    // URL to everyone waiting on it. This map is the deduplication layer, and
    // it is the only state the downloader keeps beyond the cache.
    private val waiting = mutableMapOf<String, MutableList<(Result<File>) -> Unit>>()

    fun request(url: String, callback: (Result<File>) -> Unit) {
        cache[url]?.let { hit ->
            callbackExecutor.execute { callback(Result.success(hit)) }
            return
        }
        synchronized(waiting) {
            waiting[url]?.let { already ->
                // Ten rows asking for the same avatar is one download and ten
                // callbacks. Joining the download that is already running,
                // rather than starting a second one, is the whole point.
                already += callback
                return
            }
            waiting[url] = mutableListOf(callback)
        }
        pool.execute { download(url) }
    }

    private fun download(url: String) {
        val host = hostLimits.getOrPut(hostOf(url)) { Semaphore(perHostLimit) }
        // A global cap is not enough. What gets you throttled is simultaneous
        // connections to one host, so the permit is per host and the surplus
        // workers wait here rather than at somebody else's server.
        host.acquireUninterruptibly()
        val outcome = runCatching {
            // The part file for this URL is the resume point. Whatever a
            // previous attempt managed to write is still there, and its length
            // is where the next request asks the server to continue. On
            // failure it is deliberately left in place for the next attempt.
            val temporary = cache.temporaryFileFor(url)
            fetcher.fetchTo(url, temporary, temporary.length())
            cache.commit(url, temporary)
        }
        host.release()
        // The waiting list is taken out of the map before it is walked, so a
        // callback that immediately asks for the same URL again starts a fresh
        // download instead of joining one that has already finished.
        val callbacks = synchronized(waiting) { waiting.remove(url) } ?: return
        callbackExecutor.execute { callbacks.forEach { it(outcome) } }
    }

    private fun hostOf(url: String) = URI.create(url).host.orEmpty()
}