AFFINITY-22: Change the API to support more the 64 CPUs; better JNI

testing; use C++ for JNI instead of C

affinity-test/pom.xml

@@ -28,7 +28,7 @@
 
     <modelVersion>4.0.0</modelVersion>
     <artifactId>affinity-test</artifactId>
-    <version>2.3-SNAPSHOT</version>
+    <version>3.0-SNAPSHOT</version>
     <packaging>bundle</packaging>
 
     <name>OpenHFT/Java-Thread-Affinity/affinity-test</name>

affinity/pom.xml

@@ -28,7 +28,7 @@
 
     <modelVersion>4.0.0</modelVersion>
     <artifactId>affinity</artifactId>
-    <version>2.3-SNAPSHOT</version>
+    <version>3.0-SNAPSHOT</version>
     <packaging>bundle</packaging>
 
     <name>OpenHFT/Java-Thread-Affinity/affinity</name>

affinity/src/main/c/Makefile

@@ -14,7 +14,7 @@ JAVA_CLASSES = software.chronicle.enterprise.internals.impl.NativeAffinity
 
 JNI_STUBS := $(subst .,_,$(JAVA_CLASSES))
 JNI_HEADERS := $(patsubst %,$(WORKING_DIR)/%.h,$(JNI_STUBS))
-JNI_SOURCES := $(patsubst %,%.c,$(JNI_STUBS))
+JNI_SOURCES := $(patsubst %,%.cpp,$(JNI_STUBS))
 JNI_JAVA_SOURCES := $(patsubst %,$(TARGET_DIR)/%.class,$(subst .,/,$(JAVA_CLASSES)))
 
 JAVA_BUILD_DIR := $(TARGET_DIR)
@@ -23,6 +23,7 @@ JAVA_HOME ?= /usr/java/default
 JAVA_LIB := $(JAVA_HOME)/jre/lib
 JVM_SHARED_LIBS := -L$(JAVA_LIB)/amd64/server -L$(JAVA_LIB)/i386/server  -L$(JAVA_LIB)/amd64/jrockit/ -L$(JAVA_LIB)/i386/jrockit/
 
+CXX=g++
 INCLUDES := -I $(JAVA_HOME)/include -I $(JAVA_HOME)/include/linux -I $(WORKING_DIR)
 
 # classpath for javah
@@ -37,7 +38,7 @@ endif
 all: $(TARGET)
 
 $(TARGET): $(JNI_HEADERS) $(JNI_SOURCES)
-	gcc -O3 -Wall -shared -fPIC $(JVM_SHARED_LIBS) -ljvm -lrt $(INCLUDES) $(JNI_SOURCES) -o $(TARGET)
+	$(CXX) -O3 -Wall -shared -fPIC $(JVM_SHARED_LIBS) -ljvm -lrt $(INCLUDES) $(JNI_SOURCES) -o $(TARGET)
 
 $(JNI_HEADERS): $(JNI_JAVA_SOURCES)
 	mkdir -p $(TARGET_DIR)/jni

affinity/src/main/c/software_chronicle_enterprise_internals_impl_NativeAffinity.c → affinity/src/main/c/software_chronicle_enterprise_internals_impl_NativeAffinity.cpp

@@ -1,4 +1,4 @@
-/*
+/* vim: syntax=cpp
  * Copyright 2011 Peter Lawrey
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,12 +14,16 @@
  * limitations under the License.
  */
 
+#ifndef _GNU_SOURCE
 #define _GNU_SOURCE
+#endif
+
 #include <jni.h>
 #include <sched.h>
 #include <sys/syscall.h>
 #include <sys/types.h>
 #include <unistd.h>
+#include <string.h>
 
 #include "software_chronicle_enterprise_internals_impl_NativeAffinity.h"
 
@@ -28,16 +32,27 @@
  * Method:    getAffinity0
  * Signature: ()J
  */
-JNIEXPORT jlong JNICALL Java_software_chronicle_enterprise_internals_impl_NativeAffinity_getAffinity0
-  (JNIEnv *env, jclass c) {
+JNIEXPORT jbyteArray JNICALL Java_software_chronicle_enterprise_internals_impl_NativeAffinity_getAffinity0
+  (JNIEnv *env, jclass c) 
+{
+    // The default size of the structure supports 1024 CPUs, should be enough
+    // for now In the future we can use dynamic sets, which can support more
+    // CPUs, given OS can handle them as well
     cpu_set_t mask;
-    int ret = sched_getaffinity(0, sizeof(mask), &mask);
-    if (ret < 0) return ~0LL;
-    long long mask2 = 0, i;
-    for(i=0;i<sizeof(mask2)*8;i++)
-        if (CPU_ISSET(i, &mask))
-            mask2 |= 1L << i;
-    return (jlong) mask2;
+    const size_t size = sizeof(mask);
+
+    int res = sched_getaffinity(0, size, &mask);
+    if (res < 0)
+    {
+        return NULL;
+    }
+
+    jbyteArray ret = env->NewByteArray(size);
+    jbyte* bytes = env->GetByteArrayElements(ret, 0);
+    memcpy(bytes, &mask, size);
+    env->SetByteArrayRegion(ret, 0, size, bytes);
+
+    return ret;
 }
 
 /*
@@ -46,14 +61,16 @@ JNIEXPORT jlong JNICALL Java_software_chronicle_enterprise_internals_impl_Native
  * Signature: (J)V
  */
 JNIEXPORT void JNICALL Java_software_chronicle_enterprise_internals_impl_NativeAffinity_setAffinity0
-  (JNIEnv *env, jclass c, jlong affinity) {
-    int i;
+  (JNIEnv *env, jclass c, jbyteArray affinity)
+{
     cpu_set_t mask;
+    const size_t size = sizeof(mask);
     CPU_ZERO(&mask);
-    for(i=0;i<sizeof(affinity)*8;i++)
-        if ((affinity >> i) & 1)
-            CPU_SET(i, &mask);
-    sched_setaffinity(0, sizeof(mask), &mask);
+
+    jbyte* bytes = env->GetByteArrayElements(affinity, 0);
+    memcpy(&mask, bytes, size);
+
+    sched_setaffinity(0, size, &mask);
 }
 
 /*

affinity/src/main/java/net/openhft/affinity/AffinityLock.java

@@ -26,6 +26,7 @@
 import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
+import java.util.BitSet;
 
 /**
  * This utility class support locking a thread to a single core, or reserving a whole core for a thread.
@@ -41,8 +42,8 @@ public class AffinityLock implements Closeable {
     // TODO It seems like on virtualized platforms .availableProcessors() value can change at
     // TODO runtime. We should think about how to adopt to such change
     public static final int PROCESSORS = Runtime.getRuntime().availableProcessors();
-    public static final long BASE_AFFINITY = AffinitySupport.getAffinity();
-    public static final long RESERVED_AFFINITY = getReservedAffinity0();
+    public static final BitSet BASE_AFFINITY = AffinitySupport.getAffinity();
+    public static final BitSet RESERVED_AFFINITY = getReservedAffinity0();
     private static final LockInventory LOCK_INVENTORY = new LockInventory(new NoCpuLayout(PROCESSORS));
 
     static {
@@ -102,17 +103,29 @@ public static CpuLayout cpuLayout() {
         return LOCK_INVENTORY.getCpuLayout();
     }
 
-    private static long getReservedAffinity0() {
+    private static BitSet getReservedAffinity0()
+    {
         String reservedAffinity = System.getProperty(AFFINITY_RESERVED);
-        if (reservedAffinity == null || reservedAffinity.trim().isEmpty()) {
-            long reserverable = ((1 << PROCESSORS) - 1) ^ BASE_AFFINITY;
-            if (reserverable == 0 && PROCESSORS > 1) {
+        if (reservedAffinity == null || reservedAffinity.trim().isEmpty())
+        {
+            BitSet reserverable = new BitSet(PROCESSORS);
+            reserverable.set(0, PROCESSORS - 1, true);
+            reserverable.and(BASE_AFFINITY);
+            if (reserverable.isEmpty() && PROCESSORS > 1)
+            {
                 LoggerFactory.getLogger(AffinityLock.class).info("No isolated CPUs found, so assuming CPUs 1 to {} available.", (PROCESSORS - 1));
-                return ((1 << PROCESSORS) - 2);
+                reserverable = new BitSet(PROCESSORS);
+                // make the first CPU unavailable
+                reserverable.set(1, PROCESSORS - 1, true);
+                reserverable.set(0, false);
+                return reserverable;
             }
             return reserverable;
         }
-        return Long.parseLong(reservedAffinity, 16);
+
+        long[] longs = new long[1];
+        longs[0] = Long.parseLong(reservedAffinity, 16);
+        return BitSet.valueOf(longs);
     }
 
     /**
@@ -212,7 +225,11 @@ public void bind(boolean wholeCore) {
             LOGGER.info("Assigning cpu {} to {}", cpuId, assignedThread);
         }
         if (cpuId >= 0)
-            AffinitySupport.setAffinity(1L << cpuId);
+        {
+            BitSet affinity = new BitSet();
+            affinity.set(cpuId, true);
+            AffinitySupport.setAffinity(affinity);
+        }
     }
 
     final boolean canReserve() {

affinity/src/main/java/net/openhft/affinity/AffinitySupport.java

@@ -24,6 +24,7 @@
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.lang.reflect.Field;
+import java.util.BitSet;
 
 /**
  * Library to wrap low level JNI or JNA calls.  Can be called without needing to know the actual implementation used.
@@ -144,11 +145,11 @@ private static void logThrowable(Throwable t, String description) {
         LOGGER.warn(sw.toString());
     }
 
-    public static long getAffinity() {
+    public static BitSet getAffinity() {
         return AFFINITY_IMPL.getAffinity();
     }
 
-    public static void setAffinity(final long affinity) {
+    public static void setAffinity(final BitSet affinity) {
         AFFINITY_IMPL.setAffinity(affinity);
     }
 

affinity/src/main/java/net/openhft/affinity/IAffinity.java

@@ -16,6 +16,8 @@
 
 package net.openhft.affinity;
 
+import java.util.BitSet;
+
 /**
  * Implementation interface
  *
@@ -24,14 +26,14 @@
  */
 public interface IAffinity {
     /**
-     * @return returns affinity mask for current thread, or -1 if unknown
+     * @return returns affinity mask for current thread, or null if unknown
      */
-    long getAffinity();
+    BitSet getAffinity();
 
     /**
      * @param affinity sets affinity mask of current thread to specified value
      */
-    void setAffinity(final long affinity);
+    void setAffinity(final BitSet affinity);
 
     /**
      * @return the current cpu id, or -1 if unknown.

affinity/src/main/java/net/openhft/affinity/LockInventory.java

@@ -53,9 +53,10 @@ public final synchronized void set(CpuLayout cpuLayout) {
             return;
         }
         reset(cpuLayout);
-        for (int i = 0; i < cpuLayout.cpus(); i++) {
-            boolean base = ((AffinityLock.BASE_AFFINITY >> i) & 1) != 0;
-            boolean reservable = ((AffinityLock.RESERVED_AFFINITY >> i) & 1) != 0;
+        for (int i = 0; i < cpuLayout.cpus(); i++)
+        {
+            final boolean base = AffinityLock.BASE_AFFINITY.get(i);;
+            final boolean reservable = AffinityLock.RESERVED_AFFINITY.get(i);
 
             LOGGER.trace("cpu " + i + " base={} reservable= {}", i, base, reservable);
             AffinityLock lock = logicalCoreLocks[i] = newLock(i, base, reservable);

affinity/src/main/java/net/openhft/affinity/impl/LinuxHelper.java

@@ -21,6 +21,7 @@
 import org.jetbrains.annotations.NotNull;
 
 import java.util.Arrays;
+import java.util.BitSet;
 import java.util.List;
 
 public class LinuxHelper {
@@ -222,25 +223,27 @@ int sched_getaffinity(final int pid,
         return cpuset;
     }
 
-    public static void sched_setaffinity(final long affinity) {
+    public static void sched_setaffinity(final BitSet affinity) {
         final CLibrary lib = CLibrary.INSTANCE;
         final cpu_set_t cpuset = new cpu_set_t();
         final int size = version.isSameOrNewer(VERSION_2_6) ? cpu_set_t.SIZE_OF_CPU_SET_T : NativeLong.SIZE;
-        if(Platform.is64Bit()) {
-            cpuset.__bits[0].setValue(affinity);
-
-        } else {
-            cpuset.__bits[0].setValue(affinity & 0xFFFFFFFFL);
-            cpuset.__bits[1].setValue((affinity >>> 32) & 0xFFFFFFFFL);
+        final long[] bits = affinity.toLongArray();
+        for (int i = 0; i < bits.length; i++) {
+            if (Platform.is64Bit()) {
+                cpuset.__bits[i].setValue(bits[i]);
+            } else {
+                cpuset.__bits[i*2].setValue(bits[i] & 0xFFFFFFFFL);
+                cpuset.__bits[i*2+1].setValue((bits[i] >>> 32) & 0xFFFFFFFFL);
+            }
         }
         try {
             if(lib.sched_setaffinity(0, size, cpuset) != 0) {
                 throw new IllegalStateException("sched_setaffinity(0, " + size +
-                        ", 0x" + Long.toHexString(affinity) + " failed; errno=" + Native.getLastError());
+                        ", 0x" + Utilities.toHexString(affinity) + " failed; errno=" + Native.getLastError());
             }
         } catch (LastErrorException e) {
             throw new IllegalStateException("sched_setaffinity(0, " + size +
-                    ", 0x" + Long.toHexString(affinity) + " failed; errno=" + e.getErrorCode(), e);
+                    ", 0x" + Utilities.toHexString(affinity) + " failed; errno=" + e.getErrorCode(), e);
         }
     }
 

affinity/src/main/java/net/openhft/affinity/impl/LinuxJNAAffinity.java

@@ -21,21 +21,81 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+import java.util.Collections;
+import java.util.LinkedList;
+
 public enum LinuxJNAAffinity implements IAffinity {
     INSTANCE;
     private static final Logger LOGGER = LoggerFactory.getLogger(LinuxJNAAffinity.class);
     public static final boolean LOADED;
 
     // TODO: FIXME!!! CHANGE IAffinity TO SUPPORT PLATFORMS WITH 64+ CORES FIXME!!!
     @Override
-    public long getAffinity() {
+    public BitSet getAffinity() {
         final LinuxHelper.cpu_set_t cpuset = LinuxHelper.sched_getaffinity();
-        return cpuset.__bits[0].longValue();
+
+        boolean collect = false;
+        ArrayList<Byte> bytes = new ArrayList<Byte>();
+
+        ByteBuffer buff = null;
+        if (Platform.is64Bit())
+        {
+            buff = ByteBuffer.allocate(Long.SIZE / 8);
+        }
+        else
+        {
+            buff = ByteBuffer.allocate(Integer.SIZE / 8);
+        }
+
+        for (int i = cpuset.__bits.length - 1; i >= 0; --i)
+        {
+            if (!collect && cpuset.__bits[i].longValue() != 0)
+            {
+                collect = true;
+            }
+
+            if (collect)
+            {
+                if (Platform.is64Bit())
+                {
+                    buff.putLong(cpuset.__bits[i].longValue());
+                }
+                else
+                {
+                    buff.putInt((int) cpuset.__bits[i].longValue());
+                }
+
+                final byte[] arr = buff.array();
+                //for (int j = arr.length - 1; j >= 0; --j)
+                for (int j = 0; j < arr.length; j++)
+                {
+                    bytes.add(arr[j]);
+                }
+            }
+        }
+
+        if (!bytes.isEmpty())
+        {
+            byte[] data = new byte[bytes.size()];
+            for (int i = 0; i < bytes.size(); i++)
+            {
+                // don't forget to reverse the order of long values
+                data[data.length - i - 1] = bytes.get(i);
+            }
+            return BitSet.valueOf(data);
+        }
+        else
+        {
+            return new BitSet();
+        }
     }
 
     // TODO: FIXME!!! CHANGE IAffinity TO SUPPORT PLATFORMS WITH 64+ CORES FIXME!!!
     @Override
-    public void setAffinity(final long affinity) {
+    public void setAffinity(final BitSet affinity) {
         LinuxHelper.sched_setaffinity(affinity);
     }