[core] add logic of high and low priority requests to OnlineFileSource

- only process low priority requests (=offline requests) if there are no outstanding regular requests
- favour regular requests over low priority ones

1 files changed, 85 insertions, 22 deletions

diff --git a/platform/default/online_file_source.cpp b/platform/default/online_file_source.cpp
index d685109b95..47ebbbeb45 100644
--- a/platform/default/online_file_source.cpp
+++ b/platform/default/online_file_source.cpp
@@ -89,13 +89,8 @@ public:
         if (activeRequests.erase(request)) {
             activatePendingRequest();
         } else {
-            auto it = pendingRequestsMap.find(request);
-            if (it != pendingRequestsMap.end()) {
-                pendingRequestsList.erase(it->second);
-                pendingRequestsMap.erase(it);
-            }
+            pendingRequests.remove(request);
         }
-        assert(pendingRequestsMap.size() == pendingRequestsList.size());
     }
 
     void activateOrQueueRequest(OnlineFileRequest* request) {
@@ -111,9 +106,7 @@ public:
     }
 
     void queueRequest(OnlineFileRequest* request) {
-        auto it = pendingRequestsList.insert(pendingRequestsList.end(), request);
-        pendingRequestsMap.emplace(request, std::move(it));
-        assert(pendingRequestsMap.size() == pendingRequestsList.size());
+        pendingRequests.insert(request);
     }
 
     void activateRequest(OnlineFileRequest* request) {
@@ -135,25 +128,17 @@ public:
             callback(response);
         }
 
-        assert(pendingRequestsMap.size() == pendingRequestsList.size());
     }
 
     void activatePendingRequest() {
-        if (pendingRequestsList.empty()) {
-            return;
-        }
 
-        OnlineFileRequest* request = pendingRequestsList.front();
-        pendingRequestsList.pop_front();
+        auto request = pendingRequests.pop();
 
-        pendingRequestsMap.erase(request);
-
-        activateRequest(request);
-        assert(pendingRequestsMap.size() == pendingRequestsList.size());
+        if (request) activateRequest(*request);
     }
 
     bool isPending(OnlineFileRequest* request) {
-        return pendingRequestsMap.find(request) != pendingRequestsMap.end();
+        return pendingRequests.contains(request);
     }
 
     bool isActive(OnlineFileRequest* request) {
@@ -169,13 +154,90 @@ public:
         networkIsReachableAgain();
     }
 
+    void setMaximumConcurrentRequestsOverride(const uint32_t override) {
+        maximumConcurrentRequestsOverride = override;
+    }
+
 private:
+
+    uint32_t getMaximumConcurrentRequests() {
+        if (maximumConcurrentRequestsOverride > 0) return maximumConcurrentRequestsOverride;
+        else return HTTPFileSource::maximumConcurrentRequests();
+    }
+
     void networkIsReachableAgain() {
         for (auto& request : allRequests) {
             request->networkIsReachableAgain();
         }
     }
 
+    // Using Pending Requests as an priority queue which processes
+    // file requests in a FIFO manner but prefers regular requests
+    // over offline requests with a low priority such that low priority
+    // requests do not throttle regular requests.
+    //
+    // The order of a queue is therefore:
+    //
+    // hi0 -- hi1 -- hi2 -- hi3 -- lo0 -- lo1 --lo2
+    //                              ^
+    //                              firstLowPriorityRequest
+
+    struct PendingRequests {
+        PendingRequests() : queue(), firstLowPriorityRequest(queue.begin()) {}
+
+        std::list<OnlineFileRequest*> queue;
+        std::list<OnlineFileRequest*>::iterator firstLowPriorityRequest;
+
+        void remove(OnlineFileRequest* request) {
+            auto it = std::find(queue.begin(), queue.end(), request);
+            if (it != queue.end()) {
+                if (it == firstLowPriorityRequest) {
+                    firstLowPriorityRequest++;
+                }
+                queue.erase(it);
+
+                if (queue.empty()) {
+                    first_low = queue.begin();
+                }
+            }
+        }
+
+        void insert(OnlineFileRequest* request) {
+            if (request->resource.priority == Resource::Priority::Regular) {
+                firstLowPriorityRequest = queue.insert(firstLowPriorityRequest, request);
+                firstLowPriorityRequest++;
+            }
+            else {
+                if (firstLowPriorityRequest == queue.end()) {
+                    firstLowPriorityRequest = queue.insert(queue.end(), request);
+                }
+                else {
+                    queue.insert(queue.end(), request);
+                }
+            }
+        }
+
+
+        optional<OnlineFileRequest*> pop() {
+            if (queue.empty()) {
+                return optional<OnlineFileRequest*>();
+            }
+
+            if (queue.begin() == firstLowPriorityRequest) {
+                firstLowPriorityRequest++;
+            }
+
+            OnlineFileRequest* next = queue.front();
+            queue.pop_front();
+            return optional<OnlineFileRequest*>(next);
+        }
+
+        bool contains(OnlineFileRequest* request) {
+            return (std::find(queue.begin(), queue.end(), request) != queue.end());
+        }
+
+    };
+
     optional<ActorRef<ResourceTransform>> resourceTransform;
 
     /**
@@ -190,8 +252,9 @@ private:
      * `pendingRequests`. Requests in the active state are in `activeRequests`.
      */
     std::unordered_set<OnlineFileRequest*> allRequests;
-    std::list<OnlineFileRequest*> pendingRequestsList;
-    std::unordered_map<OnlineFileRequest*, std::list<OnlineFileRequest*>::iterator> pendingRequestsMap;
+
+    PendingRequests pendingRequests;
+
     std::unordered_set<OnlineFileRequest*> activeRequests;
 
     bool online = true;