Restore thread starvation tests

[lorban]

WIP

testReadStarvation still is broken in ee9, ee10 and core.

[gregw]

@lorban I've pushed a fixed for the read side of servlets in ee10. Thoughts?

[gregw]

@lorban @sbordet we could extend CompletableFuture to an InvocableCompletableFuture that either:

• intercepts calls to methods like andThen which pass in functional references. If any such method is called, then the invocation type will be blocking, otherwise it is non-blocking
• intercepts calls to methods like andThen and looks at the invocation type of the functional references passed, which it combines as the invocation type.
• intercepts calls to methods like andThen and redirects them to andThenAsync. The invocation type can then always be non-blocking.

Note the returned completablefuture might also need to be wrapped (maybe not for the andThenAsync redirection).

[gregw]

@lorban @sbordet would this help:

    class NonBlockingCompletableFuture<V> extends CompletableFuture<V> implements Invocable
    {
        private final Executor _executor;

        public NonBlockingCompletableFuture(Executor executor)
        {
            _executor = executor;
        }

        @Override
        public InvocationType getInvocationType()
        {
            return InvocationType.NON_BLOCKING;
        }

        @Override
        public CompletableFuture<Void> acceptEither(CompletionStage<? extends V> other, Consumer<? super V> action)
        {
            return super.acceptEitherAsync(other, action, _executor);
        }

        @Override
        public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends V> other, Function<? super V, U> fn)
        {
            return super.applyToEitherAsync(other, fn, _executor);
        }

        @Override
        public CompletableFuture<V> exceptionallyCompose(Function<Throwable, ? extends CompletionStage<V>> fn)
        {
            return super.exceptionallyComposeAsync(fn, _executor);
        }

        @Override
        public <U> CompletableFuture<U> handle(BiFunction<? super V, Throwable, ? extends U> fn)
        {
            return super.handleAsync(fn, _executor);
        }

        @Override
        public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)
        {
            return super.runAfterBothAsync(other, action, _executor);
        }

        @Override
        public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action)
        {
            return super.runAfterEitherAsync(other, action, _executor);
        }

        @Override
        public CompletableFuture<Void> thenAccept(Consumer<? super V> action)
        {
            return super.thenAcceptAsync(action, _executor);
        }

        @Override
        public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super V, ? super U> action)
        {
            return super.thenAcceptBothAsync(other, action, _executor);
        }

        @Override
        public <U> CompletableFuture<U> thenApply(Function<? super V, ? extends U> fn)
        {
            return super.thenApplyAsync(fn, _executor);
        }

        @Override
        public <U, V1> CompletableFuture<V1> thenCombine(CompletionStage<? extends U> other, BiFunction<? super V, ? super U, ? extends V1> fn)
        {
            return super.thenCombineAsync(other, fn, _executor);
        }

        @Override
        public <U> CompletableFuture<U> thenCompose(Function<? super V, ? extends CompletionStage<U>> fn)
        {
            return super.thenComposeAsync(fn, _executor);
        }

        @Override
        public CompletableFuture<Void> thenRun(Runnable action)
        {
            return super.thenRunAsync(action, _executor);
        }

        @Override
        public CompletableFuture<V> whenComplete(BiConsumer<? super V, ? super Throwable> action)
        {
            return super.whenCompleteAsync(action, _executor);
        }
    }

[sbordet]

I honestly would not go there.
Changing the semantic of thenRun() into a thenRunAsync() is breaking the least surprise.

I think we can just say in the javadocs to never block on the CF returned, or deprecate the API.

[gregw]

I honestly would not go there. Changing the semantic of thenRun() into a thenRunAsync() is breaking the least surprise.

I think we can just say in the javadocs to never block on the CF returned, or deprecate the API.

We need more than javadocs, as we need the Runnable or Callback created from the CF to be Invocable. So how about:

    /**
     * An extension of {@link java.util.concurrent.CompletableFuture} that is an {@link Invocable}.
     * The {@link InvocationType} is initially the type used in construction (default NON_BLOCKING).
     * If a non async method is called, then the invocation type of any passed function is used.
     * @param <V>
     */
    class InvocableCompletableFuture<V> extends java.util.concurrent.CompletableFuture<V> implements Invocable
    {
        private final AtomicReference<InvocationType> _invocationType = new AtomicReference<>();

        public InvocableCompletableFuture()
        {
            this(null);
        }

        public InvocableCompletableFuture(InvocationType invocationType)
        {
            _invocationType.set(Objects.requireNonNullElse(invocationType, InvocationType.NON_BLOCKING));
        }

        @Override
        public InvocationType getInvocationType()
        {
            return _invocationType.get();
        }

        @Override
        public java.util.concurrent.CompletableFuture<Void> acceptEither(CompletionStage<? extends V> other, Consumer<? super V> action)
        {
            _invocationType.set(Invocable.combine(Invocable.getInvocationType(other), Invocable.getInvocationType(action)));
            return super.acceptEither(other, action);
        }

        @Override
        public <U> java.util.concurrent.CompletableFuture<U> applyToEither(CompletionStage<? extends V> other, Function<? super V, U> fn)
        {
            _invocationType.set(Invocable.combine(Invocable.getInvocationType(other), Invocable.getInvocationType(fn)));
            return super.applyToEither(other, fn);
        }

        @Override
        public <U> java.util.concurrent.CompletableFuture<U> handle(BiFunction<? super V, Throwable, ? extends U> fn)
        {
            _invocationType.set(Invocable.getInvocationType(fn));
            return super.handle(fn);
        }

        @Override
        public java.util.concurrent.CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)
        {
            _invocationType.set(Invocable.combine(Invocable.getInvocationType(other), Invocable.getInvocationType(action)));
            return super.runAfterBoth(other, action);
        }

        @Override
        public java.util.concurrent.CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action)
        {
            _invocationType.set(Invocable.combine(Invocable.getInvocationType(other), Invocable.getInvocationType(action)));
            return super.runAfterEither(other, action);
        }

        @Override
        public java.util.concurrent.CompletableFuture<Void> thenAccept(Consumer<? super V> action)
        {
            _invocationType.set(Invocable.getInvocationType(action));
            return super.thenAccept(action);
        }

        @Override
        public <U> java.util.concurrent.CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super V, ? super U> action)
        {
            _invocationType.set(Invocable.combine(Invocable.getInvocationType(other), Invocable.getInvocationType(action)));
            return super.thenAcceptBoth(other, action);
        }

        @Override
        public <U> java.util.concurrent.CompletableFuture<U> thenApply(Function<? super V, ? extends U> fn)
        {
            _invocationType.set(Invocable.getInvocationType(fn));
            return super.thenApply(fn);
        }

        @Override
        public <U, V1> java.util.concurrent.CompletableFuture<V1> thenCombine(CompletionStage<? extends U> other, BiFunction<? super V, ? super U, ? extends V1> fn)
        {
            _invocationType.set(Invocable.combine(Invocable.getInvocationType(other), Invocable.getInvocationType(fn)));
            return super.thenCombine(other, fn);
        }

        @Override
        public <U> java.util.concurrent.CompletableFuture<U> thenCompose(Function<? super V, ? extends CompletionStage<U>> fn)
        {
            _invocationType.set(Invocable.getInvocationType(fn));
            return super.thenCompose(fn);
        }

        @Override
        public java.util.concurrent.CompletableFuture<Void> thenRun(Runnable action)
        {
            _invocationType.set(Invocable.getInvocationType(action));
            return super.thenRun(action);
        }

        @Override
        public java.util.concurrent.CompletableFuture<V> whenComplete(BiConsumer<? super V, ? super Throwable> action)
        {
            _invocationType.set(Invocable.getInvocationType(action));
            return super.whenComplete(action);
        }
    }

[gregw]

That would then be used with:

public abstract class ContentSourceCompletableFuture<X> extends Invocable.InvocableCompletableFuture<X> implements Runnable
{
    private final Content.Source _content;

    public ContentSourceCompletableFuture(Content.Source content)
    {
        _content = content;
    }

    /**
     * <p>Initiates the parsing of the {@link Content.Source}.</p>
     * <p>For every valid chunk that is read, {@link #parse(Content.Chunk)}
     * is called, until a result is produced that is used to
     * complete this {@link CompletableFuture}.</p>
     * <p>Internally, this method is called multiple times to progress
     * the parsing in response to {@link Content.Source#demand(Runnable)}
     * calls.</p>
     * <p>Exceptions thrown during parsing result in this
     * {@link CompletableFuture} to be completed exceptionally.</p>
     */
    public void parse()
    {
        while (true)
        {
            Content.Chunk chunk = _content.read();
            if (chunk == null)
            {
                _content.demand(this);
                return;
            }
            if (Content.Chunk.isFailure(chunk))
            {
                if (chunk.isLast())
                {
                    completeExceptionally(chunk.getFailure());
                }
                else
                {
                    if (onTransientFailure(chunk.getFailure()))
                        continue;
                    _content.fail(chunk.getFailure());
                    completeExceptionally(chunk.getFailure());
                }
                return;
            }

            try
            {
                X x = parse(chunk);
                if (x != null)
                {
                    complete(x);
                    return;
                }
            }
            catch (Throwable failure)
            {
                completeExceptionally(failure);
                return;
            }
            finally
            {
                chunk.release();
            }

            if (chunk.isLast())
            {
                completeExceptionally(new EOFException());
                return;
            }
        }
    }

    @Override
    public void run()
    {
        parse();
    }

    /**
     * <p>Called by {@link #parse()} to parse a {@link org.eclipse.jetty.io.Content.Chunk}.</p>
     *
     * @param chunk The chunk containing content to parse. The chunk will never be {@code null} nor a
     *              {@link org.eclipse.jetty.io.Content.Chunk#isFailure(Content.Chunk) failure chunk}.
     *              If the chunk is stored away to be used later beyond the scope of this call,
     *              then implementations must call {@link Content.Chunk#retain()} and
     *              {@link Content.Chunk#release()} as appropriate.
     * @return The parsed {@code X} result instance or {@code null} if parsing is not yet complete
     * @throws Throwable If there is an error parsing
     */
    protected abstract X parse(Content.Chunk chunk) throws Throwable;

    /**
     * <p>Callback method that informs the parsing about how to handle transient failures.</p>
     *
     * @param cause A transient failure obtained by reading a {@link Content.Chunk#isLast() non-last}
     *             {@link org.eclipse.jetty.io.Content.Chunk#isFailure(Content.Chunk) failure chunk}
     * @return {@code true} if the transient failure can be ignored, {@code false} otherwise
     */
    protected boolean onTransientFailure(Throwable cause)
    {
        return false;
    }
}

[gregw]

@sbordet @lorban I've used the approach above to extend Invocable.InvocableCompletableFuture for ContentSourceCompletableFuture.
I've tested this with a new form based starvation test, that fails with the current future and passes with this fix.

[gregw]

I honestly would not go there. Changing the semantic of thenRun() into a thenRunAsync() is breaking the least surprise.

I think there is no problem changing thenRun() into thenRunAsync(), as the user of a CF has no control of what thread calls them in the first place, so the differences will be transparent to them. I don't think anybody would be surprised.

But, the current approach doesn't need this.

[lorban]

IMHO the CompletableFuture API is orthogonal to the main problem, so we should not distract ourselves too much with this side problem.

That being said, I think the main problem I have with that API is that it makes it so easy to write broken code that I'm tempted to say it encourages. This simple example brings back the original problem in a non-obvious way:

String s = Content.Source.asStringAsync(source, charset).thenRun(() -> { /* whatever */ }).get();

The above is my main motivation to say we should deprecate all our CompletableFuture API.

Back to the main problem, I remember a few months ago I said that we made a mistake when we designed the Content.Source.demand() call to take a Runnable: it should have taken a Invocable.Task instead to make it apparent that InvocationType is meaningful. I wonder if we could do that change in 12.1 without breaking backward compatibility, as it's source-compatible but not binary compatible?

[sbordet]

@lorban unfortunately it's not a compatible change: we have several occurrences of allocating a Runnable anonymous, and then calling demand(this) inside that Runnable.

[sbordet]

I think there is no problem changing thenRun() into thenRunAsync(), as the user of a CF has no control of what thread calls them in the first place, so the differences will be transparent to them. I don't think anybody would be surprised.

They will be, as thenRunAsync() will occupy a common pool thread, and after N of them they won't be available, so tasks won't be run.
In the other case I would block some other thread, but not the precious ones in the common pool.

Furthermore, it would not solve the problem of a double composition, e.g. async().thenApply(...).thenRun(<JDBC>).

I really would not go into trying to put another smartness to fix this, but rather just document clearly the API or deprecate it.

[lorban]

I've just noticed some other complicating factor we must be careful about: a lot of places where an Invocable is passed to demand(), the InvocationType is pre-computed at the time demand() is called, not when the demand callback is to be executed.

This means if we ever decide to manipulate the InvocationType dynamically, we have to be extra careful that all demand() calls take that into account, which isn't the case now.

[sbordet]

@gregw, also, Promise.Completable and Callback.Completable are already Invocable, so perhaps we just need a way to specify the InvocationType, or just assume it is non-blocking.

[sbordet]

See alternative at #12406.

[gregw]

@lorban

a lot of places where an Invocable is passed to demand(), the InvocationType is pre-computed at the time demand() is called, not when the demand callback is to be executed.

This is somewhat of an issue, but it is something we have elsewhere in the code (see dyanamic handler invocation type).
Ultimately it is a race we can never win, if a passed callback is non-blocking, but can asynchronously become blocking, then we can always lose that race. Basically if code that self described as asynchronous suddenly decides it is not, changes its self description and blocks, then we can do nothing.

It is something worth javadocing, but I do not think it is fixable.

[gregw]

@sbordet

I really would not go into trying to put another smartness to fix this, but rather just document clearly the API or deprecate it.

"This fix" is not attempting this smartness we are discussing here.... but I think it is worthwhile discussing it so that we fully understand the problem and possible ways out.
Let's not cede too much authority to the application by saying it would be surprised if it was called back by one thread or another. The thread we use to call back a CF is entirely our choice, either before we call succeeded, or within the implementation of any CF that we create.

I think there is no problem changing thenRun() into thenRunAsync(), as the user of a CF has no control of what thread calls them in the first place, so the differences will be transparent to them. I don't think anybody would be surprised.
They will be, as thenRunAsync() will occupy a common pool thread, and after N of them they won't be available, so tasks won't be run. In the other case I would block some other thread, but not the precious ones in the common pool.

Firstly, this uses a passed executor, so there is no common pool thread involved, only our executor.

But I do not think this changes behaviour. If our thread pool is exhausted, currently our CF is seen as blocking and just never called, so deadlock happens always. If we did a change based on converting non-async methods to async, then in the blocking case, we still hit the exhausted thread pool and are deadlocked.

Furthermore, it would not solve the problem of a double composition, e.g. async().thenApply(...).thenRun(<JDBC>).

Yeah, that would require further wrapping... hence I don't like this approach.

So I'm not proposing we do this, and this PR is not doing it. But I do believe that if we want to control the thread that does the callback, it is within our authority both as the caller and as the implementor of the CF, so there would be no surprise.

[gregw]

The above is my main motivation to say we should deprecate all our CompletableFuture API.

I think we should fix all our CF usages (it is not proving to be too difficult) and then separately consider if it is indeed the best API to use for many of our utility classes. As you say, it is orthogonal.

There are several ways we can "fix" our existing CF usage:

1. Change our CFs to automatically determine their invocation type by checking what callbacks are installed on them. This appears to work, but there is some risk of CFs dynamically changing their type after calling demand.
2. Change our CFs to have a fixed NON_BLOCKING invocation type of non-blocking and just javadoc that the andThen style APIs should not be used. This would "work", but would be fragile.
3. Change our CFs to have a fixed NON_BLOCKING invocation type and throw UnsupportedOperation exception if any andThen style APIs were used. This is kind of deprecating full CF support and a step towards coming up with our own API (which we could do as well).
4. Change our CFs to have a fixed NON_BLOCKING invocation type and throw IllegalArgumentException if any andThen style APIs are used with BLOCKING functions etc.

My contributions to this PR implement 1). They are not too complex, appear to be working, but are vulnerable to dynamic changes of invocation type.
2) is essentially the same as the alternate PR, as that it just changes to type and hopes for the best.

I don't mind 3) or 4) as they solve the dynamic type issue. 3) is a big neutering of the CF API and really shouts that we should not use it and do something else. 4) is less so, as it gives a way all the API can be used.

Ultimately, we have always known that the invocation type of our callbacks is important and previously we have always dealt with it. For some reason, we forgot that for a little while and introduced a few utility classes that didn't do the work required to ensure a good invocation type. I think least surprise is to not radically change our API/contract and to just go back to caring about invocation type in all our utility classes and better javadoc of the necessity to do so for others that implement the API directly (we could even give warning for non invocable demand callbacks?).

There is still a role for Blocking callbacks to demand. It is just that they cannot be used if their only job is to wakeup an otherwise blocked thread, as that is deadlock. If the demand callback is genuinely blocking (eg on a database), then it will just have to wait until a thread is available before it is called and then it can block on the database. It just cannot wait if its job is to free up a thread that would go back to the pool to call itself in the first place.

[gregw]

For our CF based utility classes, we have two distinct styles of usage:

1. in our adaption to blocking APIs (e.g. in ServletApiRequest) we often do:

           Fields formFields = FormFields.from(request).get();

2. in truly async (e.g. DelayedHandler) we do:

           CompletableFuture<Fields> futureFormFields = FormFields.from(getRequest(), _charset);
   
           // if we are done already, then we are still in the scope of the original process call and can
           // process directly, otherwise we must execute a call to process as we are within a serialized
           // demand callback.
           boolean done = futureFormFields.isDone();
           futureFormFields.whenComplete(done ? this::process : this::executeProcess);

We typically do not mix these styles, other than the DelayedHandler is designed to go in front of the blocking ServletHandler, so the whenComplete style usage may proceed the get() style usage. However, this is not really a problem as in this case, the get() will never block, as it is already done/complete by the time it is given an opportunity.

However, I don't think my fix here is good for the 2) usage. The parsing, and hence demanding, is commenced within the from method, which is before the call to whenComplete. So we cannot use calling that method to trigger switching the demand callback to be blocking. We need to know the invocation type during the from call so we could just pass in:

            CompletableFuture<Fields> futureFormFields = FormFields.from(getRequest(), _charset, InvocationType.BLOCKING);

            // if we are done already, then we are still in the scope of the original process call and can
            // process directly, otherwise we must execute a call to process as we are within a serialized
            // demand callback.
            boolean done = futureFormFields.isDone();
            futureFormFields.whenComplete(done ? this::process : this::executeProcess);

In this case the CF returned should have the get() methods extended to do something like:

    public T get()
    {
         if (getInvocationType() != InvocationType.NON_BLOCKING && !isDone())
             throw new IllegalStateException("Cannot call get unless NON_BLOCKING or complete");
         return super.get();
    }

But it would be a bit strange as we'd then need to do in ServletAPIRequest:

            Fields formFields = FormFields.from(request, InvocationType.NON_BLOCKING).get();

which I think is correct, but reads strangely. It is also strange that the passed invocation type is only used if this is the first call to from

As I think the current "fix" is wrong. I'll change it today to do this.

[gregw]

The CF fix works but is counter intuitive, fragile and mind blowing when combining CFs as we do going from core Parts to servlet Parts.

So I've pushed some changes that deprecate all the CF APIs in Fields and MultiPart and replace with explicit onXxx and getXxx methods. Internally, these use the "fixed" CF for now, but we can replace with something different once we can actually remove the CFs
This is still a WIP, but pushed so you can see it.

[sbordet]

@gregw I have not looked yet, but regarding this comment:

The thread we use to call back a CF is entirely our choice, either before we call succeeded, or within the implementation of any CF that we create.

That is true, but what I would like to see is actually the following:

• Use internally our Promise, Callback, Blocker, etc. that specify the InvocationType (likely NON_BLOCKING) until the operation is finished.
• Once the operation is finished, we can complete the CompletableFuture using ThreadPool.executeImmediately(), or similar.

In this way, the semantic of CompletableFuture is preserved, and we do not need any special subclass of CompletableFuture with overrides that would need to be reviewed when the JDK adds new methods to it.

Then, if the application blocks on the CompletableFuture, it's the application choice, and it won't block a selector thread.

I'll review as soon as I am back.

[gregw]

@sbordet I think we are agreed that ultimately we want to remove CF for our implementation and API. However, currently it is in our API (although deprecated by this PR), so we do need to fix it. Hence I think we should have the extended CF (also deprecated) as an interim fix until we can remove the CFs entirely.

So the key things we need to do in this PR is:

1. fix/work around the immediate problem in our usage of utility classes (mostly done)
2. Come up with a new API that will allow others to use these classes without using CFs (done - but needs work see below).

As we are close to release time, can you focus on reviewing 2), as new API is forever.

I like the general style of the new API in this PR: getParts(request) for blocking and onParts(request, consumer, consumer) for async. But the arguments of the onParts methods need some consideration.
Currently they are like:

    void onParts(Request request, BiConsumer<Parts, Throwable> immediate, InvocableBiConsumer<Parts, Throwable> future);
    default void onParts(Request request, InvocableBiConsumer<Parts, Throwable> future) 
    { onParts(request, future, future); }

Thoughts about this are:

• we do need the immediate and future arguments as often we act a bit differently if the parts are being returned in the calling thread (i.e. we don't execute)... although perhaps there is some other way to signal that.
• we have a single consumer convenience method for when they are both the same.
• Using the raw BiConsumer and the InvocableBiConsumer makes it clear which one needs to have an InvocationType. Better than javadoc.
• BiConsumer is a single method interface, so lambdas can still be used, although less so with the InvocableBiConsumer

An alternative would be:

    void onParts(Request request, Promise<Parts> immediate, Promise<Parts> future);
    default void onParts(Request request, Promise<Parts> future) 
    { onParts(request, future, future); }

Thoughts:

• This reads a bit better
• but we lose the distinction that it is the invocation type of the future promise that matters
• harder to use with lambdas
• it is an existing interface.

[gregw]

@lorban thoughts on the API comment above?

[lorban]

I like Promise<Parts> because:

• this PR adds a Blocker.Promise so it's trivial to implement the blocking API with the async one
• Promise is already part of our async API

but I do not like that it is not Invocable, but I think we can make it extend that interface without problem.

I don't mind that this API is not as fluent as CompletableFuture: you can't use lambdas and you cannot chain actions, this isn't so bad as this is already the case with Callback and we don't mind it.

I don't like InvocableBiConsumer as it really is equivalent to Promise and would IMHO just add confusions about when we would use one over the other. But maybe BiConsumer<Parts,Throwable> could make sense to clearly show in the API what needs an invocation type. Or maybe use Promise<Part> and introduce Promise.Invocable<Part>?

Oh, and there are other places where we expose CompletableFuture in our API, like in the Content.Source static helpers. Those should be fixed and deprecated too.

[gregw]

@sbordet @lorban I've converrted the FormFields class to use Promise on the new API and it works fine. Can you review that as a priority and if OK I'll convert the MultiPart classes as well.