initial commit

.classpath

+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="target/classes" path="src/main/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="output" path="target/classes"/>
+</classpath>

.gitignore

+/target

.project

+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>teetime-benchmark</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.m2e.core.maven2Builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+		<nature>org.eclipse.m2e.core.maven2Nature</nature>
+	</natures>
+</projectDescription>

.settings/org.eclipse.core.resources.prefs

+eclipse.preferences.version=1
+encoding//src/main/java=UTF-8
+encoding//src/main/resources=UTF-8
+encoding//src/test/java=UTF-8
+encoding//src/test/resources=UTF-8
+encoding/<project>=UTF-8

.settings/org.eclipse.jdt.core.prefs

+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
+org.eclipse.jdt.core.compiler.compliance=1.8
+org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
+org.eclipse.jdt.core.compiler.source=1.8

.settings/org.eclipse.m2e.core.prefs

+activeProfiles=
+eclipse.preferences.version=1
+resolveWorkspaceProjects=true
+version=1

README

+# Build jar for this project (generated to target/benchmarks.jar)
+mvn clean package
+
+# Execute all benchmarks
+java -jar target/benchmarks.jar
+
+# Execute one specific benchmark (unqualified class name is sufficient)
+java -jar target/benchmarks.jar Port2PortBenchmark
+
+# Execute multiple specific benchmarks
+java -jar target/benchmarks.jar Port2PortBenchmark Port2PortWithTermCheckBenchmark

chw-pc-results.txt

+[producer with send()]
+# Run complete. Total time: 00:04:45
+
+Benchmark                                     (capacity)  Mode  Cnt   Score   Error  Units
+
+Port2PortBenchmark.queue:remove                     1024  avgt   30   6,473 ± 0,399  ns/op
+Port2PortBenchmark.queue:remove                     4096  avgt   30   6,570 ± 0,439  ns/op
+Port2PortBenchmark.queue:remove                    16384  avgt   30   6,338 ± 0,172  ns/op
+Port2PortBenchmark.queue:remove                    65536  avgt   30   6,966 ± 0,374  ns/op
+Port2PortWithTermCheckBenchmark.queue:remove        1024  avgt   30   6,352 ± 0,102  ns/op
+Port2PortWithTermCheckBenchmark.queue:remove        4096  avgt   30   6,195 ± 0,458  ns/op
+Port2PortWithTermCheckBenchmark.queue:remove       16384  avgt   30   6,420 ± 0,281  ns/op
+Port2PortWithTermCheckBenchmark.queue:remove       65536  avgt   30   6,601 ± 0,231  ns/op
+
+[producer with sendNonBlocking()]					-wi 10 -i 10
+# Run complete. Total time: 00:02:01
+
+Benchmark                                     (capacity)  Mode  Cnt  Score   Error  Units
+Port2PortBenchmark.queue                            1024  avgt   50  8,416 ± 0,117  ns/op
+Port2PortBenchmark.queue:add                        1024  avgt   50  8,478 ± 0,125  ns/op
+Port2PortBenchmark.queue:remove                     1024  avgt   50  8,354 ± 0,134  ns/op
+Port2PortWithTermCheckBenchmark.queue               1024  avgt   50  8,508 ± 0,247  ns/op
+Port2PortWithTermCheckBenchmark.queue:add           1024  avgt   50  8,583 ± 0,258  ns/op
+Port2PortWithTermCheckBenchmark.queue:remove        1024  avgt   50  8,433 ± 0,247  ns/op
+
+[producer with sendNonBlocking()]					-f 3
+# Run complete. Total time: 00:05:20
+
+Benchmark                                               (capacity)  Mode  Cnt  Score   Error  Units
+Port2PortBenchmark.queue:remove                               1024  avgt   90  8,509 ± 0,141  ns/op
+Port2PortWithTermInstanceofCheckBenchmark.queue:remove        1024  avgt   90  8,822 ± 0,097  ns/op
+Port2PortWithTermReferenceCheckBenchmark.queue:remove         1024  avgt   90  8,736 ± 0,196  ns/op
+

pom.xml

+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<groupId>org.teetime</groupId>
+	<artifactId>teetime-benchmark</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+
+	<repositories>
+		<repository>
+			<id>sonatype.oss.snapshots</id>
+			<url>https://oss.sonatype.org/content/repositories/snapshots/</url>
+		</repository>
+	</repositories>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<jmh.version>1.9.2</jmh.version>
+		<javac.target>1.8</javac.target>
+		<uberjar.name>benchmarks</uberjar.name>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.openjdk.jmh</groupId>
+			<artifactId>jmh-core</artifactId>
+			<version>${jmh.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.openjdk.jmh</groupId>
+			<artifactId>jmh-generator-annprocess</artifactId>
+			<version>${jmh.version}</version>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>net.sourceforge.teetime</groupId>
+			<artifactId>teetime</artifactId>
+			<version>3.0-SNAPSHOT</version>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<version>3.1</version>
+				<configuration>
+					<compilerVersion>${javac.target}</compilerVersion>
+					<source>${javac.target}</source>
+					<target>${javac.target}</target>
+				</configuration>
+			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-shade-plugin</artifactId>
+				<version>2.2</version>
+				<executions>
+					<execution>
+						<phase>package</phase>
+						<goals>
+							<goal>shade</goal>
+						</goals>
+						<configuration>
+							<finalName>${uberjar.name}</finalName>
+							<transformers>
+								<transformer
+									implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+									<mainClass>org.openjdk.jmh.Main</mainClass>
+								</transformer>
+							</transformers>
+							<filters>
+								<filter>
+									<!-- Shading signed JARs will fail without this. http://stackoverflow.com/questions/999489/invalid-signature-file-when-attempting-to-run-a-jar -->
+									<artifact>*:*</artifact>
+									<excludes>
+										<exclude>META-INF/*.SF</exclude>
+										<exclude>META-INF/*.DSA</exclude>
+										<exclude>META-INF/*.RSA</exclude>
+									</excludes>
+								</filter>
+							</filters>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+		<pluginManagement>
+			<plugins>
+				<plugin>
+					<artifactId>maven-clean-plugin</artifactId>
+					<version>2.5</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-deploy-plugin</artifactId>
+					<version>2.8.1</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-install-plugin</artifactId>
+					<version>2.5.1</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-jar-plugin</artifactId>
+					<version>2.4</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-javadoc-plugin</artifactId>
+					<version>2.9.1</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-resources-plugin</artifactId>
+					<version>2.6</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-site-plugin</artifactId>
+					<version>3.3</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-source-plugin</artifactId>
+					<version>2.2.1</version>
+				</plugin>
+				<plugin>
+					<artifactId>maven-surefire-plugin</artifactId>
+					<version>2.17</version>
+				</plugin>
+			</plugins>
+		</pluginManagement>
+	</build>
+
+</project>
\ No newline at end of file

src/main/java/teetime/benchmark/Port2PortBenchmark.java

+package teetime.benchmark;
+
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.AuxCounters;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import teetime.framework.AbstractStage;
+import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
+import teetime.framework.pipe.BoundedSynchedPipe;
+import teetime.framework.pipe.IPipe;
+
+@State(Scope.Group)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Fork(3)
+@Warmup(iterations = 5)
+@Measurement(iterations = 30)
+public class Port2PortBenchmark {
+
+	private static final Integer ELEMENT = 777;
+
+	@Param(value = { "1024", /*"4096", "16384", "65536"*/ })
+	private int capacity;
+
+	static Producer<Integer> producer;
+	static Consumer<Integer> consumer;
+
+	private static class Producer<T> extends AbstractStage {
+		private final OutputPort<T> outputPort = createOutputPort();
+		private final T element;
+
+		public Producer(T element) {
+			this.element = element;
+		}
+
+		@Override
+		protected void execute() {
+			outputPort.send(element);
+		}
+	}
+
+	private static class Consumer<T> extends AbstractStage {
+		private final InputPort<T> inputPort = createInputPort();
+		private final Blackhole blackhole = new Blackhole();
+
+		@Override
+		protected void execute() {
+			T element = inputPort.receive();
+			blackhole.consume(element);
+		}
+	}
+
+	@Setup()
+	public void setup() {
+		producer = new Producer<>(ELEMENT);
+		consumer = new Consumer<>();
+
+		int capacity = 20_000;
+		createPipe(capacity);
+
+		// stretch the queue to the limit, working through resizing and full
+		for (int i = 0; i < capacity; i++) {
+			producer.outputPort.send(ELEMENT);
+		}
+		for (int i = 0; i < capacity; i++) {
+			consumer.inputPort.receive();
+		}
+
+		// make sure the important common case is exercised
+		for (int i = 0; i < capacity; i++) {
+			producer.outputPort.send(ELEMENT);
+			consumer.inputPort.receive();
+		}
+
+		capacity = this.capacity;
+		createPipe(capacity);
+	}
+
+	private IPipe<Integer> createPipe(final int capacity) {
+		IPipe<Integer> pipe = new BoundedSynchedPipe<>(producer.outputPort, consumer.inputPort, capacity);
+		return pipe;
+	}
+
+	@AuxCounters
+	@State(Scope.Thread)
+	public static class PollCounters {
+		@TearDown(Level.Iteration)
+		public void teardown() {
+			// clear uses poll: only the consumer is allowed to poll elements due to SpSc semantics
+			while (consumer.inputPort.receive() != null)
+				;
+		}
+	}
+
+	@Benchmark
+	@Group("queue")
+	public void add() {
+		producer.outputPort.sendNonBlocking(ELEMENT);
+	}
+
+	@Benchmark
+	@Group("queue")
+	public Integer remove(PollCounters counter) {
+		return consumer.inputPort.receive();
+	}
+}

src/main/java/teetime/benchmark/Port2PortWithTermInstanceofCheckBenchmark.java

+package teetime.benchmark;
+
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.AuxCounters;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import teetime.framework.AbstractStage;
+import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
+import teetime.framework.pipe.BoundedSynchedPipe;
+import teetime.framework.pipe.IPipe;
+
+@State(Scope.Group)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Fork(3)
+@Warmup(iterations = 5)
+@Measurement(iterations = 30)
+public class Port2PortWithTermInstanceofCheckBenchmark {
+
+	static class TerminateClass {}
+
+	private static final Object ELEMENT = 777;
+	private static final Object TERM_VALUE = 1;
+	private static final Object TERM = new TerminateClass();
+
+	@Param(value = { "1024", /*"4096", "16384", "65536"*/ })
+	private int capacity;
+
+	private boolean toggle;
+
+	static Producer<Object> producer;
+	static Consumer<Object> consumer;
+
+	private static class Producer<T> extends AbstractStage {
+		private final OutputPort<T> outputPort = createOutputPort();
+		private final T element;
+
+		public Producer(T element) {
+			this.element = element;
+		}
+
+		@Override
+		protected void execute() {
+			outputPort.send(element);
+		}
+	}
+
+	private static class Consumer<T> extends AbstractStage {
+		private final InputPort<T> inputPort = createInputPort();
+		private final Blackhole blackhole = new Blackhole();
+
+		@Override
+		protected void execute() {
+			T element = inputPort.receive();
+			blackhole.consume(element);
+		}
+	}
+
+	@Setup()
+	public void setup() {
+		producer = new Producer<>(ELEMENT);
+		consumer = new Consumer<>();
+
+		int capacity = 20_000;
+		createPipe(capacity);
+
+		// stretch the queue to the limit, working through resizing and full
+		for (int i = 0; i < capacity; i++) {
+			producer.outputPort.send(ELEMENT);
+		}
+		for (int i = 0; i < capacity; i++) {
+			consumer.inputPort.receive();
+		}
+
+		// make sure the important common case is exercised
+		for (int i = 0; i < capacity; i++) {
+			producer.outputPort.send(ELEMENT);
+			consumer.inputPort.receive();
+		}
+
+		capacity = this.capacity;
+		createPipe(capacity);
+	}
+
+	private IPipe<Object> createPipe(final int capacity) {
+		IPipe<Object> pipe = new BoundedSynchedPipe<>(producer.outputPort, consumer.inputPort, capacity);
+		return pipe;
+	}
+
+	@AuxCounters
+	@State(Scope.Thread)
+	public static class PollCounters {
+		@TearDown(Level.Iteration)
+		public void teardown() {
+			// clear uses poll: only the consumer is allowed to poll elements due to SpSc semantics
+			while (consumer.inputPort.receive() != null)
+				;
+		}
+	}
+
+	@Benchmark
+	@Group("queue")
+	public void add() {
+		toggle = !toggle;
+		Object element = (toggle) ? ELEMENT : TERM;
+		producer.outputPort.sendNonBlocking(element);
+	}
+
+	@Benchmark
+	@Group("queue")
+	public Object remove(PollCounters counter) {
+		Object element = consumer.inputPort.receive();
+		if (element instanceof TerminateClass) {
+			return TERM_VALUE;
+		}
+		return element;
+	}
+}

src/main/java/teetime/benchmark/Port2PortWithTermReferenceCheckBenchmark.java

+package teetime.benchmark;
+
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.AuxCounters;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import teetime.framework.AbstractStage;
+import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
+import teetime.framework.pipe.BoundedSynchedPipe;
+import teetime.framework.pipe.IPipe;
+
+@State(Scope.Group)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Fork(3)
+@Warmup(iterations = 5)
+@Measurement(iterations = 30)
+public class Port2PortWithTermReferenceCheckBenchmark {
+
+	private static final Integer ELEMENT = 777;
+	private static final Integer TERM = new Integer(42);
+	private static final Integer TERM_VALUE = new Integer(1);
+
+	@Param(value = { "1024", /*"4096", "16384", "65536"*/ })
+	private int capacity;
+
+	static Producer<Integer> producer;
+	static Consumer<Integer> consumer;
+
+	private static class Producer<T> extends AbstractStage {
+		private final OutputPort<T> outputPort = createOutputPort();
+		private final T element;
+
+		public Producer(T element) {
+			this.element = element;
+		}
+
+		@Override
+		protected void execute() {
+			outputPort.send(element);
+		}
+	}
+
+	private static class Consumer<T> extends AbstractStage {
+		private final InputPort<T> inputPort = createInputPort();
+		private final Blackhole blackhole = new Blackhole();
+
+		@Override
+		protected void execute() {
+			T element = inputPort.receive();
+			blackhole.consume(element);
+		}
+	}
+
+	@Setup()
+	public void setup() {
+		producer = new Producer<>(ELEMENT);
+		consumer = new Consumer<>();
+
+		int capacity = 20_000;
+		createPipe(capacity);
+
+		// stretch the queue to the limit, working through resizing and full
+		for (int i = 0; i < capacity; i++) {
+			producer.outputPort.send(ELEMENT);
+		}
+		for (int i = 0; i < capacity; i++) {
+			consumer.inputPort.receive();
+		}
+
+		// make sure the important common case is exercised
+		for (int i = 0; i < capacity; i++) {
+			producer.outputPort.send(ELEMENT);
+			consumer.inputPort.receive();
+		}
+
+		capacity = this.capacity;
+		createPipe(capacity);
+	}
+
+	private IPipe<Integer> createPipe(final int capacity) {
+		IPipe<Integer> pipe = new BoundedSynchedPipe<>(producer.outputPort, consumer.inputPort, capacity);
+		return pipe;
+	}
+
+	@AuxCounters
+	@State(Scope.Thread)
+	public static class PollCounters {
+		@TearDown(Level.Iteration)
+		public void teardown() {
+			// clear uses poll: only the consumer is allowed to poll elements due to SpSc semantics
+			while (consumer.inputPort.receive() != null)
+				;
+		}
+	}
+
+	@Benchmark
+	@Group("queue")
+	public void add() {
+		producer.outputPort.sendNonBlocking(ELEMENT);
+	}
+
+	@Benchmark
+	@Group("queue")
+	public Integer remove(PollCounters counter) {
+		Integer element = consumer.inputPort.receive();
+		if (TERM == element) {
+			return TERM_VALUE;
+		}
+		return element;
+	}
+}

src/main/resources/logback.xml

+<configuration>
+<!-- 	log off -->
+</configuration>