diff --git a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java index d601146..035e3fa 100644 --- a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java +++ b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java @@ -28,7 +28,9 @@ import java.net.Socket; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.LinkOption; import java.nio.file.Path; +import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -38,6 +40,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.jar.JarEntry; @@ -87,22 +90,43 @@ static String encodeBytecodeFrame(String buildTag, String manifestJson, byte[] z // The unix control socket's path. Not final only so tests can bind one of // their own; the agent never changes it. static volatile String controlSocketPath = "/tmp/coverage_control.sock"; - private static final String DATA_SOCKET_PATH = "/tmp/coverage_data.sock"; + // The unix data socket each test's coverage is sent on. Not final for the + // same reason. + static volatile String dataSocketPath = "/tmp/coverage_data.sock"; private static final String DEFAULT_JACOCO_HOST = "127.0.0.1"; private static final int DEFAULT_JACOCO_PORT = 36320; private static final int SOCKET_TIMEOUT_MILLIS = 3000; private static final int SOCKET_BACKLOG = 50; - // The answer to an END not taken (no START for that test came first, as - // after a restart): nothing was published for it, so Keploy must not use - // the test's coverage. On the unix socket it is the whole answer; on TCP - // it is a frame naming the test, and ACK still follows (CoverageTcpClient). + // The answer to a command not taken (a START whose reset failed, an END + // with no START for that test, as after a restart): nothing is published + // for that test, so Keploy must not use its coverage. On the unix socket + // it starts the whole answer (UNIX_REPLY_IGNORED); on TCP it is a frame + // naming the test, and ACK still follows (CoverageTcpClient). static final String IGNORED_REPLY = "IGNORED"; + // The answers on the unix control socket: one line for every command the + // SDK reads, written before it closes the connection. ACK says it took the + // command: it reset the counters for a START, or sent the test's coverage + // for an END. IGNORED says it read the command and did not take it. + // proto=2 says this SDK answers every command it reads, so Keploy can read + // a close with no answer as a command not taken: on macOS a connection + // closed with its command unread gives Keploy no error, only the close. + static final String UNIX_REPLY_TAKEN = "ACK proto=2"; + static final String UNIX_REPLY_IGNORED = IGNORED_REPLY + " proto=2"; private static final AtomicBoolean STARTED = new AtomicBoolean(false); private static final AtomicBoolean SHUTDOWN_HOOK_REGISTERED = new AtomicBoolean(false); // The active transport worker: a CommandServer (unix, local/docker) or a - // CoverageTcpClient (TCP, k8s). Both are Closeable so stop() is transport-agnostic. - private static volatile Closeable coverageWorker; + // CoverageTcpClient (TCP, k8s), so stop() is transport-agnostic. + private static volatile CoverageWorker coverageWorker; + // The worker's thread. stop() waits for it to end, and start() does too + // when stop() gave up waiting (a command still running): until it ends it + // owns its socket, and a second worker on the same path would lose its + // socket file to the first one's close. Before the first start() it is a + // thread never started, which join() does not wait for. + private static volatile Thread coverageThread = new Thread("keploy-java-dedup-none"); + // How long stop(), and a start() after it, wait for the worker's thread. + // Not final: the tests shorten it. + static volatile long workerStopWaitMillis = 5000; private KeployDedupAgent() { } @@ -139,12 +163,17 @@ public static boolean start() { if (!STARTED.compareAndSet(false, true)) { return true; } + if (!awaitWorkerThread()) { + STARTED.set(false); + log(Level.SEVERE, "Java dedup is not started: the worker stopped before is still running a command", null); + return false; + } CoverageCollector collector = new CoverageCollector( new JacocoClient(resolveHost(), resolvePort()), new CoverageIndex()); - Runnable worker; + CoverageWorker worker; String threadName; String endpoint = resolveEndpoint(); if (endpoint != null) { @@ -185,12 +214,12 @@ public static boolean start() { // up after the app answered and Keploy had fired its first tests, // whose START and END then reached no one: their coverage was lost. CommandServer server = new CommandServer(collector, - new CoveragePublisher(new File(DATA_SOCKET_PATH)), new File(controlSocketPath)); + new CoveragePublisher(new File(dataSocketPath)), new File(controlSocketPath)); try { server.bind(); } catch (Throwable t) { // Never let premain fail the app's start: run it without dedup. - server.close(); + // bind() has closed the socket it opened. STARTED.set(false); log(Level.SEVERE, "Java dedup control socket server is unavailable", t); return false; @@ -202,6 +231,7 @@ public static boolean start() { Thread thread = new Thread(worker, threadName); thread.setDaemon(true); + coverageThread = thread; thread.start(); registerShutdownHook(); return true; @@ -220,18 +250,39 @@ public static boolean isStarted() { * Stops the background control socket listener. */ public static void stop() { - Closeable worker = coverageWorker; + CoverageWorker worker = coverageWorker; + coverageWorker = null; if (worker != null) { - try { - worker.close(); - } catch (IOException e) { - log(Level.FINE, "Failed to close Java dedup coverage worker", e); + worker.close(); + // Wait for the worker to end, so that a start() after this one + // does not overlap it. Still running after the wait, it is in a + // command or in an accept close()'s wake-up did not reach (the + // socket file was bound again by someone else). Either way it is + // past the point where closing its socket from here races its + // accept, so close it: an accept ends, and a command ends on its + // own once it is answered. A start() waits for it too. + if (!awaitWorkerThread()) { + worker.closeNow(); + log(Level.WARNING, "Java dedup worker did not stop within " + workerStopWaitMillis + + " ms (a command still running, or its socket file bound again by another process);" + + " its socket is closed, and it stops once its command ends", null); } } - coverageWorker = null; STARTED.set(false); } + // awaitWorkerThread waits up to workerStopWaitMillis for the last + // worker's thread to end and says whether it has. + private static boolean awaitWorkerThread() { + Thread thread = coverageThread; + try { + thread.join(workerStopWaitMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return !thread.isAlive(); + } + private static boolean isDisabled() { return isTruthy(System.getenv("KEPLOY_JAVA_DEDUP_DISABLED")) || isTruthy(System.getProperty("keploy.java.dedup.disabled")); @@ -316,6 +367,17 @@ private static final class ManifestEntry { private int probeCount; } + // fileKey is the file's identity (its device and inode on Linux and + // macOS), or null when it is gone or the platform has none. + private static Object fileKey(File file) { + try { + return Files.readAttributes(file.toPath(), BasicFileAttributes.class, LinkOption.NOFOLLOW_LINKS) + .fileKey(); + } catch (IOException e) { + return null; + } + } + private static void deleteSocketFile(File file) { if (file.exists() && !file.delete()) { log(Level.FINE, "Failed to delete socket file " + file.getAbsolutePath(), null); @@ -352,7 +414,18 @@ private static void diagnostic(String message) { LOGGER.log(Level.INFO, formatted); } - private static final class CommandServer implements Runnable, Closeable { + // A transport worker: run() serves Keploy's commands until close(). + private interface CoverageWorker extends Runnable, Closeable { + // close asks run() to end; it does not wait for it. + @Override + void close(); + + // closeNow closes the worker's socket from the caller's thread, for + // stop() once run() has outlived close(). + void closeNow(); + } + + private static final class CommandServer implements CoverageWorker { private final CoverageCollector collector; private final CoveragePublisher publisher; @@ -360,6 +433,9 @@ private static final class CommandServer implements Runnable, Closeable { private final AtomicBoolean running = new AtomicBoolean(true); private final Object testCaseLock = new Object(); private volatile AFUNIXServerSocket serverSocket; + // The socket file's identity once bound (fileKey), so that run() + // deletes the file only while it is still this socket's. + private volatile Object socketFileKey; private String activeTestId = ""; CommandServer(CoverageCollector collector, CoveragePublisher publisher, File controlSocket) { @@ -371,99 +447,150 @@ private static final class CommandServer implements Runnable, Closeable { // bind replaces a stale socket file and listens on the control socket. // start() calls it before it returns, so a command sent once start() // has returned waits in the backlog for run() instead of being refused. - // On a failure the caller closes this server, which closes the socket. + // On a failure it closes the socket it opened, as no run() will. void bind() throws IOException { deleteSocketFile(controlSocket); - serverSocket = AFUNIXServerSocket.newInstance(); - serverSocket.bind(AFUNIXSocketAddress.of(controlSocket), SOCKET_BACKLOG); - relaxSocketPermissions(controlSocket); + AFUNIXServerSocket socket = AFUNIXServerSocket.newInstance(); + // Not junixsocket's delete on close: it deletes by path, so it + // deleted the file of a server that had bound the path since. + socket.setDeleteOnClose(false); + try { + socket.bind(AFUNIXSocketAddress.of(controlSocket), SOCKET_BACKLOG); + socketFileKey = fileKey(controlSocket); + relaxSocketPermissions(controlSocket); + } catch (Throwable t) { + closeQuietly(socket); + throw t; + } + serverSocket = socket; } + // run accepts commands until close(). Only this thread closes the + // server socket, and it does so once it has left accept: junixsocket's + // accept holds the socket's raw fd number, and when it finds the + // socket file gone it closes that number. Closed from another thread + // while this one was inside accept, the number could already belong + // to the next server's socket (start() after stop()), which was then + // closed with its file still there: every command was refused. @Override public void run() { try (AFUNIXServerSocket localServer = serverSocket) { while (running.get()) { + Socket accepted; try { - handle(localServer.accept()); + accepted = localServer.accept(); } catch (IOException e) { - if (running.get()) { - log(Level.SEVERE, "Failed to accept Java dedup coverage command", e); + if (!running.get()) { + break; } + if (localServer.isClosed()) { + // junixsocket closes the socket when its file was + // deleted or bound again by someone else: no + // command can reach it. Said once; looping on it + // spun a core and logged without end. + STARTED.set(false); + log(Level.SEVERE, "Java dedup control socket " + controlSocket.getAbsolutePath() + + " was closed: its file was deleted or replaced, so Keploy's coverage" + + " commands no longer reach this app", e); + break; + } + log(Level.SEVERE, "Failed to accept Java dedup coverage command", e); + continue; } + // close()'s wake-up is a connection with nothing sent: + // handle() answers none, and the loop ends. + handle(accepted); } } catch (Throwable t) { STARTED.set(false); log(Level.SEVERE, "Java dedup control socket server is unavailable", t); - } finally { + } + // The socket file goes once the socket is closed, only if it is + // still this socket's: deleted or bound again by someone else, it + // is theirs. A start() after stop() binds only once this thread + // has ended (awaitWorkerThread), so it is never the next server's. + if (Objects.equals(socketFileKey, fileKey(controlSocket))) { deleteSocketFile(controlSocket); } } + // handle reads one command and answers it before the close: ACK if it + // was taken, IGNORED if not. Only a connection closed with nothing sent + // gets no answer, as there is no command to answer. private void handle(Socket socket) { try (Socket commandSocket = socket; BufferedReader reader = new BufferedReader( new InputStreamReader(commandSocket.getInputStream(), StandardCharsets.UTF_8))) { String line = reader.readLine(); - if (line == null || line.trim().isEmpty()) { + if (line == null) { return; } - CoverageCommand command = CoverageCommand.parse(line.trim()); - if (command == null) { - return; + boolean taken; + try { + taken = run(line.trim()); + } catch (RuntimeException e) { + // Not taken, and the server goes on: thrown past here, it + // would end the accept loop and every later test's command. + log(Level.SEVERE, "Failed to run Java dedup coverage command: " + line.trim(), e); + taken = false; } - - dispatch(command, commandSocket.getOutputStream()); + writeReply(commandSocket.getOutputStream(), taken ? UNIX_REPLY_TAKEN : UNIX_REPLY_IGNORED); } catch (IOException e) { log(Level.SEVERE, "Failed to handle Java dedup coverage command", e); } } - private void dispatch(CoverageCommand command, OutputStream outputStream) { + // run runs one command line ("START id", "END id") and says whether it + // was taken. + private boolean run(String line) { + CoverageCommand command = CoverageCommand.parse(line); + if (command == null) { + return false; + } synchronized (testCaseLock) { if (command.action == CommandAction.START) { - activeTestId = command.testId; + // Set only once the counters are reset: counters not reset + // hold the lines of the tests before this one, so this START + // is not taken and its END is ignored too. + activeTestId = ""; // Warm up app classes once before the first measured window so // one-time lines aren't charged to the first test. collector.warmup(); - collector.reset(); - writeAck(outputStream); - return; + if (!collector.reset()) { + return false; + } + activeTestId = command.testId; + return true; } - if (command.action == CommandAction.END) { - if (!command.testId.equals(activeTestId)) { - log(Level.SEVERE, - "Ignoring mismatched END command. expected=" + activeTestId + ", actual=" - + command.testId, - null); - // Not ACK: this END publishes nothing, and an ACK told - // Keploy it had, so it waited for a record that never came. - writeReply(outputStream, IGNORED_REPLY); - return; - } + if (!command.testId.equals(activeTestId)) { + log(Level.SEVERE, + "Ignoring mismatched END command. expected=" + activeTestId + ", actual=" + + command.testId, + null); + // Not ACK: this END publishes nothing, and an ACK told + // Keploy it had, so it waited for a record that never came. + return false; + } - try { - Map> executedProbesByClass = collector.capture(); - if (executedProbesByClass.isEmpty()) { - log(Level.FINE, "No Java coverage collected for " + command.testId, null); - } - publisher.publish(command.testId, executedProbesByClass); - } catch (Exception e) { - log(Level.SEVERE, "Failed to collect Java coverage for " + command.testId, e); - } finally { - activeTestId = ""; - writeAck(outputStream); + try { + Map> executedProbesByClass = collector.capture(); + if (executedProbesByClass.isEmpty()) { + log(Level.FINE, "No Java coverage collected for " + command.testId, null); } - return; + publisher.publish(command.testId, executedProbesByClass); + return true; + } catch (Exception e) { + // Not ACK for the same reason: nothing reached Keploy. + log(Level.SEVERE, "Failed to collect Java coverage for " + command.testId, e); + return false; + } finally { + activeTestId = ""; } } } - private void writeAck(OutputStream outputStream) { - writeReply(outputStream, "ACK"); - } - private void writeReply(OutputStream outputStream, String reply) { try { outputStream.write((reply + "\n").getBytes(StandardCharsets.UTF_8)); @@ -473,18 +600,34 @@ private void writeReply(OutputStream outputStream, String reply) { } } + // close stops run() and leaves the socket for run() to close: it + // wakes run()'s accept with a connection of its own. Only when that + // connection cannot be made (the file is gone, so no command can come + // either) does it close the socket from here, as nothing else can + // wake that accept. @Override public void close() { running.set(false); - AFUNIXServerSocket localServer = serverSocket; - if (localServer != null) { - try { - localServer.close(); - } catch (IOException e) { - log(Level.FINE, "Failed to close Java dedup control socket", e); - } + try (AFUNIXSocket wake = AFUNIXSocket.newInstance()) { + wake.connect(AFUNIXSocketAddress.of(controlSocket), SOCKET_TIMEOUT_MILLIS); + } catch (IOException e) { + log(Level.FINE, "Could not wake the Java dedup control socket; closing it", e); + closeNow(); } } + + @Override + public void closeNow() { + closeQuietly(serverSocket); + } + } + + private static void closeQuietly(Closeable closeable) { + try { + closeable.close(); + } catch (IOException e) { + log(Level.FINE, "Failed to close a Java dedup socket", e); + } } /** @@ -495,12 +638,14 @@ public void close() { * collector -> SDK : "START <id>" | "END <id>" * SDK -> collector : "ACK" (after START reset) * "COV <compact-json>" + "ACK" (after END dump) - * "IGNORED <id>" + "ACK" (after END with no START) + * "IGNORED <id>" + "ACK" (after a START whose reset + * failed, an END with no START, + * or an END whose dump failed) * * The collector starts listening only when replay begins, so the connect loop * retries until it is reachable. */ - private static final class CoverageTcpClient implements Runnable, Closeable { + private static final class CoverageTcpClient implements CoverageWorker { private static final long RECONNECT_DELAY_MILLIS = 1000; @@ -581,11 +726,31 @@ private void connectAndServe() throws IOException { private void dispatch(CoverageCommand command, OutputStream out) throws IOException { synchronized (testCaseLock) { if (command.action == CommandAction.START) { - activeTestId = command.testId; - // Warm up app classes once before the first measured window so - // one-time lines aren't charged to the first test. - collector.warmup(); - collector.reset(); + // As on the unix socket, the test is set only once the + // counters are reset: counters not reset hold the lines of + // the tests before this one, so this START is not taken and + // its END is ignored too. + activeTestId = ""; + boolean reset; + try { + // Warm up app classes once before the first measured window so + // one-time lines aren't charged to the first test. + collector.warmup(); + reset = collector.reset(); + } catch (RuntimeException e) { + // Not taken, and the connection goes on: thrown past + // here, it ended this thread, and no later command + // was answered. + log(Level.SEVERE, "Failed to reset Java coverage for " + command.testId, e); + reset = false; + } + if (reset) { + activeTestId = command.testId; + } else { + // Said before the ACK, as for an END not taken: the + // collector has no other sign this test went unmeasured. + writeLine(out, IGNORED_REPLY + " " + command.testId); + } writeLine(out, "ACK"); return; } @@ -613,8 +778,13 @@ private void dispatch(CoverageCommand command, OutputStream out) throws IOExcept try { executedProbesByClass = collector.capture(); } catch (Exception e) { + // Not taken, as on the unix socket: an empty COV is + // the frame of a test that ran no line of the app. log(Level.SEVERE, "Failed to collect Java coverage for " + command.testId, e); - executedProbesByClass = Collections.emptyMap(); + activeTestId = ""; + writeLine(out, IGNORED_REPLY + " " + command.testId); + writeLine(out, "ACK"); + return; } if (executedProbesByClass.isEmpty()) { log(Level.FINE, "No Java coverage collected for " + command.testId, null); @@ -660,6 +830,12 @@ public void close() { } } } + // closeNow is close(): it already closes the socket from the caller's + // thread, which no accept races here. + @Override + public void closeNow() { + close(); + } } private enum CommandAction { @@ -735,11 +911,14 @@ private CoverageCollector(JacocoClient jacocoClient, CoverageIndex coverageIndex this.coverageIndex = coverageIndex; } - private void reset() { + // reset zeroes JaCoCo's counters and says whether it did. + private boolean reset() { try { jacocoClient.dump(false, true); + return true; } catch (IOException e) { - log(Level.FINE, "Failed to reset JaCoCo counters", e); + log(Level.WARNING, "Failed to reset JaCoCo counters", e); + return false; } } diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java index 70c9a76..9ebdedb 100644 --- a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java +++ b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java @@ -2,44 +2,114 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -import java.io.BufferedReader; +import java.io.ByteArrayOutputStream; +import java.io.Closeable; import java.io.File; import java.io.IOException; -import java.io.InputStreamReader; +import java.io.InputStream; import java.io.OutputStream; +import java.net.Socket; +import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.attribute.PosixFilePermission; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.logging.Handler; +import java.util.logging.Level; +import java.util.logging.LogRecord; +import java.util.logging.Logger; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.newsclub.net.unix.AFUNIXServerSocket; import org.newsclub.net.unix.AFUNIXSocket; import org.newsclub.net.unix.AFUNIXSocketAddress; +import org.newsclub.net.unix.AFUNIXSocketChannel; /** * The unix control socket Keploy sends START and END on. Keploy fires its first * test as soon as the app answers its health check, so the socket must listen - * by the time start() (premain, before the app's main) returns; and an END the - * SDK does not take must not be answered as taken. + * by the time start() (premain, before the app's main) returns. Every command + * the SDK reads gets one answer line before the close: "ACK proto=2" if it was + * taken, "IGNORED proto=2" if not. Keploy reads a close with no answer from an + * SDK that says proto=2 as a command not taken, so only a connection closed + * with nothing sent may go unanswered. + * + *

The answers are compared with literals, not the SDK's constants: Keploy + * reads the words, so a change to them must fail here. */ public class ControlSocketTest { - private String savedPath; + private static final String JACOCO_PORT_PROPERTY = "keploy.jacoco.port"; + private static final String WARMUP_DISABLED_PROPERTY = "keploy.java.dedup.warmup.disabled"; + private static final String TAKEN = "ACK proto=2\n"; + private static final String IGNORED = "IGNORED proto=2\n"; + + private String savedControlPath; + private String savedDataPath; private File dir; + private FakeJacoco jacoco; + private DataReceiver data; + // What the agent logged at SEVERE during the test: the first 20 of them, + // so that a server looping on a closed socket fails a row, not the JVM. + private final List severe = new CopyOnWriteArrayList<>(); + private final Handler severeLog = new Handler() { + @Override + public void publish(LogRecord record) { + if (record.getLevel().intValue() >= Level.SEVERE.intValue() && severe.size() < 20) { + severe.add(record.getMessage()); + } + } + + @Override + public void flush() { + } + + @Override + public void close() { + } + }; + private AFUNIXServerSocket foreign; @Before public void setUp() throws IOException { - savedPath = KeployDedupAgent.controlSocketPath; + Logger.getLogger(KeployDedupAgent.class.getName()).addHandler(severeLog); + savedControlPath = KeployDedupAgent.controlSocketPath; + savedDataPath = KeployDedupAgent.dataSocketPath; dir = Files.createTempDirectory("keploy-dedup-control").toFile(); KeployDedupAgent.controlSocketPath = new File(dir, "control.sock").getAbsolutePath(); + KeployDedupAgent.dataSocketPath = new File(dir, "data.sock").getAbsolutePath(); + // The warmup loads every class on the test classpath; nothing here needs it. + System.setProperty(WARMUP_DISABLED_PROPERTY, "true"); } @After - public void tearDown() { + public void tearDown() throws Exception { KeployDedupAgent.stop(); - KeployDedupAgent.controlSocketPath = savedPath; + KeployDedupAgent.workerStopWaitMillis = 5000; + Logger.getLogger(KeployDedupAgent.class.getName()).removeHandler(severeLog); + Logger.getLogger(KeployDedupAgent.class.getName()).setUseParentHandlers(true); + if (foreign != null) { + foreign.close(); + } + KeployDedupAgent.controlSocketPath = savedControlPath; + KeployDedupAgent.dataSocketPath = savedDataPath; + System.clearProperty(JACOCO_PORT_PROPERTY); + System.clearProperty(WARMUP_DISABLED_PROPERTY); + if (jacoco != null) { + jacoco.close(); + } + if (data != null) { + data.close(); + } File[] left = dir.listFiles(); if (left != null) { for (File f : left) { @@ -69,7 +139,7 @@ public void startReplacesAStaleSocketFile() throws Exception { File socket = new File(KeployDedupAgent.controlSocketPath); assertTrue(socket.createNewFile()); assertTrue(KeployDedupAgent.start()); - assertEquals("IGNORED", send("END test-set-0/test-1")); + assertEquals(IGNORED, send("END test-set-0/test-1")); } @Test @@ -82,19 +152,402 @@ public void startReportsFailureWhenTheControlSocketCannotBind() { @Test public void endWithNoStartIsAnsweredIgnoredNotAck() throws Exception { assertTrue(KeployDedupAgent.start()); - assertEquals("IGNORED", send("END test-set-0/test-47")); + assertEquals(IGNORED, send("END test-set-0/test-47")); + } + + @Test + public void aTestStartedAndEndedIsAnsweredTakenAndItsCoverageIsSent() throws Exception { + startWithJacoco(); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals(TAKEN, send("START test-set-0/test-1")); + assertEquals(1, jacoco.served.get()); + assertEquals(TAKEN, send("END test-set-0/test-1")); + String payload = data.next(); + assertNotNull("the END's coverage reached the data socket", payload); + assertTrue(payload, payload.contains("\"id\":\"test-set-0/test-1\"")); + } + + @Test + public void anEndForAnotherTestIsIgnoredAndTheTestRunningGoesOn() throws Exception { + startWithJacoco(); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals(TAKEN, send("START test-set-0/test-1")); + assertEquals(IGNORED, send("END test-set-0/test-2")); + assertNull("an ignored END sends no coverage", data.poll()); + assertEquals(TAKEN, send("END test-set-0/test-1")); + assertTrue(data.next().contains("\"id\":\"test-set-0/test-1\"")); + } + + @Test + public void aStartWhoseResetFailedIsIgnoredAndSoIsItsEnd() throws Exception { + startWithJacoco(); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals(TAKEN, send("START test-set-0/test-1")); + jacoco.dropNext.set(1); + assertEquals(IGNORED, send("START test-set-0/test-2")); + // JaCoCo answers again, so only the failed START can make these ENDs + // not taken: the counters still hold test-1's lines and whatever ran + // after, and test-2's window never began. + assertEquals(IGNORED, send("END test-set-0/test-1")); + assertEquals(IGNORED, send("END test-set-0/test-2")); + assertNull("an ignored END sends no coverage", data.poll()); + assertEquals(TAKEN, send("START test-set-0/test-3")); + assertEquals(TAKEN, send("END test-set-0/test-3")); + } + + @Test + public void anEndWhoseCoverageCouldNotBeSentIsIgnoredAndEndsTheTest() throws Exception { + startWithJacoco(); + assertEquals(TAKEN, send("START test-set-0/test-1")); + // No one listens on the data socket. + assertEquals(IGNORED, send("END test-set-0/test-1")); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals("the failed END ended the test", IGNORED, send("END test-set-0/test-1")); + assertNull(data.poll()); + assertEquals(TAKEN, send("START test-set-0/test-2")); + assertEquals(TAKEN, send("END test-set-0/test-2")); + } + + @Test + public void aLineThatIsNoCommandIsIgnored() throws Exception { + startWithJacoco(); + String[] lines = {"", " ", "START", "START ", "END", "PING test-set-0/test-1", + "start test-set-0/test-1", "STARTtest-set-0/test-1"}; + for (String line : lines) { + assertEquals("[" + line + "]", IGNORED, send(line)); + } + byte[] notUtf8 = {(byte) 0xff, (byte) 0xfe, ' ', 'x', '\n'}; + assertEquals("bytes that are not UTF-8", IGNORED, sendBytes(notUtf8)); + assertEquals("none of them reset the counters", 0, jacoco.served.get()); + } + + @Test(timeout = 60000) + public void aCommandWithNoNewlineBeforeTheCloseIsRunAndAnswered() throws Exception { + startWithJacoco(); + assertEquals(TAKEN, sendThenCloseOutput("START test-set-0/test-1".getBytes(StandardCharsets.UTF_8))); + assertEquals(1, jacoco.served.get()); + } + + @Test(timeout = 60000) + public void aConnectionClosedWithNothingSentGetsNoAnswer() throws Exception { + assertTrue(KeployDedupAgent.start()); + assertEquals("", sendThenCloseOutput(new byte[0])); + assertEquals("the server still answers", IGNORED, send("END test-set-0/test-1")); + } + + @Test + public void aCommandThatThrowsIsIgnoredAndTheServerGoesOn() throws Exception { + // A port no socket can have: the reset throws IllegalArgumentException. + System.setProperty(JACOCO_PORT_PROPERTY, "70000"); + assertTrue(KeployDedupAgent.start()); + assertEquals(IGNORED, send("START test-set-0/test-1")); + assertEquals("the server still answers", IGNORED, send("END test-set-0/test-1")); + assertTrue(KeployDedupAgent.isStarted()); + } + + @Test(timeout = 60000) + public void stopWaitsForTheCommandInFlightAndForItsThread() throws Exception { + startWithJacoco(); + jacoco.holdMillis = 1000; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + // The START is in the SDK, waiting on JaCoCo, when stop() is called. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + KeployDedupAgent.stop(); + // Before, stop() closed the socket and returned at once, with the + // thread still inside the START. + assertFalse("the control thread ended before stop() returned", controlThreadAlive()); + assertEquals("the START was answered", TAKEN, answer.poll(30, TimeUnit.SECONDS)); + client.join(); + } + + @Test(timeout = 60000) + public void aStartAfterStopOnTheSamePathKeepsItsSocket() throws Exception { + for (int i = 0; i < 20; i++) { + assertTrue(KeployDedupAgent.start()); + assertEquals("round " + i, IGNORED, send("END test-set-0/test-1")); + KeployDedupAgent.stop(); + assertTrue(KeployDedupAgent.start()); + // Bound by the closing server's thread, the file was gone or its + // socket closed: these were refused. + assertEquals("round " + i, IGNORED, send("END test-set-0/test-1")); + assertEquals("round " + i, IGNORED, send("END test-set-0/test-1")); + KeployDedupAgent.stop(); + } + } + + @Test(timeout = 60000) + public void aSocketFileDeletedUnderTheServerEndsItOnceAndSaysSo() throws Exception { + startWithJacoco(); + jacoco.holdMillis = 500; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + // Deleted while the START runs: the next accept finds the file gone, + // and junixsocket closes the socket. + assertTrue(new File(KeployDedupAgent.controlSocketPath).delete()); + // A server that loops on the closed socket logs without end: kept + // off the console, so that it fails this row rather than the run. + Logger.getLogger(KeployDedupAgent.class.getName()).setUseParentHandlers(false); + assertEquals(TAKEN, answer.poll(30, TimeUnit.SECONDS)); + long ends = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while ((KeployDedupAgent.isStarted() || controlThreadAlive()) && System.nanoTime() < ends) { + Thread.sleep(5); + } + // Before, the accept loop went on on the closed socket: it spun a core + // and logged without end, and the agent still said it was started. + assertFalse("the agent says it is no longer started", KeployDedupAgent.isStarted()); + assertFalse("the control thread ended", controlThreadAlive()); + assertEquals("said once, and nothing else at SEVERE: " + severe, 1, severe.size()); + assertTrue(severe.get(0), severe.get(0).contains("was closed: its file was deleted or replaced")); + client.join(); + } + + @Test(timeout = 60000) + public void aStopDuringACommandLeavesTheSocketToItsOwnThread() throws Exception { + startWithJacoco(); + jacoco.holdMillis = 1500; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + Thread stopper = new Thread(KeployDedupAgent::stop); + stopper.start(); + Thread.sleep(300); + // stop() is waiting for the START. The socket is still open, as only + // its own thread closes it: closed from here, junixsocket's accept on + // that thread could close whatever socket took its fd number next. + assertTrue("stop() is still waiting for the command", stopper.isAlive()); + File socket = new File(KeployDedupAgent.controlSocketPath); + assertTrue("the socket file is still there", socket.exists()); + assertConnects("the socket still listens", socket); + stopper.join(); + assertEquals("the START was answered", TAKEN, answer.poll(30, TimeUnit.SECONDS)); + assertFalse("the control thread ended", controlThreadAlive()); + assertFalse("its own thread deleted the socket file", socket.exists()); + assertEquals("nothing at SEVERE: " + severe, 0, severe.size()); + client.join(); + } + + @Test(timeout = 60000) + public void aStopAfterAnotherServerTookThePathEndsTheThreadAndLeavesTheirSocket() throws Exception { + KeployDedupAgent.workerStopWaitMillis = 300; + assertTrue(KeployDedupAgent.start()); + assertEquals(IGNORED, send("END test-set-0/test-1")); + Thread.sleep(100); + severe.clear(); + // Another server takes the path while the control thread waits in its + // accept: close()'s wake-up reaches that server, not this one. + File socket = new File(KeployDedupAgent.controlSocketPath); + assertTrue(socket.delete()); + foreign = AFUNIXServerSocket.newInstance(); + foreign.bind(AFUNIXSocketAddress.of(socket)); + KeployDedupAgent.stop(); + assertFalse(KeployDedupAgent.isStarted()); + // stop() closed the socket once its wait was over: the thread ends. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (controlThreadAlive() && System.nanoTime() < deadline) { + Thread.sleep(5); + } + assertFalse("the control thread ended", controlThreadAlive()); + assertTrue("the other server's socket file is left alone", socket.exists()); + assertConnects("the other server still listens", socket); + assertEquals("a stop says nothing at SEVERE: " + severe, 0, severe.size()); + } + + @Test(timeout = 60000) + public void aStopAfterTheSocketFileWasDeletedEndsAtOnce() throws Exception { + KeployDedupAgent.workerStopWaitMillis = 20000; + assertTrue(KeployDedupAgent.start()); + assertEquals(IGNORED, send("END test-set-0/test-1")); + Thread.sleep(100); + assertTrue(new File(KeployDedupAgent.controlSocketPath).delete()); + long began = System.nanoTime(); + KeployDedupAgent.stop(); + long tookMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - began); + // No wake-up can reach the accept, so close() closes the socket at + // once; left to stop()'s second try, it took the whole wait. + assertTrue("stop() took " + tookMillis + " ms", tookMillis < 10000); + assertFalse("the control thread ended", controlThreadAlive()); + } + + @Test(timeout = 60000) + public void aStartWhileTheStoppedWorkerStillRunsACommandIsRefused() throws Exception { + KeployDedupAgent.workerStopWaitMillis = 100; + startWithJacoco(); + // Under the SDK's 3 s read timeout, so the START is answered ACK. + jacoco.holdMillis = 2000; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + KeployDedupAgent.stop(); + assertTrue("stop() gave up while the START runs", controlThreadAlive()); + // A second server on the path now would lose its socket file when + // the first one's thread ends and closes its own. + assertFalse("start() refuses while the old worker runs", KeployDedupAgent.start()); + assertFalse(KeployDedupAgent.isStarted()); + assertEquals("the START was answered", TAKEN, answer.poll(30, TimeUnit.SECONDS)); + while (controlThreadAlive() && System.nanoTime() < deadline) { + Thread.sleep(5); + } + assertTrue("start() binds once the old worker has ended", KeployDedupAgent.start()); + assertEquals(IGNORED, send("END test-set-0/test-1")); + client.join(); + } + + private static void assertConnects(String what, File socket) { + try (AFUNIXSocket probe = AFUNIXSocket.newInstance()) { + probe.connect(AFUNIXSocketAddress.of(socket), 3000); + } catch (IOException e) { + throw new AssertionError(what + ": " + e, e); + } + } + + private int jacocoConnections() { + return jacoco.accepted.get(); + } + + private static boolean controlThreadAlive() { + for (Thread t : Thread.getAllStackTraces().keySet()) { + if (t.getName().equals("keploy-java-dedup-control") && t.isAlive()) { + return true; + } + } + return false; + } + + private void startWithJacoco() throws IOException { + jacoco = new FakeJacoco(); + System.setProperty(JACOCO_PORT_PROPERTY, String.valueOf(jacoco.server.getLocalPort())); + assertTrue(KeployDedupAgent.start()); } private static String send(String command) throws IOException { + return sendBytes((command + "\n").getBytes(StandardCharsets.UTF_8)); + } + + // sendBytes sends raw as one command, as Keploy does, and returns all the + // SDK answered before its close. + private static String sendBytes(byte[] raw) throws IOException { try (AFUNIXSocket client = AFUNIXSocket.newInstance()) { client.connect(AFUNIXSocketAddress.of(new File(KeployDedupAgent.controlSocketPath)), 3000); client.setSoTimeout(30000); OutputStream out = client.getOutputStream(); - out.write((command + "\n").getBytes(StandardCharsets.UTF_8)); + out.write(raw); out.flush(); - BufferedReader in = new BufferedReader( - new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); - return in.readLine(); + return readAll(client.getInputStream()); + } + } + + // sendThenCloseOutput sends raw, closes its own side, and returns all the + // SDK answered before its close. A channel: on JDK 21 java.net.Socket's + // shutdownOutput refuses a junixsocket socket as not connected. + private static String sendThenCloseOutput(byte[] raw) throws IOException { + try (AFUNIXSocketChannel client = AFUNIXSocketChannel.open( + AFUNIXSocketAddress.of(new File(KeployDedupAgent.controlSocketPath)))) { + ByteBuffer out = ByteBuffer.wrap(raw); + while (out.hasRemaining()) { + client.write(out); + } + client.shutdownOutput(); + ByteArrayOutputStream all = new ByteArrayOutputStream(); + ByteBuffer in = ByteBuffer.allocate(512); + while (client.read(in) != -1) { + all.write(in.array(), 0, in.position()); + in.clear(); + } + return new String(all.toByteArray(), StandardCharsets.UTF_8); + } + } + + private static String readAll(InputStream in) throws IOException { + ByteArrayOutputStream all = new ByteArrayOutputStream(); + byte[] buf = new byte[512]; + int n; + while ((n = in.read(buf)) != -1) { + all.write(buf, 0, n); + } + return new String(all.toByteArray(), StandardCharsets.UTF_8); + } + + /** Keploy's data socket: it keeps each payload sent to it. */ + private static final class DataReceiver implements Closeable { + + final AFUNIXServerSocket server; + final BlockingQueue payloads = new LinkedBlockingQueue<>(); + + DataReceiver(String path) throws IOException { + server = AFUNIXServerSocket.newInstance(); + server.bind(AFUNIXSocketAddress.of(new File(path)), 50); + Thread thread = new Thread(this::serve, "data-receiver"); + thread.setDaemon(true); + thread.start(); + } + + private void serve() { + while (!server.isClosed()) { + try (Socket socket = server.accept()) { + socket.setSoTimeout(30000); + payloads.add(readAll(socket.getInputStream())); + } catch (IOException e) { + // Closed. + } + } + } + + // next waits for the next payload. The SDK sends it before it answers + // the END, so it is there, or on its way, once the END is answered. + String next() throws InterruptedException { + return payloads.poll(30, TimeUnit.SECONDS); + } + + // poll returns a payload already sent, or null. + String poll() throws InterruptedException { + return payloads.poll(200, TimeUnit.MILLISECONDS); + } + + @Override + public void close() throws IOException { + server.close(); } } } diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java b/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java index ddb83dc..1d48f5f 100644 --- a/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java +++ b/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java @@ -16,28 +16,39 @@ import org.junit.Test; /** - * The TCP transport, with the test as the collector the SDK dials (k8s). An END - * the SDK does not take publishes no COV, so it must say which test went - * unmeasured, and must still answer ACK, which is what ends the collector's wait. + * The TCP transport, with the test as the collector the SDK dials (k8s). A + * command the SDK does not take (an END with no START, a START whose reset + * failed) publishes no COV, so it must say which test went unmeasured, and must + * still answer ACK, which is what ends the collector's wait. */ public class CoverageTcpClientTest { private static final String ENDPOINT_PROPERTY = "keploy.coverage.endpoint"; + private static final String JACOCO_PORT_PROPERTY = "keploy.jacoco.port"; + private static final String WARMUP_DISABLED_PROPERTY = "keploy.java.dedup.warmup.disabled"; private ServerSocket collector; + private FakeJacoco jacoco; @Before public void setUp() throws IOException { collector = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); collector.setSoTimeout(30000); System.setProperty(ENDPOINT_PROPERTY, "127.0.0.1:" + collector.getLocalPort()); + // The warmup loads every class on the test classpath; nothing here needs it. + System.setProperty(WARMUP_DISABLED_PROPERTY, "true"); } @After public void tearDown() throws IOException { KeployDedupAgent.stop(); System.clearProperty(ENDPOINT_PROPERTY); + System.clearProperty(JACOCO_PORT_PROPERTY); + System.clearProperty(WARMUP_DISABLED_PROPERTY); collector.close(); + if (jacoco != null) { + jacoco.close(); + } } @Test @@ -59,4 +70,102 @@ public void endWithNoStartNamesTheTestThenAnswersAck() throws Exception { assertEquals("ACK", in.readLine()); } } + + @Test + public void aStartWhoseResetFailedNamesTheTestThenAnswersAckAndSoDoesItsEnd() throws Exception { + jacoco = new FakeJacoco(); + System.setProperty(JACOCO_PORT_PROPERTY, String.valueOf(jacoco.server.getLocalPort())); + assertTrue(KeployDedupAgent.start()); + try (Socket sdk = collector.accept()) { + sdk.setSoTimeout(30000); + OutputStream out = sdk.getOutputStream(); + BufferedReader in = new BufferedReader( + new InputStreamReader(sdk.getInputStream(), StandardCharsets.UTF_8)); + + send(out, "START test-set-0/test-1"); + assertEquals("ACK", in.readLine()); + // JaCoCo goes away for the next reset: the counters still hold + // test-1's lines, so test-2 is not measured. + jacoco.dropNext.set(1); + send(out, "START test-set-0/test-2"); + assertEquals("IGNORED test-set-0/test-2", in.readLine()); + assertEquals("ACK", in.readLine()); + // test-1 was left when test-2 started, and test-2 never began. + send(out, "END test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-2"); + assertEquals("IGNORED test-set-0/test-2", in.readLine()); + assertEquals("ACK", in.readLine()); + + // JaCoCo is back: the next test is measured as before. + send(out, "START test-set-0/test-3"); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-3"); + String cov = in.readLine(); + assertTrue("a COV for test-3, got " + cov, + cov != null && cov.startsWith("COV ") && cov.contains("test-set-0/test-3")); + assertEquals("ACK", in.readLine()); + } + } + + @Test + public void anEndWhoseDumpFailedNamesTheTestThenAnswersAckAndEndsTheTest() throws Exception { + jacoco = new FakeJacoco(); + System.setProperty(JACOCO_PORT_PROPERTY, String.valueOf(jacoco.server.getLocalPort())); + assertTrue(KeployDedupAgent.start()); + try (Socket sdk = collector.accept()) { + sdk.setSoTimeout(30000); + OutputStream out = sdk.getOutputStream(); + BufferedReader in = new BufferedReader( + new InputStreamReader(sdk.getInputStream(), StandardCharsets.UTF_8)); + + send(out, "START test-set-0/test-1"); + assertEquals("ACK", in.readLine()); + // JaCoCo goes away for the dump: test-1's lines never reach the + // SDK. Before, it sent an empty COV, as for a test that ran no + // line of the app. + jacoco.dropNext.set(1); + send(out, "END test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + // That END ended test-1. + send(out, "END test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + + send(out, "START test-set-0/test-2"); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-2"); + String cov = in.readLine(); + assertTrue("a COV for test-2, got " + cov, + cov != null && cov.startsWith("COV ") && cov.contains("test-set-0/test-2")); + assertEquals("ACK", in.readLine()); + } + } + + @Test + public void aStartThatThrowsNamesTheTestThenAnswersAckAndTheConnectionGoesOn() throws Exception { + // A port no socket can have: the reset throws IllegalArgumentException. + System.setProperty(JACOCO_PORT_PROPERTY, "70000"); + assertTrue(KeployDedupAgent.start()); + try (Socket sdk = collector.accept()) { + sdk.setSoTimeout(30000); + OutputStream out = sdk.getOutputStream(); + BufferedReader in = new BufferedReader( + new InputStreamReader(sdk.getInputStream(), StandardCharsets.UTF_8)); + + send(out, "START test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-1"); + assertEquals("the connection still answers", "IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + } + } + + private static void send(OutputStream out, String command) throws IOException { + out.write((command + "\n").getBytes(StandardCharsets.UTF_8)); + out.flush(); + } } diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/FakeJacoco.java b/keploy-sdk/src/test/java/io/keploy/dedup/FakeJacoco.java new file mode 100644 index 0000000..dfdd28e --- /dev/null +++ b/keploy-sdk/src/test/java/io/keploy/dedup/FakeJacoco.java @@ -0,0 +1,67 @@ +package io.keploy.dedup; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.concurrent.atomic.AtomicInteger; +import org.jacoco.core.runtime.RemoteControlReader; +import org.jacoco.core.runtime.RemoteControlWriter; + +/** + * A JaCoCo tcpserver: it answers each dump or reset with an empty dump. + * dropNext closes that many connections first with no answer, as a JaCoCo + * that went away does. + */ +final class FakeJacoco implements Closeable { + + final ServerSocket server; + final AtomicInteger dropNext = new AtomicInteger(); + final AtomicInteger served = new AtomicInteger(); + final AtomicInteger accepted = new AtomicInteger(); + // holdMillis holds each answer that long, as a JaCoCo busy with a big dump does. + volatile long holdMillis; + + FakeJacoco() throws IOException { + server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread thread = new Thread(this::serve, "fake-jacoco"); + thread.setDaemon(true); + thread.start(); + } + + private void serve() { + while (!server.isClosed()) { + try (Socket socket = server.accept()) { + accepted.incrementAndGet(); + socket.setSoTimeout(30000); + if (dropNext.getAndUpdate(n -> n > 0 ? n - 1 : 0) > 0) { + continue; + } + RemoteControlWriter writer = new RemoteControlWriter(socket.getOutputStream()); + RemoteControlReader reader = new RemoteControlReader(socket.getInputStream()); + reader.setRemoteCommandVisitor((dump, reset) -> { + hold(); + served.incrementAndGet(); + writer.sendCmdOk(); + }); + reader.read(); + } catch (IOException e) { + // Closed, or the client went away. + } + } + } + + private void hold() { + try { + Thread.sleep(holdMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + @Override + public void close() throws IOException { + server.close(); + } +} diff --git a/scripts/smoke-javaagent.sh b/scripts/smoke-javaagent.sh index 74a3e9e..fd00a46 100755 --- a/scripts/smoke-javaagent.sh +++ b/scripts/smoke-javaagent.sh @@ -129,9 +129,11 @@ public final class SmokeHarness { OutputStream output = socket.getOutputStream(); output.write((command + "\n").getBytes(StandardCharsets.UTF_8)); output.flush(); + // The whole answer: one line saying the command was taken, then + // the close. String ack = readAll(socket.getInputStream()); - if (!ack.contains("ACK")) { - throw new IllegalStateException("missing ACK for " + command + ": " + ack); + if (!"ACK proto=2\n".equals(ack)) { + throw new IllegalStateException("not taken: " + command + ": " + ack); } } }