IDEA-390527 Update buffering strategy

src/main/java/com/intellij/rt/debugger/agent/CaptureAgent.java

@@ -13,7 +13,7 @@
 import java.security.ProtectionDomain;
 import java.util.*;
 
-@SuppressWarnings({"UseOfSystemOutOrSystemErr", "CallToPrintStackTrace", "rawtypes"})
+@SuppressWarnings("rawtypes")
 public final class CaptureAgent {
   private static Instrumentation ourInstrumentation;
   private static final Set<Class> mySkipped = new HashSet<>();
@@ -122,7 +122,7 @@ public byte[] transform(ClassLoader loader,
   static void storeClassForDebug(String className, byte[] bytes) {
     if (CaptureStorage.DEBUG) {
       try {
-        FileOutputStream stream = new FileOutputStream("instrumented_" + className.replaceAll("/", "_") + ".class");
+        FileOutputStream stream = new FileOutputStream("instrumented_" + className.replace("/", "_") + ".class");
         try {
           stream.write(bytes);
         }

src/main/java/com/intellij/rt/debugger/agent/CaptureStorage.java

@@ -12,7 +12,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
-@SuppressWarnings({"UseOfSystemOutOrSystemErr"})
 public final class CaptureStorage {
   public static final String GENERATED_INSERT_METHOD_POSTFIX = "$$$capture";
   private static final ConcurrentIdentityWeakHashMap<Object, CapturedStack> STORAGE_GENERAL = new ConcurrentIdentityWeakHashMap<>();
@@ -32,6 +31,7 @@ protected Deque<CapturedStack> initialValue() {
   );
 
   static final double DEFAULT_OVERHEAD_PERCENT = 50;
+  private static final String PACKAGE_PREFIX = CaptureStorage.class.getPackage().getName();
   private static OverheadDetector ourOverheadDetector = new OverheadDetector(DEFAULT_OVERHEAD_PERCENT, true);
 
   static void init(Properties properties) {
@@ -645,12 +645,11 @@ public static void setEnabled(boolean enabled) {
   private static void handleException(Throwable e) {
     ENABLED = false;
     System.err.println("Critical error in IDEA Async Stacktraces instrumenting agent. Agent is now disabled. Please report to IDEA support:");
-    //noinspection CallToPrintStackTrace
     e.printStackTrace();
   }
 
   static boolean isAgentFrame(StackTraceElement elem) {
-    return elem.getClassName().startsWith(CaptureStorage.class.getPackage().getName());
+    return elem.getClassName().startsWith(PACKAGE_PREFIX);
   }
 
   static List<StackTraceElement> getCurrentStackTraceWithoutAgentFrames() {

src/main/java/com/intellij/rt/debugger/agent/CapturedStackDeduplicator.java

@@ -29,21 +29,19 @@ private StackRef(int id, List<StackTraceElement> stack) {
 
     private final List<LogCaptureStorage.Event> events;
     private final int maxStackDepth;
-    private final ArrayList<List<StackTraceElement>> stacks;
+    private final ArrayList<List<StackTraceElement>> stacks = new ArrayList<>();
     private final int[] throwableStackIds;
     private final int[] capturedStackIds;
-    private final IdentityHashMap<Throwable, StackRef> throwableStacks;
-    private final IdentityHashMap<CaptureStorage.CapturedStack, CapturedStackInfo> capturedStacks;
+    private final IdentityHashMap<Throwable, StackRef> throwableStacks = new IdentityHashMap<>();
+    private final IdentityHashMap<CaptureStorage.CapturedStack, CapturedStackInfo> capturedStacks = new IdentityHashMap<>();
 
     private CapturedStackDeduplicator(List<LogCaptureStorage.Event> events, int maxStackDepth) {
         this.events = events;
         this.maxStackDepth = maxStackDepth;
+
         int eventCount = events.size();
-        stacks = new ArrayList<>();
         throwableStackIds = new int[eventCount];
         capturedStackIds = new int[eventCount];
-        throwableStacks = new IdentityHashMap<>();
-        capturedStacks = new IdentityHashMap<>();
     }
 
     static StackDictionary createStackDictionary(List<LogCaptureStorage.Event> events, int maxStackDepth) {

src/main/java/com/intellij/rt/debugger/agent/CollectionBreakpointInstrumentor.java

@@ -18,7 +18,7 @@
 
 import static com.intellij.rt.debugger.agent.CaptureAgent.getInternalClsName;
 
-@SuppressWarnings({"UseOfSystemOutOrSystemErr", "CallToPrintStackTrace", "rawtypes"})
+@SuppressWarnings("rawtypes")
 public class CollectionBreakpointInstrumentor {
   private static final String OBJECT_TYPE = "Ljava/lang/Object;";
   private static final String STRING_TYPE = "Ljava/lang/String;";
@@ -139,13 +139,9 @@ private static void processFailedToInstrumentError(String className, Exception e
   private static void writeDebugInfo(String className, byte[] bytes) {
     try {
       System.out.println("instrumented: " + className);
-      FileOutputStream stream = new FileOutputStream("instrumented_" + className.replaceAll("/", "_") + ".class");
-      try {
-        stream.write(bytes);
-      }
-      finally {
-        stream.close();
-      }
+        try (FileOutputStream stream = new FileOutputStream("instrumented_" + className.replace("/", "_") + ".class")) {
+            stream.write(bytes);
+        }
     }
     catch (IOException e) {
       e.printStackTrace();

src/main/java/com/intellij/rt/debugger/agent/DebuggerAgent.java

@@ -14,7 +14,6 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 
-@SuppressWarnings("UseOfSystemOutOrSystemErr")
 public class DebuggerAgent {
   private static final String KEEP_SUFFIX = "[keep]";
 

src/main/java/com/intellij/rt/debugger/agent/LogCaptureStorage.java

@@ -10,6 +10,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.zip.Deflater;
 import java.util.zip.GZIPOutputStream;
 
 public class LogCaptureStorage {
@@ -27,36 +28,57 @@ protected Boolean initialValue() {
     static final String BATCHING_ENABLED_PROPERTY = "logCaptureBatchingEnabled";
     static final String BATCHING_FLUSH_PERIOD_PROPERTY = "logCaptureBatchingFlushPeriod";
     static final String BATCHING_BUFFER_SIZE_PROPERTY = "logCaptureBatchingBufferSize";
-    private static final long DEFAULT_BUFFER_SIZE = calculateDefaultBufferSize();
-    private static final int ESTIMATED_THROWABLE_BYTES = 3000;
+    static final String FORCE_BATCHING_BUFFER_SIZE_PROPERTY = "logCaptureForceBatchingBufferSize";
+    private static final int ESTIMATED_THROWABLE_BYTES = 3000; // See the comment at the top of ThrowableCapacityOverhead
 
     private static boolean BATCHING_ENABLED;
     private static long BUFFER_SIZE;
     private static boolean STDOUT_CAPTURE_ENABLED;
 
-    // It's used by the debugger.
+    // The debugger reads this value via evaluation to learn how many events have been created by the agent.
+    // The agent owns all writes.
     static final AtomicLong EVENT_COUNTER = new AtomicLong();
 
+    /**
+     * The debugger writes the greatest event id that was received and decoded successfully via the {@link #packBatchedData()} call.
+     * The confirmation is done outside the method itself because it can be called multiple times (see javadoc).
+     * The confirmation is not guaranteed, it is used only as an optimization against repeated sending of the same events.
+     * <p>
+     * The agent checks the flushed state based on this field and {@link #LAST_FLUSHED_EVENT_ID}.
+     */
+    static volatile long CONFIRMED_FLUSHED_EVENT_ID = -1;
+
     // It contains raw events that are waiting to be packed.
     // New ones could be added concurrently.
-    // Raw or packed data can be flushed concurrently, leading to sending the same events multiple times.
-    // It's ok and is handled by the debugger using IDs.
     static final ConcurrentLinkedQueue<Event> EVENTS = new ConcurrentLinkedQueue<>();
     static final AtomicLong EVENTS_PAYLOAD_BYTES = new AtomicLong();
+
+    // It contains packed batches that are waiting to be sent.
+    // New ones could be added concurrently.
     static final ConcurrentLinkedQueue<PackedBatch> PACKED_BATCHES = new ConcurrentLinkedQueue<>();
     static final AtomicLong PACKED_BATCHES_BYTES = new AtomicLong();
 
+    // Raw or packed data can be flushed concurrently, leading to sending the same events multiple times.
+    // It's ok and is handled by the debugger using IDs.
     static final AtomicLong LAST_FLUSHED_EVENT_ID = new AtomicLong(-1);
     static final AtomicLong LAST_PACKED_EVENT_ID = new AtomicLong(-1);
     static final AtomicLong LAST_LOGGING_BREAKPOINT_EVENT_ID = new AtomicLong(-1);
 
-    private interface MemoryFootprintEstimate {
-        int memoryFootprintEstimate();
+    private abstract static class MemoryFootprintEstimate {
+        private final AtomicBoolean removed = new AtomicBoolean();
+
+        public abstract int memoryFootprintEstimate();
 
-        boolean markRemoved();
+        /** Returns true if the event was removed for the first time. */
+        public boolean markRemoved() {
+            return removed.compareAndSet(false, true);
+        }
     }
 
-    static class Event implements MemoryFootprintEstimate {
+    /**
+     * A single event.
+     */
+    static class Event extends MemoryFootprintEstimate {
         public static final byte STD_OUTPUT_TYPE = 0;
         public static final byte LOGGING_BREAKPOINT_TYPE = 1;
 
@@ -65,7 +87,6 @@ static class Event implements MemoryFootprintEstimate {
         public final byte[] payload;
         public final Throwable throwable;
         public final CaptureStorage.CapturedStack stack;
-        private final AtomicBoolean removed = new AtomicBoolean();
 
         public Event(long id, byte type, byte[] payload, Throwable throwable, CaptureStorage.CapturedStack stack) {
             this.id = id;
@@ -81,15 +102,26 @@ public int memoryFootprintEstimate() {
         }
 
         @Override
-        public boolean markRemoved() {
-            return removed.compareAndSet(false, true);
+        public int hashCode() {
+            // This override is not necessary for correctness, but it is a bit faster than Object.hashCode.
+            return (int) (id ^ (id >>> 32));
+        }
+
+        @Override
+        public final boolean equals(Object o) {
+            if (!(o instanceof Event)) return false;
+
+            Event event = (Event) o;
+            return id == event.id;
         }
     }
 
-    static class PackedBatch implements MemoryFootprintEstimate {
+    /**
+     * Multiple events, compressed and packed. Batch is prepared to be sent to the debugger.
+     */
+    static class PackedBatch extends MemoryFootprintEstimate {
         public final byte[] data;
         public final long lastEventId;
-        private final AtomicBoolean removed = new AtomicBoolean();
 
         public PackedBatch(byte[] data, long lastEventId) {
             this.data = data;
@@ -100,11 +132,6 @@ public PackedBatch(byte[] data, long lastEventId) {
         public int memoryFootprintEstimate() {
             return data.length;
         }
-
-        @Override
-        public boolean markRemoved() {
-            return removed.compareAndSet(false, true);
-        }
     }
 
     private static final FileDescriptor FD_OUT = FileDescriptor.out;
@@ -120,8 +147,7 @@ public static boolean init(Properties properties, boolean logCaptureEnabled) {
         ENABLED = true;
         STDOUT_CAPTURE_ENABLED = logCaptureEnabled;
         BATCHING_ENABLED = Boolean.parseBoolean(properties.getProperty(BATCHING_ENABLED_PROPERTY, "true"));
-        // Split in 2 halves for raw events and compressed batches.
-        BUFFER_SIZE = getBufferSize(properties) / 2;
+        BUFFER_SIZE = getBufferSize(properties);
         if (BATCHING_ENABLED && !batchingSchedulerStarted) {
             batchingSchedulerStarted = true;
 
@@ -150,16 +176,17 @@ public void run() {
         return true;
     }
 
-    static long calculateDefaultBufferSize() {
-        long defaultMax = 5L * 1024L * 1024L; // 5 MB
-        long runtimeMax = Runtime.getRuntime().maxMemory() / 100; // 1% of max heap
-        return Math.min(defaultMax, runtimeMax);
-    }
-
     static long getBufferSize(Properties properties) {
-        return Long.parseLong(properties.getProperty(
-                BATCHING_BUFFER_SIZE_PROPERTY,
-                String.valueOf(DEFAULT_BUFFER_SIZE)));
+        String forcedBufferSize = properties.getProperty(FORCE_BATCHING_BUFFER_SIZE_PROPERTY);
+        if (forcedBufferSize != null) {
+            return Long.parseLong(forcedBufferSize);
+        }
+        String stringMaxSize = properties.getProperty(BATCHING_BUFFER_SIZE_PROPERTY);
+        long maxBufferSize = stringMaxSize == null
+                ? 5 * 1024L * 1024L // 5MB
+                : Long.parseLong(stringMaxSize);
+        long runtimeMax = Runtime.getRuntime().maxMemory() / 100; // 1% of max heap
+        return Math.min(maxBufferSize, runtimeMax);
     }
 
     private static long createNextEventId(int eventType) {
@@ -234,6 +261,7 @@ private static void handleException(Throwable e) {
      * before it appears on the debugger side. The clearing happens in the periodic flush cycle.
      */
     static String packBatchedData() throws IOException {
+        clearConfirmedBatches();
         packRawEvents();
         List<PackedBatch> packedBatchesSnapshot = new ArrayList<>(PACKED_BATCHES);
         if (packedBatchesSnapshot.isEmpty()) return null;
@@ -271,24 +299,52 @@ private static void enqueuePackedBatch(List<Event> events) throws IOException {
     }
 
     private static void flushBatchedData(boolean forceOutput) throws IOException {
-        if (forceOutput || currentEventsEstimatedBytes() > BUFFER_SIZE) {
+        clearConfirmedBatches();
+        if (forceOutput) {
             packRawEvents();
+        } else {
+            long eventsBytes = currentEventsEstimatedBytes();
+            long packedBytes = PACKED_BATCHES_BYTES.get();
+            if (eventsBytes + packedBytes <= BUFFER_SIZE) {
+                return;
+            }
+            packRawEvents();
+            packedBytes = PACKED_BATCHES_BYTES.get();
+            // We will zip too often if there is little space for raw events.
+            if (packedBytes <= BUFFER_SIZE * 9 / 10) {
+                return;
+            }
         }
-        if (!forceOutput && PACKED_BATCHES_BYTES.get() <= BUFFER_SIZE) return;
         List<PackedBatch> packedBatchesSnapshot = new ArrayList<>(PACKED_BATCHES);
         if (packedBatchesSnapshot.isEmpty()) return;
 
         outputWritten(packBatches(packedBatchesSnapshot));
-        long removedBytes = removeItems(PACKED_BATCHES, packedBatchesSnapshot);
+        markBatchesFlushed(packedBatchesSnapshot);
+    }
+
+    private static void markBatchesFlushed(Collection<PackedBatch> batches) {
+        long removedBytes = removeItems(PACKED_BATCHES, batches);
         PACKED_BATCHES_BYTES.addAndGet(-removedBytes);
-        setIfGreater(LAST_FLUSHED_EVENT_ID, findMaxPackedEventId(packedBatchesSnapshot));
+        setIfGreater(LAST_FLUSHED_EVENT_ID, findMaxPackedEventId(batches));
+    }
+
+    private static void clearConfirmedBatches() {
+        long confirmedId = CONFIRMED_FLUSHED_EVENT_ID;
+        if (confirmedId <= LAST_FLUSHED_EVENT_ID.get()) return;
+        Set<PackedBatch> alreadyConfirmed = new HashSet<>();
+        for (PackedBatch batch : PACKED_BATCHES) {
+            if (batch.lastEventId <= confirmedId) {
+                alreadyConfirmed.add(batch);
+            }
+        }
+        markBatchesFlushed(alreadyConfirmed);
     }
 
     private static byte[] packBytes(List<Event> events) throws IOException {
         assert !events.isEmpty();
 
         ByteArrayOutputStream bas = new ByteArrayOutputStream(); // no need to close it
-        try (GZIPOutputStream gos = new GZIPOutputStream(bas);
+        try (GZIPOutputStream gos = new FastGzipOutputStream(bas);
              DataOutputStream dos = new DataOutputStream(gos)) {
             CapturedStackDeduplicator.StackDictionary stackDictionary =
                     CapturedStackDeduplicator.createStackDictionary(events, MAX_STACK_DEPTH);
@@ -316,6 +372,13 @@ private static byte[] packBytes(List<Event> events) throws IOException {
         return bas.toByteArray();
     }
 
+    private static class FastGzipOutputStream extends GZIPOutputStream {
+        FastGzipOutputStream(ByteArrayOutputStream out) throws IOException {
+            super(out);
+            def.setLevel(Deflater.BEST_SPEED);
+        }
+    }
+
     private static byte[] packStack(List<StackTraceElement> stackTrace) throws IOException {
         ByteArrayOutputStream bas = new ByteArrayOutputStream(); // no need to close it
         try (DataOutputStream dos = new DataOutputStream(bas)) {
@@ -387,7 +450,7 @@ private static long findMaxPackedEventId(Collection<PackedBatch> packedBatches)
     }
 
     private static <T extends MemoryFootprintEstimate> long removeItems(ConcurrentLinkedQueue<T> queue, Collection<T> items) {
-        Set<T> itemsToRemove = new HashSet<>(items);
+        Set<T> itemsToRemove = items instanceof Set ? (Set<T>) items : new HashSet<>(items);
 
         long removedBytes = 0;
         for (Iterator<T> queueIterator = queue.iterator(); queueIterator.hasNext(); ) {