Merge pull request #18265 from amcasey/ThrottledOperations

Limit the number of unanswered Typings Installer requests

src/server/server.ts

@@ -236,25 +236,40 @@ namespace ts.server {
         return `${d.getHours()}:${d.getMinutes()}:${d.getSeconds()}.${d.getMilliseconds()}`;
     }
 
+    interface QueuedOperation {
+        operationId: string;
+        operation: () => void;
+    }
+
     class NodeTypingsInstaller implements ITypingsInstaller {
         private installer: NodeChildProcess;
         private installerPidReported = false;
         private socket: NodeSocket;
         private projectService: ProjectService;
-        private throttledOperations: ThrottledOperations;
         private eventSender: EventSender;
+        private activeRequestCount = 0;
+        private requestQueue: QueuedOperation[] = [];
+        private requestMap = createMap<QueuedOperation>(); // Maps operation ID to newest requestQueue entry with that ID
+
+        // This number is essentially arbitrary.  Processing more than one typings request
+        // at a time makes sense, but having too many in the pipe results in a hang
+        // (see https://github.com/nodejs/node/issues/7657).
+        // It would be preferable to base our limit on the amount of space left in the
+        // buffer, but we have yet to find a way to retrieve that value.
+        private static readonly maxActiveRequestCount = 10;
+        private static readonly requestDelayMillis = 100;
+
 
         constructor(
             private readonly telemetryEnabled: boolean,
             private readonly logger: server.Logger,
-            host: ServerHost,
+            private readonly host: ServerHost,
             eventPort: number,
             readonly globalTypingsCacheLocation: string,
             readonly typingSafeListLocation: string,
             readonly typesMapLocation: string,
             private readonly npmLocation: string | undefined,
             private newLine: string) {
-            this.throttledOperations = new ThrottledOperations(host);
             if (eventPort) {
                 const s = net.connect({ port: eventPort }, () => {
                     this.socket = s;
@@ -338,12 +353,26 @@ namespace ts.server {
                     this.logger.info(`Scheduling throttled operation: ${JSON.stringify(request)}`);
                 }
             }
-            this.throttledOperations.schedule(project.getProjectName(), /*ms*/ 250, () => {
+
+            const operationId = project.getProjectName();
+            const operation = () => {
                 if (this.logger.hasLevel(LogLevel.verbose)) {
                     this.logger.info(`Sending request: ${JSON.stringify(request)}`);
                 }
                 this.installer.send(request);
-            });
+            };
+            const queuedRequest: QueuedOperation = { operationId, operation };
+
+            if (this.activeRequestCount < NodeTypingsInstaller.maxActiveRequestCount) {
+                this.scheduleRequest(queuedRequest);
+            }
+            else {
+                if (this.logger.hasLevel(LogLevel.verbose)) {
+                    this.logger.info(`Deferring request for: ${operationId}`);
+                }
+                this.requestQueue.push(queuedRequest);
+                this.requestMap.set(operationId, queuedRequest);
+            }
         }
 
         private handleMessage(response: SetTypings | InvalidateCachedTypings | BeginInstallTypes | EndInstallTypes | InitializationFailedResponse) {
@@ -404,11 +433,39 @@ namespace ts.server {
                 return;
             }
 
+            if (this.activeRequestCount > 0) {
+                this.activeRequestCount--;
+            }
+            else {
+                Debug.fail("Received too many responses");
+            }
+
+            while (this.requestQueue.length > 0) {
+                const queuedRequest = this.requestQueue.shift();
+                if (this.requestMap.get(queuedRequest.operationId) === queuedRequest) {
+                    this.requestMap.delete(queuedRequest.operationId);
+                    this.scheduleRequest(queuedRequest);
+                    break;
+                }
+
+                if (this.logger.hasLevel(LogLevel.verbose)) {
+                    this.logger.info(`Skipping defunct request for: ${queuedRequest.operationId}`);
+                }
+            }
+
             this.projectService.updateTypingsForProject(response);
             if (response.kind === ActionSet && this.socket) {
                 this.sendEvent(0, "setTypings", response);
             }
         }
+
+        private scheduleRequest(request: QueuedOperation) {
+            if (this.logger.hasLevel(LogLevel.verbose)) {
+                this.logger.info(`Scheduling request for: ${request.operationId}`);
+            }
+            this.activeRequestCount++;
+            this.host.setTimeout(request.operation, NodeTypingsInstaller.requestDelayMillis);
+        }
     }
 
     class IOSession extends Session {

src/server/utilities.ts

@@ -179,6 +179,12 @@ namespace ts.server {
         constructor(private readonly host: ServerHost) {
         }
 
+        /**
+         * Wait `number` milliseconds and then invoke `cb`.  If, while waiting, schedule
+         * is called again with the same `operationId`, cancel this operation in favor
+         * of the new one.  (Note that the amount of time the canceled operation had been
+         * waiting does not affect the amount of time that the new operation waits.)
+         */
         public schedule(operationId: string, delay: number, cb: () => void) {
             const pendingTimeout = this.pendingTimeouts.get(operationId);
             if (pendingTimeout) {