AffinityLock.acquireCore should lock/unlock all cpus ie 2 when hypert…

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

@@ -69,6 +69,7 @@ public class AffinityLock implements Closeable {
      * Logical ID of the CPU to which this lock belongs to.
      */
     private final int cpuId;
+    private final int cpuId2;
     /**
      * CPU to which this lock belongs to is of general use.
      */
@@ -88,9 +89,10 @@ public class AffinityLock implements Closeable {
     Throwable boundHere;
     private boolean resetAffinity = true;
 
-    AffinityLock(int cpuId, boolean base, boolean reservable, LockInventory lockInventory) {
+    AffinityLock(int cpuId, int cpuId2, boolean base, boolean reservable, LockInventory lockInventory) {
         this.lockInventory = lockInventory;
         this.cpuId = cpuId;
+        this.cpuId2 = cpuId2;
         this.base = base;
         this.reservable = reservable;
     }
@@ -133,7 +135,7 @@ private static BitSet getReservedAffinity0() {
         int end = reservedAffinity.length();
         for (int i = 0; i < longs.length; i++) {
             int begin = Math.max(0, end - 16);
-            longs[i] = Long.parseLong(reservedAffinity.substring(begin, end), 16);
+            longs[i] = Long.parseUnsignedLong(reservedAffinity.substring(begin, end), 16);
             end = begin;
         }
         return BitSet.valueOf(longs);
@@ -183,11 +185,11 @@ public static AffinityLock acquireLock(boolean bind) {
      * for defining your thread layout centrally and passing the handle via dependency injection.
      *
      * @param cpuId the CPU id to bind to
-     * @return A handle for an affinity lock.
+     * @return A handle for an affinity lock, or no lock if no available CPU in the array
      */
     public static AffinityLock acquireLock(int cpuId) {
         if (cpuId < 0 || cpuId >= PROCESSORS) {
-            throw new IllegalArgumentException("cpuId must be between 0 and " + (PROCESSORS - 1) + ": " + cpuId);
+            return LOCK_INVENTORY.noLock();
         }
         return acquireLock(true, cpuId, AffinityStrategies.ANY);
     }
@@ -204,7 +206,8 @@ public static AffinityLock acquireLock(int cpuId) {
     public static AffinityLock acquireLock(int[] cpus) {
         for (int cpu : cpus) {
             if (cpu < 0 || cpu >= PROCESSORS) {
-                throw new IllegalArgumentException("cpuId must be between 0 and " + (PROCESSORS - 1) + ": " + cpu);
+                LOGGER.warn("cpuId {} is out of range", cpu);
+                continue;
             }
             AffinityLock lock = tryAcquireLock(true, cpu);
             if (lock != null) {
@@ -265,7 +268,7 @@ public static AffinityLock acquireLock(String desc) {
 
         } else if (desc.startsWith("csv:")) {
             String content = desc.substring(4);
-            int[] cpus = Arrays.asList(content.split(",")).stream()
+            int[] cpus = Arrays.stream(content.split(","))
                     .map(String::trim)
                     .mapToInt(Integer::parseInt).toArray();
 
@@ -469,6 +472,10 @@ public int cpuId() {
         return cpuId;
     }
 
+    public int cpuId2() {
+        return cpuId2;
+    }
+
     /**
      * @return Was a cpu found to bind this lock to.
      */

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

@@ -49,4 +49,10 @@ public interface CpuLayout {
      * @return which thread on a core this cpu is on.
      */
     int threadId(int cpuId);
+
+    /**
+     * @param cpuId the logical processor number
+     * @return the hyperthreaded pair number or 0 if not hyperthreaded.
+     */
+    int pair(int cpuId);
 }

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

@@ -55,8 +55,8 @@ public static boolean isCpuFree(int cpu) {
         return isLockFree(cpu);
     }
 
-    static boolean replacePid(int cpu, long processID) throws IOException {
-        return storePid(processID, cpu);
+    static boolean replacePid(int cpu, int cpu2, long processID) throws IOException {
+        return storePid(processID, cpu, cpu2);
     }
 
     public static boolean isProcessRunning(long pid) {
@@ -70,8 +70,8 @@ public static boolean isProcessRunning(long pid) {
      * stores the pid in a file, named by the core, the pid is written to the file with the date
      * below
      */
-    private synchronized static boolean storePid(long processID, int cpu) throws IOException {
-        return lockChecker.obtainLock(cpu, Long.toString(processID));
+    private synchronized static boolean storePid(long processID, int cpu, int cpu2) throws IOException {
+        return lockChecker.obtainLock(cpu, cpu2, Long.toString(processID));
     }
 
     private synchronized static boolean isLockFree(int id) {
@@ -91,10 +91,10 @@ public static int getProcessForCpu(int core) throws IOException {
         return EMPTY_PID;
     }
 
-    static boolean updateCpu(int cpu) throws IOException {
+    static boolean updateCpu(int cpu, int cpu2) throws IOException {
         if (!canOSSupportOperation())
             return true;
-        return replacePid(cpu, getPID());
+        return replacePid(cpu, cpu2, getPID());
     }
 
     public static void releaseLock(int cpu) {

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

@@ -82,7 +82,7 @@ private static boolean isAnyCpu(final int cpuId) {
      */
     private static boolean updateLockForCurrentThread(final boolean bind, final AffinityLock al, final boolean wholeCore) throws ClosedByInterruptException {
         try {
-            if (LockCheck.updateCpu(al.cpuId())) {
+            if (LockCheck.updateCpu(al.cpuId(), wholeCore ? al.cpuId2() : 0)) {
                 al.assignCurrentThread(bind, wholeCore);
                 return true;
             }
@@ -108,7 +108,9 @@ public final synchronized void set(CpuLayout cpuLayout) {
             final boolean base = AffinityLock.BASE_AFFINITY.get(i);
             final boolean reservable = AffinityLock.RESERVED_AFFINITY.get(i);
             LOGGER.trace("cpu {} base={} reservable= {}", i, base, reservable);
-            AffinityLock lock = logicalCoreLocks[i] = newLock(i, base, reservable);
+            assert logicalCoreLocks != null;
+            @SuppressWarnings("resource")
+            AffinityLock lock = logicalCoreLocks[i] = newLock(i, cpuLayout.pair(i), base, reservable);
 
             int layoutId = lock.cpuId();
             int physicalCore = toPhysicalCore(layoutId);
@@ -277,8 +279,8 @@ public final synchronized String dumpLocks() {
         return dumpLocks(logicalCoreLocks);
     }
 
-    protected AffinityLock newLock(int cpuId, boolean base, boolean reservable) {
-        return new AffinityLock(cpuId, base, reservable, this);
+    protected AffinityLock newLock(int cpuId, int cpuId2, boolean base, boolean reservable) {
+        return new AffinityLock(cpuId, cpuId2, base, reservable, this);
     }
 
     private void reset(CpuLayout cpuLayout) {
@@ -301,6 +303,6 @@ private void releaseAffinityLock(final Thread t, final AffinityLock al, final St
     }
 
     public AffinityLock noLock() {
-        return newLock(AffinityLock.ANY_CPU, false, false);
+        return newLock(AffinityLock.ANY_CPU, 0, false, false);
     }
 }

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

@@ -64,4 +64,9 @@ public int coreId(int cpuId) {
     public int threadId(int cpuId) {
         return 0;
     }
+
+    @Override
+    public int pair(int cpuId) {
+        return 0;
+    }
 }

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

@@ -177,6 +177,19 @@ public int threadId(int cpuId) {
         return cpuDetails.get(cpuId).threadId;
     }
 
+    @Override
+    public int pair(int cpuId) {
+        for (int i = 0; i < cpuDetails.size(); i++) {
+            CpuInfo info = cpuDetails.get(i);
+            if (info.socketId == cpuDetails.get(cpuId).socketId &&
+                    info.coreId == cpuDetails.get(cpuId).coreId &&
+                    info.threadId != cpuDetails.get(cpuId).threadId) {
+                return i;
+            }
+        }
+        return 0;
+    }
+
     @NotNull
     @Override
     public String toString() {

affinity/src/main/java/net/openhft/affinity/lockchecker/FileLockBasedLockChecker.java

@@ -98,14 +98,25 @@ public synchronized boolean isLockFree(int id) {
     }
 
     @Override
-    public synchronized boolean obtainLock(int id, String metaInfo) throws IOException {
+    public synchronized boolean obtainLock(int id, int id2, String metaInfo) throws IOException {
         int attempt = 0;
         while (attempt < MAX_LOCK_RETRIES) {
             try {
                 LockReference lockReference = tryAcquireLockOnFile(id, metaInfo);
                 if (lockReference != null) {
-                    locks[id] = lockReference;
-                    return true;
+                    if (id2 <= 0) {
+                        // no second lock to acquire, return success
+                        locks[id] = lockReference;
+                        return true;
+                    }
+                    LockReference lockReference2 = tryAcquireLockOnFile(id2, metaInfo);
+                    if (lockReference2 != null) {
+                        locks[id] = lockReference;
+                        locks[id2] = lockReference2;
+                        return true;
+                    } else {
+                        releaseLock(id);
+                    }
                 }
                 return false;
             } catch (ConcurrentLockFileDeletionException e) {
@@ -163,6 +174,7 @@ private void writeMetaInfoToFile(FileChannel fc, String metaInfo) throws IOExcep
         byte[] content = String.format("%s%n%s", metaInfo, dfTL.get().format(new Date())).getBytes();
         ByteBuffer buffer = ByteBuffer.wrap(content);
         while (buffer.hasRemaining()) {
+            //noinspection ResultOfMethodCallIgnored
             fc.write(buffer);
         }
     }

affinity/src/main/java/net/openhft/affinity/lockchecker/LockChecker.java

@@ -26,7 +26,7 @@ public interface LockChecker {
 
     boolean isLockFree(int id);
 
-    boolean obtainLock(int id, String metaInfo) throws IOException;
+    boolean obtainLock(int id, int id2, String metaInfo) throws IOException;
 
     boolean releaseLock(int id);
 

affinity/src/test/java/net/openhft/affinity/AffinityLockTest.java

@@ -19,7 +19,6 @@
 
 import net.openhft.affinity.impl.Utilities;
 import net.openhft.affinity.impl.VanillaCpuLayout;
-import net.openhft.affinity.testimpl.TestFileLockBasedLockChecker;
 import org.hamcrest.MatcherAssert;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -28,6 +27,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
@@ -43,20 +43,19 @@
 public class AffinityLockTest extends BaseAffinityTest {
     private static final Logger logger = LoggerFactory.getLogger(AffinityLockTest.class);
 
-    private final TestFileLockBasedLockChecker lockChecker = new TestFileLockBasedLockChecker();
 
     @Test
     public void dumpLocksI7() throws IOException {
         LockInventory lockInventory = new LockInventory(VanillaCpuLayout.fromCpuInfo("i7.cpuinfo"));
         AffinityLock[] locks = {
-                new AffinityLock(0, true, false, lockInventory),
-                new AffinityLock(1, false, false, lockInventory),
-                new AffinityLock(2, false, true, lockInventory),
-                new AffinityLock(3, false, true, lockInventory),
-                new AffinityLock(4, true, false, lockInventory),
-                new AffinityLock(5, false, false, lockInventory),
-                new AffinityLock(6, false, true, lockInventory),
-                new AffinityLock(7, false, true, lockInventory),
+                new AffinityLock(0, 0, true, false, lockInventory),
+                new AffinityLock(1, 5, false, false, lockInventory),
+                new AffinityLock(2, 6, false, true, lockInventory),
+                new AffinityLock(3, 7, false, true, lockInventory),
+                new AffinityLock(4, 0, true, false, lockInventory),
+                new AffinityLock(5, 1, false, false, lockInventory),
+                new AffinityLock(6, 2, false, true, lockInventory),
+                new AffinityLock(7, 3, false, true, lockInventory),
         };
         locks[2].assignedThread = new Thread(new InterrupedThread(), "logger");
         locks[2].assignedThread.start();
@@ -86,10 +85,10 @@ public void dumpLocksI7() throws IOException {
     public void dumpLocksI3() throws IOException {
         LockInventory lockInventory = new LockInventory(VanillaCpuLayout.fromCpuInfo("i3.cpuinfo"));
         AffinityLock[] locks = {
-                new AffinityLock(0, true, false, lockInventory),
-                new AffinityLock(1, false, true, lockInventory),
-                new AffinityLock(2, true, false, lockInventory),
-                new AffinityLock(3, false, true, lockInventory),
+                new AffinityLock(0, 0, true, false, lockInventory),
+                new AffinityLock(1, 3, false, true, lockInventory),
+                new AffinityLock(2, 0, true, false, lockInventory),
+                new AffinityLock(3, 1, false, true, lockInventory),
         };
         locks[1].assignedThread = new Thread(new InterrupedThread(), "engine");
         locks[1].assignedThread.start();
@@ -109,8 +108,8 @@ public void dumpLocksI3() throws IOException {
     public void dumpLocksCoreDuo() throws IOException {
         LockInventory lockInventory = new LockInventory(VanillaCpuLayout.fromCpuInfo("core.duo.cpuinfo"));
         AffinityLock[] locks = {
-                new AffinityLock(0, true, false, lockInventory),
-                new AffinityLock(1, false, true, lockInventory),
+                new AffinityLock(0, 0, true, false, lockInventory),
+                new AffinityLock(1, 0, false, true, lockInventory),
         };
         locks[1].assignedThread = new Thread(new InterrupedThread(), "engine");
         locks[1].assignedThread.start();
@@ -253,11 +252,34 @@ public void lockFilesShouldBeRemovedOnRelease() {
         }
         final AffinityLock lock = AffinityLock.acquireLock();
 
-        assertTrue(Files.exists(Paths.get(lockChecker.doToFile(lock.cpuId()).getAbsolutePath())));
+        Path lockFile = Paths.get(System.getProperty("java.io.tmpdir"), "cpu-" + lock.cpuId() + ".lock");
+        assertTrue(Files.exists(lockFile));
 
         lock.release();
 
-        assertFalse(Files.exists(Paths.get(lockChecker.doToFile(lock.cpuId()).getAbsolutePath())));
+        assertFalse(Files.exists(lockFile));
+    }
+
+    @Test
+    public void wholeCoreLockReservesAllLogicalCpus() throws IOException {
+        if (!Utilities.ISLINUX || !new File("/proc/cpuinfo").exists()) {
+            return;
+        }
+        AffinityLock.cpuLayout(VanillaCpuLayout.fromCpuInfo());
+
+        CpuLayout layout = AffinityLock.cpuLayout();
+        try (AffinityLock lock = AffinityLock.acquireCore()) {
+            int socketId = layout.socketId(lock.cpuId());
+            int coreId = layout.coreId(lock.cpuId());
+            for (int i = 0; i < layout.cpus(); i++) {
+                if (layout.socketId(i) == socketId && layout.coreId(i) == coreId) {
+                    assertFalse("CPU " + i + " should be reserved", LockCheck.isCpuFree(i));
+                }
+            }
+        }
+        for (int i = 0; i < layout.cpus(); i++) {
+            assertTrue("CPU " + i + " should not be reserved", LockCheck.isCpuFree(i));
+        }
     }
 
     private void displayStatus() {
@@ -308,7 +330,7 @@ public void testTooHighCpuId() {
 
     @Test
     public void testTooHighCpuId2() {
-        try (AffinityLock ignored = AffinityLock.acquireLock(new int[] {123456})) {
+        try (AffinityLock ignored = AffinityLock.acquireLock(new int[]{123456})) {
             assertNotNull(ignored);
         }
     }

affinity/src/test/java/net/openhft/affinity/FileLockLockCheckTest.java

@@ -45,7 +45,7 @@ public void before() {
     @Test
     public void test() throws IOException {
         Assert.assertTrue(LockCheck.isCpuFree(cpu));
-        LockCheck.updateCpu(cpu);
+        LockCheck.updateCpu(cpu, 0);
         Assert.assertEquals(LockCheck.getPID(), LockCheck.getProcessForCpu(cpu));
     }
 
@@ -58,13 +58,13 @@ public void testPidOnLinux() {
     public void testReplace() throws IOException {
         cpu++;
         Assert.assertTrue(LockCheck.isCpuFree(cpu + 1));
-        LockCheck.replacePid(cpu, 123L);
+        LockCheck.replacePid(cpu, 0, 123L);
         Assert.assertEquals(123L, LockCheck.getProcessForCpu(cpu));
     }
 
     @Test
     public void shouldNotBlowUpIfPidFileIsEmpty() throws Exception {
-        LockCheck.updateCpu(cpu);
+        LockCheck.updateCpu(cpu, 0);
 
         final File file = lockChecker.doToFile(cpu);
         new RandomAccessFile(file, "rw").setLength(0);