Make StreamSubscription be the active part of a stream.

pkg/scheduled_test/lib/scheduled_process.dart

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 library scheduled_test.scheduled_process;
 
 import 'dart:async';
 import 'dart:io';
 
 import 'scheduled_test.dart';
@@ skipping 21 matching lines @@
   final Encoding _encoding;
 
   /// The process that's scheduled to run.
   ValueFuture<Process> _process;
 
   /// A fork of [_stdout] that records the standard output of the process. Used
   /// for debugging information.
   Stream<String> _stdoutLog;
 
   /// A line-by-line view of the standard output stream of the process.
-  Stream<String> _stdout;
+  StreamIterator<String> _stdout;

[nweiz, 2013/05/22 18:42:41]

I'm not sure I understand why using a StreamIterator here is better than using a
Stream. It makes the implementation of [remainingStdout] and [remainingStderr]
more complicated, and I don't see a corresponding benefit.

[floitsch, 2013/05/22 20:23:35]

The moment you cancel a stream it is dead. We have had a bug for some time where
you could still subscribe to it.
This CL fixes the bug. Unfortunately there is code that relied on the broken
behavior. The StreamIterator class provides an interface that usually allows to
simulate the old behavior.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Exactly. 
The current code subscribed to the same stream more than once (using streamFirst
one or more time). That does not work any more.

[nweiz, 2013/05/24 20:12:31]

Okay, I suppose that makes sense. It seems a little weird to me that having an
Iterator-style API and supporting multiple subscriptions are conflated concepts,
but that's a different discussion and I don't feel too strongly about it.

If streamFirst is completely broken now, please remove it from utils.dart.

[Lasse Reichstein Nielsen, 2013/05/27 08:04:53]

Done.

 
   /// A canceller that controls both [_stdout] and [_stdoutLog].
   StreamCanceller _stdoutCanceller;
 
   /// A fork of [_stderr] that records the standard error of the process. Used
   /// for debugging information.
   Stream<String> _stderrLog;
 
   /// A line-by-line view of the standard error stream of the process.
-  Stream<String> _stderr;
+  StreamIterator<String> _stderr;
 
   /// A canceller that controls both [_stderr] and [_stderrLog].
   StreamCanceller _stderrCanceller;
 
   /// The exit code of the process that's scheduled to run. This will naturally
   /// only complete once the process has terminated.
   ValueFuture<int> _exitCode;
 
   /// Whether the user has scheduled the end of this process by calling either
   /// [shouldExit] or [kill].
@@ skipping 32 matching lines @@
     var stdoutWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => p.stdout));
     _stdoutCanceller = stdoutWithCanceller.last;
     _stdoutLog = stdoutWithCanceller.first;
 
     var stderrWithCanceller = _lineStreamWithCanceller(
         _process.then((p) => p.stderr));
     _stderrCanceller = stderrWithCanceller.last;
     _stderrLog = stderrWithCanceller.first;
 
-    _stdout = stdoutStream();
-    _stderr = stderrStream();
+    _stdout = new StreamIterator<String>(stdoutStream());
+    _stderr = new StreamIterator<String>(stderrStream());
   }
 
   /// Updates [_description] to reflect [executable] and [arguments], which are
   /// the same values as in [start].
   void _updateDescription(executable, arguments) {
     if (_explicitDescription) return;
     if (executable is Future) {
       _description = "future process";
     } else if (arguments is Future || arguments.any((e) => e is Future)) {
       _description = executable;
@@ skipping 115 matching lines @@
             "Results of running '$description':\n"
             "$exitDescription\n"
             "Standard output:\n"
             "${prefixLines(stdout)}\n"
             "Standard error:\n"
             "${prefixLines(stderr)}");
       });
     }, "cleaning up process '$description'");
   }
 
+  // Similar to [streamFirst], but works on streamIterators.
+  // And it works at all, which streamFirst doesn't.

[floitsch, 2013/05/22 16:26:29]

Reword. This could be read like an insult.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Ack, working comment left in :)

+  Future<String> _streamIteratorFirst(StreamIterator<String> streamIterator) {

[nweiz, 2013/05/22 18:42:41]

This belongs in utils.dart.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Ok, will move it.

+    StackTrace stackTrace;
+    try {
+      throw 0;
+    } catch (_, thrownStackTrace) {
+      stackTrace = thrownStackTrace;
+    }

[Bob Nystrom, 2013/05/22 16:23:46]

Instead of doing this by hand, use new Trace.current() in the stack_trace
package.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Will do.

+    return streamIterator.moveNext().then((hasNext) {
+      if (hasNext) {
+        return streamIterator.current;
+      } else {
+        return new Future.error(new StateError("No elements"), stackTrace);
+      }
+    });
+  }
+
   /// Reads the next line of stdout from the process.
-  Future<String> nextLine() => schedule(() => streamFirst(_stdout),
+  Future<String> nextLine() => schedule(() => _streamIteratorFirst(_stdout),
       "reading the next stdout line from process '$description'");
 
   /// Reads the next line of stderr from the process.
-  Future<String> nextErrLine() => schedule(() => streamFirst(_stderr),
+  Future<String> nextErrLine() => schedule(() => _streamIteratorFirst(_stderr),
       "reading the next stderr line from process '$description'");
 
+  /// Collects all remaining lines from a [StreamIterator] of lines.
+  ///
+  /// Returns the concatenation of the collected lines joined by newlines.
+  static Future<String> _concatRest(StreamIterator<String> streamIterator) {

[nweiz, 2013/05/22 18:42:41]

This also belongs in utils.dart.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Done.

+    Future<String> collectAll(List<String> accumulator) {
+      return streamIterator.moveNext().then((bool hasNext) {

[Bob Nystrom, 2013/05/22 16:23:46]

Don't bother type annotating parameters in fn expressions.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

ok.
Grudgingly.
How else would I know the type of hasNext from looking at the code? :(

[Bob Nystrom, 2013/05/24 15:54:38]

1. You know what you're passing that callback to, so you probably know the fn
type it expects.
2. It's name starts with "has" -> probably a bool.
3. The type doesn't matter that much, its name tells you what it is.

[Lasse Reichstein Nielsen, 2013/05/27 08:04:53]

I don't believe in "knowing" being good enough as documentation.
Writing code is easy. Reading it is hard.
You can't convince me that types are irrelevant. Redundancy is *good* when it
comes to documentation. It dispels any doubt you may have over either of the
parts in isolation.

+        if (hasNext) {
+          accumulator.add(streamIterator.current);
+          return collectAll(accumulator);

[Bob Nystrom, 2013/05/22 16:23:46]

If the stream has a lot of output, this can build a very long chain of futures.
They're async, so they won't blow the stack, but I've seen long future chains
exhaust the heap or consume a lot of CPU traversing the chain.

Maybe we should look to add something reduce-like to dart:async specifically to
address this? You can handle it yourself using a completer, but it's tedious and
error-prone to do so.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

I've rewritten this to not chain the futures, and just complete a completer at
the end.

[nweiz, 2013/05/24 20:12:31]

Working around this here before we know it's actually a problem seems like
premature optimization. So far (as far as I know) the only case where we've seen
this actually become an issue is in pathologically-exponential cases in the Pub
version solver. I don't want to add a workaround here unless we've got good
reason to believe the issue will come up in practice.

If it does come up in practice, that means that Futures are memory-hungry enough
that something like 10,000 of them can blow the heap. This seems like a serious
problem external to this code. In general, it's not unreasonable to want to
write code that iterates over a collection or stream and creates a Future for
each member. Sometimes it's unavoidable.

It's going to be untenable to try to identify every case where a Future chain
could become sufficiently long to tickle this issue, even for those of us aware
of it. For people coming to dart:async for the first time, there's no way
they'll intuit that this issue exists. We need to find a more general solution.

[Lasse Reichstein Nielsen, 2013/05/27 08:04:53]

In this case, I actually think the new version is also more readable, so it's a
pure win.

In general, I doubt memory is going to be the constraining factor, but stack
space might.

We can consider special casing future-chaining, so if a future.then returns a
new future, the original future isn't just made a listener on the new future -
instead all its listeners, current and future, are moved to the new future
directly. That should reduce the nesting.

+        } else {
+          return accumulator.join("\n");

[Bob Nystrom, 2013/05/22 16:23:46]

A StringBuffer seems like a more natural fit for this than a List.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Good point.

+        }
+      });
+    }
+    return collectAll(<String>[]);
+  }
+
   /// Reads the remaining stdout from the process. This should only be called
   /// after kill() or shouldExit().
   Future<String> remainingStdout() {
     if (!_endScheduled) {
       throw new StateError("remainingStdout() should only be called after "
           "kill() or shouldExit().");
     }
-
-    return schedule(() => _stdout.toList().then((lines) => lines.join("\n")),
+    List<String> accumulator = <String>[];

[Bob Nystrom, 2013/05/22 16:23:46]

Unused.

[floitsch, 2013/05/22 16:26:29]

Looks unused.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

Well spotted.

+    return schedule(() => _concatRest(_stdout),
         "reading the remaining stdout from process '$description'");
   }
 
   /// Reads the remaining stderr from the process. This should only be called
   /// after kill() or shouldExit().
   Future<String> remainingStderr() {
     if (!_endScheduled) {
       throw new StateError("remainingStderr() should only be called after "
           "kill() or shouldExit().");
     }
 
-    return schedule(() => _stderr.toList().then((lines) => lines.join("\n")),
+    return schedule(() => _concatRest(_stderr),
         "reading the remaining stderr from process '$description'");
   }
 
   /// Returns a stream that will emit anything the process emits via the
   /// process's standard output from now on.
   ///
   /// This stream will be independent from any other methods that deal with
   /// standard output, including other calls to [stdoutStream].
   ///
   /// This can be overridden by subclasses to return a derived standard output
@@ skipping 55 matching lines @@
     schedule(() {
       _endExpected = true;
       return _exitCode.then((exitCode) {
         if (expectedExitCode != null) {
           expect(exitCode, equals(expectedExitCode));
         }
       });
     }, "waiting for process '$description' to exit");
   }
 }