Allow limiting number of concurrent vector tile requests to prevent exhaustion of browser resources

src/source/load_vector_tile.ts

@@ -1,10 +1,12 @@
 import {VectorTile} from '@mapbox/vector-tile';
 import Protobuf from 'pbf';
 import {getArrayBuffer} from '../util/ajax';
+import assert from "assert";
 
 import type {Callback} from '../types/callback';
 import type {RequestedTileParameters} from './worker_source';
 import type Scheduler from '../util/scheduler';
+import type {Cancelable} from '../types/cancelable';
 
 export type LoadVectorTileResult = {
     rawData: ArrayBuffer;
@@ -23,7 +25,29 @@ export type LoadVectorTileResult = {
 export type LoadVectorDataCallback = Callback<LoadVectorTileResult | null | undefined>;
 
 export type AbortVectorData = () => void;
-export type LoadVectorData = (params: RequestedTileParameters, callback: LoadVectorDataCallback) => AbortVectorData | undefined;
+export type LoadVectorData = (params: RequestedTileParameters, callback: LoadVectorDataCallback, deduped: DedupedRequest) => AbortVectorData | undefined;
+export type DedupedRequestInput = {key : string,
+    metadata: any,
+    requestFunc: any,
+    callback: LoadVectorDataCallback,
+    fromQueue?: boolean
+};
+export type VectorTileQueueEntry = DedupedRequestInput & {
+    cancelled: boolean,
+    cancel: () => void
+};
+
+let requestQueue: Map<string, VectorTileQueueEntry>, numRequests: number;
+export const resetRequestQueue = () => {
+    requestQueue = new Map();
+    numRequests = 0;
+};
+resetRequestQueue();
+
+const filterQueue = (key: string) => {
+    requestQueue.delete(key);
+};
+
 export class DedupedRequest {
     entries: {
         [key: string]: any;
@@ -35,62 +59,153 @@ export class DedupedRequest {
         this.scheduler = scheduler;
     }
 
-    request(key: string, metadata: any, request: any, callback: LoadVectorDataCallback): () => void {
-        const entry = this.entries[key] = this.entries[key] || {callbacks: []};
+    addToSchedulerOrCallDirectly({
+        callback,
+        metadata,
+        err,
+        result,
+    }: {
+        callback: LoadVectorDataCallback;
+        metadata: any;
+        err: Error | null | undefined;
+        result: any;
+    }) {
+        if (this.scheduler) {
+            this.scheduler.add(() => {
+                callback(err, result);
+            }, metadata);
+        } else {
+            callback(err, result);
+        }
+    }
+
+    getEntry = (key: string) => {
+        return (
+            this.entries[key] || {
+                // use a set to avoid duplicate callbacks being added when calling from queue
+                callbacks: new Set(),
+            }
+        );
+    };
+
+    request({key, metadata, requestFunc, callback, fromQueue}: DedupedRequestInput): Cancelable {
+        const entry = (this.entries[key] = this.getEntry(key));
+
+        const removeCallbackFromEntry = ({key, requestCallback}) => {
+            const entry = this.getEntry(key);
+            if (entry.result) {
+                return;
+            }
+            entry.callbacks.delete(requestCallback);
+            if (entry.callbacks.size) {
+                return;
+            }
+            if (entry.cancel) {
+                entry.cancel();
+            }
+            filterQueue(key);
+            delete this.entries[key];
+        };
+
+        let advanced = false;
+        const advanceRequestQueue = () => {
+            if (advanced) {
+                return;
+            }
+            advanced = true;
+            numRequests--;
+            assert(numRequests >= 0);
+            while (requestQueue.size && numRequests < 50) {
+                const request = requestQueue.values().next().value;
+                const {key, metadata, requestFunc, callback, cancelled} = request;
+                filterQueue(key);
+                if (!cancelled) {
+                    request.cancel = this.request({
+                        key,
+                        metadata,
+                        requestFunc,
+                        callback,
+                        fromQueue: true
+                    }).cancel;
+                }
+            }
+        };
 
         if (entry.result) {
             const [err, result] = entry.result;
-            if (this.scheduler) {
-                this.scheduler.add(() => {
-                    callback(err, result);
-                }, metadata);
-            } else {
-                callback(err, result);
-            }
-            return () => {};
+            this.addToSchedulerOrCallDirectly({
+                callback,
+                metadata,
+                err,
+                result,
+            });
+            return {cancel: () => {}};
         }
 
-        entry.callbacks.push(callback);
+        entry.callbacks.add(callback);
+
+        const inQueue = requestQueue.has(key);
+        if ((!entry.cancel && !inQueue) || fromQueue) {
+            // Lack of attached cancel handler means this is the first request for this resource
+            if (numRequests >= 50) {
+                const queued = {
+                    key,
+                    metadata,
+                    requestFunc,
+                    callback,
+                    cancelled: false,
+                    cancel() {},
+                };
+                const cancelFunc = () => {
+                    queued.cancelled = true;
+                    removeCallbackFromEntry({
+                        key,
+                        requestCallback: callback,
+                    });
+                };
+                queued.cancel = cancelFunc;
+                requestQueue.set(key, queued);
+                return queued;
+            }
+            numRequests++;
 
-        if (!entry.cancel) {
-            entry.cancel = request((err, result) => {
+            const actualRequestCancel = requestFunc((err, result) => {
                 entry.result = [err, result];
+
                 for (const cb of entry.callbacks) {
-                    if (this.scheduler) {
-                        this.scheduler.add(() => {
-                            cb(err, result);
-                        }, metadata);
-                    } else {
-                        cb(err, result);
-                    }
+                    this.addToSchedulerOrCallDirectly({
+                        callback: cb,
+                        metadata,
+                        err,
+                        result,
+                    });
                 }
-                setTimeout(() => delete this.entries[key], 1000 * 3);
+
+                filterQueue(key);
+                advanceRequestQueue();
+
+                setTimeout(() => {
+                    delete this.entries[key];
+                }, 1000 * 3);
             });
+            entry.cancel = actualRequestCancel;
         }
 
-        return () => {
-            if (entry.result) return;
-            entry.callbacks = entry.callbacks.filter(cb => cb !== callback);
-            if (!entry.callbacks.length) {
-                entry.cancel();
-                delete this.entries[key];
-            }
+        return {
+            cancel() {
+                removeCallbackFromEntry({
+                    key,
+                    requestCallback: callback,
+                });
+            },
         };
     }
 }
 
-/**
- * @private
- */
-export function loadVectorTile(
-    params: RequestedTileParameters,
-    callback: LoadVectorDataCallback,
-    skipParse?: boolean,
-): () => void {
-    const key = JSON.stringify(params.request);
+const makeArrayBufferHandler = ({requestParams, skipParse}) => {
 
     const makeRequest = (callback: LoadVectorDataCallback) => {
-        const request = getArrayBuffer(params.request, (err?: Error | null, data?: ArrayBuffer | null, cacheControl?: string | null, expires?: string | null) => {
+        const request = getArrayBuffer(requestParams, (err?: Error | null, data?: ArrayBuffer | null, cacheControl?: string | null, expires?: string | null) => {
             if (err) {
                 callback(err);
             } else if (data) {
@@ -108,11 +223,37 @@ export function loadVectorTile(
         };
     };
 
+    return makeRequest;
+};
+
+/**
+ * @private
+ */
+export function loadVectorTile(
+    params: RequestedTileParameters,
+    callback: LoadVectorDataCallback,
+    deduped: DedupedRequest,
+    skipParse?: boolean,
+    providedArrayBufferHandlerMaker?: any
+): () => void {
+    const key = JSON.stringify(params.request);
+
+    const arrayBufferCallbackMaker = providedArrayBufferHandlerMaker || makeArrayBufferHandler;
+    const makeRequest = arrayBufferCallbackMaker({requestParams: params.request, skipParse});
+
     if (params.data) {
         // if we already got the result earlier (on the main thread), return it directly
-        (this.deduped as DedupedRequest).entries[key] = {result: [null, params.data]};
+        deduped.entries[key] = {result: [null, params.data]};
     }
 
     const callbackMetadata = {type: 'parseTile', isSymbolTile: params.isSymbolTile, zoom: params.tileZoom};
-    return (this.deduped as DedupedRequest).request(key, callbackMetadata, makeRequest, callback);
+    const dedupedAndQueuedRequest = deduped.request({
+        key,
+        metadata: callbackMetadata,
+        requestFunc: makeRequest,
+        callback,
+        fromQueue: false
+    });
+
+    return dedupedAndQueuedRequest.cancel;
 }

src/source/vector_tile_source.ts

@@ -290,7 +290,7 @@ class VectorTileSource extends Evented<SourceEvents> implements ISource {
             // if workers are not ready to receive messages yet, use the idle time to preemptively
             // load tiles on the main thread and pass the result instead of requesting a worker to do so
             if (!this.dispatcher.ready) {
-                const cancel = loadVectorTile.call({deduped: this._deduped}, params, (err?: Error | null, data?: LoadVectorTileResult | null) => {
+                const cancel = loadVectorTile.call({}, params, (err?: Error | null, data?: LoadVectorTileResult | null) => {
                     if (err || !data) {
                         done.call(this, err);
                     } else {
@@ -302,7 +302,7 @@ class VectorTileSource extends Evented<SourceEvents> implements ISource {
                         };
                         if (tile.actor) tile.actor.send('loadTile', params, done.bind(this), undefined, true);
                     }
-                }, true);
+                }, this._deduped, true);
                 tile.request = {cancel};
 
             } else {

src/source/vector_tile_worker_source.ts

@@ -69,16 +69,18 @@ class VectorTileWorkerSource extends Evented implements WorkerSource {
      */
     loadTile(params: WorkerTileParameters, callback: WorkerTileCallback) {
         const uid = params.uid;
-
         const requestParam = params && params.request;
         const perf = requestParam && requestParam.collectResourceTiming;
 
         const workerTile = this.loading[uid] = new WorkerTile(params);
         workerTile.abort = this.loadVectorData(params, (err, response) => {
             const aborted = !this.loading[uid];
-
             delete this.loading[uid];
 
+            if (workerTile.status === 'done') {
+                return;
+            }
+
             if (aborted || err || !response) {
                 workerTile.status = 'done';
                 if (!aborted) this.loaded[uid] = workerTile;
@@ -128,7 +130,7 @@ class VectorTileWorkerSource extends Evented implements WorkerSource {
 
             this.loaded = this.loaded || {};
             this.loaded[uid] = workerTile;
-        });
+        }, this.deduped);
     }
 
     /**