JBTM-4038 JGroupsSlots and JGroupsRaftSlots benchmarks

.github/scripts/benchmark.sh

@@ -110,7 +110,7 @@ BM4b="ArjunaJTA/jta io.narayana.perf.product.GeronimoComparison.* 1"
 BM4c="ArjunaJTA/jta io.narayana.perf.product.NarayanaComparison.* 1"
 BM4d="ArjunaJTA/jta io.narayana.perf.product.AtomikosComparison.* 1"
 #ArjunaJTA/jta/tests/classes/io/narayana/perf/product/ProductComparison.java
-BM5="ArjunaJTA/jta com.arjuna.ats.jta.xa.performance.*StoreBenchmark.* 5"
+BM5="ArjunaJTA/jta com.arjuna.ats.jta.xa.performance.*StoreBenchmark.* 9"
 #ArjunaJTA/jta/tests/classes/com/arjuna/ats/jta/xa/performance/HQStoreBenchmark.java
 #ArjunaJTA/jta/tests/classes/com/arjuna/ats/jta/xa/performance/VolatileStoreBenchmark.java
 #ArjunaJTA/jta/tests/classes/com/arjuna/ats/jta/xa/performance/ShadowNoFileLockStoreBenchmark.java

narayana/ArjunaJTA/jta/etc/jgroups-raft.xml

@@ -0,0 +1,108 @@
+<!--
+  JGroups configuration for JGroups-Raft consensus with a persistent write-ahead log.
+
+  This configuration enables:
+  - Raft consensus algorithm for strong consistency
+  - FileBasedLog for persistent write-ahead logging
+  - Leader election with automatic failover
+  - SHARED_LOOPBACK_PING for localhost testing
+
+  For production use:
+  - Replace SHARED_LOOPBACK_PING with TCPPING or JDBC_PING
+  - Adjust raft_id and members list
+  - Set appropriate log_name directory
+
+  It defines the JGroups protocol stack config for JGroupsRaftSlots building on the same transport and reliability protocols
+  as jgroups.xml, but adds four Raft-specific protocols towards the bottom of the file:
+  - raft.NO_DUPES — deduplicates messages to prevent double-application to the state machine
+  - raft.ELECTION — handles Raft leader election (heartbeats, election timeouts, vote requests)
+  - raft.RAFT — the core Raft consensus protocol (log replication, commit index tracking). members and raft_id are set programmatically by JGroupsRaftSlots before connecting
+  - raft.REDIRECT — forwards writes that arrive at a follower to the current leader, so any node can accept writes
+-->
+<config xmlns="urn:org:jgroups"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/jgroups.xsd">
+
+    <!-- Transport layer - SHARED_LOOPBACK for localhost testing -->
+    <SHARED_LOOPBACK/>
+
+    <!-- Discovery protocol for localhost testing -->
+    <org.jgroups.protocols.SHARED_LOOPBACK_PING/>
+
+    <!--
+      For production, use one of these instead:
+
+      <TCPPING initial_hosts="node1[7800],node2[7800],node3[7800]"
+               port_range="2"/>
+
+      <JDBC_PING connection_driver="org.postgresql.Driver"
+                 connection_url="jdbc:postgresql://localhost/jgroups"
+                 connection_username="jgroups"
+                 connection_password="secret"
+                 initialize_sql="CREATE TABLE IF NOT EXISTS JGROUPSPING ..."/>
+    -->
+
+    <!-- Merging partitions -->
+    <MERGE3 max_interval="30000"
+            min_interval="10000"/>
+
+    <!-- Failure detection -->
+    <FD_SOCK/>
+    <FD_ALL3 timeout="3000"
+             interval="1000"/>
+    <VERIFY_SUSPECT timeout="1500"/>
+
+    <!-- Barrier for flush protocol -->
+    <BARRIER/>
+
+    <!-- Reliable multicast -->
+    <pbcast.NAKACK2 use_mcast_xmit="false"
+                    discard_delivered_msgs="true"
+                    log_discard_msgs="false"
+                    log_not_found_msgs="false"/>
+
+    <!-- Reliable unicast -->
+    <UNICAST3 conn_expiry_timeout="0"/>
+
+    <!-- Stability protocol -->
+    <pbcast.STABLE desired_avg_gossip="50000"
+                   max_bytes="8M"/>
+
+    <!-- Group membership -->
+    <!-- Duplicate message detection -->
+    <raft.NO_DUPES/>
+    <pbcast.GMS print_local_addr="true"
+                join_timeout="3000"/>
+
+    <!-- Flow control -->
+    <UFC max_credits="2M"
+         min_threshold="0.4"/>
+    <MFC max_credits="2M"
+         min_threshold="0.4"/>
+
+    <!-- Message fragmentation -->
+    <FRAG4 frag_size="60K"/>
+
+    <!-- Raft-specific protocols -->
+
+    <!-- Leader election protocol - CRITICAL for leader election to work -->
+    <raft.ELECTION/>
+
+    <!-- Main Raft protocol
+
+         Note: members and raft_id will be set programmatically before connecting
+    -->
+    <raft.RAFT/>
+
+    <!-- Redirect non-leader requests to leader -->
+    <raft.REDIRECT/>
+
+    <!--
+      Optional: CLIENT protocol for external clients
+      Uncomment if you need external client connections
+
+      <raft.CLIENT bind_addr="0.0.0.0"
+                   bind_port="9000"/>
+    -->
+
+</config>

narayana/ArjunaJTA/jta/etc/jgroups.xml

@@ -0,0 +1,16 @@
+<config xmlns="urn:org:jgroups"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/jgroups.xsd">
+    <SHARED_LOOPBACK/>
+    <SHARED_LOOPBACK_PING/>
+    <MERGE3 min_interval="1000" max_interval="5000"/>
+    <FD_ALL3 timeout="3000" interval="1000"/>
+    <VERIFY_SUSPECT2 timeout="1000"/>
+    <pbcast.NAKACK2 use_mcast_xmit="false"/>
+    <UNICAST3/>
+    <pbcast.STABLE desired_avg_gossip="5000" max_bytes="1M"/>
+    <pbcast.GMS print_local_addr="false" join_timeout="500"/>
+    <UFC max_credits="4M" min_threshold="0.4"/>
+    <MFC max_credits="4M" min_threshold="0.4"/>
+    <FRAG4/>
+</config>

narayana/ArjunaJTA/jta/pom.xml

@@ -43,6 +43,7 @@
         <version.com.atomikos.transactions-jta>6.0.0</version.com.atomikos.transactions-jta>
         <version.com.atomikos.transactions-jdbc>6.0.0</version.com.atomikos.transactions-jdbc>
         <version.org.codehaus.btm.btm>2.1.4</version.org.codehaus.btm.btm>
+        <version.org.jgroups.raft>1.1.4.Final</version.org.jgroups.raft>
         <version.jfreechart>1.0.9</version.jfreechart>
         <version.jfreecommon>1.0.16</version.jfreecommon>
     </properties>
@@ -380,6 +381,12 @@
             <artifactId>jcommon</artifactId>
             <version>${version.jfreecommon}</version>
         </dependency>
+        <dependency>
+            <groupId>org.jgroups</groupId>
+            <artifactId>jgroups-raft</artifactId>
+            <version>${version.org.jgroups.raft}</version>
+            <optional>true</optional>
+        </dependency>
     </dependencies>
 
 </project>

narayana/ArjunaJTA/jta/tests/classes/com/arjuna/ats/jta/xa/performance/JGroupsRaftClusterStoreBenchmark.java

@@ -0,0 +1,216 @@
+/*
+ * Copyright The Narayana Authors
+ * SPDX short identifier: Apache-2.0
+ */
+
+package com.arjuna.ats.jta.xa.performance;
+
+import com.arjuna.ats.internal.arjuna.objectstore.slot.jgroups.JGroupsRaftSlots;
+import com.arjuna.ats.internal.arjuna.objectstore.slot.jgroups.JGroupsRaftStoreEnvironmentBean;
+import org.jgroups.JChannel;
+import org.jgroups.protocols.raft.RAFT;
+import org.jgroups.protocols.raft.Role;
+import org.jgroups.raft.blocks.ReplicatedStateMachine;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Mode;
+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.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.openjdk.jmh.runner.options.TimeValue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Multi-node Raft cluster benchmark measuring the impact of {@code allowDirtyReads}.
+ * <p>
+ * Creates a 3-node Raft cluster in-process (via SHARED_LOOPBACK), writes pre-populated
+ * data through the leader, then benchmarks reads from a follower node. With dirty reads
+ * enabled, reads hit the follower's local in-memory map; with dirty reads disabled,
+ * reads go through the Raft leader via consensus.
+ * <p>
+ * The benchmark measures raw slot I/O (slots.read() and slots.write()) with no transaction overhead.
+ * The *StoreBenchmark classes measure end-to-end transaction throughput through the full JTA stack.
+ * Consequently one should expect significant performance difference between the benchmarks.
+ * <p>
+ * Run: {@code java -jar benchmarks.jar "JGroupsRaftClusterStoreBenchmark" -t 10 -f 1 -i 5 -wi 2 -r 3}
+ */
+@State(Scope.Benchmark)
+public class JGroupsRaftClusterStoreBenchmark {
+
+    private static final String STORE_BASE_DIR = System.getProperty("user.dir") + "/target/jgroups-raft-cluster-benchmark";
+    private static final String[] NODE_NAMES = {"node1", "node2", "node3"};
+    private static final String RAFT_MEMBERS = String.join(",", NODE_NAMES);
+    private static final int NUM_SLOTS = 256;
+    private static final int PREPOPULATE_SLOTS = 64;
+    private static final byte[] SLOT_DATA = "benchmark-data-payload".getBytes();
+
+    @Param({"true", "false"})
+    private boolean allowDirtyReads;
+
+    private JChannel[] channels;
+    private JGroupsRaftSlots[] allSlots;
+    private JGroupsRaftSlots leaderSlots;
+    private JGroupsRaftSlots followerSlots;
+
+    private final AtomicInteger readCounter = new AtomicInteger(0);
+    private final AtomicInteger writeCounter = new AtomicInteger(0);
+
+    public static void main(String[] args) throws RunnerException {
+        Options opt = new OptionsBuilder()
+                .include(JGroupsRaftClusterStoreBenchmark.class.getSimpleName())
+                .timeUnit(TimeUnit.SECONDS)
+                .threads(10)
+                .forks(1)
+                .mode(Mode.Throughput)
+                .warmupIterations(2)
+                .warmupTime(TimeValue.seconds(2))
+                .measurementIterations(5)
+                .measurementTime(TimeValue.seconds(3))
+                .shouldDoGC(true)
+                .build();
+
+        new Runner(opt).run();
+    }
+
+    @Setup(Level.Trial)
+    @SuppressWarnings("unchecked")
+    public void setup() throws Exception {
+        purgeDirectory(new File(STORE_BASE_DIR));
+
+        String clusterName = "raft-cluster-bm-" + System.currentTimeMillis();
+        channels = new JChannel[3];
+        allSlots = new JGroupsRaftSlots[3];
+        ReplicatedStateMachine<Integer, byte[]>[] stateMachines = new ReplicatedStateMachine[3];
+        JGroupsRaftStoreEnvironmentBean[] configs = new JGroupsRaftStoreEnvironmentBean[3];
+
+        // Phase 1: Create and connect all channels before init — Raft needs a quorum
+        // for leader election, so all nodes must be connected before any can elect.
+        for (int i = 0; i < 3; i++) {
+            String storeDir = STORE_BASE_DIR + "/" + NODE_NAMES[i];
+            channels[i] = new JChannel("jgroups-raft.xml").name(NODE_NAMES[i]);
+
+            RAFT raft = channels[i].getProtocolStack().findProtocol(RAFT.class);
+            raft.logDir(storeDir);
+            raft.logUseFsync(false);
+            raft.members(Arrays.asList(RAFT_MEMBERS.split(",")));
+
+            stateMachines[i] = new ReplicatedStateMachine<>(channels[i]);
+            stateMachines[i].raftId(NODE_NAMES[i]);
+            stateMachines[i].allowDirtyReads(allowDirtyReads);
+            stateMachines[i].timeout(5000);
+
+            channels[i].connect(clusterName);
+        }
+
+        // Wait for leader election across the cluster
+        waitForLeader(channels, 10_000);
+
+        // Phase 2: Create JGroupsRaftSlots with pre-configured channels
+        for (int i = 0; i < 3; i++) {
+            configs[i] = new JGroupsRaftStoreEnvironmentBean();
+            configs[i].setJGroupsConfigFileName("jgroups-raft.xml");
+            configs[i].setNodeAddress(NODE_NAMES[i]);
+            configs[i].setClusterName(clusterName);
+            configs[i].setCacheName(clusterName);
+            configs[i].setStoreDir(STORE_BASE_DIR + "/" + NODE_NAMES[i]);
+            configs[i].setNumberOfSlots(NUM_SLOTS);
+            configs[i].setRaftMembers(RAFT_MEMBERS);
+            configs[i].setRaftLogFsync(false);
+            configs[i].setRaftTimeout(5000);
+            configs[i].setAllowDirtyReads(allowDirtyReads);
+            configs[i].setPreConfiguredChannel(channels[i]);
+            configs[i].setPreConfiguredStateMachine(stateMachines[i]);
+
+            allSlots[i] = new JGroupsRaftSlots();
+            allSlots[i].init(configs[i]);
+        }
+
+        // Identify leader and a follower
+        for (JGroupsRaftSlots slots : allSlots) {
+            String role = slots.getRole();
+            if (Role.Leader.name().equals(role) && leaderSlots == null) {
+                leaderSlots = slots;
+            } else if (Role.Follower.name().equals(role) && followerSlots == null) {
+                followerSlots = slots;
+            }
+        }
+
+        if (leaderSlots == null || followerSlots == null) {
+            throw new IllegalStateException("Cluster did not form correctly: leader=" + leaderSlots + " follower=" + followerSlots);
+        }
+
+        // Pre-populate slots via the leader
+        for (int i = 0; i < PREPOPULATE_SLOTS; i++) {
+            leaderSlots.write(i, SLOT_DATA, true);
+        }
+
+        // Brief pause for replication to followers
+        TimeUnit.MILLISECONDS.sleep(500);
+    }
+
+    @TearDown(Level.Trial)
+    public void tearDown() {
+        if (channels != null) {
+            for (JChannel ch : channels) {
+                if (ch != null) {
+                    try {
+                        ch.close();
+                    } catch (Exception ignore) {
+                    }
+                }
+            }
+        }
+
+        purgeDirectory(new File(STORE_BASE_DIR));
+    }
+
+    @Benchmark
+    public void readFromFollower(Blackhole bh) throws IOException {
+        int slotId = readCounter.getAndIncrement() % PREPOPULATE_SLOTS;
+        bh.consume(followerSlots.read(slotId));
+    }
+
+    @Benchmark
+    public void writeToLeader(Blackhole bh) throws IOException {
+        int slotId = writeCounter.getAndIncrement() % NUM_SLOTS;
+        leaderSlots.write(slotId, SLOT_DATA, true);
+    }
+
+    private static void waitForLeader(JChannel[] channels, long timeoutMs) throws InterruptedException {
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        while (System.currentTimeMillis() < deadline) {
+            for (JChannel ch : channels) {
+                RAFT raft = ch.getProtocolStack().findProtocol(RAFT.class);
+                if (raft != null && raft.leader() != null) {
+                    return;
+                }
+            }
+            TimeUnit.MILLISECONDS.sleep(50);
+        }
+        throw new IllegalStateException("No Raft leader elected within " + timeoutMs + "ms");
+    }
+
+    private static void purgeDirectory(File dir) {
+        if (dir.exists()) {
+            File[] files = dir.listFiles();
+            if (files != null) {
+                for (File file : files) {
+                    purgeDirectory(file);
+                }
+            }
+            dir.delete();
+        }
+    }
+}