Merge pull request #80 from OpenHFT/79_fix_readMetaInfo_when_another_process_holds_the_lock

Fix getMetaInfo when another process holds the lock, Fixes #79

Author: nicktindall

affinity/src/main/java/net/openhft/affinity/common/ProcessRunner.java

@@ -0,0 +1,113 @@
+package net.openhft.affinity.common;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ManagementFactory;
+import java.lang.management.RuntimeMXBean;
+import java.nio.charset.Charset;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class ProcessRunner {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ProcessRunner.class);
+
+    /**
+     * Spawn a process running the main method of a specified class
+     *
+     * @param clazz The class to execute
+     * @param args  Any arguments to pass to the process
+     * @return the Process spawned
+     * @throws IOException if there is an error starting the process
+     */
+    public static Process runClass(Class<?> clazz, String... args) throws IOException {
+        // Because Java17 must be run using various module flags, these must be propagated
+        // to the child processes
+        // https://stackoverflow.com/questions/1490869/how-to-get-vm-arguments-from-inside-of-java-application
+        final RuntimeMXBean runtimeMxBean = ManagementFactory.getRuntimeMXBean();
+
+        // filter out javaagent params, or this confuses the IntelliJ debugger
+        final List<String> jvmArgsWithoutJavaAgents = runtimeMxBean.getInputArguments().stream()
+                .filter(arg -> !arg.startsWith("-javaagent:"))
+                .filter(arg -> !arg.startsWith("-agentlib:"))
+                .collect(Collectors.toList());
+
+        String classPath = System.getProperty("java.class.path");
+        String className = clazz.getName();
+        String javaBin = findJavaBinPath().toString();
+        List<String> allArgs = new ArrayList<>();
+        allArgs.add(javaBin);
+        allArgs.addAll(jvmArgsWithoutJavaAgents);
+        allArgs.add("-cp");
+        allArgs.add(classPath);
+        allArgs.add(className);
+        allArgs.addAll(Arrays.asList(args));
+        ProcessBuilder processBuilder = new ProcessBuilder(allArgs.toArray(new String[]{}));
+//        processBuilder.inheritIO(); // this doesn't place nice with surefire
+        return processBuilder.start();
+    }
+
+    /**
+     * Log stdout and stderr for a process
+     * <p>
+     * ProcessBuilder.inheritIO() didn't play nicely with Maven failsafe plugin
+     * <p>
+     * https://maven.apache.org/surefire/maven-failsafe-plugin/faq.html#corruptedstream
+     */
+    public static void printProcessOutput(String processName, Process process) {
+        LOGGER.info("\n"
+                + "Output for " + processName + "\n"
+                + "stdout:\n"
+                + copyStreamToString(process.getInputStream()) + "\n"
+                + "stderr:\n"
+                + copyStreamToString(process.getErrorStream()));
+    }
+
+    /**
+     * Copies a stream to a string, up to the point where reading more would block
+     *
+     * @param inputStream The stream to read from
+     * @return The output as a string
+     */
+    private static String copyStreamToString(InputStream inputStream) {
+        ByteArrayOutputStream os = new ByteArrayOutputStream();
+        byte[] buffer = new byte[1024];
+        int read;
+        try {
+            while (inputStream.available() > 0 && (read = inputStream.read(buffer)) >= 0) {
+                os.write(buffer, 0, read);
+            }
+        } catch (IOException e) {
+            // Ignore
+        }
+        return new String(os.toByteArray(), Charset.defaultCharset());
+    }
+
+    /**
+     * Try and work out what the java executable is cross platform
+     *
+     * @return the Path to the java executable
+     * @throws IllegalStateException if the executable couldn't be located
+     */
+    private static Path findJavaBinPath() {
+        final Path javaBinPath = Paths.get(System.getProperty("java.home")).resolve("bin");
+        final Path linuxJavaExecutable = javaBinPath.resolve("java");
+        if (linuxJavaExecutable.toFile().exists()) {
+            return linuxJavaExecutable;
+        } else {
+            Path windowsJavaExecutable = javaBinPath.resolve("java.exe");
+            if (windowsJavaExecutable.toFile().exists()) {
+                return windowsJavaExecutable;
+            }
+        }
+        throw new IllegalStateException("Couldn't locate java executable!");
+    }
+}

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

@@ -134,15 +134,21 @@ public String getMetaInfo(int id) throws IOException {
         }
 
         LockReference lr = locks[id];
-        if (lr == null) {
-            return null;
+        if (lr != null) {
+            return readMetaInfoFromLockFileChannel(file, lr.channel);
+        } else {
+            try (FileChannel fc = FileChannel.open(file.toPath(), READ)) {
+                return readMetaInfoFromLockFileChannel(file, fc);
+            }
         }
-        FileChannel fc = lr.channel;
+    }
+
+    private String readMetaInfoFromLockFileChannel(File lockFile, FileChannel lockFileChannel) throws IOException {
         ByteBuffer buffer = ByteBuffer.allocate(64);
-        int len = fc.read(buffer, 0);
+        int len = lockFileChannel.read(buffer, 0);
         String content = len < 1 ? "" : new String(buffer.array(), 0, len);
         if (content.isEmpty()) {
-            LOGGER.warn("Empty lock file {}", file.getAbsolutePath());
+            LOGGER.warn("Empty lock file {}", lockFile.getAbsolutePath());
             return null;
         }
         return content.substring(0, content.indexOf("\n"));

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

@@ -0,0 +1,59 @@
+package net.openhft.affinity;
+
+import net.openhft.affinity.common.ProcessRunner;
+import net.openhft.affinity.lockchecker.FileLockBasedLockChecker;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.fail;
+
+public class MultiProcessAffinityTest {
+
+    @Test
+    public void shouldNotAcquireLockOnCoresLockedByOtherProcesses() throws IOException, InterruptedException {
+        // run the separate affinity locker
+        final Process last = ProcessRunner.runClass(AffinityLockerProcess.class, "last");
+        try {
+            int lastCpuId = AffinityLock.PROCESSORS - 1;
+
+            // wait for the CPU to be locked
+            long endTime = System.currentTimeMillis() + 5_000;
+            while (FileLockBasedLockChecker.getInstance().isLockFree(lastCpuId)) {
+                Thread.sleep(100);
+                if (System.currentTimeMillis() > endTime) {
+                    fail("Timed out waiting for the sub-process to acquire the lock");
+                }
+            }
+
+            try (AffinityLock lock = AffinityLock.acquireLock("last")) {
+                assertNotEquals(lastCpuId, lock.cpuId());
+            }
+        } finally {
+            last.destroy();
+            if (!last.waitFor(5, TimeUnit.SECONDS)) {
+                fail("Sub-process didn't terminate!");
+            }
+        }
+    }
+
+    static class AffinityLockerProcess {
+
+        private static final Logger LOGGER = LoggerFactory.getLogger(AffinityLockerProcess.class);
+
+        public static void main(String[] args) {
+            String cpuIdToLock = args[0];
+
+            try (final AffinityLock affinityLock = AffinityLock.acquireLock(cpuIdToLock)) {
+                LOGGER.info("Got affinity lock " + affinityLock);
+                Thread.sleep(Integer.MAX_VALUE);
+            } catch (InterruptedException e) {
+                // expected, just end
+            }
+        }
+    }
+}