Stdout/stderr pause/resume

[vietj]

There is no way to pause/resume the standard output/error when the NuProcessHandler cannot handle the buffers.

It would be a useful addition for the Vert.x 3 integration https://github.com/vietj/vertx-childprocess

Are there plans to implement such feature ?

[brettwooldridge]

@vietj I haven't tried it, but this should work...

If you simply don't read any data when you are called back, the pipe will fill-up and the child process itself will block. You would need to keep track of the fact that you have pending reads, and read the ByteBuffer to create room in the pipe. That will unblock the child process.

Let me know if it works.

[vietj]

ok, you mean I leave the byte buffer untouched. I will let you know.

[vietj]

so that would work to pause the stream, however how do I unpause it and request data again ? i.e how do I get a new callback to drain the pipe ?

[brettwooldridge]

You won't get called back again. Presumably you somehow know when you want to pause and resume, so your callback will need to retain a reference to the ByteBuffer and you will need some other thread or timer to drain the buffer.

It's not exactly clear to me why vertx would need to suspend a stream in the first place...

[vietj]

I see, that sounds doable and I will try.

Vert.x would do that for propagating flow control to another stream : you can read that short part of the doc http://vertx.io/docs/vertx-core/java/#streams

[vietj]

Hi,

I'm looking further at this and the javadoc says You do not own the ByteBuffer provided to you. You should not retain a reference to this buffer. which seems contradictory with retaining a reference to the ByteBuffer.

[brettwooldridge]

@vietj Opps, I apologize, the JavaDoc is out of date. It used to be that buffers were reused across processes, but after refactors by @bhamiltoncx this is no longer true. Each process instance has its own stdout, stderr, and stdin ByteBuffer instances.

[pfxuan]

@brettwooldridge Does NuProcess have any plan to implement Reactive Streams?

[vietj]

@brettwooldridge how can we ensure that the buffer is not modified concurrently by NuProcess eventLoop and a possible drain by another thread ?

[vietj]

@brettwooldridge also now when I access the buffer outside NuProcess event loop I get buffer.remaining() == 0 (the buffer is full) which means the buffer was flipped so when using outside should we flip again this buffer ?

[brettwooldridge]

@pfxuan This is the first that I've heard of the Reactive Streams initiative, but NuProcess would certainly welcome any contribution. If someone were to make such an effort, I would recommend the implementation go into a com.zaxxer.nuprocess.streams or com.zaxxer.nuprocess.reactive package.

[brettwooldridge]

@vietj Can you post your NuProcessHandler implementation, either here or linked as a gist?

[pfxuan]

@brettwooldridge Sounds a good plan! If @vietj can do this, the integration is going to become very natural: http://vertx.io/docs/vertx-reactive-streams/java/

[vietj]

@brettwooldridge here is the handler https://github.com/vietj/vertx-childprocess/blob/stdout-control-flow/src/main/java/io/vertx/ext/childprocess/impl/ChildProcessImpl.java

here is a test https://github.com/vietj/vertx-childprocess/blob/stdout-control-flow/src/test/java/io/vertx/ext/childprocess/SpawnTest.java#L214

if you run this test (on osx at least), after resume the buffer size is 0.

[vietj]

@pfxuan that's something possible, however it would be best to come first with a solution that works with the current api :-) (unless I'm using not well the current API).

[pfxuan]

@vietj Agree! I'm looking forward to seeing the first prototype. This integration is very useful.

[brettwooldridge]

@vietj Ok, looking at the code, this is going to be a little tricky. ByteBuffer is not a thread-safe class, so having multiple threads (NuProcess's and yours) accessing the buffer (at the same time) is going to be trouble.

NuProcess is going to compact the buffer after calling your handler. The contract is that the handler should read as much as it can. If you want to create back-pressure, to block the spawned process from writing more data, you need to let that buffer fill-up. Which you seem to be doing. If you flip the buffer, the back-pressure goes away, so don't do that.

If your handler is called, and remaining != 0, it is not safe to interact with the ByteBuffer from another thread. Once your handler is called with remaining == 0, it should be safe to interact with the ByteBuffer. Note, once remaining == 0, you may not ever get called again (but you may be called multiple times with remaining == 0).

My concern now is, without a change in NuProcess, the NuProcess thread may spin -- consuming lots of CPU.

[brettwooldridge]

@vietj The more I think about it, the more I think NuProcess will need to support a suspend()/resume() of some sort. The epoll (Linux), kqueue (Mac), and IO Completion Port (Windows) triggers need to be deregistered so that they aren't constantly firing that data is available -- but nothing is reading it -- causing NuProcess to spin and consume CPU.

[vietj]

@brettwooldridge indeed when using the ByteBuffer outside of NuProcess event loop sounds like an abuse and pause/resume should be supported natively. Whenever you come up with such feature, I can contribute a reactive-streams implementation on top of it. WDYT ?

[brettwooldridge]

I'm reviewing the specifications now. I'll let you know shortly.

[brettwooldridge]

@vietj @pfxuan I will implement the Reactive Streams specification. Please allow a week or so for me to cleanly refactor some of the internals.

@bhamiltoncx As I get closer, I may ask for your input. In particular, NuProcess does not elegantly implement back-pressure -- though I have a reasonably well-formed idea how to do it. What is missing from NuProcess is something that more closely mirrors the epoll/kqueue model that it is wrapping -- ie. while there is wantWrite() to express intention to write to stdin, there is no corresponding wantRead() to express intention to read (or not read). stdout/stderr are basically a hose with no shutoff -- we expect the implementer of the NuProcessHandler to keep-up, buffer, or otherwise discard the data the process is shoving at it.

This is the part that will change. I will attempt to do so in the least intrusive way possible to the API.

[pfxuan]

@brettwooldridge I'm so excited for receiving your direct support!

[bhamiltoncx]

Yeah, happy to help brainstorm this.

I think we'll want to generalize and extend the wantWrite()
infrastructure to allow manipulating (and querying?) the paused / running
state of each of the I/O streams.

On Fri, Nov 6, 2015 at 8:01 AM Pengfei Xuan [email protected]
wrote:

@brettwooldridge https://github.com/brettwooldridge I'm so excited for
receiving your direct support!

—
Reply to this email directly or view it on GitHub
#53 (comment)
.

[brettwooldridge]

I have completed the initial implementation of Reactive Streams in a branch called streams. The principal classes are in the com.zaxxer.nuprocess.streams package.

@pfxuan @vietj Would you mind creating some junit tests for com.zaxxer.nuprocess.streams? Take a look at NuStreamProcessBuilder as a place to start.

@bhamiltoncx There is a "substantial" (not really) change in the core API. NuProcess now requires expressing wants not only for STDIN, but now also all STDOUT and STDERR. And similar to the previous NuProcessHandler.onStdinReady() callback, now both onStdout() and onStderr() have a boolean return value that expresses whether more data is desired. In order to maintain current behavior, the simplest implementation is to return !closed; from those methods. Because you probably use the codecs, you may not need to change anything on your side (Buck). But your review is helpful.

Oh, and the streams branch does not currently run on Windows (haven't even looked at it yet). I will need to add support in Windows to mimic the "oneshot" events that epoll and kqueue provide.

[vietj]

@brettwooldridge good job I will have a look soon. I see that the implementation itself does not need reactive-streams lib out of the box (i.e one can use it without the reactive-streams lib) : would you mind to declare the reactive-streams library as optional=true ?

[brettwooldridge]

No problem on the optional dependency.

[pfxuan]

@brettwooldridge So awesome! I'll look at it and see if I can do something.

[pfxuan]

@brettwooldridge I can not find the implementation on specification of 3.10 in your NuStreamSubscription#request(). It looks like 3.10 requires a synchronous call on subscriber's onNext().

[brettwooldridge]

@pfxuan I pushed a minor change to streams to ensure that after onComplete() is called that no more Subscription.request(n) calls will be honored.

[pfxuan]

@brettwooldridge Got it. Do you know how to setup the number of elements on Publisher? I'm trying to construct a scenario where Publisher has a bounded amount of elements: NaivePublisher

[brettwooldridge]

@pfxuan Oh, maybe I missed it, but the user (you) does not construct a Publisher. NuProcess is the publisher. You only need to implement a Subscriber.

Construction goes something like this:

NuProcessBuilder builder = new NuProcessBuilder(commands);
NuStreamProcessBuilder streamBuilder = new NuStreamProcessBuilder(builder);
NuStreamProcess process = streamBuilder.start();

NuStreamPublisher stdoutPublisher = process.getStdoutPublisher();

NaiveSubscriber subscriber = new NaiveSubscriber();
stdoutPublisher.subscribe(subscriber);
...
process.waitFor(0, TimeUnit.Seconds); // wait until the process exists

NaiveSubscriber:

public class NaiveSubscriber implements Subscriber<ByteBuffer> {
   private Executor executor;
   private Subscription subscription;

   public NaiveSubscriber() {
      executor = Executors.newSingleThreadedExecutor();
   }

   public onSubscribe(Subscription sub) {
      subscription = sub;
      executor( () -> { subscription.request(1) } );
   }

   public void onNext(ByteBuffer buffer) {
      // consume buffer
      executor( () -> { subscription.request(1) } );
   }
   ...

[pfxuan]

@brettwooldridge How does NuProcess handle back-pressure? Say the command 'cat in.txt' is able to produce 100MB/s textural traffic, but my NaiveSubscriber only can accept/process 30MB/s textual data. Can NuProcess pause or stall the "cat" process internally when receiving a back-pressure signal from Subscriber?

[brettwooldridge]

@pfxuan Another minor set of commits to streams.

Also, minor edits to the above example code (the no-arg start() method was added).

With respect to a unit test with a bounded number of published elements, consider something like the CatTest using a text file as a parameter to the cat command. You should get a limited amount of data, and eventually if you keep calling request(1) your onComplete() method should be called.

[pfxuan]

@brettwooldridge Yeah! This is a hot feature I need now.

[brettwooldridge]

@pfxuan Yes, there is a pipe (operating system pipe) between the process and NuProcess. This OS pipe has limited capacity, typically a few kilobytes. When the process (for example cat) tries to write more data into the pipe, but NuProcess is not pulling it out of the other end, the cat process will stall. It will be blocked at the OS level until NuProcess drains some of the data from the pipe, making space for more.

[vietj]

to ease implementation of subscriber, wouldn't it make sense to configure the NuProcess to have buffers of a fixed (configurable) size ?

[brettwooldridge]

@vietj It may make sense, but it also adds additional configuration and testing vectors that I would rather not have unless absolutely necessary. I tend to lean toward a very small API surface area, and well published capabilities and limitations. Most programmers are sophisticated enough to work within constraints if they are aware of them.

Note that NuProcess already advertises buffer capacity. And due to the nature of asynchronous I/O, the onNext(ByteBuffer) callback might get anywhere from 1 byte of data up to NuProcess.BUFFER_CAPACITY bytes of data. At a default of 64k, it is unlikely to overwhelm a subscriber, and the subscriber has the opportunity to request more data or wait until it is ready to handle more.

[pfxuan]

@brettwooldridge I see. Under the hood, NuProcess still uses Unix Pipes to do interprocess communication. This is why you haven't implemented Windows version yet. Is that correct?

[brettwooldridge]

@pfxuan Mostly correct. Windows works fine (pre-streams) using Windows pipes and IO Completion Ports. Linux uses pipes and epoll. Mac OS X and FreeBSD use pipes and kqueue.

However, in order to support back-pressure, I had to change how NuProcess used epoll and kqueue slightly. Windows can do it as well, I just haven't taken the time to change it yet. It is probably about 4-8 hours of work, and I'll probably get it done this week.

[brettwooldridge]

@pfxuan Just wanted to let you know, if something is not working like it seems it should, there is not necessarily something wrong with your unit test. Because the streams implementation is completely untested at this point, it is possible there are bugs in NuProcess.

[vietj]

@pfxuan reactive-streams has a TCK : https://github.com/reactive-streams/reactive-streams-jvm/tree/master/tck/src/main/java/org/reactivestreams/tck perhaps it can be implemented in this project. I know it is implemented in vertx-reactive-streams project.

[pfxuan]

@brettwooldridge Got it! I'll let you know when I find the problem.

[pfxuan]

@vietj Yes! I happened to know the TCK this morning and also found a simple test example from rstoyanchev's talk. Would you like to add TCK tests for NuProcess? I see you are a member of vertx-reactive-streams and should have a much strong background in this part.

[brettwooldridge]

@pfxuan I wrote a single unit test, and found a few bugs. They're fixed. You can update your version of the streams branch if you want the latest.

The test is here.

[vietj]

I will update soon the vertx-childprocess project to use it and provide feedback - it will consume mainly the original api as Vert.x is based on pause/resume backpressure system and not request system.

[pfxuan]

@brettwooldridge @vietj I integrated junit tests with the TCK test to get a further verification on NuStreamPublisher. The TCK detects lots of low-level details for us based-on specification rules. So far, I got 11 passed, 12 skipped and 13 failed tests on NuStreamPublisher. After unit tests, you can look through the detailed reports at path target/surefire-reports/testng-native-results/emailable-report.html. Since this is an initial commit to work with TCK, it could include some bugs and design issues. For example, the verification on Publisher requires Publisher can support the number of elements. So I did a minor change on NuStreamPublisher to match the requirement, I'm not sure if this is a feasible solution. Please help me check.

You can review my changes at pfxuan@d9adac6.

[brettwooldridge]

@pfxuan Thanks for the start. Note that the requirements of the TCK framework itself are different than the Specification per se. The specification does not require that a Publisher accept a number of elements during construction. However, the TCK requires that a Publisher used within the TCK supports that, to simplify testing by the TCK harness.

I have modified the TckFiniteStdoutPublisherTest class to contain a publisher that is useful for use within the TCK. It accepts a number of elements during construction, and proxies to the NuProcess publisher.

Note that the tests are still failing, at least on OS X (I have not tried the TCK on Linux yet). The tests have uncovered further errors on OS X that will need to be fixed before they can pass.

[pfxuan]

@brettwooldridge Your modification is the way better than my original version. Especially, the using of proxy pattern made everything so clear. I was trying to write a self-contained test, but didn't get it. Thanks for pointing it out!

In addition, I have tried the modified and original tests on OS X and Linux and summarized the failed cases as followings:

1. Modified version on OS X (8 Passed, 13 Skipped, 15 Failed):
   required_createPublisher1MustProduceAStreamOfExactly1Element
   required_createPublisher3MustProduceAStreamOfExactly3Elements
   required_spec101_subscriptionRequestMustResultInTheCorrectNumberOfProducedElements
   required_spec102_maySignalLessThanRequestedAndTerminateSubscription
   required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates
   required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled
   required_spec109_mustIssueOnSubscribeForNonNullSubscriber
   required_spec306_afterSubscriptionIsCancelledRequestMustBeNops
   required_spec307_afterSubscriptionIsCancelledAdditionalCancelationsMustBeNops
   required_spec309_requestNegativeNumberMustSignalIllegalArgumentException
   required_spec309_requestZeroMustSignalIllegalArgumentException
   required_spec313_cancelMustMakeThePublisherEventuallyDropAllReferencesToTheSubscriber
   required_spec317_mustSupportACumulativePendingElementCountUpToLongMaxValue
   required_spec317_mustSupportAPendingElementCountUpToLongMaxValue
   stochastic_spec103_mustSignalOnMethodsSequentially
2. Modified version on Linux (8 Passed, 13 Skipped, 15 Failed):
   required_createPublisher1MustProduceAStreamOfExactly1Element
   required_createPublisher3MustProduceAStreamOfExactly3Elements
   required_spec101_subscriptionRequestMustResultInTheCorrectNumberOfProducedElements
   required_spec102_maySignalLessThanRequestedAndTerminateSubscription
   required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates
   required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled
   required_spec109_mustIssueOnSubscribeForNonNullSubscriber
   required_spec306_afterSubscriptionIsCancelledRequestMustBeNops
   required_spec307_afterSubscriptionIsCancelledAdditionalCancelationsMustBeNops
   required_spec309_requestNegativeNumberMustSignalIllegalArgumentException
   required_spec309_requestZeroMustSignalIllegalArgumentException
   required_spec313_cancelMustMakeThePublisherEventuallyDropAllReferencesToTheSubscriber
   required_spec317_mustSupportACumulativePendingElementCountUpToLongMaxValue
   required_spec317_mustSupportAPendingElementCountUpToLongMaxValue
   stochastic_spec103_mustSignalOnMethodsSequentially
3. Original version on OS X (11 Passed, 12 Skipped, 13 Failed):
   required_createPublisher1MustProduceAStreamOfExactly1Element
   required_createPublisher3MustProduceAStreamOfExactly3Elements
   required_spec102_maySignalLessThanRequestedAndTerminateSubscription
   required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates
   required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled
   required_spec302_mustAllowSynchronousRequestCallsFromOnNextAndOnSubscribe
   required_spec309_requestNegativeNumberMustSignalIllegalArgumentException
   required_spec309_requestZeroMustSignalIllegalArgumentException
   required_spec313_cancelMustMakeThePublisherEventuallyDropAllReferencesToTheSubscriber
   required_spec317_mustNotSignalOnErrorWhenPendingAboveLongMaxValue
   required_spec317_mustSupportACumulativePendingElementCountUpToLongMaxValue
   required_spec317_mustSupportAPendingElementCountUpToLongMaxValue
   stochastic_spec103_mustSignalOnMethodsSequentially
4. Original version on Linux (10 Passed, 12 Skipped, 14 Failed):
   required_createPublisher1MustProduceAStreamOfExactly1Element
   required_createPublisher3MustProduceAStreamOfExactly3Elements
   required_spec102_maySignalLessThanRequestedAndTerminateSubscription
   required_spec105_mustSignalOnCompleteWhenFiniteStreamTerminates
   required_spec107_mustNotEmitFurtherSignalsOnceOnCompleteHasBeenSignalled
   required_spec302_mustAllowSynchronousRequestCallsFromOnNextAndOnSubscribe
   required_spec309_requestNegativeNumberMustSignalIllegalArgumentException
   required_spec309_requestZeroMustSignalIllegalArgumentException
   required_spec312_cancelMustMakeThePublisherToEventuallyStopSignaling
   required_spec313_cancelMustMakeThePublisherEventuallyDropAllReferencesToTheSubscriber
   required_spec317_mustNotSignalOnErrorWhenPendingAboveLongMaxValue
   required_spec317_mustSupportACumulativePendingElementCountUpToLongMaxValue
   required_spec317_mustSupportAPendingElementCountUpToLongMaxValue
   stochastic_spec103_mustSignalOnMethodsSequentially

Any thoughts and suggestions for the next moving?

[brettwooldridge]

@pfxuan Until I get a chance to revisit some of the NuProcess internals, there may not be much to be done. It will take a day or three.

[pfxuan]

@brettwooldridge Take your time and feel free to assign a task to me if you need any help.

[brettwooldridge]

@pfxuan All tests are now passing on Linux and OS X.

[pfxuan]

@brettwooldridge it's super awesome!!! I'll test it and let you know if there is a problem.

[pfxuan]

@brettwooldridge I want to do blackbox and whitebox verifications on NuProcess's subscriber. But I found STDIN is actually working as publisher. It seems STDIN should work as subscriber rather than publisher, right?

void setSubscriber(final Stream stream, final Subscriber<? super ByteBuffer> subscriber)
{
   switch (stream)
   {
      case STDIN:
         stdinSubscriber = subscriber;
         stdinRequests.set(0);
         break;
      case STDOUT:
         stdoutSubscriber = subscriber;
         stdoutRequests.set(0);
         break;
      case STDERR:
         stderrSubscriber = subscriber;
         stderrRequests.set(0);
         break;
   }
}

[vietj]

@brettwooldridge current I see a bug with stderr stream that prevents the stderr close and exit to be called on OSX, here is a reproducer:

    NuProcess process = new NuProcessBuilder(new NuProcessHandler() {
      @Override
      public void onPreStart(NuProcess nuProcess) {}

      @Override
      public void onStart(NuProcess nuProcess) {
        System.out.println("started");
        nuProcess.want(NuProcess.Stream.STDERR);
      }

      @Override
      public void onExit(int exitCode) {
        System.out.println("Exit");
      }

      @Override
      public boolean onStdout(ByteBuffer buffer, boolean closed) {
        return false;
      }

      @Override
      public boolean onStderr(ByteBuffer buffer, boolean closed) {
        int len = buffer.remaining();
        byte[] bytes = new byte[len];
        buffer.get(bytes);
        System.out.println("buffer=" + new String(bytes));
        System.out.println("closed=" + closed);
        return true;
      }

      @Override
      public boolean onStdinReady(ByteBuffer buffer) {
        return false;
      }
    }, Arrays.asList("/usr/bin/java", "-cp", "target/test-classes", "EchoStderr", "the_echoed_value")).start();

    process.waitFor(10, TimeUnit.SECONDS);

The EchoStderr is:

public class EchoStderr {

  public static void main(String[] args) {
    for (String arg : args) {
      System.err.print(arg);
    }
  }
}

Exit is never called, nor a callback with closed=true.

This does not happen with stdout.

[huntc]

I think this issue has conflated both reactive streams support and the support for back pressure. I'd like to draw it back to its original discussion around just providing support for back pressure and let #71 deal with reactive streams.

My view is that it'd be great to have just have a wantRead method on the process in order to signal that we're ready to read again. My suggestion would be for the caller to first signal the desire for this behaviour when building the process. In the case where the ByteBuffer hasn't been consumed on return from onStdout or onStderr, then the process should wait for wantRead before making the calls again. WDYT? Something like that anyhow...

[huntc]

Another (perhaps better) solution would be to improve NuProcess so that it waits up to 1 second (say) before calling the stdout/err callbacks again. Meanwhile, if wantRead is called then this cancels the associated timers... the solution would be backwardly compatible and support back-pressure.