Use a pool to restrict access to file descriptors in pub.

sdk/lib/_internal/pub/lib/src/io.dart

Index: sdk/lib/_internal/pub/lib/src/io.dart
diff --git a/sdk/lib/_internal/pub/lib/src/io.dart b/sdk/lib/_internal/pub/lib/src/io.dart
index 89179ebdac869de0b23d9f0d32789c3c08526c95..6a5ab568c57b454bdbe411769b76a641970a6217 100644
--- a/sdk/lib/_internal/pub/lib/src/io.dart
+++ b/sdk/lib/_internal/pub/lib/src/io.dart
@@ -16,11 +16,20 @@ import 'package:stack_trace/stack_trace.dart';
 
 import 'error_group.dart';
 import 'log.dart' as log;
+import 'pool.dart';
 import 'sdk.dart' as sdk;
 import 'utils.dart';
 
 export 'package:http/http.dart' show ByteStream;
 
+/// The pool used for restricting access to asynchronous operations that consume
+/// file descriptors.
+///
+/// The maximum number of allocated descriptors is based on empirical tests that
+/// indicate that beyond 32, additional file reads don't provide substantial
+/// additional throughput.
+final _descriptorPool = new Pool(32);
+
 /// Returns whether or not [entry] is nested somewhere within [dir]. This just
 /// performs a path comparison; it doesn't look at the actual filesystem.
 bool isBeneath(String entry, String dir) {
@@ -179,9 +188,11 @@ String writeBinaryFile(String file, List<int> contents) {
 Future<String> createFileFromStream(Stream<List<int>> stream, String file) {
   log.io("Creating $file from stream.");
 
-  return stream.pipe(new File(file).openWrite()).then((_) {
-    log.fine("Created $file from stream.");
-    return file;
+  return _descriptorPool.withResource(() {
+    return stream.pipe(new File(file).openWrite()).then((_) {
+      log.fine("Created $file from stream.");
+      return file;
+    });
   });
 }
 
@@ -486,21 +497,23 @@ Future store(Stream stream, EventSink sink,
 /// the inherited variables.
 Future<PubProcessResult> runProcess(String executable, List<String> args,
     {workingDir, Map<String, String> environment}) {
-  return _doProcess(Process.run, executable, args, workingDir, environment)
-      .then((result) {
-    // TODO(rnystrom): Remove this and change to returning one string.
-    List<String> toLines(String output) {
-      var lines = splitLines(output);
-      if (!lines.isEmpty && lines.last == "") lines.removeLast();
-      return lines;
-    }
+  return _descriptorPool.withResource(() {
+    return _doProcess(Process.run, executable, args, workingDir, environment)
+        .then((result) {
+      // TODO(rnystrom): Remove this and change to returning one string.
+      List<String> toLines(String output) {
+        var lines = splitLines(output);
+        if (!lines.isEmpty && lines.last == "") lines.removeLast();
+        return lines;
+      }
 
-    var pubResult = new PubProcessResult(toLines(result.stdout),
-                                toLines(result.stderr),
-                                result.exitCode);
+      var pubResult = new PubProcessResult(toLines(result.stdout),
+                                  toLines(result.stderr),
+                                  result.exitCode);
 
-    log.processResult(executable, pubResult);
-    return pubResult;
+      log.processResult(executable, pubResult);
+      return pubResult;
+    });
   });
 }
 
@@ -511,9 +524,16 @@ Future<PubProcessResult> runProcess(String executable, List<String> args,
 /// [environment] is provided, that will be used to augment (not replace) the
 /// the inherited variables.
 Future<PubProcess> startProcess(String executable, List<String> args,
-    {workingDir, Map<String, String> environment}) =>
-  _doProcess(Process.start, executable, args, workingDir, environment)
-      .then((process) => new PubProcess(process));
+    {workingDir, Map<String, String> environment}) {
+  return _descriptorPool.request().then((resource) {
+    return _doProcess(Process.start, executable, args, workingDir, environment)
+        .then((ioProcess) {
+      var process = new PubProcess(ioProcess);
+      process.exitCode.whenComplete(resource.release);
+      return process;
+    });
+  });
+}
 
 /// A wrapper around [Process] that exposes `dart:async`-style APIs.
 class PubProcess {
@@ -668,28 +688,31 @@ Future withTempDir(Future fn(String path)) {
 Future<bool> extractTarGz(Stream<List<int>> stream, String destination) {
   log.fine("Extracting .tar.gz stream to $destination.");
 
-  if (Platform.operatingSystem == "windows") {
-    return _extractTarGzWindows(stream, destination);
-  }
-
-  return startProcess("tar",
-      ["--extract", "--gunzip", "--directory", destination]).then((process) {
-    // Ignore errors on process.std{out,err}. They'll be passed to
-    // process.exitCode, and we don't want them being top-levelled by
-    // std{out,err}Sink.
-    store(process.stdout.handleError((_) {}), stdout, closeSink: false);
-    store(process.stderr.handleError((_) {}), stderr, closeSink: false);
-    return Future.wait([
-      store(stream, process.stdin),
-      process.exitCode
-    ]);
-  }).then((results) {
-    var exitCode = results[1];
-    if (exitCode != 0) {
-      throw new Exception("Failed to extract .tar.gz stream to $destination "
-          "(exit code $exitCode).");
+  return _descriptorPool.withResource(() {
+    if (Platform.operatingSystem == "windows") {
+      return _extractTarGzWindows(stream, destination);
     }
-    log.fine("Extracted .tar.gz stream to $destination. Exit code $exitCode.");
+
+    return startProcess("tar",
+        ["--extract", "--gunzip", "--directory", destination]).then((process) {
+      // Ignore errors on process.std{out,err}. They'll be passed to
+      // process.exitCode, and we don't want them being top-levelled by
+      // std{out,err}Sink.
+      store(process.stdout.handleError((_) {}), stdout, closeSink: false);
+      store(process.stderr.handleError((_) {}), stderr, closeSink: false);
+      return Future.wait([
+        store(stream, process.stdin),
+        process.exitCode
+      ]);
+    }).then((results) {
+      var exitCode = results[1];
+      if (exitCode != 0) {
+        throw new Exception("Failed to extract .tar.gz stream to $destination "
+            "(exit code $exitCode).");
+      }
+      log.fine("Extracted .tar.gz stream to $destination. Exit code "
+          "$exitCode.");
+    });
   });
 }
 
@@ -750,70 +773,60 @@ Future<bool> _extractTarGzWindows(Stream<List<int>> stream,
 /// considered to be [baseDir], which defaults to the current working directory.
 /// Returns a [ByteStream] that will emit the contents of the archive.
 ByteStream createTarGz(List contents, {baseDir}) {
-  var buffer = new StringBuffer();
-  buffer.write('Creating .tag.gz stream containing:\n');
-  contents.forEach((file) => buffer.write('$file\n'));
-  log.fine(buffer.toString());
-
-  var controller = new StreamController<List<int>>(sync: true);
-
-  if (baseDir == null) baseDir = path.current;
-  baseDir = path.absolute(baseDir);
-  contents = contents.map((entry) {
-    entry = path.absolute(entry);
-    if (!isBeneath(entry, baseDir)) {
-      throw new ArgumentError('Entry $entry is not inside $baseDir.');
-    }
-    return path.relative(entry, from: baseDir);
-  }).toList();
-
-  if (Platform.operatingSystem != "windows") {
-    var args = ["--create", "--gzip", "--directory", baseDir];
-    args.addAll(contents);
-    // TODO(nweiz): It's possible that enough command-line arguments will make
-    // the process choke, so at some point we should save the arguments to a
-    // file and pass them in via --files-from for tar and -i@filename for 7zip.
-    startProcess("tar", args).then((process) {
-      store(process.stdout, controller);
-    }).catchError((e, stackTrace) {
-      // We don't have to worry about double-signaling here, since the store()
-      // above will only be reached if startProcess succeeds.
-      controller.addError(e, stackTrace);
-      controller.close();
-    });
-    return new ByteStream(controller.stream);
-  }
+  return new ByteStream(futureStream(_descriptorPool.request().then((resource) {
+    return new Future.sync(() {
+      var buffer = new StringBuffer();
+      buffer.write('Creating .tag.gz stream containing:\n');
+      contents.forEach((file) => buffer.write('$file\n'));
+      log.fine(buffer.toString());
+
+      var controller = new StreamController<List<int>>(sync: true);
+
+      if (baseDir == null) baseDir = path.current;
+      baseDir = path.absolute(baseDir);
+      contents = contents.map((entry) {
+        entry = path.absolute(entry);
+        if (!isBeneath(entry, baseDir)) {
+          throw new ArgumentError('Entry $entry is not inside $baseDir.');
+        }
+        return path.relative(entry, from: baseDir);
+      }).toList();
+
+      if (Platform.operatingSystem != "windows") {
+        var args = ["--create", "--gzip", "--directory", baseDir];
+        args.addAll(contents);
+        // TODO(nweiz): It's possible that enough command-line arguments will
+        // make the process choke, so at some point we should save the arguments
+        // to a file and pass them in via --files-from for tar and -i@filename
+        // for 7zip.
+        return startProcess("tar", args).then((process) => process.stdout);
+      }
 
-  withTempDir((tempDir) {
-    // Create the tar file.
-    var tarFile = path.join(tempDir, "intermediate.tar");
-    var args = ["a", "-w$baseDir", tarFile];
-    args.addAll(contents.map((entry) => '-i!$entry'));
-
-    // We're passing 'baseDir' both as '-w' and setting it as the working
-    // directory explicitly here intentionally. The former ensures that the
-    // files added to the archive have the correct relative path in the archive.
-    // The latter enables relative paths in the "-i" args to be resolved.
-    return runProcess(pathTo7zip, args, workingDir: baseDir).then((_) {
-      // GZIP it. 7zip doesn't support doing both as a single operation. Send
-      // the output to stdout.
-      args = ["a", "unused", "-tgzip", "-so", tarFile];
-      return startProcess(pathTo7zip, args);
-    }).then((process) {
-      // Ignore 7zip's stderr. 7zip writes its normal output to stderr. We don't
-      // want to show that since it's meaningless.
-      //
-      // TODO(rnystrom): Should log the stderr and display it if an actual error
-      // occurs.
-      return store(process.stdout, controller);
+      return withTempDir((tempDir) {
+        // Create the tar file.
+        var tarFile = path.join(tempDir, "intermediate.tar");
+        var args = ["a", "-w$baseDir", tarFile];
+        args.addAll(contents.map((entry) => '-i!$entry'));
+
+        // We're passing 'baseDir' both as '-w' and setting it as the working
+        // directory explicitly here intentionally. The former ensures that the
+        // files added to the archive have the correct relative path in the
+        // archive. The latter enables relative paths in the "-i" args to be
+        // resolved.
+        return runProcess(pathTo7zip, args, workingDir: baseDir).then((_) {
+          // GZIP it. 7zip doesn't support doing both as a single operation.
+          // Send the output to stdout.
+          args = ["a", "unused", "-tgzip", "-so", tarFile];
+          return startProcess(pathTo7zip, args);
+        }).then((process) => process.stdout);
+      });
+    }).then((stream) {
+      return stream.transform(onDoneTransformer(() => resource.release()));
+    }).catchError((e) {
+      resource.release();
+      throw e;
     });
-  }).catchError((e, stackTrace) {
-    // We don't have to worry about double-signaling here, since the store()
-    // above will only be reached if everything succeeds.
-    controller.addError(e, stackTrace);
-    controller.close();
-  });
-  return new ByteStream(controller.stream);
+  })));
 }
 
 /// Exception thrown when an operation times out.