RFR: 8294916: Cancelling a request must eventually cause its response body subscriber to be unregistered [v4]

[Daniel Fuchs]

On Sat, 15 Oct 2022 05:54:12 GMT, Jaikiran Pai <jpai at openjdk.org> wrote:

>> Daniel Fuchs has updated the pull request incrementally with one additional commit since the last revision:
>> 
>>   More cleanup to the test
>
> src/java.net.http/share/classes/jdk/internal/net/http/common/HttpBodySubscriberWrapper.java line 71:
> 
>> 69:     }
>> 70: 
>> 71:     class SubscriptionWrapper implements Subscription {
> 
> Perhaps this can be made `private` class?

good point

> src/java.net.http/share/classes/jdk/internal/net/http/common/HttpBodySubscriberWrapper.java line 83:
> 
>> 81:         @Override
>> 82:         public void cancel() {
>> 83:             HttpBodySubscriberWrapper.this.cancel(subscription);
> 
> Hello Daniel, at first when I noticed this code, it looked odd to me that we aren't calling the underlying `subscription.cancel()`. I then noticed that the `HttpBodySubscriberWrapper.cancel()` actually does the cancellation of this underyling `subscription`. It's slightly odd to me that we are doing it there instead of here. Furthermore, the `HttpBodySubscriberWrapper` (which implements the `Flow` interface) has APIs like `onComplete()`, `onError()`. Perhaps it would be more consistent to have an `onCancel` in `HttpBodySubscriberWrapper` and then have the actual cancellation of the `subscription` done here in the `SubscriptionWrapper` and then call the `onCancel(Subscription)` on the `HttpBodySubscriberWrapper`? Something like:
> 
> 
> 
> 
> class SubscriptionWrapper ... {
>   @Override
>   public void cancel() {
>     try {
>       subscription.cancel();
>     } finally {
>          HttpBodySubscriberWrapper.this.onCancel(subscription);
>     }
>   }
> }
> 
> 
> 
> class HttpBodySubscriberWrapper {
>     protected void onCancel(Subscription cancelledSubscription) {
>         
>     }
> }

good idea. done.

> src/java.net.http/share/classes/jdk/internal/net/http/common/HttpBodySubscriberWrapper.java line 166:
> 
>> 164:             if (subscribed.compareAndSet(false, true)) {
>> 165:                 SubscriptionWrapper wrapped = new SubscriptionWrapper(subscription);
>> 166:                 userSubscriber.onSubscribe(this.subscription = wrapped);
> 
> To make it simpler to read, perhaps change it to:
> 
> this.subscription = new SubscriptionWrapper(subscription);
> userSubscriber.onSubscribe(this.subscription);
> 
> 
> It's OK with me if you prefer to have it in the current form.

I would prefer to leave it like that because this.subscription is volatile.

> src/java.net.http/share/classes/jdk/internal/net/http/common/HttpBodySubscriberWrapper.java line 181:
> 
>> 179:         if (completed.get()) {
>> 180:             if (subscription != null) {
>> 181:                 subscription.subscription.cancel();
> 
> Are we intentionally accessing and cancelling the underlying subscription here, instead of the wrapper? Do we intentionally want to bypass the on-cancellation logic in this flow?

Yes. completed.get() is true here which means that we're receiving onNext after onError or onComplete have been called. Which also means that the stream is guaranteed to be unregistered (at some point). Receiving onNext after onError or onComplete have been called should not happen according to reactive stream specification. But if it does, we really want to make sure the underlying subscription is cancelled. Though I agree it's belt and braces.

> test/jdk/java/net/httpclient/CancelStreamedBodyTest.java line 212:
> 
>> 210:         String[] uris = uris();
>> 211:         Object[][] result = new Object[uris.length * 2][];
>> 212:         //Object[][] result = new Object[uris.length][];
> 
> Leftover comment?

Yes. Thanks.

> test/jdk/java/net/httpclient/CancelStreamedBodyTest.java line 215:
> 
>> 213:         int i = 0;
>> 214:         for (boolean sameClient : List.of(false, true)) {
>> 215:             //if (!sameClient) continue;
> 
> Same here

Thanks.

> test/jdk/java/net/httpclient/CancelStreamedBodyTest.java line 232:
> 
>> 230:                 .sslContext(sslContext)
>> 231:                 .build();
>> 232:         return shared ? client : TRACKER.track(client);
> 
> The implementation of this method looks a bit odd to me. We seem to be creating a new client and then checking if the caller passed `true` for `shared` and if they did then we are returning a brand new client?
> 
> Looking at the usage of this `makeNewClient()` method, we should perhaps not even accept a parameter for this method and instead just return a newly created client?

OK - let me see what I can do. What happens here is that we have a big hammer, the TRACKER, that checks shutdown for all clients. If we track the shared client we can't use the hammer properly because until teardown() is called the shared client will remain alive. But in this particular, test - we still need to track some properties of the shared client, so I believe we should track it. But we don't want to track that it's shutdown (gah). I will need to add some changes to the ReferenceTracker to support this.

> test/jdk/java/net/httpclient/CancelStreamedBodyTest.java line 295:
> 
>> 293:         uri = uri + "/testInputStream";
>> 294:         out.printf("%n%s testInputStream(%s, %b)%n", now(), uri, sameClient);
>> 295:         for (int i=0; i< ITERATION_COUNT; i++) {
> 
> Should we reconsider running each test more than once? Here it looks like we are running 3 * 2 = 6 times for testing this use case.

Standard procedure for the HttpClient. The first request is somewhat special because it's the first. It may trigger creation of the connection, protocol upgrade, etc... The second should find the connection in the pool - it can be special as well in the sense that it's the first time we fetch something from the pool ;-) . So in many tests we test the same things 3 times (granted: we could probably reduce the ITERATION_COUNT to 2, but having it go up to 3 helps convince ourselves that yes, it didn't pass the first time only by chance).

> test/jdk/java/net/httpclient/CancelStreamedBodyTest.java line 402:
> 
>> 400:                     req = is.readAllBytes();
>> 401:                 }
>> 402:                 t.sendResponseHeaders(200, -1); // chunked/variable
> 
> Unrelated to this PR - but this is extra confusing that our `HttpTestExchange` expects `-1` for chunked responses whereas the actual server `com.sun.net.httpserver.HttpExchange` expects `0` for chunked responses.

Sigh. Yes - I am to blame. The com.sun.httpserver API always frustrated me because I couldn't do `sendResponseHeaders(200, resp.length);`. I always had to special case `resp.length == 0 ? -1 : resp.length` . So when I created the adapters I fixed it there. But I agree - if you don't know the history it adds some confusion. I'm not sure I want to change it ;-)

> test/jdk/java/net/httpclient/CancelStreamedBodyTest.java line 407:
> 
>> 405:                     String msg = (req != null && req.length != 0)
>> 406:                             ? new String(req, UTF_8)
>> 407:                             : BODY;
> 
> As far as I can see, none of the test requests send a request body. So perhaps we can simplify this part to just always send this static `BODY` content without these additional checks?

Right - I have changed that code to return 418 (I'm a teapot) if any bytes are received.

> test/jdk/java/net/httpclient/CancelStreamedBodyTest.java line 418:
> 
>> 416:                             // OK
>> 417:                         }
>> 418:                         out.printf("Server wrote %d bytes%n", req.length);
> 
> I think it would be better to move this printf before the call to sleep, since that's when the write completes.

good point

-------------

PR: https://git.openjdk.org/jdk/pull/10659