Fix soft TTL for duplicate requests (google#73)

* Fix soft TTL for duplicate requests: when a soft TTL cache entry exists, that response should be served immediately rather than having duplicate requests wait for the first request's network response.

Author: uhager

src/main/java/com/android/volley/CacheDispatcher.java

@@ -18,6 +18,10 @@
 
 import android.os.Process;
 
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Queue;
 import java.util.concurrent.BlockingQueue;
 
 /**
@@ -48,6 +52,9 @@ public class CacheDispatcher extends Thread {
     /** Used for telling us to die. */
     private volatile boolean mQuit = false;
 
+    /** Manage list of waiting requests and de-duplicate requests with same cache key. */
+    private final WaitingRequestManager mWaitingRequestManager;
+
     /**
      * Creates a new cache triage dispatcher thread.  You must call {@link #start()}
      * in order to begin processing.
@@ -64,6 +71,7 @@ public CacheDispatcher(
         mNetworkQueue = networkQueue;
         mCache = cache;
         mDelivery = delivery;
+        mWaitingRequestManager = new WaitingRequestManager(this);
     }
 
     /**
@@ -101,15 +109,19 @@ public void run() {
                 if (entry == null) {
                     request.addMarker("cache-miss");
                     // Cache miss; send off to the network dispatcher.
-                    mNetworkQueue.put(request);
+                    if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
+                        mNetworkQueue.put(request);
+                    }
                     continue;
                 }
 
                 // If it is completely expired, just send it to the network.
                 if (entry.isExpired()) {
                     request.addMarker("cache-hit-expired");
                     request.setCacheEntry(entry);
-                    mNetworkQueue.put(request);
+                    if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
+                        mNetworkQueue.put(request);
+                    }
                     continue;
                 }
 
@@ -128,22 +140,28 @@ public void run() {
                     // refreshing.
                     request.addMarker("cache-hit-refresh-needed");
                     request.setCacheEntry(entry);
-
                     // Mark the response as intermediate.
                     response.intermediate = true;
 
-                    // Post the intermediate response back to the user and have
-                    // the delivery then forward the request along to the network.
-                    mDelivery.postResponse(request, response, new Runnable() {
-                        @Override
-                        public void run() {
-                            try {
-                                mNetworkQueue.put(request);
-                            } catch (InterruptedException e) {
-                                // Not much we can do about this.
+                    if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
+                        // Post the intermediate response back to the user and have
+                        // the delivery then forward the request along to the network.
+                        mDelivery.postResponse(request, response, new Runnable() {
+                            @Override
+                            public void run() {
+                                try {
+                                    mNetworkQueue.put(request);
+                                } catch (InterruptedException e) {
+                                    // Restore the interrupted status
+                                    Thread.currentThread().interrupt();
+                                }
                             }
-                        }
-                    });
+                        });
+                    } else {
+                        // request has been added to list of waiting requests
+                        // to receive the network response from the first request once it returns.
+                        mDelivery.postResponse(request, response);
+                    }
                 }
 
             } catch (InterruptedException e) {
@@ -154,4 +172,112 @@ public void run() {
             }
         }
     }
+
+    private static class WaitingRequestManager implements Request.NetworkRequestCompleteListener {
+
+        /**
+         * Staging area for requests that already have a duplicate request in flight.
+         *
+         * <ul>
+         *     <li>containsKey(cacheKey) indicates that there is a request in flight for the given cache
+         *          key.</li>
+         *     <li>get(cacheKey) returns waiting requests for the given cache key. The in flight request
+         *          is <em>not</em> contained in that list. Is null if no requests are staged.</li>
+         * </ul>
+         */
+        private final Map<String, Queue<Request<?>>> mWaitingRequests = new HashMap<>();
+
+        private final CacheDispatcher mCacheDispatcher;
+
+        WaitingRequestManager(CacheDispatcher cacheDispatcher) {
+            mCacheDispatcher = cacheDispatcher;
+        }
+
+        /** Request received a valid response that can be used by other waiting requests. */
+        @Override
+        public void onResponseReceived(Request<?> request, Response<?> response) {
+            if (response.cacheEntry == null || response.cacheEntry.isExpired()) {
+                onNoUsableResponseReceived(request);
+                return;
+            }
+            String cacheKey = request.getCacheKey();
+            Queue<Request<?>> waitingRequests;
+            synchronized (this) {
+                waitingRequests = mWaitingRequests.remove(cacheKey);
+            }
+            if (waitingRequests != null) {
+                if (VolleyLog.DEBUG) {
+                    VolleyLog.v("Releasing %d waiting requests for cacheKey=%s.",
+                            waitingRequests.size(), cacheKey);
+                }
+                // Process all queued up requests.
+                for (Request<?> waiting : waitingRequests) {
+                    mCacheDispatcher.mDelivery.postResponse(waiting, response);
+                }
+            }
+        }
+
+        /** No valid response received from network, release waiting requests. */
+        @Override
+        public synchronized void onNoUsableResponseReceived(Request<?> request) {
+            String cacheKey = request.getCacheKey();
+            Queue<Request<?>> waitingRequests = mWaitingRequests.remove(cacheKey);
+            if (waitingRequests != null) {
+                if (VolleyLog.DEBUG) {
+                    VolleyLog.v("%d waiting requests for cacheKey=%s; resend to network",
+                            waitingRequests.size(), cacheKey);
+                }
+                Request<?> nextInLine = waitingRequests.remove();
+                    if (nextInLine == null) {
+                        return;
+                    }
+                    mWaitingRequests.put(cacheKey, waitingRequests);
+                try {
+                    mCacheDispatcher.mNetworkQueue.put(nextInLine);
+                } catch (InterruptedException iex) {
+                    VolleyLog.e("Couldn't add request to queue. %s", iex.toString());
+                    // Restore the interrupted status of the calling thread (i.e. NetworkDispatcher)
+                    Thread.currentThread().interrupt();
+                    // Quit the current CacheDispatcher thread.
+                    mCacheDispatcher.quit();
+                }
+            }
+        }
+
+        /**
+         * For cacheable requests, if a request for the same cache key is already in flight,
+         * add it to a queue to wait for that in-flight request to finish.
+         * @return whether the request was queued. If false, we should continue issuing the request
+         * over the network. If true, we should put the request on hold to be processed when
+         * the in-flight request finishes.
+         */
+        private synchronized boolean maybeAddToWaitingRequests(Request<?> request) {
+            String cacheKey = request.getCacheKey();
+            // Insert request into stage if there's already a request with the same cache key
+            // in flight.
+            if (mWaitingRequests.containsKey(cacheKey)) {
+                // There is already a request in flight. Queue up.
+                Queue<Request<?>> stagedRequests = mWaitingRequests.get(cacheKey);
+                if (stagedRequests == null) {
+                    stagedRequests = new LinkedList<Request<?>>();
+                }
+                request.addMarker("waiting-for-response");
+                stagedRequests.add(request);
+                mWaitingRequests.put(cacheKey, stagedRequests);
+                if (VolleyLog.DEBUG) {
+                    VolleyLog.d("Request for cacheKey=%s is in flight, putting on hold.", cacheKey);
+                }
+                return true;
+            } else {
+                // Insert 'null' queue for this cacheKey, indicating there is now a request in
+                // flight.
+                mWaitingRequests.put(cacheKey, null);
+                request.setNetworkRequestCompleteListener(this);
+                if (VolleyLog.DEBUG) {
+                    VolleyLog.d("new request, sending to network %s", cacheKey);
+                }
+                return false;
+            }
+        }
+    }
 }

src/main/java/com/android/volley/NetworkDispatcher.java

@@ -33,6 +33,7 @@
  * errors are posted back to the caller via a {@link ResponseDelivery}.
  */
 public class NetworkDispatcher extends Thread {
+
     /** The queue of requests to service. */
     private final BlockingQueue<Request<?>> mQueue;
     /** The network interface for processing requests. */
@@ -54,8 +55,7 @@ public class NetworkDispatcher extends Thread {
      * @param delivery Delivery interface to use for posting responses
      */
     public NetworkDispatcher(BlockingQueue<Request<?>> queue,
-            Network network, Cache cache,
-            ResponseDelivery delivery) {
+            Network network, Cache cache, ResponseDelivery delivery) {
         mQueue = queue;
         mNetwork = network;
         mCache = cache;
@@ -103,6 +103,7 @@ public void run() {
                 // network request.
                 if (request.isCanceled()) {
                     request.finish("network-discard-cancelled");
+                    request.notifyListenerResponseNotUsable();
                     continue;
                 }
 
@@ -116,6 +117,7 @@ public void run() {
                 // we're done -- don't deliver a second identical response.
                 if (networkResponse.notModified && request.hasHadResponseDelivered()) {
                     request.finish("not-modified");
+                    request.notifyListenerResponseNotUsable();
                     continue;
                 }
 
@@ -133,14 +135,17 @@ public void run() {
                 // Post the response back.
                 request.markDelivered();
                 mDelivery.postResponse(request, response);
+                request.notifyListenerResponseReceived(response);
             } catch (VolleyError volleyError) {
                 volleyError.setNetworkTimeMs(SystemClock.elapsedRealtime() - startTimeMs);
                 parseAndDeliverNetworkError(request, volleyError);
+                request.notifyListenerResponseNotUsable();
             } catch (Exception e) {
                 VolleyLog.e(e, "Unhandled exception %s", e.toString());
                 VolleyError volleyError = new VolleyError(e);
                 volleyError.setNetworkTimeMs(SystemClock.elapsedRealtime() - startTimeMs);
                 mDelivery.postError(request, volleyError);
+                request.notifyListenerResponseNotUsable();
             }
         }
     }

src/main/java/com/android/volley/Request.java

@@ -56,6 +56,18 @@ public interface Method {
         int PATCH = 7;
     }
 
+    /**
+     * Callback to notify when the network request returns.
+     */
+    /* package */ interface NetworkRequestCompleteListener {
+
+        /** Callback when a network response has been received. */
+        void onResponseReceived(Request<?> request, Response<?> response);
+
+        /** Callback when request returns from network without valid response. */
+        void onNoUsableResponseReceived(Request<?> request);
+    }
+
     /** An event log tracing the lifetime of this request; for debugging. */
     private final MarkerLog mEventLog = MarkerLog.ENABLED ? new MarkerLog() : null;
 
@@ -105,6 +117,12 @@ public interface Method {
     /** An opaque token tagging this request; used for bulk cancellation. */
     private Object mTag;
 
+    /** Listener that will be notified when a response has been delivered. */
+    private NetworkRequestCompleteListener mRequestCompleteListener;
+
+    /** Object to guard access to mRequestCompleteListener. */
+    private final Object mLock = new Object();
+
     /**
      * Creates a new request with the given URL and error listener.  Note that
      * the normal response listener is not provided here as delivery of responses
@@ -584,6 +602,42 @@ public void deliverError(VolleyError error) {
         }
     }
 
+    /**
+     * {@link NetworkRequestCompleteListener} that will receive callbacks when the request
+     * returns from the network.
+     */
+    /* package */ void setNetworkRequestCompleteListener(
+            NetworkRequestCompleteListener requestCompleteListener) {
+        synchronized (mLock) {
+            mRequestCompleteListener = requestCompleteListener;
+        }
+    }
+
+    /**
+     * Notify NetworkRequestCompleteListener that a valid response has been received
+     * which can be used for other, waiting requests.
+     * @param response received from the network
+     */
+    /* package */ void notifyListenerResponseReceived(Response<?> response) {
+        synchronized (mLock) {
+            if (mRequestCompleteListener != null) {
+               mRequestCompleteListener.onResponseReceived(this, response);
+            }
+        }
+    }
+
+    /**
+     * Notify NetworkRequestCompleteListener that the network request did not result in
+     * a response which can be used for other, waiting requests.
+     */
+    /* package */ void notifyListenerResponseNotUsable() {
+        synchronized (mLock) {
+            if (mRequestCompleteListener != null) {
+                mRequestCompleteListener.onNoUsableResponseReceived(this);
+            }
+        }
+    }
+
     /**
      * Our comparator sorts from high to low priority, and secondarily by
      * sequence number to provide FIFO ordering.