diff --git a/src/main/java/groovy/concurrent/AsyncChannel.java b/src/main/java/groovy/concurrent/AsyncChannel.java index dc8cfdbde4d..a286ec689a2 100644 --- a/src/main/java/groovy/concurrent/AsyncChannel.java +++ b/src/main/java/groovy/concurrent/AsyncChannel.java @@ -25,8 +25,10 @@ import java.time.Instant; import java.util.List; import java.util.Objects; +import java.util.concurrent.CancellationException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Predicate; @@ -55,6 +57,20 @@ * println item // prints 'a', then 'b' * } * } + * {@link #produce(int, Consumer) produce} runs a producer and closes the + * channel when it finishes, normally or with the failure it throws. + * A producer that does not use {@code produce} catches that failure and + * passes it to {@link #close(Throwable)}. {@code for await} throws that + * failure after + * the values already sent, and the code that reads the channel writes the + * {@code try}/{@code catch}. A plain {@code for} loop, {@code each}, + * {@code collect} and {@code toList} use the same iterator and throw it + * too. A Java enhanced {@code for} calls {@code Iterator.hasNext()}, which + * may throw a checked exception that {@code Iterator} does not declare. + * {@code receive} and {@code send} still fail with + * {@link ChannelClosedException}; the failure is that exception's + * {@linkplain Throwable#getCause() cause}, so a hand-written loop has to + * read the cause or the failure is dropped. * * @param the payload type * @see Awaitable @@ -78,6 +94,70 @@ static AsyncChannel create(int capacity) { return new DefaultAsyncChannel<>(capacity); } + /** + * Starts a producer for a new unbuffered channel. + *

+ * The body runs asynchronously. {@link Producer#send(Object)} waits until + * the channel accepts each value. When the body returns, the channel is + * closed normally. When the body throws, the channel is closed with that + * failure, after the values already accepted. A {@link ChannelClosedException} + * from {@code send} means the channel was closed by its consumer; that is + * not recorded as a producer failure. + * + * @param body the producer; must not be {@code null} + * @param the payload type + * @return the channel the body sends to + * @throws NullPointerException if {@code body} is null + * @see #produce(int, Consumer) + * @since 6.0.1 + */ + static AsyncChannel produce(Consumer> body) { + return produce(0, body); + } + + /** + * Starts a producer for a new channel of the given capacity. + * See {@link #produce(Consumer)}. + * + * @param capacity the maximum buffer size; 0 for unbuffered + * @param body the producer; must not be {@code null} + * @param the payload type + * @return the channel the body sends to + * @throws NullPointerException if {@code body} is null + * @throws IllegalArgumentException if {@code capacity} is negative + * @since 6.0.1 + */ + static AsyncChannel produce(int capacity, Consumer> body) { + Objects.requireNonNull(body, "body must not be null"); + AsyncChannel channel = create(capacity); + Producer producer = new Producer<>(channel); + AsyncSupport.getExecutor().execute(() -> runProducer(channel, producer, body)); + return channel; + } + + /** + * Runs {@code body} and closes {@code channel}. The body can throw an + * {@link Error} or a checked exception; that failure has to close the + * channel on this thread rather than escape the executor. + */ + private static void runProducer(AsyncChannel channel, Producer producer, + Consumer> body) { + Throwable failure = null; + try { + body.accept(producer); + } catch (Throwable thrown) { + Throwable unwrapped = AsyncSupport.unwrap(thrown); + // send throws ChannelClosedException only after the consumer has + // already closed the channel. A ChannelClosedException thrown while + // the channel is still open is the producer's own failure. + if (!(unwrapped instanceof ChannelClosedException) || !channel.isClosed()) { + failure = unwrapped; + } + } finally { + channel.close(failure); + } + } + /** * Creates a timer channel: a capacity-1 channel that delivers a single * value, the {@link Instant} at which it fired, once {@code millis} has @@ -150,7 +230,8 @@ static AsyncChannel after(Duration duration) { *

* The returned {@link Awaitable} completes when the value has been * delivered to a receiver or buffered. Sending to a closed channel - * fails immediately with {@link ChannelClosedException}. + * fails immediately with {@link ChannelClosedException}. When the channel + * was closed with a failure, that failure is the exception's cause. * * @param value the value to send; must not be {@code null} * @return an Awaitable that completes when the send succeeds @@ -163,7 +244,11 @@ static AsyncChannel after(Duration duration) { *

* The returned {@link Awaitable} completes when a value is available. * Receiving from a closed, empty channel fails with - * {@link ChannelClosedException}. + * {@link ChannelClosedException}. When the channel was closed with a + * failure, that failure is the exception's cause. A loop that catches + * only {@code ChannelClosedException} drops the failure unless it reads + * {@link Throwable#getCause()}. {@code for await} throws the failure + * itself; the code that reads the channel writes the {@code try}/{@code catch}. * * @return an Awaitable that yields the next value */ @@ -177,19 +262,117 @@ static AsyncChannel after(Duration duration) { * drained, subsequent receives also fail. * * @return {@code true} if this call actually closed the channel + * @see #close(Throwable) */ boolean close(); + /** + * Closes this channel because {@code cause} failed the producer. + *

+ * Values already buffered are still delivered. {@code receive} and + * {@code send} then fail with {@link ChannelClosedException}, and + * {@code cause} is that exception's cause. {@code for await} throws + * {@code cause} after those values. The reading code writes the + * {@code try}/{@code catch}. A normal {@link #close()} still ends the + * loop without throwing. + *

+ * {@code null} means a normal close, the same as {@link #close()}. + * Returns {@code true} when this call closes the channel. The first + * close wins, so a later call returns {@code false} and leaves any + * retained failure unchanged. + *

+ * The default implementation calls {@link #close()} and returns its + * result. It does not retain {@code cause}. An implementation that does + * not override this method therefore ends iteration without throwing + * {@code cause}. The channel returned by {@link #create(int)} retains + * {@code cause}. An implementation of {@link #close()} must not call + * this method unless it also overrides it: the default calls + * {@link #close()} again. + * + * @param cause the failure iteration should throw, or {@code null} for a normal close + * @return {@code true} if this call closed the channel + * @since 6.0.1 + */ + default boolean close(Throwable cause) { + return close(); + } + + /** + * Returns the failure passed to {@link #close(Throwable)}, or {@code null} + * when this channel is open, was closed normally, or does not retain a cause. + *

+ * The default is {@code null}. {@code receive} and {@code send} do not + * throw this failure; it is the {@linkplain Throwable#getCause() cause} + * of the {@link ChannelClosedException} they throw. + * + * @return the retained failure, or {@code null} + * @since 6.0.1 + */ + default Throwable getCloseCause() { + return null; + } + + /** + * Reads {@code source} until it closes, {@code step} returns {@code false}, + * or {@code step} throws, then runs {@code close}. A close of {@code source} + * keeps that close's cause, and {@code null} means a normal close. A + * throwable from {@code step} is the failure to retain. A closed output is + * not one of those: {@code step} reports it by returning {@code false} + * (see {@link #offer}), and that output's own close is left as it is. + */ + private static void pump(AsyncChannel source, Predicate step, + Consumer close) { + Throwable failure = null; + try { + while (true) { + T item; + try { + item = AsyncSupport.await(source.receive()); + } catch (ChannelClosedException closed) { + failure = closed.getCause(); + break; + } + if (!step.test(item)) break; + } + } catch (Throwable thrown) { + failure = AsyncSupport.unwrap(thrown); + } finally { + close.accept(failure); + } + } + + /** + * Sends {@code item} to {@code out}. + * + * @return {@code false} when {@code out} is already closed, so the caller + * stops without adopting {@code out}'s close cause + */ + private static boolean offer(AsyncChannel out, T item) { + try { + AsyncSupport.await(out.send(item)); + return true; + } catch (ChannelClosedException ignored) { + return false; + } + } + // ---- Composition (each returns a new channel) ----------------------- // // Each method spawns a background task that reads from this channel // using explicit receive()/await, writes to the output channel(s), // and closes them when the source is exhausted. We use receive() // directly rather than the Iterable iterator to avoid blocking - // iterator issues on JDK 17 (no virtual threads). + // iterator issues on JDK 17 (no virtual threads). A failure from the + // source, or thrown by the callback, is retained on the output. A normal + // close is not. merge forwards a value a relay has already taken, then + // retains a failure without waiting for the other input to end. An output + // the consumer has already closed stops that send and is not treated as + // a failure of the source. /** * Returns a new channel that passes only elements matching the predicate. + * The result closes when this channel does. A failure from this channel + * or from {@code predicate} is retained, so {@code for await} throws it. * * @param predicate the filter function * @return a new filtered channel @@ -197,24 +380,17 @@ static AsyncChannel after(Duration duration) { */ default AsyncChannel filter(Predicate predicate) { AsyncChannel out = create(getCapacity()); - AsyncSupport.getExecutor().execute(() -> { - try { - while (true) { - T item = AsyncSupport.await(receive()); - if (predicate.test(item)) { - AsyncSupport.await(out.send(item)); - } - } - } catch (ChannelClosedException ignored) { - } finally { - out.close(); - } - }); + AsyncSupport.getExecutor().execute(() -> pump(this, item -> { + if (!predicate.test(item)) return true; + return offer(out, item); + }, out::close)); return out; } /** * Returns a new channel that transforms each element using the function. + * The result closes when this channel does. A failure from this channel + * or from {@code transform} is retained, so {@code for await} throws it. * * @param transform the mapping function * @param the output element type @@ -223,58 +399,105 @@ default AsyncChannel filter(Predicate predicate) { */ default AsyncChannel map(Function transform) { AsyncChannel out = create(getCapacity()); - AsyncSupport.getExecutor().execute(() -> { - try { - while (true) { - T item = AsyncSupport.await(receive()); - AsyncSupport.await(out.send(transform.apply(item))); - } - } catch (ChannelClosedException ignored) { - } finally { - out.close(); - } - }); + AsyncSupport.getExecutor().execute(() -> pump(this, + item -> offer(out, transform.apply(item)), out::close)); return out; } /** * Returns a new channel that receives values from both this channel and - * the other channel. Values are interleaved as they arrive. The output - * closes when both inputs are exhausted. + * {@code other}. Values are interleaved as they arrive. + *

+ * The output closes when both inputs have closed normally. A failure + * from either input closes the output with that failure once every value + * a relay has already taken, and every value already buffered on the + * other input, has been forwarded. The relay keeps taking from that + * input while its buffer is non-empty, so a producer that refills the + * buffer delays the failure until a gap appears. The other input is not + * closed. A producer still waiting to hand a value over is not released. + * When both inputs fail, the first failure retained on the output wins. + * A later normal close does not replace it. * * @param other the channel to merge with * @return a new merged channel + * @throws NullPointerException if {@code other} is null * @since 6.0.0 */ default AsyncChannel merge(AsyncChannel other) { + Objects.requireNonNull(other, "other must not be null"); AsyncChannel out = create(getCapacity()); - var remaining = new AtomicInteger(2); - Runnable closer = () -> { - if (remaining.decrementAndGet() == 0) out.close(); - }; - AsyncSupport.getExecutor().execute(() -> { - try { - while (true) AsyncSupport.await(out.send(AsyncSupport.await(this.receive()))); - } catch (ChannelClosedException ignored) { - } finally { - closer.run(); - } - }); + // A failure must not take a count. If it did, the sibling's normal + // end could be the decrement that reaches zero and close the output + // with no cause before the failure is retained. + var openInputs = new AtomicInteger(2); + var left = new MergeRelay(); + var right = new MergeRelay(); + relay(this, out, openInputs, left, right); + relay(other, out, openInputs, right, left); + return out; + } + + /** + * Forwards {@code in} to {@code out}. A failure first lets {@code sibling} + * finish values it has already taken or that are already buffered, then + * closes {@code out}. A normal end closes {@code out} only after every + * input has ended the same way. + */ + private static void relay(AsyncChannel in, AsyncChannel out, + AtomicInteger openInputs, MergeRelay self, MergeRelay sibling) { AsyncSupport.getExecutor().execute(() -> { + Throwable failure = null; try { - while (true) AsyncSupport.await(out.send(AsyncSupport.await(other.receive()))); - } catch (ChannelClosedException ignored) { + while (true) { + // An empty input must not be received once the sibling has + // failed: that receive would remove a later value and then + // find the output closed. Buffered values are already the + // sender's and are forwarded. + if (self.isStopping() && in.getBufferedSize() == 0) break; + Awaitable pending = in.receive(); + self.publish(pending); + // A sender can commit between isDone() and cancel(). + // cancel() then returns false and the value is already + // out of the input; awaiting it forwards that value. + if (self.isStopping() && !pending.isDone() && pending.cancel()) { + break; + } + T item; + try { + item = AsyncSupport.await(pending); + } catch (ChannelClosedException closed) { + failure = closed.getCause(); + break; + } catch (CancellationException cancelled) { + break; + } finally { + self.clearPending(pending); + } + if (!offer(out, item)) break; + } + } catch (Throwable thrown) { + Throwable unwrapped = AsyncSupport.unwrap(thrown); + if (!(unwrapped instanceof CancellationException)) { + failure = unwrapped; + } } finally { - closer.run(); + self.markIdle(); + if (failure != null) { + sibling.stopAndWait(); + out.close(failure); + } else if (openInputs.decrementAndGet() == 0) { + out.close(); + } } }); - return out; } /** * Returns two new channels: elements matching the predicate go to the - * first, non-matching to the second. Both are closed when this channel - * is exhausted. + * first, non-matching to the second. Both close when this channel does, + * and both retain a failure from this channel or from {@code predicate}. + * Closing one result stops the split and closes the other normally; a + * failure close of one result is not copied onto the other. * * @param predicate the split condition * @return a list of two channels: [matching, non-matching] @@ -283,29 +506,20 @@ default AsyncChannel merge(AsyncChannel other) { default List> split(Predicate predicate) { AsyncChannel trueOut = create(getCapacity()); AsyncChannel falseOut = create(getCapacity()); - AsyncSupport.getExecutor().execute(() -> { - try { - while (true) { - T item = AsyncSupport.await(receive()); - if (predicate.test(item)) { - AsyncSupport.await(trueOut.send(item)); - } else { - AsyncSupport.await(falseOut.send(item)); - } - } - } catch (ChannelClosedException ignored) { - } finally { - trueOut.close(); - falseOut.close(); - } - }); + AsyncSupport.getExecutor().execute(() -> pump(this, item -> + offer(predicate.test(item) ? trueOut : falseOut, item), failure -> { + trueOut.close(failure); + falseOut.close(failure); + })); return List.of(trueOut, falseOut); } /** * Returns a new channel that receives all values from this channel while * also sending a copy of each value to the tap channel. Useful for - * logging, monitoring, or forking a side pipeline. + * logging, monitoring, or forking a side pipeline. A closed tap is + * ignored and the pass-through continues. A failure from this channel + * is retained on the result. * * @param tap the channel to send copies to * @return a new pass-through channel @@ -313,19 +527,107 @@ default List> split(Predicate predicate) { */ default AsyncChannel tap(AsyncChannel tap) { AsyncChannel out = create(getCapacity()); - AsyncSupport.getExecutor().execute(() -> { - try { - while (true) { - T item = AsyncSupport.await(receive()); - try { AsyncSupport.await(tap.send(item)); } - catch (ChannelClosedException ignored) { } - AsyncSupport.await(out.send(item)); + AsyncSupport.getExecutor().execute(() -> pump(this, item -> { + offer(tap, item); // a closed tap does not stop the pass-through + return offer(out, item); + }, out::close)); + return out; + } + + /** + * Blocking sender for {@link AsyncChannel#produce(int, Consumer)}. + *

+ * {@link #send(Object)} waits until the channel accepts the value, so a + * rendezvous producer does not run ahead of its consumer. + * + * @param the payload type + * @since 6.0.1 + */ + final class Producer { + private final AsyncChannel channel; + + private Producer(AsyncChannel channel) { + this.channel = channel; + } + + /** + * Sends {@code value} and waits until it is buffered or delivered. + * + * @param value the value to send; must not be {@code null} + * @throws ChannelClosedException if the channel is closed + * @throws NullPointerException if {@code value} is null + */ + public void send(T value) { + AsyncSupport.await(channel.send(value)); + } + + /** + * Returns the channel this producer sends to. + * + * @return the underlying channel + */ + public AsyncChannel getChannel() { + return channel; + } + } +} + +/** + * Handshake for the two relays of {@link AsyncChannel#merge(AsyncChannel)}. + * A failure waits until the sibling is idle, after cancelling a receive that + * has not taken a value, so a value already removed from that input is + * forwarded before the output closes. + */ +final class MergeRelay { + private final Object lock = new Object(); + /** False until the relay task exits. Starts false so a failure waits for it. */ + private boolean idle; + private boolean stopping; + private Awaitable pending; + + boolean isStopping() { + synchronized (lock) { + return stopping; + } + } + + void publish(Awaitable receive) { + synchronized (lock) { + pending = receive; + } + } + + void clearPending(Awaitable receive) { + synchronized (lock) { + if (pending == receive) pending = null; + } + } + + void markIdle() { + synchronized (lock) { + pending = null; + idle = true; + lock.notifyAll(); + } + } + + /** Cancels a receive that has not completed, then waits until the relay exits. */ + void stopAndWait() { + Awaitable toCancel; + synchronized (lock) { + stopping = true; + toCancel = pending; + } + if (toCancel != null) toCancel.cancel(); + synchronized (lock) { + while (!idle) { + try { + lock.wait(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + return; } - } catch (ChannelClosedException ignored) { - } finally { - out.close(); } - }); - return out; + } } } diff --git a/src/main/java/groovy/concurrent/BroadcastChannel.java b/src/main/java/groovy/concurrent/BroadcastChannel.java index be8e233d1cd..b9825f46789 100644 --- a/src/main/java/groovy/concurrent/BroadcastChannel.java +++ b/src/main/java/groovy/concurrent/BroadcastChannel.java @@ -62,6 +62,10 @@ public final class BroadcastChannel { private final CopyOnWriteArrayList> subscribers = new CopyOnWriteArrayList<>(); + /** Guards {@link #closed} together with membership of {@link #subscribers}. */ + private final Object lock = new Object(); + /** Failure passed to {@link #close(Throwable)}, published before {@link #closed}. */ + private volatile Throwable closeCause; // NOSONAR java:S3077 — published as a reference; the Throwable is not mutated private volatile boolean closed; private BroadcastChannel() { } @@ -94,9 +98,11 @@ public AsyncChannel subscribe() { * @return a new subscriber channel */ public AsyncChannel subscribe(int bufferSize) { - if (closed) throw new ChannelClosedException("BroadcastChannel is closed"); AsyncChannel sub = AsyncChannel.create(bufferSize); - subscribers.add(sub); + synchronized (lock) { + if (closed) throw closedNow(); + subscribers.add(sub); + } return sub; } @@ -105,11 +111,12 @@ public AsyncChannel subscribe(int bufferSize) { * * @param value the value to broadcast * @return an Awaitable that completes when all subscribers have accepted the value - * @throws ChannelClosedException if the broadcast channel is closed + * @throws ChannelClosedException if the broadcast channel is closed. + * The cause is the failure passed to {@link #close(Throwable)}, when there is one. */ public Awaitable send(T value) { Objects.requireNonNull(value, "value must not be null"); - if (closed) throw new ChannelClosedException("BroadcastChannel is closed"); + if (closed) throw closedNow(); CompletableFuture[] futures = subscribers.stream() .map(sub -> { try { @@ -124,12 +131,56 @@ public Awaitable send(T value) { /** * Closes this broadcast channel and all subscriber channels. + * Idempotent. A later {@link #close(Throwable)} does not add a failure. */ public void close() { - closed = true; + close(null); + } + + /** + * Closes this broadcast channel because {@code cause} failed the producer. + *

+ * Values already accepted by a subscriber remain receivable. Iteration + * on each subscriber throws {@code cause}. {@code null} is a normal close, + * the same as {@link #close()}. Returns {@code true} when this call closes + * the broadcast. The first close wins, so a later call returns {@code false} + * and leaves the retained failure unchanged. A subscriber that arrives + * afterwards fails at once with {@link ChannelClosedException} carrying + * this cause. + * + * @param cause the failure each subscriber should surface, or {@code null} + * @return {@code true} if this call closed the broadcast + * @since 6.0.1 + */ + public boolean close(Throwable cause) { + synchronized (lock) { + if (closed) { + return false; + } + closeCause = cause; + closed = true; + } for (AsyncChannel sub : subscribers) { - sub.close(); + sub.close(cause); } + return true; + } + + /** + * Returns the failure passed to {@link #close(Throwable)}, or {@code null} + * when this broadcast is open or was closed normally. + * + * @return the retained failure, or {@code null} + * @since 6.0.1 + */ + public Throwable getCloseCause() { + return closeCause; + } + + private ChannelClosedException closedNow() { + Throwable cause = closeCause; + if (cause == null) return new ChannelClosedException("BroadcastChannel is closed"); + return new ChannelClosedException("BroadcastChannel is closed", cause); } /** @@ -162,8 +213,9 @@ public int getSubscriberCount() { * broadcast send when no demand exists (sender-side backpressure). *

  • Cancellation: closes this subscriber's channel and removes it * from the broadcast's subscriber set.
    • - *
  • Completion: signals {@code onComplete} when the broadcast channel - * is closed and the per-subscriber buffer drained.
    • + *
  • Completion: signals {@code onComplete} when the subscriber channel + * is closed normally and drained. {@link #close(Throwable)} signals + * {@code onError} with that failure instead.
    • * *

    * Backpressure policy (important). This bridge uses lossless, @@ -302,8 +354,7 @@ private void drain() { // If the upstream channel has closed and drained, emit // onComplete even when demand == 0. if (channel.isClosed() && channel.getBufferedSize() == 0) { - subscriber.onComplete(); - cancel(); + completeOrFail(null); return; } if (demand.get() == 0) { @@ -322,10 +373,7 @@ private void drain() { try { item = AsyncSupport.await(channel.receive()); } catch (ChannelClosedException e) { - if (!cancelled.get()) { - subscriber.onComplete(); - cancel(); - } + completeOrFail(e.getCause()); return; } if (cancelled.get()) return; @@ -346,5 +394,21 @@ private void drain() { } } } + + /** A retained failure fails the subscriber. A normal close completes it. */ + private void completeOrFail(Throwable cause) { + if (cancelled.get()) { + return; + } + if (cause == null) { + cause = channel.getCloseCause(); + } + if (cause != null) { + subscriber.onError(cause); + } else { + subscriber.onComplete(); + } + cancel(); + } } } diff --git a/src/main/java/groovy/concurrent/ChannelClosedException.java b/src/main/java/groovy/concurrent/ChannelClosedException.java index 8d22bccb580..2a08dc33601 100644 --- a/src/main/java/groovy/concurrent/ChannelClosedException.java +++ b/src/main/java/groovy/concurrent/ChannelClosedException.java @@ -30,14 +30,16 @@ * or during the send attempt. Pending senders that were waiting for buffer * space when the channel closed also receive this exception. *

  • {@link AsyncChannel#receive() receive()} — the channel was closed and all - * buffered values have been drained. Note that values buffered before - * closure are still delivered normally; only once the buffer is exhausted - * does this exception appear.
    • + * buffered values have been drained. Values buffered before closure are + * still delivered. A failure passed to {@link AsyncChannel#close(Throwable)} + * is this exception's {@linkplain Throwable#getCause() cause}, not a + * different exception type. A hand-written loop must read that cause. * *

    - * When used with {@code for await}, the loop infrastructure translates - * {@code ChannelClosedException} into a clean end-of-stream signal (i.e., - * the loop exits normally rather than propagating the exception): + * When used with {@code for await}, a plain {@code for}, {@code each}, + * {@code collect} or {@code toList}, a close with no cause ends iteration. + * A close that carries a cause throws that cause after the values already + * sent. The code that reads the channel writes the {@code try}/{@code catch}: * 

      * def ch = AsyncChannel.create()
  * // ... producer sends values, then calls ch.close()
@@ -46,6 +48,8 @@
  * }
  * // loop exits cleanly after the channel is closed and drained
  *
    + * A Java enhanced {@code for} calls {@code Iterator.hasNext()}, which may + * throw a checked exception that {@code Iterator} does not declare. * * @see AsyncChannel#send(Object) * @see AsyncChannel#receive() diff --git a/src/main/java/groovy/concurrent/ChannelSelect.java b/src/main/java/groovy/concurrent/ChannelSelect.java index 612fe3356dd..f9d0871418b 100644 --- a/src/main/java/groovy/concurrent/ChannelSelect.java +++ b/src/main/java/groovy/concurrent/ChannelSelect.java @@ -332,8 +332,16 @@ public ChannelSelect random() { * exception, add a timer offer from {@link #after(long)} or receive * from a timer channel from {@link AsyncChannel#after(long)}. *

    - * If every offer's channel is closed, the result fails with - * {@link ChannelClosedException}. If every offer is guarded off (see + * If every offer's channel is closed normally, the result fails with + * {@link ChannelClosedException}. A channel closed with a failure still + * delivers a value that is already buffered. The failure fails the + * select only when that offer's next operation is the close. Which of + * a buffered value and such a failure is met first follows registration + * order: list order here, rotation for {@link #fair()}, and a shuffle + * for {@link #random()}. A transfer that has already committed stays + * the result: the claim admits one outcome, so that value is not + * discarded when a close races it. + * If every offer is guarded off (see * {@link Offer#when(BooleanSupplier)}) there is nothing to wait for and * the result fails with {@link IllegalStateException}. *

    @@ -441,8 +449,14 @@ private Awaitable start(int[] order) { } else { resend(ch, value); } - } else if (error instanceof ChannelClosedException && closedCount.incrementAndGet() == count) { - winner.completeExceptionally(new ChannelClosedException("all channels are closed")); + } else if (error instanceof ChannelClosedException closed) { + Throwable cause = closed.getCause(); + if (cause != null) { + failIfUnclaimed(winner, branches, timers, cause); + } else if (closedCount.incrementAndGet() == count) { + failIfUnclaimed(winner, branches, timers, + new ChannelClosedException("all channels are closed")); + } } }); } @@ -500,6 +514,19 @@ private boolean isEnabled(int index, boolean[] enabled) { return (enabled == null || enabled[index]) && offers.get(index).isEnabled(); } + /** + * Fails {@code winner} with {@code cause} only when no offer has committed. + * The claim is the same gate {@link Winner#cancel} uses, so a transfer + * that already holds it stays the result and its value is not discarded. + * Losers are withdrawn before the caller observes the failure. + */ + private static void failIfUnclaimed(Winner winner, Awaitable[] branches, + AsyncChannel[] timers, Throwable cause) { + if (!winner.claim.tryCommitCancel()) return; + withdraw(branches, timers); + winner.completeExceptionally(cause); + } + /** * Withdraws every registered branch, and closes every timer armed for * the call so its scheduled firing is cancelled. Branches go first: a diff --git a/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java b/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java index d86b0ec453d..dd6f91dda2e 100644 --- a/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java +++ b/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java @@ -546,6 +546,14 @@ public static Throwable unwrap(Throwable t) { return t; } + /** + * Rethrows {@code failure} as itself, including a checked exception. + * Kept in this package so the pure-Java concurrent jar stays self-contained. + */ + static void rethrow(Throwable failure) { + sneakyThrow(failure); + } + @SuppressWarnings("unchecked") private static void sneakyThrow(Throwable t) throws T { throw (T) t; diff --git a/src/main/java/org/apache/groovy/runtime/async/DefaultAsyncChannel.java b/src/main/java/org/apache/groovy/runtime/async/DefaultAsyncChannel.java index b2d7aafad75..cfc175d385e 100644 --- a/src/main/java/org/apache/groovy/runtime/async/DefaultAsyncChannel.java +++ b/src/main/java/org/apache/groovy/runtime/async/DefaultAsyncChannel.java @@ -71,6 +71,8 @@ public final class DefaultAsyncChannel implements AsyncChannel { private final Deque> waitingReceivers = new ConcurrentLinkedDeque<>(); private final int capacity; private volatile boolean closed; + /** Failure passed to {@link #close(Throwable)}, published before {@link #closed}. */ + private volatile Throwable closeCause; // NOSONAR java:S3077 — published as a reference; the Throwable is not mutated /** * Run once, outside the lock, by the call that closes the channel; {@code null} for none. * {@link AtomicReference} rather than {@code volatile}: java:S3077 flags a volatile @@ -286,9 +288,27 @@ private Awaitable receive(SelectClaim claim) { @Override public boolean close() { + return finishClose(null); + } + + @Override + public Throwable getCloseCause() { + return closeCause; + } + + /** Retains {@code cause} and closes. Does not call the interface default. */ + @Override + public boolean close(Throwable cause) { + return finishClose(cause); + } + + private boolean finishClose(Throwable cause) { lock.lock(); try { - if (closed) return false; + if (closed) { + return false; + } + closeCause = cause; closed = true; finishCloseLocked(); } finally { @@ -344,7 +364,13 @@ private void runCloseHook() { /** * Returns a blocking iterator that receives values until the channel * is closed and drained. Each {@code next()} call blocks until a value - * is available. {@link ChannelClosedException} signals end-of-iteration. + * is available. A {@link ChannelClosedException} with no cause ends + * iteration. A failure retained by {@link #close(Throwable)} is that + * exception's cause and is thrown here, once, for the caller to catch. + * {@code for}, {@code each}, {@code collect} and {@code toList} all use + * this iterator. A Java enhanced {@code for} calls {@code hasNext()}, + * which may throw a checked exception that {@link Iterator} does not + * declare. */ @Override public Iterator iterator() { @@ -364,6 +390,10 @@ public boolean hasNext() { return true; } catch (ChannelClosedException e) { done = true; + Throwable failure = e.getCause(); + if (failure != null) { + AsyncSupport.rethrow(failure); + } return false; } } @@ -510,12 +540,21 @@ private void refillBufferFromWaitingSenders() { } } - private static ChannelClosedException closedForSend() { - return new ChannelClosedException("channel is closed for send"); + private ChannelClosedException closedForSend() { + return closed("channel is closed for send"); + } + + private ChannelClosedException closedForReceive() { + return closed("channel is closed"); } - private static ChannelClosedException closedForReceive() { - return new ChannelClosedException("channel is closed"); + /** One terminal signal. A retained failure rides on the cause. */ + private ChannelClosedException closed(String message) { + Throwable cause = closeCause; + if (cause == null) { + return new ChannelClosedException(message); + } + return new ChannelClosedException(message, cause); } /** diff --git a/src/spec/doc/_core-async-channels.adoc b/src/spec/doc/_core-async-channels.adoc index db2aa310d9a..cadf1dc56d8 100644 --- a/src/spec/doc/_core-async-channels.adoc +++ b/src/spec/doc/_core-async-channels.adoc @@ -38,6 +38,56 @@ With virtual threads, this blocking is essentially free. Since channels implement `Iterable`, they also work with regular `for` loops and Groovy collection methods. +[[async-channel-failure]] +=== When the producer fails + +`AsyncChannel.produce` runs the producer and closes the channel when the +body finishes. A normal return is a normal close. A thrown failure is +passed to `close(cause)`. `send` waits until the channel accepts the value: + +[source,groovy] +---- +include::../test/AsyncAwaitSpecTest.groovy[tags=channel_failure,indent=0] +---- + +`close(cause)` is that underlying call. It returns `true` when that call +closes the channel. The first close wins, so a later close returns `false` +and leaves the retained failure unchanged. `close(null)` is a normal close. +A producer that does not use `produce` writes the same try / close / catch +around its own sends. + +Values already sent are still delivered. `receive` and `send` then fail +with `ChannelClosedException`, and the producer's exception is its cause. +A loop written with `receive` has to read that cause. Catching +`ChannelClosedException` and ignoring `getCause()` drops the failure. + +`for await` throws the producer's exception after the values already sent. +The loop does not catch it. The code that reads the channel writes the +`try` / `catch`. + +The same iterator backs a plain `for` loop and the collection methods +`each`, `collect` and `toList`, so those throw the failure too. A Java +enhanced `for` calls `Iterator.hasNext()`, which can throw a checked +exception that `Iterator` does not declare. Java code cannot catch that +exception by its checked type. + +A custom `AsyncChannel` that does not override `close(Throwable)` still +just closes, so iteration ends instead of throwing. `defer { channel.close() }` +is a normal close: it stops the loop, and the task's exception stays on +the `Awaitable` returned by `async`. +`map`, `filter`, `split` and `tap` carry a failure through to their +output, so the same `try` / `catch` sees it at the end of the pipeline. +Closing one side of a `split` does not fail the other. `merge` closes its +output when either input fails, after it has forwarded values a relay has +already taken and values already buffered on the other input. It keeps +taking from that input while the buffer is non-empty, so a producer that +refills the buffer delays the failure until a gap appears. It does not +close that input. A producer still waiting to hand a value over is not +released. When both inputs fail, the first failure retained on the output +wins. +`BroadcastChannel.close(failure)` returns the same boolean and closes every +subscriber the same way. + [[async-channel-composition]] == Channel Composition @@ -82,7 +132,10 @@ for await (val in pipeline) { process(val) } === merge -Combines two channels into one, interleaving values as they arrive: +Combines two channels into one, interleaving values as they arrive. +The output closes when both inputs have closed normally. If either input +fails, values already taken or buffered are forwarded and the output then +closes with that failure: [source,groovy] ---- @@ -131,8 +184,13 @@ println "Channel ${result.index}: ${result.value}" A select takes exactly one value from exactly one channel; the other channels are left untouched. When several channels already have a value -waiting, the one listed first is taken. If every channel is closed and -drained, `select()` fails with `ChannelClosedException`. +waiting, the one met first in registration order is taken. Priority order +meets them as listed. `fair()` and `random()` change that order, as +described below. A channel closed with a failure still delivers a value +that was already buffered. The failure fails the select only when nothing +buffered remains. A normal close does not fail the select. That channel +is skipped until every channel has closed normally, and only then does +`select()` fail with `ChannelClosedException`. That priority by list order means a channel that always has a value waiting starves the channels listed after it. `fair()` returns a select diff --git a/src/spec/doc/core-async-await.adoc b/src/spec/doc/core-async-await.adoc index 028110afaf2..c75e3cd4f19 100644 --- a/src/spec/doc/core-async-await.adoc +++ b/src/spec/doc/core-async-await.adoc @@ -210,6 +210,12 @@ If a deferred action returns an `Awaitable` or `Future`, the result is awaited before the next deferred action runs, ensuring orderly cleanup of asynchronous resources. +`defer { channel.close() }` closes the channel when the task ends, including +when the task throws, so `for await` does not wait forever. That close has +no cause, so the loop ends normally. The task's exception stays on the +`Awaitable` from `async`. To make `for await` throw it, catch the failure +and call `channel.close(failure)`. See <>. + [[async-structured]] == Structured Concurrency @@ -529,6 +535,9 @@ type, e.g. `AtomicInteger` for a shared counter, or thread-safe types from | `AsyncChannel.create(n)` | Create a buffered (or unbuffered) channel for inter-task communication. +| `channel.close(failure)` +| Close a channel with a failure. `for await` throws that failure; the reader writes `try` / `catch`. + | `ChannelSelect.after(ms)` | Timer offer: a `ChannelSelect` branch that re-arms each round and commits with the `Instant` it fired. diff --git a/src/spec/test/AsyncAwaitSpecTest.groovy b/src/spec/test/AsyncAwaitSpecTest.groovy index 40830e13c3a..a3cba17c339 100644 --- a/src/spec/test/AsyncAwaitSpecTest.groovy +++ b/src/spec/test/AsyncAwaitSpecTest.groovy @@ -260,6 +260,32 @@ final class AsyncAwaitSpecTest { ''' } + @Test + void testChannelFailureReachesForAwait() { + assertScript ''' + import groovy.concurrent.AsyncChannel + + // tag::channel_failure[] + def channel = AsyncChannel.produce(2) { + for (int i = 1; i <= 3; i++) { + if (i == 3) throw new IllegalStateException('sensor failed') + it.send(i * i) + } + } + def seen = [] + try { + for await (value in channel) { + seen << value + } + assert false : 'for await should throw the producer failure' + } catch (IllegalStateException e) { + assert e.message == 'sensor failed' + } + assert seen == [1, 4] + // end::channel_failure[] + ''' + } + @Test void testCspProcessNetwork() { assertScript ''' diff --git a/src/test/groovy/groovy/concurrent/AsyncChannelFailureTest.groovy b/src/test/groovy/groovy/concurrent/AsyncChannelFailureTest.groovy new file mode 100644 index 00000000000..47add5ad826 --- /dev/null +++ b/src/test/groovy/groovy/concurrent/AsyncChannelFailureTest.groovy @@ -0,0 +1,945 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package groovy.concurrent + +import org.apache.groovy.runtime.async.AsyncSupport +import org.apache.groovy.runtime.async.GroovyPromise +import org.junit.jupiter.api.Test + +import java.util.concurrent.CompletableFuture +import java.util.concurrent.Executor +import java.util.concurrent.LinkedBlockingQueue +import java.util.function.Predicate + +import static org.junit.jupiter.api.Assertions.assertThrows + +/** + * A producer closes the channel with the failure it caught. + * {@code for await} throws that failure. The reader writes the catch. + * {@code await receive()} still fails with {@link ChannelClosedException}. + */ +final class AsyncChannelFailureTest { + + @Test + void normalCloseEndsTheLoop() { + def channel = AsyncChannel.create(2) + await channel.send(1) + await channel.send(2) + assert channel.close() + assert !channel.close() + assert channel.closeCause == null + + def seen = [] + for await (value in channel) { + seen << value + } + assert seen == [1, 2] + } + + @Test + void closeNullIsANormalClose() { + def channel = AsyncChannel.create(1) + assert channel.close(null) + assert !channel.close(null) + assert channel.closed + assert channel.closeCause == null + def received = assertThrows(ChannelClosedException) { + await channel.receive() + } + assert received.cause == null + def sent = assertThrows(ChannelClosedException) { + await channel.send('late') + } + assert sent.cause == null + } + + @Test + void callerCatchesTheCheckedFailureFromForAwait() { + def channel = AsyncChannel.create(1) + def failure = new IOException('disk') + async { + try { + throw failure + } catch (Exception caught) { + channel.close(caught) + } + } + def error = assertThrows(IOException) { + for await (ignored in channel) { + } + } + assert error.is(failure) + } + + @Test + void forAwaitThrowsTheFailureAfterBufferedValues() { + def channel = AsyncChannel.create(2) + def failure = new IllegalStateException('boom') + await channel.send(1) + await channel.send(2) + channel.close(failure) + + assert await(channel.receive()) == 1 + assert await(channel.receive()) == 2 + def closed = assertThrows(ChannelClosedException) { + await channel.receive() + } + assert closed.cause.is(failure) + + def again = AsyncChannel.create(2) + await again.send(1) + await again.send(2) + again.close(failure) + def seen = [] + def error = assertThrows(IllegalStateException) { + for await (value in again) { + seen << value + } + } + assert seen == [1, 2] + assert error.is(failure) + assert again.closeCause.is(failure) + } + + @Test + void forAwaitThrowsAChannelClosedExceptionUsedAsTheFailure() { + def channel = AsyncChannel.create(1) + def failure = new ChannelClosedException('not a clean end') + channel.close(failure) + + def iterator = channel.iterator() + def error = assertThrows(ChannelClosedException) { + iterator.hasNext() + } + assert error.is(failure) + assert !iterator.hasNext() + } + + @Test + void forAwaitThrowsCheckedExceptionsAndErrors() { + def checked = AsyncChannel.create(1) + def io = new IOException('disk') + checked.close(io) + def received = assertThrows(ChannelClosedException) { + await checked.receive() + } + assert received.cause.is(io) + + def iterated = AsyncChannel.create(1) + iterated.close(io) + def iterator = iterated.iterator() + def thrown = assertThrows(IOException) { + iterator.hasNext() + } + assert thrown.is(io) + assert !iterator.hasNext() + + def serious = AsyncChannel.create(1) + def failure = new AssertionError('ae') + serious.close(failure) + def seriousIterator = serious.iterator() + def error = assertThrows(AssertionError) { + seriousIterator.hasNext() + } + assert error.is(failure) + assert !seriousIterator.hasNext() + } + + @Test + void sendAndPendingOperationsCarryTheFailure() { + def closed = AsyncChannel.create(1) + def failure = new IllegalStateException('boom') + closed.close(failure) + def sent = assertThrows(ChannelClosedException) { + await closed.send('x') + } + assert sent.cause.is(failure) + + def senders = AsyncChannel.create(0) + def pendingSend = senders.send('x') + def sendFailure = new IllegalStateException('send-boom') + senders.close(sendFailure) + def sendError = assertThrows(ChannelClosedException) { + await pendingSend + } + assert sendError.cause.is(sendFailure) + + def receivers = AsyncChannel.create(0) + def pendingReceive = receivers.receive() + def receiveFailure = new IllegalStateException('receive-boom') + receivers.close(receiveFailure) + def receiveError = assertThrows(ChannelClosedException) { + await pendingReceive + } + assert receiveError.cause.is(receiveFailure) + } + + @Test + void firstCloseWins() { + def channel = AsyncChannel.create(0) + def first = new IllegalStateException('first') + def second = new IllegalStateException('second') + assert channel.close(first) + assert !channel.close(second) + assert !channel.close() + assert channel.closeCause.is(first) + + def later = AsyncChannel.create(0) + assert later.close() + assert !later.close(new IllegalStateException('late')) + assert later.closeCause == null + def error = assertThrows(ChannelClosedException) { + await later.receive() + } + assert error.cause == null + } + + @Test + void closingATimerStillCancelsIt() { + def timer = AsyncChannel.after(60_000) + assert timer.close() + assert timer.closed + assert timer.closeCause == null + assert !timer.close(new IllegalStateException('late')) + assert timer.closeCause == null + } + + @Test + void defaultCloseDoesNotRetainAFailure() { + def channel = new CloseOnlyChannel() + assert channel.close(new IllegalStateException('not retained')) + assert channel.closed + assert channel.closeCause == null + assert !channel.close(null) + assert channel.closed + } + + @Test + void stagesDeliverTheFailureToForAwait() { + def fed = AsyncChannel.create(4) + def filtered = fed.filter { + if (it == 2) throw new IllegalStateException('bad-item') + it % 2 == 0 + } + async { + await fed.send(1) + await fed.send(2) + fed.close() + } + def seen = [] + def error = assertThrows(IllegalStateException) { + for await (value in filtered) { + seen << value + } + } + assert seen.isEmpty() + assert error.message == 'bad-item' + + def upstream = AsyncChannel.create(4) + def failure = new IllegalStateException('map-boom') + def mapped = upstream.map { it * 10 } + async { + await upstream.send(2) + upstream.close(failure) + } + def mappedSeen = [] + def mappedError = assertThrows(IllegalStateException) { + for await (value in mapped) { + mappedSeen << value + } + } + assert mappedSeen == [20] + assert mappedError.is(failure) + } + + @Test + void mergeSplitAndTapCarryTheFailure() { + def left = AsyncChannel.create(2) + def right = AsyncChannel.create(0) + def merged = left.merge(right) + def mergeFailure = new IllegalStateException('merge-boom') + async { + await left.send(1) + left.close(mergeFailure) + } + def mergedSeen = [] + def mergedConsumption = async { + for await (value in merged) { + mergedSeen << value + } + } + try { + def mergedError = assertThrows(IllegalStateException) { + await mergedConsumption.orTimeoutMillis(2000) + } + assert mergedSeen == [1] + assert mergedError.is(mergeFailure) + assert !right.closed + } finally { + right.close() + } + + def splitSource = AsyncChannel.create(4) + def parts = splitSource.split({ it >= 5 } as Predicate) + async { + await splitSource.send(1) + await splitSource.send(6) + splitSource.close(new IllegalStateException('split-boom')) + } + def high = [] + def highError = assertThrows(IllegalStateException) { + for await (value in parts[0]) { + high << value + } + } + def low = [] + def lowError = assertThrows(IllegalStateException) { + for await (value in parts[1]) { + low << value + } + } + assert high == [6] + assert low == [1] + assert highError.message == 'split-boom' + assert lowError.message == 'split-boom' + + def tapSource = AsyncChannel.create(2) + def monitor = AsyncChannel.create(2) + def passed = tapSource.tap(monitor) + async { + await tapSource.send('a') + tapSource.close(new IllegalStateException('tap-boom')) + } + def passedSeen = [] + def tapError = assertThrows(IllegalStateException) { + for await (value in passed) { + passedSeen << value + } + } + assert passedSeen == ['a'] + assert tapError.message == 'tap-boom' + monitor.close() + def copied = [] + for (value in monitor) { + copied << value + } + assert copied == ['a'] + } + + @Test + void selectThrowsTheFailureAndIgnoresAWithdrawnBranch() { + def failed = AsyncChannel.create(0) + def open = AsyncChannel.create(1) + def pending = ChannelSelect.from(failed, open).select() + def failure = new IllegalStateException('boom') + failed.close(failure) + def error = assertThrows(IllegalStateException) { + await pending + } + assert error.is(failure) + await open.send('kept') + assert await(open.receive().orTimeoutMillis(1000)) == 'kept' + + def marked = AsyncChannel.create(0) + def marker = new ChannelClosedException('not a clean end') + def markedSelect = ChannelSelect.from(marked).select() + marked.close(marker) + def markedError = assertThrows(ChannelClosedException) { + await markedSelect + } + assert markedError.is(marker) + + def first = AsyncChannel.create(0) + def second = AsyncChannel.create(0) + first.close() + second.close() + def closed = assertThrows(ChannelClosedException) { + await ChannelSelect.from(first, second).select() + } + assert closed.message == 'all channels are closed' + + def ready = AsyncChannel.create(1) + def other = AsyncChannel.create(1) + def race = ChannelSelect.from(ready, other).select() + await ready.send('win') + def result = await race + assert result.value == 'win' + assert result.index == 0 + await other.send('still-open') + assert await(other.receive()) == 'still-open' + } + + @Test + void selectKeepsACommittedValueWhenAnotherChannelFailsLater() { + def left = AsyncChannel.create(0) + def right = AsyncChannel.create(0) + def pending = ChannelSelect.from(left, right).select() + await left.send('win') + def failure = new IllegalStateException('late') + right.close(failure) + + def result = await pending + assert result.value == 'win' + assert result.index == 0 + assert right.closeCause.is(failure) + + def failedFirst = AsyncChannel.create(0) + def buffered = AsyncChannel.create(1) + def earlier = new IllegalStateException('earlier') + failedFirst.close(earlier) + await buffered.send('kept') + def error = assertThrows(IllegalStateException) { + await ChannelSelect.from(failedFirst, buffered).select() + } + assert error.is(earlier) + assert await(buffered.receive()) == 'kept' + + def sendFailed = AsyncChannel.create(0) + sendFailed.close(failure) + def sendError = assertThrows(IllegalStateException) { + await ChannelSelect.offers(ChannelSelect.send(sendFailed, 'x')).select() + } + assert sendError.is(failure) + + def ready = AsyncChannel.create(1) + def failedLater = AsyncChannel.create(0) + await ready.send('win') + assert failedLater.close(failure) + def preferred = await ChannelSelect.from(ready, failedLater).select() + assert preferred.value == 'win' + assert preferred.index == 0 + assert failedLater.closeCause.is(failure) + + def bufferedFailure = AsyncChannel.create(1) + def untouched = AsyncChannel.create(1) + await bufferedFailure.send('kept') + assert bufferedFailure.close(failure) + await untouched.send('later') + def bufferedResult = await ChannelSelect.from(bufferedFailure, untouched).select() + assert bufferedResult.index == 0 + assert bufferedResult.value == 'kept' + assert await(untouched.receive()) == 'later' + } + + @Test + void selectKeepsACommittedTransferWhenACloseLosesTheClaim() { + // Not a DefaultAsyncChannel, so select completes the branch future + // before it can withdraw the sibling. Cancel is a no-op, which lets + // the sibling still report its close after the transfer has committed. + def value = new ManualChannel() + def failed = new ManualChannel() + def pending = ChannelSelect.from(value, failed).select() + value.future.complete('win') + failed.future.completeExceptionally( + new ChannelClosedException('closed', new IllegalStateException('late'))) + + def result = await pending + assert result.value == 'win' + assert result.index == 0 + } + + @Test + void closingOneSplitOutputDoesNotFailTheOther() { + def source = AsyncChannel.create(4) + def parts = source.split({ it >= 5 } as Predicate) + def side = new IllegalStateException('only-high') + parts[0].close(side) + async { + await source.send(1) + await source.send(6) + source.close() + } + def low = [] + for await (value in parts[1]) { + low << value + } + assert low == [1] + assert parts[1].closeCause == null + def highError = assertThrows(IllegalStateException) { + for await (ignored in parts[0]) { + } + } + assert highError.is(side) + } + + @Test + void tapIgnoresAClosedSideChannel() { + def source = AsyncChannel.create(2) + def monitor = AsyncChannel.create(1) + def side = new IllegalStateException('side') + monitor.close(side) + def passed = source.tap(monitor) + def sourceFailure = new IllegalStateException('source') + async { + await source.send('a') + await source.send('b') + source.close(sourceFailure) + } + def seen = [] + def error = assertThrows(IllegalStateException) { + for await (value in passed) { + seen << value + } + } + assert seen == ['a', 'b'] + assert error.is(sourceFailure) + assert monitor.closeCause.is(side) + } + + @Test + void produceClosesNormallyAndOnFailure() { + def squares = AsyncChannel.produce { + it.send(1) + it.send(2) + assert it.channel.capacity == 0 + } + def seen = [] + for await (value in squares) { + seen << value + } + assert seen == [1, 2] + assert squares.closeCause == null + assert !squares.close() + + def channel = AsyncChannel.produce(2) { + it.send(1) + throw new IOException('disk') + } + def buffered = [] + def error = assertThrows(IOException) { + for await (value in channel) { + buffered << value + } + } + assert buffered == [1] + assert error.message == 'disk' + assert channel.closeCause.is(error) + + def serious = AsyncChannel.produce(1) { + throw new AssertionError('ae') + } + def seriousError = assertThrows(AssertionError) { + for await (ignored in serious) { + } + } + assert seriousError.message == 'ae' + assert serious.closeCause.is(seriousError) + + def marked = AsyncChannel.produce(1) { + throw new ChannelClosedException('not a clean end') + } + def markedError = assertThrows(ChannelClosedException) { + for await (ignored in marked) { + } + } + assert markedError.message == 'not a clean end' + assert marked.closeCause.is(markedError) + } + + @Test + void produceSendUnblocksWhenTheConsumerCloses() { + def channel = AsyncChannel.produce(0) { + it.send('a') + it.send('b') + } + assert await(channel.receive()) == 'a' + assert channel.close() + assert channel.closeCause == null + def closed = assertThrows(ChannelClosedException) { + await channel.receive().orTimeoutMillis(2000) + } + assert closed.cause == null + } + + @Test + void produceRejectsANullBodyAndANegativeCapacity() { + assertThrows(NullPointerException) { + AsyncChannel.produce(null) + } + assertThrows(NullPointerException) { + AsyncChannel.produce(1, null) + } + assertThrows(IllegalArgumentException) { + AsyncChannel.produce(-1) { } + } + } + + @Test + void produceSendRejectsNull() { + def channel = AsyncChannel.produce(1) { + it.send(null) + } + def error = assertThrows(NullPointerException) { + for await (ignored in channel) { + } + } + assert channel.closeCause.is(error) + } + + @Test + void mergeKeepsOneFailureWhenBothInputsFail() { + def left = AsyncChannel.create(1) + def right = AsyncChannel.create(1) + def leftFailure = new IllegalStateException('left') + def rightFailure = new IllegalStateException('right') + left.close(leftFailure) + right.close(rightFailure) + def merged = left.merge(right) + def error = assertThrows(IllegalStateException) { + for await (ignored in merged) { + } + } + assert error.is(leftFailure) || error.is(rightFailure) + } + + @Test + void mergeForwardsAValueWhenCancelLosesTheCommit() { + def left = AsyncChannel.create(0) + def failure = new IllegalStateException('left') + assert left.close(failure) + def right = new CancelLosesChannel() + def queue = new LinkedBlockingQueue() + AsyncSupport.setExecutor({ Runnable task -> queue.add(task) } as Executor) + AsyncChannel merged + try { + merged = left.merge(right) + assert queue.size() == 2 + def leftTask = queue.take() + def rightTask = queue.take() + def leftThread = Thread.startDaemon { leftTask.run() } + waitUntil { leftThread.state == Thread.State.WAITING } + def rightThread = Thread.startDaemon { rightTask.run() } + // The relays are already running. Restore the executor so the + // consumer below is not queued behind them. + AsyncSupport.resetExecutor() + def seen = [] + def consumption = async { + for await (value in merged) { + seen << value + } + } + def error = assertThrows(IllegalStateException) { + await consumption.orTimeoutMillis(2000) + } + assert seen == ['kept'] + assert error.is(failure) + leftThread.join(2000) + rightThread.join(2000) + assert !leftThread.alive + assert !rightThread.alive + } finally { + AsyncSupport.resetExecutor() + right.close() + merged?.close() + } + } + + private static void waitUntil(Closure condition) { + long deadline = System.currentTimeMillis() + 2000 + while (!condition.call()) { + if (System.currentTimeMillis() > deadline) { + throw new AssertionError('waitUntil timed out after 2000ms') + } + Thread.sleep(2) + } + } + + @Test + void mergeClosesAsSoonAsOneInputFails() { + def left = AsyncChannel.create(1) + def right = AsyncChannel.create(0) + def failure = new IOException('sensor failed') + await left.send(7) + assert left.close(failure) + def merged = left.merge(right) + def seen = [] + def consumption = async { + for await (value in merged) { + seen << value + } + } + try { + def error = assertThrows(IOException) { + await consumption.orTimeoutMillis(2000) + } + assert seen == [7] + assert error.is(failure) + assert merged.closeCause.is(failure) + assert !right.closed + } finally { + right.close() + merged.close() + } + } + + @Test + void mergeForwardsABufferedSiblingValueBeforeTheFailure() { + def left = AsyncChannel.create(1) + def right = AsyncChannel.create(1) + def failure = new IllegalStateException('left') + await left.send(1) + await right.send(2) + assert left.close(failure) + def merged = left.merge(right) + def seen = [] + def consumption = async { + for await (value in merged) { + seen << value + } + } + try { + def error = assertThrows(IllegalStateException) { + await consumption.orTimeoutMillis(2000) + } + assert seen.sort() == [1, 2] + assert error.is(failure) + assert merged.closeCause.is(failure) + assert !right.closed + assert right.bufferedSize == 0 + right.close() + assert merged.closeCause.is(failure) + } finally { + right.close() + merged.close() + } + } + + @Test + void mergeRejectsANullInput() { + def channel = AsyncChannel.create(0) + assertThrows(NullPointerException) { + channel.merge(null) + } + assert !channel.closed + } + + @Test + void iterationHelpersThrowTheFailure() { + def failure = new IOException('disk') + + def forChannel = AsyncChannel.create(2) + await forChannel.send(1) + await forChannel.send(2) + assert forChannel.close(failure) + def forSeen = [] + def forError = assertThrows(IOException) { + for (value in forChannel) { + forSeen << value + } + } + assert forSeen == [1, 2] + assert forError.is(failure) + + def eachChannel = AsyncChannel.create(2) + await eachChannel.send(1) + await eachChannel.send(2) + assert eachChannel.close(failure) + def eachSeen = [] + def eachError = assertThrows(IOException) { + eachChannel.each { eachSeen << it } + } + assert eachSeen == [1, 2] + assert eachError.is(failure) + + def collectChannel = AsyncChannel.create(2) + await collectChannel.send(1) + await collectChannel.send(2) + assert collectChannel.close(failure) + def collectSeen = [] + def collectError = assertThrows(IOException) { + collectChannel.collect { collectSeen << it; it } + } + assert collectSeen == [1, 2] + assert collectError.is(failure) + + def listChannel = AsyncChannel.create(1) + await listChannel.send(1) + assert listChannel.close(failure) + def listError = assertThrows(IOException) { + listChannel.toList() + } + assert listError.is(failure) + } + + @Test + void receiveLoopReadsTheFailureFromTheCause() { + def channel = AsyncChannel.create(2) + await channel.send(1) + await channel.send(2) + def failure = new IOException('disk') + assert channel.close(failure) + def seen = [] + while (true) { + try { + seen << await(channel.receive()) + } catch (ChannelClosedException closed) { + assert closed.cause.is(failure) + break + } + } + assert seen == [1, 2] + def sent = assertThrows(ChannelClosedException) { + await channel.send('late') + } + assert sent.cause.is(failure) + } + + /** + * Reports a buffered value, then a receive whose {@code cancel()} loses + * to a sender that has already committed. {@code isDone()} stays false + * so {@code relay} takes the cancel branch. + */ + private static final class CancelLosesChannel implements AsyncChannel { + private final LateCommitPromise promise = new LateCommitPromise() + private int buffered = 1 + + @Override + int getCapacity() { 1 } + + @Override + int getBufferedSize() { buffered } + + @Override + boolean isClosed() { false } + + @Override + Awaitable send(String value) { + throw new UnsupportedOperationException('send') + } + + @Override + Awaitable receive() { + buffered = 0 + promise + } + + @Override + boolean close() { false } + + @Override + Iterator iterator() { + throw new UnsupportedOperationException('iterator') + } + } + + private static final class LateCommitPromise extends GroovyPromise { + private final CompletableFuture committed = new CompletableFuture<>() + + LateCommitPromise() { + super(new CompletableFuture<>()) + } + + @Override + boolean isDone() { false } + + @Override + boolean cancel() { + committed.complete('kept') + return false + } + + @Override + String get() { + committed.get() + } + } + + /** + * A channel whose one receive completes only when the test completes + * {@link #future}. {@link #cancel()} does nothing, so a select cannot + * retire the branch by withdrawing it. + */ + private static final class ManualChannel implements AsyncChannel { + final CompletableFuture future = new CompletableFuture<>() + + @Override + int getCapacity() { 0 } + + @Override + int getBufferedSize() { 0 } + + @Override + boolean isClosed() { false } + + @Override + Awaitable send(String value) { + throw new UnsupportedOperationException('send') + } + + @Override + Awaitable receive() { + new StickyPromise(future) + } + + @Override + boolean close() { false } + + @Override + Iterator iterator() { + throw new UnsupportedOperationException('iterator') + } + } + + private static final class StickyPromise extends GroovyPromise { + StickyPromise(CompletableFuture future) { + super(future) + } + + @Override + boolean cancel() { false } + } + + /** Does not override {@code close(Throwable)}, so the interface default applies. */ + private static final class CloseOnlyChannel implements AsyncChannel { + private boolean closed + + @Override + int getCapacity() { 0 } + + @Override + int getBufferedSize() { 0 } + + @Override + boolean isClosed() { closed } + + @Override + Awaitable send(String value) { + throw new UnsupportedOperationException('send') + } + + @Override + Awaitable receive() { + throw new UnsupportedOperationException('receive') + } + + @Override + boolean close() { + if (closed) { + return false + } + closed = true + return true + } + + @Override + Iterator iterator() { + throw new UnsupportedOperationException('iterator') + } + } +} diff --git a/src/test/groovy/groovy/concurrent/BroadcastChannelAsPublisherTest.groovy b/src/test/groovy/groovy/concurrent/BroadcastChannelAsPublisherTest.groovy index 0cea5af7a21..26ce3ea6ab3 100644 --- a/src/test/groovy/groovy/concurrent/BroadcastChannelAsPublisherTest.groovy +++ b/src/test/groovy/groovy/concurrent/BroadcastChannelAsPublisherTest.groovy @@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicReference import static groovy.test.GroovyAssert.assertScript +import static org.junit.jupiter.api.Assertions.assertThrows final class BroadcastChannelAsPublisherTest { @@ -219,4 +220,60 @@ final class BroadcastChannelAsPublisherTest { assert collected == ['x', 'y', 'z'] ''' } + + @Test + void forAwaitOverPublisherThrowsARetainedFailure() { + def broadcast = BroadcastChannel.create() + def publisher = broadcast.asPublisher() + def failure = new IllegalStateException('pub-boom') + async { + waitUntil { broadcast.subscriberCount > 0 } + await broadcast.send('a') + broadcast.close(failure) + } + def seen = [] + def error = assertThrows(IllegalStateException) { + for await (item in publisher) { + seen << item + } + } + assert seen == ['a'] + assert error.is(failure) + assert broadcast.closeCause.is(failure) + } + + @Test + void drainReceivesTheFailureFromTheClosedChannel() { + // The drain task is blocked in receive() when the broadcast fails, + // so the failure arrives as the cause of ChannelClosedException + // rather than from a later isClosed() check. + def broadcast = BroadcastChannel.create() + def publisher = broadcast.asPublisher() + def failure = new IllegalStateException('from-receive') + def error = new AtomicReference() + def done = new CountDownLatch(1) + publisher.subscribe(new Flow.Subscriber() { + @Override void onSubscribe(Flow.Subscription subscription) { + subscription.request(Long.MAX_VALUE) + } + @Override void onNext(String item) { } + @Override void onError(Throwable thrown) { + error.set(thrown) + done.countDown() + } + @Override void onComplete() { done.countDown() } + }) + waitUntil { subscriberWaiting(broadcast) } + assert broadcast.close(failure) + assert done.await(2, TimeUnit.SECONDS) + assert error.get().is(failure) + } + + /** The publisher's backing channel has a receiver parked in {@code receive()}. */ + private static boolean subscriberWaiting(BroadcastChannel broadcast) { + def subscribers = BroadcastChannel.getDeclaredField('subscribers') + subscribers.accessible = true + def channels = subscribers.get(broadcast) as List> + return !channels.isEmpty() && channels[0].toString().contains('waitingReceivers=1') + } } diff --git a/src/test/groovy/groovy/concurrent/BroadcastChannelTest.groovy b/src/test/groovy/groovy/concurrent/BroadcastChannelTest.groovy index 568678cdf4f..88372b4474c 100644 --- a/src/test/groovy/groovy/concurrent/BroadcastChannelTest.groovy +++ b/src/test/groovy/groovy/concurrent/BroadcastChannelTest.groovy @@ -21,6 +21,7 @@ package groovy.concurrent import org.junit.jupiter.api.Test import static groovy.test.GroovyAssert.assertScript +import static org.junit.jupiter.api.Assertions.assertThrows final class BroadcastChannelTest { @@ -92,4 +93,77 @@ final class BroadcastChannelTest { assert lateResults == ['second'] ''' } + + @Test + void closeWithAFailureReachesEverySubscriber() { + def broadcast = BroadcastChannel.create() + def first = broadcast.subscribe() + def second = broadcast.subscribe() + def failure = new IllegalStateException('broadcast-boom') + await broadcast.send('a') + assert broadcast.close(failure) + + for (sub in [first, second]) { + assert await(sub.receive()) == 'a' + def error = assertThrows(IllegalStateException) { + for await (ignored in sub) { + } + } + assert error.is(failure) + } + assert broadcast.closeCause.is(failure) + + def late = assertThrows(ChannelClosedException) { + broadcast.subscribe() + } + assert late.cause.is(failure) + def sent = assertThrows(ChannelClosedException) { + await broadcast.send('late') + } + assert sent.cause.is(failure) + assert sent.message == 'BroadcastChannel is closed' + + broadcast.close() + assert !broadcast.close(new IllegalStateException('later')) + assert broadcast.closeCause.is(failure) + } + + @Test + void closeNullEndsSubscribersNormally() { + def broadcast = BroadcastChannel.create() + def sub = broadcast.subscribe() + await broadcast.send('a') + assert broadcast.close(null) + def seen = [] + for await (value in sub) { + seen << value + } + assert seen == ['a'] + assert broadcast.closeCause == null + + def closed = BroadcastChannel.create() + def child = closed.subscribe() + closed.close() + assert !closed.close(new IllegalStateException('late')) + def drained = [] + for await (value in child) { + drained << value + } + assert drained.isEmpty() + assert closed.closeCause == null + def rejected = assertThrows(ChannelClosedException) { + closed.subscribe() + } + assert rejected.cause == null + + def empty = BroadcastChannel.create() + def only = new IllegalStateException('only') + assert empty.close(only) + assert !empty.close(null) + assert empty.closeCause.is(only) + def late = assertThrows(ChannelClosedException) { + empty.subscribe() + } + assert late.cause.is(only) + } }