probe queue for activity on network io to prevent stuck message - fixes #139 (#141)

Author: Sebastian McKenzie

src/resolvers/exotics/_hosted-git.js

@@ -90,7 +90,8 @@ export default class HostedGitResolver extends ExoticResolver {
     let client = new Git(this.config, url, this.hash);
 
     let out = await this.config.requestManager.request({
-      url: `${url}/info/refs?service=git-upload-pack`
+      url: `${url}/info/refs?service=git-upload-pack`,
+      queue: this.resolver.fetchingQueue
     });
 
     if (out) {
@@ -120,11 +121,12 @@ export default class HostedGitResolver extends ExoticResolver {
     let self = this; // TODO: babel bug...
     let commit = await this.getRefOverHTTP(url);
 
-    async function tryRegistry(registry) {
+    let tryRegistry = async (registry) => {
       let filenames = registries[registry].filenames;
       for (let filename of filenames) {
         let file = await self.config.requestManager.request({
-          url: self.constructor.getHTTPFileUrl(self.exploded, filename, commit)
+          url: self.constructor.getHTTPFileUrl(self.exploded, filename, commit),
+          queue: this.resolver.fetchingQueue
         });
         if (!file) continue;
 
@@ -139,7 +141,7 @@ export default class HostedGitResolver extends ExoticResolver {
 
         return json;
       }
-    }
+    };
 
     let file = await tryRegistry(this.registry);
     if (file) return file;
@@ -155,7 +157,11 @@ export default class HostedGitResolver extends ExoticResolver {
   }
 
   async hasHTTPCapability(url: string): Promise<boolean> {
-    return (await this.config.requestManager.request({ url, method: "HEAD" })) !== false;
+    return (await this.config.requestManager.request({
+      url,
+      method: "HEAD",
+      queue: this.resolver.fetchingQueue
+    })) !== false;
   }
 
   async resolve(): Promise<Manifest> {

src/resolvers/registries/bower.js

@@ -22,7 +22,8 @@ export default class BowerResolver extends RegistryResolver {
   }> {
     return this.config.requestManager.request({
       url: `${this.registryConfig.registry}/packages/${this.name}`,
-      json: true
+      json: true,
+      queue: this.resolver.fetchingQueue
     });
   }
 

src/util/blocking-queue.js

@@ -22,6 +22,9 @@ export default class BlockingQueue {
 
     this.running = map();
     this.queue   = map();
+
+    // $FlowFixMe: for performance we refer to this in `stillActive`
+    this.stuckTick = this.stuckTick.bind(this);
   }
 
   concurrencyQueue: Array<Function>;
@@ -44,27 +47,29 @@ export default class BlockingQueue {
     [key: string]: boolean
   };
 
-  startStuckTimer() {
+  stillActive() {
     if (this.stuckTimer) {
       clearTimeout(this.stuckTimer);
     }
 
-    this.stuckTimer = setTimeout(() => {
-      if (this.runningCount === 1) {
-        this.warnedStuck = true;
-        console.warn(
-          `[fbkpm] The ${JSON.stringify(this.alias)} blocking queue may be stuck. 5 seconds ` +
-          `without any activity with 1 worker: ${Object.keys(this.running)[0]}`
-        );
-      }
-    }, 5000);
+    this.stuckTimer = setTimeout(this.stuckTick, 5000);
+  }
+
+  stuckTick() {
+    if (this.runningCount === 1) {
+      this.warnedStuck = true;
+      console.warn(
+        `[fbkpm] The ${JSON.stringify(this.alias)} blocking queue may be stuck. 5 seconds ` +
+        `without any activity with 1 worker: ${Object.keys(this.running)[0]}`
+      );
+    }
   }
 
   push<T>(key: string, factory: () => Promise<T>): Promise<T> {
     if (this.first) {
       this.first = false;
     } else {
-      this.startStuckTimer();
+      this.stillActive();
     }
 
     return new Promise((resolve, reject) => {

src/util/request-manager.js

@@ -10,6 +10,7 @@
  */
 
 import type { Reporter } from "../reporters/index.js";
+import BlockingQueue from "./blocking-queue.js";
 import * as constants from "../constants.js";
 import * as network from "./network.js";
 import map from "../util/map.js";
@@ -28,6 +29,7 @@ declare class RequestError extends Error {
 type RequestParams<T> = {
   url: string,
   method?: "GET" | "HEAD" | "POST" | "PUT",
+  queue?: BlockingQueue,
   json?: boolean,
   forever?: boolean,
   headers?: {
@@ -218,6 +220,9 @@ export default class RequestManager {
       }
     });
 
+    let queue = params.queue;
+    if (queue) req.on("data", queue.stillActive);
+
     if (params.process) {
       params.process(req, resolve, reject);
     }