diff --git a/src/main/java/com/intellij/rt/debugger/agent/DebuggerAgent.java b/src/main/java/com/intellij/rt/debugger/agent/DebuggerAgent.java index f75c8bb..fa1a7f8 100644 --- a/src/main/java/com/intellij/rt/debugger/agent/DebuggerAgent.java +++ b/src/main/java/com/intellij/rt/debugger/agent/DebuggerAgent.java @@ -64,7 +64,13 @@ private static void initAll(Instrumentation instrumentation, Properties properti CollectionBreakpointInstrumentor.init(properties, instrumentation); SpilledVariablesTransformer.init(instrumentation); TailCallContinuationTransformer.init(instrumentation); - LogCaptureTransformer.init(properties, instrumentation); + + boolean logCaptureEnabled = Boolean.getBoolean("debugger.agent.enable.log.capture"); + LogCaptureStorage.init(properties, logCaptureEnabled); + if (logCaptureEnabled) { + LogCaptureTransformer.init(instrumentation); + } + InstrumentationBreakpointTransformer.init(properties, instrumentation); } diff --git a/src/main/java/com/intellij/rt/debugger/agent/LogCaptureStorage.java b/src/main/java/com/intellij/rt/debugger/agent/LogCaptureStorage.java index e94d2c7..aa3f89c 100644 --- a/src/main/java/com/intellij/rt/debugger/agent/LogCaptureStorage.java +++ b/src/main/java/com/intellij/rt/debugger/agent/LogCaptureStorage.java @@ -29,23 +29,32 @@ protected Boolean initialValue() { private static boolean BATCHING_ENABLED; private static int MAX_BATCHED_EVENTS_COUNT; + private static boolean STDOUT_CAPTURE_ENABLED; // It's used by the debugger. - private static final AtomicLong EVENT_COUNTER = new AtomicLong(); + static final AtomicLong EVENT_COUNTER = new AtomicLong(); // It contains events that are waiting to be flushed. // New ones could be added concurrently. // They can also be flushed concurrently, leading to sending the same events multiple times. // It's ok and is handed by the debugger using IDs. // Event is removed from the queue only after it's guaranteed to be received by the debugger. - private static final ConcurrentLinkedQueue EVENTS = new ConcurrentLinkedQueue<>(); + static final ConcurrentLinkedQueue EVENTS = new ConcurrentLinkedQueue<>(); + + static final AtomicLong LAST_FLUSHED_EVENT_ID = new AtomicLong(-1); + static final AtomicLong LAST_LOGGING_BREAKPOINT_EVENT_ID = new AtomicLong(-1); + + static class Event { + public static final byte STD_OUTPUT_TYPE = 0; + public static final byte LOGGING_BREAKPOINT_TYPE = 1; - private static class Event { public final long id; + public final byte type; public final byte[] payload; - public Event(long id, byte[] payload) { + public Event(long id, byte type, byte[] payload) { this.id = id; + this.type = type; this.payload = payload; } } @@ -55,12 +64,17 @@ public Event(long id, byte[] payload) { private static final int MAX_STACK_DEPTH = 100; // It should be enough, we usually need only a few first frames. - public static boolean init(Properties properties) { + + private static boolean batchingSchedulerStarted; + static ArrayList outputWrittenDumpForTests = null; + + public static boolean init(Properties properties, boolean logCaptureEnabled) { ENABLED = true; + STDOUT_CAPTURE_ENABLED = logCaptureEnabled; BATCHING_ENABLED = Boolean.parseBoolean(properties.getProperty(BATCHING_ENABLED_PROPERTY, "true")); - if (BATCHING_ENABLED) { - MAX_BATCHED_EVENTS_COUNT = Integer.parseInt( - properties.getProperty(BATCHING_MAX_EVENTS_PROPERTY, "100")); + MAX_BATCHED_EVENTS_COUNT = Integer.parseInt(properties.getProperty(BATCHING_MAX_EVENTS_PROPERTY, "100")); + if (BATCHING_ENABLED && !batchingSchedulerStarted) { + batchingSchedulerStarted = true; final Runnable flushAction = new Runnable() { @Override @@ -87,6 +101,15 @@ public void run() { return true; } + private static long createNextEventId(int eventType) { + if (!BATCHING_ENABLED) return -1; + long id = EVENT_COUNTER.getAndIncrement(); + if (eventType == Event.LOGGING_BREAKPOINT_TYPE) { + setIfGreater(LAST_LOGGING_BREAKPOINT_EVENT_ID, id); + } + return id; + } + public static void capture(FileDescriptor fd, byte[] bytes) { capture(fd, bytes, 0, bytes.length); } @@ -98,12 +121,19 @@ public static void capture(FileDescriptor fd, byte[] bytes, int off, int len) { if (fd != FD_OUT && fd != FD_ERR) return; if (len == 0) return; - List regularStack = CaptureStorage.getCurrentStackTraceWithoutAgentFrames(); - List capturedStack = CaptureStorage.getCurrentCapturedStack(MAX_STACK_DEPTH - regularStack.size()); - - byte[] captured = encodeMessageAndStacks(bytes, off, len, regularStack, capturedStack); - captureEvent(captured); + // Avoid logging breakpoint's output reorder with stdout. + if (hasBatchedLoggingBreakpointEvents()) { + flushBatchedData(); + } + if (!STDOUT_CAPTURE_ENABLED) return; + long id = createNextEventId(Event.STD_OUTPUT_TYPE); + ByteArrayOutputStream bas = new ByteArrayOutputStream(); // no need to close it + try (DataOutputStream dos = new DataOutputStream(bas)) { + encodeMessageAndCurrentStacks(dos, bytes, off, len); + } + byte[] payload = bas.toByteArray(); + captureEvent(new Event(id, Event.STD_OUTPUT_TYPE, payload)); } catch (Throwable e) { handleException(e); } finally { @@ -111,31 +141,29 @@ public static void capture(FileDescriptor fd, byte[] bytes, int off, int len) { } } - private static void captureEvent(byte[] captured) throws IOException { + private static boolean hasBatchedLoggingBreakpointEvents() { + return LAST_LOGGING_BREAKPOINT_EVENT_ID.get() > LAST_FLUSHED_EVENT_ID.get(); + } + + private static void captureEvent(Event event) throws IOException { if (BATCHING_ENABLED) { - long id = EVENT_COUNTER.getAndIncrement(); - Event event = new Event(id, captured); EVENTS.add(event); flushBatchedDataIfMoreThan(MAX_BATCHED_EVENTS_COUNT); } else { - packAndSend(Collections.singletonList(new Event(-1, captured))); + packAndSend(Collections.singletonList(event)); } } - private static byte[] encodeMessageAndStacks(byte[] bytes, int off, int len, - List regularStack, - List capturedStack) throws IOException { - ByteArrayOutputStream bas = new ByteArrayOutputStream(); // no need to close it - try (DataOutputStream dos = new DataOutputStream(bas)) { - dos.writeInt(len); - dos.write(bytes, off, len); - CaptureStorage.writeAsyncStackTraceToStream(regularStack, dos); - if (capturedStack != null) { - CaptureStorage.writeAsyncStackTraceElementToStream(CaptureStorage.ASYNC_STACK_ELEMENT, dos); - CaptureStorage.writeAsyncStackTraceToStream(capturedStack, dos); - } + private static void encodeMessageAndCurrentStacks(DataOutputStream dos, byte[] bytes, int off, int len) throws IOException { + List regularStack = CaptureStorage.getCurrentStackTraceWithoutAgentFrames(); + List capturedStack = CaptureStorage.getCurrentCapturedStack(MAX_STACK_DEPTH - regularStack.size()); + dos.writeInt(len); + dos.write(bytes, off, len); + CaptureStorage.writeAsyncStackTraceToStream(regularStack, dos); + if (capturedStack != null) { + CaptureStorage.writeAsyncStackTraceElementToStream(CaptureStorage.ASYNC_STACK_ELEMENT, dos); + CaptureStorage.writeAsyncStackTraceToStream(capturedStack, dos); } - return bas.toByteArray(); } private static void handleException(Throwable e) { @@ -155,6 +183,8 @@ private static void flushBatchedDataIfMoreThan(int eventsCountLimit) throws IOEx if (eventsSnapshot.size() <= eventsCountLimit) return; packAndSend(eventsSnapshot); EVENTS.removeAll(new HashSet<>(eventsSnapshot)); + long lastFlushedId = findMaxId(eventsSnapshot); + setIfGreater(LAST_FLUSHED_EVENT_ID, lastFlushedId); } private static void packAndSend(Collection events) throws IOException { @@ -166,6 +196,7 @@ private static void packAndSend(Collection events) throws IOException { dos.writeInt(events.size()); for (Event event : events) { dos.writeLong(event.id); + dos.writeByte(event.type); byte[] bytes = event.payload; dos.writeInt(bytes.length); dos.write(bytes); @@ -176,8 +207,6 @@ private static void packAndSend(Collection events) throws IOException { outputWritten(packed); } - static ArrayList outputWrittenDumpForTests = null; - // It's used by the debugger. @SuppressWarnings("unused") private static void outputWritten(String captured) { @@ -186,9 +215,44 @@ private static void outputWritten(String captured) { } } - // It's used by the debugger and instrumentation. - @SuppressWarnings("unused") + // It's used in instrumentation. public static void loggingBreakpointHit(int instrumentationId, String message) { + if (!ENABLED || CAPTURING.get()) return; + CAPTURING.set(true); + try { + long id = createNextEventId(Event.LOGGING_BREAKPOINT_TYPE); + byte[] messageBytes = message.getBytes(StandardCharsets.UTF_8); + ByteArrayOutputStream bas = new ByteArrayOutputStream(); // no need to close it + try (DataOutputStream dos = new DataOutputStream(bas)) { + dos.writeInt(instrumentationId); + encodeMessageAndCurrentStacks(dos, messageBytes, 0, messageBytes.length); + } + byte[] payload = bas.toByteArray(); + captureEvent(new Event(id, Event.LOGGING_BREAKPOINT_TYPE, payload)); + } catch (Throwable e) { + handleException(e); + } finally { + CAPTURING.set(false); + } + } + + private static long findMaxId(ArrayList events) { + long lastFlushedId = -1; + for (int i = events.size() - 1; i >= 0; i--) { + long id = events.get(i).id; + if (id > lastFlushedId) { + lastFlushedId = id; + } + } + return lastFlushedId; } + private static void setIfGreater(AtomicLong maxValue, long newValue) { + while (true) { + long current = maxValue.get(); + if (current >= newValue || maxValue.compareAndSet(current, newValue)) { + break; + } + } + } } diff --git a/src/main/java/com/intellij/rt/debugger/agent/LogCaptureTransformer.java b/src/main/java/com/intellij/rt/debugger/agent/LogCaptureTransformer.java index fc4ade0..475a32a 100644 --- a/src/main/java/com/intellij/rt/debugger/agent/LogCaptureTransformer.java +++ b/src/main/java/com/intellij/rt/debugger/agent/LogCaptureTransformer.java @@ -10,16 +10,11 @@ import java.lang.instrument.Instrumentation; import java.lang.instrument.UnmodifiableClassException; import java.security.ProtectionDomain; -import java.util.Properties; import static com.intellij.rt.debugger.agent.CaptureAgent.getInternalClsName; class LogCaptureTransformer implements ClassFileTransformer { - public static void init(Properties properties, Instrumentation instrumentation) { - if (!Boolean.getBoolean("debugger.agent.enable.log.capture")) return; - - if (!LogCaptureStorage.init(properties)) return; - + public static void init(Instrumentation instrumentation) { instrumentation.addTransformer(new LogCaptureTransformer(), true); for (Class aClass : instrumentation.getAllLoadedClasses()) { if (CLASS_NAME.equals(getInternalClsName(aClass))) { diff --git a/src/test/java/com/intellij/rt/debugger/agent/LogCaptureEncodingTest.java b/src/test/java/com/intellij/rt/debugger/agent/LogCaptureEncodingTest.java index aeb12f3..15f86ac 100644 --- a/src/test/java/com/intellij/rt/debugger/agent/LogCaptureEncodingTest.java +++ b/src/test/java/com/intellij/rt/debugger/agent/LogCaptureEncodingTest.java @@ -1,5 +1,6 @@ package com.intellij.rt.debugger.agent; +import org.junit.Before; import org.junit.Test; import java.io.ByteArrayInputStream; @@ -12,46 +13,146 @@ import java.util.zip.GZIPInputStream; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; public class LogCaptureEncodingTest { + private final Properties properties = new Properties(); - @Test - public void test() throws Exception { - // Please disable the agent if you try to debug this test. - // Otherwise, you debug the bundled agent and not the code in the project. - assertEquals(this.getClass().getClassLoader(), LogCaptureStorage.class.getClassLoader()); - - Properties properties = new Properties(); + @Before + public void setUp() { + assertEquals( + "Please disable the agent if you try to debug this test. " + + "Otherwise, you debug the bundled agent and not the code in the project.", + this.getClass().getClassLoader(), LogCaptureStorage.class.getClassLoader()); + resetLogCaptureStorage(); properties.put(LogCaptureStorage.BATCHING_ENABLED_PROPERTY, "true"); - properties.put(LogCaptureStorage.BATCHING_MAX_EVENTS_PROPERTY, "1"); // 1 is ok, 2 is a signal to flush properties.put(LogCaptureStorage.BATCHING_FLUSH_PERIOD_PROPERTY, "999999999"); // never - LogCaptureStorage.init(properties); LogCaptureStorage.outputWrittenDumpForTests = new ArrayList<>(); + } + + @Test + public void batchesCapturedStdoutEvents() throws Exception { + properties.put(LogCaptureStorage.BATCHING_MAX_EVENTS_PROPERTY, "1"); // 1 is ok, 2 is a signal to flush + LogCaptureStorage.init(properties, true); + LogCaptureStorage.capture(FileDescriptor.out, "aaa\n".getBytes(StandardCharsets.UTF_8)); assertEquals(0, LogCaptureStorage.outputWrittenDumpForTests.size()); LogCaptureStorage.capture(FileDescriptor.out, "bbb\n".getBytes(StandardCharsets.UTF_8)); assertEquals(1, LogCaptureStorage.outputWrittenDumpForTests.size()); - String output = LogCaptureStorage.outputWrittenDumpForTests.get(0); - try (DataInputStream is = new DataInputStream(new GZIPInputStream(new ByteArrayInputStream(output.getBytes(StandardCharsets.ISO_8859_1))))) { + try (DataInputStream is = openDump(0)) { + assertEquals(2, is.readInt()); // count + readAndCheckStdoutEvent(0, "aaa\n", is); + readAndCheckStdoutEvent(1, "bbb\n", is); + } + } + + @Test + public void batchesLoggingBreakpointEvents() throws Exception { + properties.put(LogCaptureStorage.BATCHING_MAX_EVENTS_PROPERTY, "1"); // 1 is ok, 2 is a signal to flush + LogCaptureStorage.init(properties, false); + + LogCaptureStorage.loggingBreakpointHit(11, "first message"); + assertEquals(0, LogCaptureStorage.outputWrittenDumpForTests.size()); + LogCaptureStorage.loggingBreakpointHit(22, "second message"); + assertEquals(1, LogCaptureStorage.outputWrittenDumpForTests.size()); + + try (DataInputStream is = openDump(0)) { assertEquals(2, is.readInt()); // count - readAndCheckEvent(0, "aaa\n", is); - readAndCheckEvent(1, "bbb\n", is); + readAndCheckLoggingBreakpointEvent(0, 11, "first message", is); + readAndCheckLoggingBreakpointEvent(1, 22, "second message", is); + } + } + + @Test + public void stdoutCaptureFlushesPendingLoggingBreakpointEventsFirst() throws Exception { + properties.put(LogCaptureStorage.BATCHING_MAX_EVENTS_PROPERTY, "100"); + LogCaptureStorage.init(properties, true); + + LogCaptureStorage.loggingBreakpointHit(33, "before stdout"); + assertEquals(0, LogCaptureStorage.outputWrittenDumpForTests.size()); + + LogCaptureStorage.capture(FileDescriptor.out, "stdout\n".getBytes(StandardCharsets.UTF_8)); + + assertEquals(1, LogCaptureStorage.outputWrittenDumpForTests.size()); + try (DataInputStream is = openDump(0)) { + assertEquals(1, is.readInt()); // count + readAndCheckLoggingBreakpointEvent(0, 33, "before stdout", is); } } - private static void readAndCheckEvent(int expectedId, String expectedMsg, DataInputStream is) throws IOException { + @Test + public void stdoutFlushHookWorksWhenStdoutCaptureIsDisabled() throws Exception { + properties.put(LogCaptureStorage.BATCHING_MAX_EVENTS_PROPERTY, "100"); + LogCaptureStorage.init(properties, false); + + LogCaptureStorage.loggingBreakpointHit(44, "before ignored stdout"); + LogCaptureStorage.capture(FileDescriptor.out, "stdout\n".getBytes(StandardCharsets.UTF_8)); + + assertEquals(1, LogCaptureStorage.outputWrittenDumpForTests.size()); + try (DataInputStream is = openDump(0)) { + assertEquals(1, is.readInt()); // count + readAndCheckLoggingBreakpointEvent(0, 44, "before ignored stdout", is); + } + } + + @Test + public void keepsEventIdsAndOrderingAcrossMultipleFlushes() throws Exception { + properties.put(LogCaptureStorage.BATCHING_MAX_EVENTS_PROPERTY, "1"); // 1 is ok, 2 is a signal to flush + LogCaptureStorage.init(properties, true); + + LogCaptureStorage.loggingBreakpointHit(55, "first log"); + LogCaptureStorage.capture(FileDescriptor.out, "first stdout\n".getBytes(StandardCharsets.UTF_8)); + assertEquals(1, LogCaptureStorage.outputWrittenDumpForTests.size()); + + LogCaptureStorage.loggingBreakpointHit(66, "second log"); + LogCaptureStorage.capture(FileDescriptor.out, "second stdout\n".getBytes(StandardCharsets.UTF_8)); + assertEquals(2, LogCaptureStorage.outputWrittenDumpForTests.size()); + + try (DataInputStream is = openDump(0)) { + assertEquals(1, is.readInt()); // count + readAndCheckLoggingBreakpointEvent(0, 55, "first log", is); + } + try (DataInputStream is = openDump(1)) { + assertEquals(2, is.readInt()); // count + readAndCheckStdoutEvent(1, "first stdout\n", is); + readAndCheckLoggingBreakpointEvent(2, 66, "second log", is); + } + } + + private static DataInputStream openDump(int index) throws IOException { + String output = LogCaptureStorage.outputWrittenDumpForTests.get(index); + return new DataInputStream(new GZIPInputStream(new ByteArrayInputStream(output.getBytes(StandardCharsets.ISO_8859_1)))); + } + + private static void readAndCheckStdoutEvent(int expectedId, String expectedMsg, DataInputStream is) throws IOException { assertEquals(expectedId, is.readLong()); - byte[] msgAndStackTrace = readBytesWithSize(is); - try (DataInputStream eis = new DataInputStream(new ByteArrayInputStream(msgAndStackTrace))) { - byte[] msgBytes = readBytesWithSize(eis); - String msg = new String(msgBytes, StandardCharsets.UTF_8); - assertEquals(expectedMsg, msg); - // don't hassle with stack trace + assertEquals(LogCaptureStorage.Event.STD_OUTPUT_TYPE, is.readByte()); + try (DataInputStream eis = new DataInputStream(new ByteArrayInputStream(readBytesWithSize(is)))) { + readAndCheckMessageAndStack(expectedMsg, eis); } } + private static void readAndCheckLoggingBreakpointEvent(int expectedId, + int expectedInstrumentationId, + String expectedMsg, + DataInputStream is) throws IOException { + assertEquals(expectedId, is.readLong()); + assertEquals(LogCaptureStorage.Event.LOGGING_BREAKPOINT_TYPE, is.readByte()); + try (DataInputStream eis = new DataInputStream(new ByteArrayInputStream(readBytesWithSize(is)))) { + assertEquals(expectedInstrumentationId, eis.readInt()); + readAndCheckMessageAndStack(expectedMsg, eis); + } + } + + private static void readAndCheckMessageAndStack(String expectedMsg, DataInputStream is) throws IOException { + byte[] msgBytes = readBytesWithSize(is); + String msg = new String(msgBytes, StandardCharsets.UTF_8); + assertEquals(expectedMsg, msg); + assertTrue("expected encoded stack trace after message", is.available() > 0); + } + private static byte[] readBytesWithSize(DataInputStream is) throws IOException { // Performance is not critical, just do it in a loop missing Java 11 readNBytes(). int size = is.readInt(); @@ -61,4 +162,12 @@ private static byte[] readBytesWithSize(DataInputStream is) throws IOException { } return bytes; } + + private static void resetLogCaptureStorage() { + LogCaptureStorage.EVENT_COUNTER.set(0); + LogCaptureStorage.LAST_FLUSHED_EVENT_ID.set(-1); + LogCaptureStorage.LAST_LOGGING_BREAKPOINT_EVENT_ID.set(-1); + LogCaptureStorage.EVENTS.clear(); + LogCaptureStorage.outputWrittenDumpForTests = null; + } }