From 1d9aa61e819f0c522ec5606ae40a4f85f764d0ef Mon Sep 17 00:00:00 2001
From: Keshava Munegowda <keshava.gowda@gmail.com>
Date: Wed, 29 May 2024 19:57:12 +0530
Subject: [PATCH] Add multi threaded micro benchmarking

Signed-off-by: Keshava Munegowda <keshava.gowda@gmail.com>
---
 .../io/perl/benchmark/QueueBenchmark.java     | 67 +++++++++++++++++--
 .../java/io/perl/api/impl/AtomicQueue.java    |  3 +-
 2 files changed, 64 insertions(+), 6 deletions(-)

diff --git a/perl/src/jmh/java/io/perl/benchmark/QueueBenchmark.java b/perl/src/jmh/java/io/perl/benchmark/QueueBenchmark.java
index 7561cf92..d3ec7cbc 100644
--- a/perl/src/jmh/java/io/perl/benchmark/QueueBenchmark.java
+++ b/perl/src/jmh/java/io/perl/benchmark/QueueBenchmark.java
@@ -18,6 +18,7 @@
 import org.openjdk.jmh.annotations.Measurement;
 import org.openjdk.jmh.annotations.Scope;
 import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Timeout;
 import org.openjdk.jmh.annotations.Warmup;
 import org.openjdk.jmh.runner.Runner;
@@ -32,11 +33,11 @@
 //@Fork(value = 2, jvmArgs = {"-Djmh.blackhole.autoDetect=false"})
 public class QueueBenchmark {
 
-    final private CQueue<Integer> cqueue;
-    final private ConcurrentLinkedQueue<Integer> clinkedQueue;
-    final private LinkedBlockingQueue<Integer> linkedbq;
-    final private AtomicQueue<Integer> atomicQueue;
-    final private SyncQueue<Integer> syncQueue;
+    final protected CQueue<Integer> cqueue;
+    final protected ConcurrentLinkedQueue<Integer> clinkedQueue;
+    final protected LinkedBlockingQueue<Integer> linkedbq;
+    final protected AtomicQueue<Integer> atomicQueue;
+    final protected SyncQueue<Integer> syncQueue;
 
     
     public QueueBenchmark() {
@@ -99,6 +100,62 @@ public void syncQueueBenchmark() {
     }
 
 
+    @Benchmark
+    @Fork(value = 1, warmups = 0)
+    @Timeout(time = 60)
+    @Warmup(iterations = 0)
+    @Measurement(iterations = 3)
+    @Threads(10)
+    public void cqueueMultiThreadBenchmark() {
+        cqueue.add(10);
+        cqueue.poll();
+    }
+
+    @Benchmark
+    @Fork(value = 1, warmups = 0)
+    @Timeout(time = 60)
+    @Warmup(iterations = 0)
+    @Measurement(iterations = 3)
+    @Threads(10)
+    public void concurrentQueueMultiThreadBenchmark() {
+        clinkedQueue.add(10);
+        clinkedQueue.poll();
+    }
+
+    @Benchmark
+    @Fork(value = 1, warmups = 0)
+    @Timeout(time = 60)
+    @Warmup(iterations = 0)
+    @Measurement(iterations = 3)
+    @Threads(10)
+    public void linkedBlockingQueueMultiThreadBenchmark() {
+        linkedbq.add(10);
+        linkedbq.poll();
+    }
+
+    @Benchmark
+    @Fork(value = 1, warmups = 0)
+    @Timeout(time = 60)
+    @Warmup(iterations = 0)
+    @Measurement(iterations = 3)
+    @Threads(10)
+    public void atomicQueueMultiThreadBenchmark() {
+        atomicQueue.add(10);
+        atomicQueue.poll();
+    }
+
+    @Benchmark
+    @Fork(value = 1, warmups = 0)
+    @Timeout(time = 60)
+    @Warmup(iterations = 0)
+    @Measurement(iterations = 3)
+    @Threads(10)
+    public void syncQueueMultiThreadBenchmark() {
+        syncQueue.add(10);
+        syncQueue.poll();
+    }
+
+
     public static void main(String[] args) throws Exception {
         Options opt = new OptionsBuilder()
                 .exclude("org.openjdk.jmh.benchmarks.*")
diff --git a/perl/src/main/java/io/perl/api/impl/AtomicQueue.java b/perl/src/main/java/io/perl/api/impl/AtomicQueue.java
index e93e6fd7..90127ddf 100644
--- a/perl/src/main/java/io/perl/api/impl/AtomicQueue.java
+++ b/perl/src/main/java/io/perl/api/impl/AtomicQueue.java
@@ -43,12 +43,13 @@ public AtomicQueue() {
 
     @Override
     public T poll() {
-        final Node<T> curHead = head.get();
+        final Node<T> curHead = head.getAndSet(null);
         if (curHead == null) {
             return null;
         }
         final Node<T> nxt = curHead.next.getAndSet(null);
         if (nxt == null) {
+            head.set(curHead);
             return null;
         }
         head.set(nxt);