Drain SinkManyEmitterProcessor buffer after cancel

reactor-core/src/jcstress/java/reactor/core/publisher/SinkManyEmitterProcessorTest.java

@@ -0,0 +1,34 @@
+package reactor.core.publisher;
+
+import org.openjdk.jcstress.annotations.*;
+import org.openjdk.jcstress.infra.results.I_Result;
+import org.openjdk.jcstress.infra.results.LI_Result;
+
+import static org.openjdk.jcstress.annotations.Expect.*;
+
+public class SinkManyEmitterProcessorTest {
+
+    @JCStressTest
+    @Outcome(id = {"0"}, expect = ACCEPTABLE, desc = "Queue is empty")
+    @Outcome(id = {"1"}, expect = ACCEPTABLE_INTERESTING, desc = "Queue is not empty")
+    @State
+    public static class ParallelSubscribersStressTest {
+
+        final SinkManyEmitterProcessor<Integer> sink = new SinkManyEmitterProcessor(true, 1);
+
+        @Actor
+        public void subscribeAndCancel() {
+            sink.subscribe().dispose();
+        }
+
+        @Actor
+        public void tryEmitNext1(LI_Result r) {
+            sink.tryEmitNext(1);
+        }
+
+        @Arbiter
+        public void arbiter(I_Result result) {
+            result.r1 = sink.queue.size();
+        }
+    }
+}

reactor-core/src/main/java/reactor/core/publisher/SinkManyEmitterProcessor.java

@@ -384,6 +384,7 @@ public Object scanUnsafe(Attr key) {
 
 	final void drain() {
 		if (WIP.getAndIncrement(this) != 0) {
+			WIP.decrementAndGet(this);
 			return;
 		}
 
@@ -398,6 +399,7 @@ final void drain() {
 			boolean empty = q == null || q.isEmpty();
 
 			if (checkTerminated(d, empty)) {
+				WIP.addAndGet(this, -missed);
 				return;
 			}
 
@@ -432,6 +434,7 @@ final void drain() {
 						v = null;
 					}
 					if (checkTerminated(d, v == null)) {
+						WIP.addAndGet(this, -missed);
 						return;
 					}
 					if (sourceMode != Fuseable.SYNC) {
@@ -459,6 +462,7 @@ final void drain() {
 					empty = v == null;
 
 					if (checkTerminated(d, empty)) {
+						WIP.addAndGet(this, -missed);
 						return;
 					}
 
@@ -593,13 +597,15 @@ final void remove(FluxPublish.PubSubInner<T> inner) {
 				//happens when the removed inner makes the subscribers array EMPTY
 				if (autoCancel && b == EMPTY && Operators.terminate(S, this)) {
 					if (WIP.getAndIncrement(this) != 0) {
+						WIP.decrementAndGet(this);
 						return;
 					}
 					terminate();
 					Queue<T> q = queue;
 					if (q != null) {
 						q.clear();
 					}
+					WIP.decrementAndGet(this);
 				}
 				return;
 			}

reactor-core/src/test/java/reactor/core/publisher/SinkManyEmitterProcessorTest.java

@@ -921,4 +921,22 @@ void emitNextWithNoSubscriberNoCapacityKeepsSinkOpenWithBuffer() {
 		            .expectTimeout(Duration.ofSeconds(1))
 		            .verify();
 	}
+
+	@Test
+	public void cancelledSinkClearsQueue() {
+		SinkManyEmitterProcessor<Integer> sinkManyEmitterProcessor = new SinkManyEmitterProcessor<>(true, 1);
+		// fill the buffer
+		assertThat(sinkManyEmitterProcessor.tryEmitNext(1)).as("filling buffer").isEqualTo(Sinks.EmitResult.OK);
+		StepVerifier.create(sinkManyEmitterProcessor)
+			.expectNext(1)
+			.expectTimeout(Duration.ofSeconds(1))
+			.verify();
+		sinkManyEmitterProcessor.asFlux().subscribe().dispose();
+
+		// fill the buffer
+		assertThat(sinkManyEmitterProcessor.tryEmitNext(1)).as("filling buffer").isEqualTo(Sinks.EmitResult.OK);
+		StepVerifier.create(sinkManyEmitterProcessor)
+			.verifyComplete();
+		assertThat(sinkManyEmitterProcessor.queue.isEmpty()).as("Buffer should be empty").isTrue();
+	}
 }