Merge #3168 into 3.5.0-M6

reactor · Aug 29, 2022 · 34e1a8d · 34e1a8d
2 parents 31b4273 + ba08c27
commit 34e1a8d
Show file tree

Hide file tree

Showing 4 changed files with 198 additions and 71 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/QueueDrainSubscriber.java b/reactor-core/src/main/java/reactor/core/publisher/QueueDrainSubscriber.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-2022 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -284,9 +284,24 @@ static <Q, S> boolean checkTerminated(boolean d, boolean empty,
 //-------------------------------------------------------------------
 
 /** Pads the header away from other fields. */
+@SuppressWarnings("unused")
 class QueueDrainSubscriberPad0 {
-	volatile long p1, p2, p3, p4, p5, p6, p7;
-	volatile long p8, p9, p10, p11, p12, p13, p14, p15;
+	byte pad000,pad001,pad002,pad003,pad004,pad005,pad006,pad007;//  8b
+	byte pad010,pad011,pad012,pad013,pad014,pad015,pad016,pad017;// 16b
+	byte pad020,pad021,pad022,pad023,pad024,pad025,pad026,pad027;// 24b
+	byte pad030,pad031,pad032,pad033,pad034,pad035,pad036,pad037;// 32b
+	byte pad040,pad041,pad042,pad043,pad044,pad045,pad046,pad047;// 40b
+	byte pad050,pad051,pad052,pad053,pad054,pad055,pad056,pad057;// 48b
+	byte pad060,pad061,pad062,pad063,pad064,pad065,pad066,pad067;// 56b
+	byte pad070,pad071,pad072,pad073,pad074,pad075,pad076,pad077;// 64b
+	byte pad100,pad101,pad102,pad103,pad104,pad105,pad106,pad107;// 72b
+	byte pad110,pad111,pad112,pad113,pad114,pad115,pad116,pad117;// 80b
+	byte pad120,pad121,pad122,pad123,pad124,pad125,pad126,pad127;// 88b
+	byte pad130,pad131,pad132,pad133,pad134,pad135,pad136,pad137;// 96b
+	byte pad140,pad141,pad142,pad143,pad144,pad145,pad146,pad147;//104b
+	byte pad150,pad151,pad152,pad153,pad154,pad155,pad156,pad157;//112b
+	byte pad160,pad161,pad162,pad163,pad164,pad165,pad166,pad167;//120b
+	byte pad170,pad171,pad172,pad173,pad174,pad175,pad176,pad177;//128b
 }
 
 /** The WIP counter. */
@@ -295,9 +310,24 @@ class QueueDrainSubscriberWip extends QueueDrainSubscriberPad0 {
 }
 
 /** Pads away the wip from the other fields. */
+@SuppressWarnings("unused")
 class QueueDrainSubscriberPad2 extends QueueDrainSubscriberWip {
-	volatile long p1a, p2a, p3a, p4a, p5a, p6a, p7a;
-	volatile long p8a, p9a, p10a, p11a, p12a, p13a, p14a, p15a;
+	byte pad000,pad001,pad002,pad003,pad004,pad005,pad006,pad007;//  8b
+	byte pad010,pad011,pad012,pad013,pad014,pad015,pad016,pad017;// 16b
+	byte pad020,pad021,pad022,pad023,pad024,pad025,pad026,pad027;// 24b
+	byte pad030,pad031,pad032,pad033,pad034,pad035,pad036,pad037;// 32b
+	byte pad040,pad041,pad042,pad043,pad044,pad045,pad046,pad047;// 40b
+	byte pad050,pad051,pad052,pad053,pad054,pad055,pad056,pad057;// 48b
+	byte pad060,pad061,pad062,pad063,pad064,pad065,pad066,pad067;// 56b
+	byte pad070,pad071,pad072,pad073,pad074,pad075,pad076,pad077;// 64b
+	byte pad100,pad101,pad102,pad103,pad104,pad105,pad106,pad107;// 72b
+	byte pad110,pad111,pad112,pad113,pad114,pad115,pad116,pad117;// 80b
+	byte pad120,pad121,pad122,pad123,pad124,pad125,pad126,pad127;// 88b
+	byte pad130,pad131,pad132,pad133,pad134,pad135,pad136,pad137;// 96b
+	byte pad140,pad141,pad142,pad143,pad144,pad145,pad146,pad147;//104b
+	byte pad150,pad151,pad152,pad153,pad154,pad155,pad156,pad157;//112b
+	byte pad160,pad161,pad162,pad163,pad164,pad165,pad166,pad167;//120b
+	byte pad170,pad171,pad172,pad173,pad174,pad175,pad176,pad177;//128b
 }
 
 /** Contains the requested field. */
@@ -310,7 +340,22 @@ class QueueDrainSubscriberPad3 extends QueueDrainSubscriberPad2 {
 }
 
 /** Pads away the requested from the other fields. */
+@SuppressWarnings("unused")
 class QueueDrainSubscriberPad4 extends QueueDrainSubscriberPad3 {
-	volatile long q1, q2, q3, q4, q5, q6, q7;
-	volatile long q8, q9, q10, q11, q12, q13, q14, q15;
+	byte pad000,pad001,pad002,pad003,pad004,pad005,pad006,pad007;//  8b
+	byte pad010,pad011,pad012,pad013,pad014,pad015,pad016,pad017;// 16b
+	byte pad020,pad021,pad022,pad023,pad024,pad025,pad026,pad027;// 24b
+	byte pad030,pad031,pad032,pad033,pad034,pad035,pad036,pad037;// 32b
+	byte pad040,pad041,pad042,pad043,pad044,pad045,pad046,pad047;// 40b
+	byte pad050,pad051,pad052,pad053,pad054,pad055,pad056,pad057;// 48b
+	byte pad060,pad061,pad062,pad063,pad064,pad065,pad066,pad067;// 56b
+	byte pad070,pad071,pad072,pad073,pad074,pad075,pad076,pad077;// 64b
+	byte pad100,pad101,pad102,pad103,pad104,pad105,pad106,pad107;// 72b
+	byte pad110,pad111,pad112,pad113,pad114,pad115,pad116,pad117;// 80b
+	byte pad120,pad121,pad122,pad123,pad124,pad125,pad126,pad127;// 88b
+	byte pad130,pad131,pad132,pad133,pad134,pad135,pad136,pad137;// 96b
+	byte pad140,pad141,pad142,pad143,pad144,pad145,pad146,pad147;//104b
+	byte pad150,pad151,pad152,pad153,pad154,pad155,pad156,pad157;//112b
+	byte pad160,pad161,pad162,pad163,pad164,pad165,pad166,pad167;//120b
+	byte pad170,pad171,pad172,pad173,pad174,pad175,pad176,pad177;//128b
 }
diff --git a/reactor-core/src/main/java/reactor/util/concurrent/SpscArrayQueue.java b/reactor-core/src/main/java/reactor/util/concurrent/SpscArrayQueue.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -175,13 +175,28 @@ public SpscArrayQueueCold(int length) {
 		mask = length - 1;
 	}
 }
+@SuppressWarnings("unused")
 class SpscArrayQueueP1<T> extends SpscArrayQueueCold<T> {
 	/** */
 	private static final long serialVersionUID = -4461305682174876914L;
-
-	long p00, p01, p02, p03, p04, p05, p06, p07;
-	long p08, p09, p0A, p0B, p0C, p0D, p0E;
 
+	byte pad000,pad001,pad002,pad003,pad004,pad005,pad006,pad007;//  8b
+	byte pad010,pad011,pad012,pad013,pad014,pad015,pad016,pad017;// 16b
+	byte pad020,pad021,pad022,pad023,pad024,pad025,pad026,pad027;// 24b
+	byte pad030,pad031,pad032,pad033,pad034,pad035,pad036,pad037;// 32b
+	byte pad040,pad041,pad042,pad043,pad044,pad045,pad046,pad047;// 40b
+	byte pad050,pad051,pad052,pad053,pad054,pad055,pad056,pad057;// 48b
+	byte pad060,pad061,pad062,pad063,pad064,pad065,pad066,pad067;// 56b
+	byte pad070,pad071,pad072,pad073,pad074,pad075,pad076,pad077;// 64b
+	byte pad100,pad101,pad102,pad103,pad104,pad105,pad106,pad107;// 72b
+	byte pad110,pad111,pad112,pad113,pad114,pad115,pad116,pad117;// 80b
+	byte pad120,pad121,pad122,pad123,pad124,pad125,pad126,pad127;// 88b
+	byte pad130,pad131,pad132,pad133,pad134,pad135,pad136,pad137;// 96b
+	byte pad140,pad141,pad142,pad143,pad144,pad145,pad146,pad147;//104b
+	byte pad150,pad151,pad152,pad153,pad154,pad155,pad156,pad157;//112b
+	byte pad160,pad161,pad162,pad163,pad164,pad165,pad166,pad167;//120b
+	byte pad170,pad171,pad172,pad173,pad174,pad175,pad176,pad177;//128b
+	byte pad200,pad201,pad202,pad203;                            //132b
 	SpscArrayQueueP1(int length) {
 		super(length);
 	}
@@ -203,12 +218,27 @@ class SpscArrayQueueProducer<T> extends SpscArrayQueueP1<T> {
 
 }
 
+@SuppressWarnings("unused")
 class SpscArrayQueueP2<T> extends SpscArrayQueueProducer<T> {
 	/** */
 	private static final long serialVersionUID = -5400235061461013116L;
-
-	long p00, p01, p02, p03, p04, p05, p06, p07;
-	long p08, p09, p0A, p0B, p0C, p0D, p0E;
+
+	byte pad000,pad001,pad002,pad003,pad004,pad005,pad006,pad007;//  8b
+	byte pad010,pad011,pad012,pad013,pad014,pad015,pad016,pad017;// 16b
+	byte pad020,pad021,pad022,pad023,pad024,pad025,pad026,pad027;// 24b
+	byte pad030,pad031,pad032,pad033,pad034,pad035,pad036,pad037;// 32b
+	byte pad040,pad041,pad042,pad043,pad044,pad045,pad046,pad047;// 40b
+	byte pad050,pad051,pad052,pad053,pad054,pad055,pad056,pad057;// 48b
+	byte pad060,pad061,pad062,pad063,pad064,pad065,pad066,pad067;// 56b
+	byte pad070,pad071,pad072,pad073,pad074,pad075,pad076,pad077;// 64b
+	byte pad100,pad101,pad102,pad103,pad104,pad105,pad106,pad107;// 72b
+	byte pad110,pad111,pad112,pad113,pad114,pad115,pad116,pad117;// 80b
+	byte pad120,pad121,pad122,pad123,pad124,pad125,pad126,pad127;// 88b
+	byte pad130,pad131,pad132,pad133,pad134,pad135,pad136,pad137;// 96b
+	byte pad140,pad141,pad142,pad143,pad144,pad145,pad146,pad147;//104b
+	byte pad150,pad151,pad152,pad153,pad154,pad155,pad156,pad157;//112b
+	byte pad160,pad161,pad162,pad163,pad164,pad165,pad166,pad167;//120b
+	byte pad170,pad171,pad172,pad173,pad174,pad175,pad176,pad177;//128b
 
 	SpscArrayQueueP2(int length) {
 		super(length);
@@ -231,12 +261,27 @@ class SpscArrayQueueConsumer<T> extends SpscArrayQueueP2<T> {
 
 }
 
+@SuppressWarnings("unused")
 class SpscArrayQueueP3<T> extends SpscArrayQueueConsumer<T> {
 	/** */
 	private static final long serialVersionUID = -2684922090021364171L;
-
-	long p00, p01, p02, p03, p04, p05, p06, p07;
-	long p08, p09, p0A, p0B, p0C, p0D, p0E;
+
+	byte pad000,pad001,pad002,pad003,pad004,pad005,pad006,pad007;//  8b
+	byte pad010,pad011,pad012,pad013,pad014,pad015,pad016,pad017;// 16b
+	byte pad020,pad021,pad022,pad023,pad024,pad025,pad026,pad027;// 24b
+	byte pad030,pad031,pad032,pad033,pad034,pad035,pad036,pad037;// 32b
+	byte pad040,pad041,pad042,pad043,pad044,pad045,pad046,pad047;// 40b
+	byte pad050,pad051,pad052,pad053,pad054,pad055,pad056,pad057;// 48b
+	byte pad060,pad061,pad062,pad063,pad064,pad065,pad066,pad067;// 56b
+	byte pad070,pad071,pad072,pad073,pad074,pad075,pad076,pad077;// 64b
+	byte pad100,pad101,pad102,pad103,pad104,pad105,pad106,pad107;// 72b
+	byte pad110,pad111,pad112,pad113,pad114,pad115,pad116,pad117;// 80b
+	byte pad120,pad121,pad122,pad123,pad124,pad125,pad126,pad127;// 88b
+	byte pad130,pad131,pad132,pad133,pad134,pad135,pad136,pad137;// 96b
+	byte pad140,pad141,pad142,pad143,pad144,pad145,pad146,pad147;//104b
+	byte pad150,pad151,pad152,pad153,pad154,pad155,pad156,pad157;//112b
+	byte pad160,pad161,pad162,pad163,pad164,pad165,pad166,pad167;//120b
+	byte pad170,pad171,pad172,pad173,pad174,pad175,pad176,pad177;//128b
 
 	SpscArrayQueueP3(int length) {
 		super(length);

diff --git a/reactor-core/src/test/java/reactor/core/publisher/QueueDrainSubscriberTest.java b/reactor-core/src/test/java/reactor/core/publisher/QueueDrainSubscriberTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-2022 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,64 +16,56 @@
 
 package reactor.core.publisher;
 
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.openjdk.jol.info.ClassLayout;
-import org.openjdk.jol.info.FieldLayout;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class QueueDrainSubscriberTest {
 
 	@Test
 	@Tag("slow")
-	public void objectPadding() {
+	void objectPadding() {
 		ClassLayout layout = ClassLayout.parseClass(QueueDrainSubscriber.class);
-		AtomicReference<FieldLayout> wip = new AtomicReference<>();
-		AtomicReference<FieldLayout> requested = new AtomicReference<>();
+
+		AtomicLong currentPaddingSize = new AtomicLong();
+		List<String> fields = new ArrayList<>();
+		List<Long> paddingSizes = new ArrayList<>();
 
 		layout.fields().forEach(f -> {
-			if ("wip".equals(f.name())) wip.set(f);
-			else if ("requested".equals(f.name())) requested.set(f);
+			if (f.name().startsWith("pad")) {
+				currentPaddingSize.addAndGet(f.size());
+			} else {
+				if (currentPaddingSize.get() > 0) {
+					fields.add("[padding]");
+					paddingSizes.add(currentPaddingSize.getAndSet(0));
+				}
+				fields.add(f.name());
+			}
 		});
+		if (currentPaddingSize.get() > 0) {
+			fields.add("[padding]");
+			paddingSizes.add(currentPaddingSize.getAndSet(0));
+		}
 
-		final FieldLayout fieldAfterRequested = layout.fields()
-		                                              .tailSet(requested.get())
-		                                              .stream()
-		                                              .skip(1)
-		                                              .filter(fl -> fl.name().length() >= 4)
-		                                              .findFirst()
-		                                              .get();
-
-		assertThat(layout.fields().headSet(wip.get()))
-				.as("wip pre-padding")
-				.hasSize(15)
-				.allSatisfy(fl -> assertThat(fl.name()).startsWith("p"));
-
-		assertThat(layout.fields().subSet(wip.get(), requested.get()).stream().skip(1))
-				.as("wip-requested padding")
-				.hasSize(15)
-				.allSatisfy(fl -> assertThat(fl.name()).startsWith("p").endsWith("a"));
+		assertThat(fields).containsExactly(
+				"[padding]",
+				"wip",
+				"[padding]",
+				"requested",
+				"[padding]",
+				"cancelled",
+				"done",
+				"actual",
+				"queue",
+				"error"
+		);
 
-		assertThat(layout.fields().subSet(requested.get(), fieldAfterRequested)
-		                 .stream()
-		                 .skip(1))
-				.as("requested post-padding")
-				.hasSize(15)
-				.allSatisfy(fl -> assertThat(fl.name()).startsWith("q").isNotEqualTo("queue"));
-
-		assertThat(wip.get().offset())
-				.as("wip offset")
-				.isEqualTo(136);
-		assertThat(requested.get().offset())
-				.as("requested offset")
-				.isEqualTo(wip.get().offset() + 128);
-
-		System.out.println(wip.get());
-		System.out.println(requested.get());
-		System.out.println(fieldAfterRequested);
+		assertThat(paddingSizes).containsExactly(128L, 128L, 128L);
 	}
-
 }