Limit bucket size by spilling

Author: robbins

java/src/com/ibm/javametrics/agent/JsonDataBucket.java renamed to java/src/com/ibm/javametrics/agent/ArrayDataBucket.java

@@ -17,12 +17,17 @@
 
 import java.util.ArrayList;
 
-public class JsonDataBucket implements Bucket {
+public class ArrayDataBucket implements Bucket {
 
-    private static final int MAX_SIZE = 4 * 1024;
-    private ArrayList<String> bucket = new ArrayList<String>(100);
+    private ArrayList<String> bucket;
     int size = 0;
     int cursor = 0;
+    private int maxBucketSize;
+
+    public ArrayDataBucket(int maxBucketSize) {
+        this.maxBucketSize = maxBucketSize;
+        bucket = new ArrayList<>(100);
+    }
 
     @Override
     public int getSize() {
@@ -39,18 +44,26 @@ public void empty() {
     }
 
     @Override
-    public void pushData(String data) {
+    public boolean addData(String data) {
         synchronized (bucket) {
-            bucket.add(data);
-            size += data.length();
 
-            System.err.println("pushed " + cursor + " " + bucket.size() + " " + size);
-            while (size > MAX_SIZE && cursor > 0) {
+            int newSize = size + data.length();
+            /*
+             * Spill data if necessary
+             */
+            while (newSize > maxBucketSize && cursor > 0) {
                 String removed = bucket.remove(0);
                 cursor--;
-                size -= removed.length();
-                System.err.println("trimmed " + cursor + " " + bucket.size() + " " + size);
+                newSize -= removed.length();
             }
+
+            if (newSize > maxBucketSize) {
+                return false;
+            }
+
+            bucket.add(data);
+            size = newSize;
+            return true;
         }
     }
 
@@ -61,7 +74,6 @@ public String getNext() {
             if (cursor < bucket.size()) {
                 data = bucket.get(cursor);
                 cursor++;
-                System.err.println("returned " + cursor + " " + bucket.size() + " " + size);
             }
         }
         return data;

java/src/com/ibm/javametrics/agent/Bucket.java

@@ -16,7 +16,7 @@
 package com.ibm.javametrics.agent;
 
 public interface Bucket {
-    public void pushData(String data);
+    public boolean addData(String data);
 
     public String getNext();
 

java/src/com/ibm/javametrics/agent/JavaAgent.java

@@ -27,7 +27,7 @@
 
 public class JavaAgent implements Agent {
 
-    private static final int MAX_BUCKET_SIZE = 1024 * 1024;
+    private static final int MAX_BUCKET_SIZE = 2 * 1024 * 1024;
     private Map<String, Bucket> buckets = new HashMap<String, Bucket>();
     private Set<Receiver> receivers = new HashSet<Receiver>();
     private int collectionInterval = 2;
@@ -51,34 +51,42 @@ public void pushData(String type, String data) {
         synchronized (buckets) {
             Bucket bucket = buckets.get(type);
             if (bucket == null) {
-                bucket = new JsonDataBucket();
+                bucket = new ArrayDataBucket(MAX_BUCKET_SIZE);
                 buckets.put(type, bucket);
             }
-
-            if ((bucket.getSize() + data.length()) > MAX_BUCKET_SIZE) {
-                drain(type, bucket);
-            }
-
-            bucket.pushData(data);
+            bucket.addData(data);
         }
     }
 
     private void drain() {
         synchronized (buckets) {
             buckets.forEach((name, bucket) -> {
-                drain(name, bucket);
+                drainBatched(name, bucket);
             });
         }
     }
 
+    @SuppressWarnings("unused")
     private void drain(String type, Bucket bucket) {
         String data = bucket.getNext();
         while (data != null) {
             emit(type, data);
             data = bucket.getNext();
         }
     }
-    
+
+    private void drainBatched(String type, Bucket bucket) {
+        StringBuffer sb = new StringBuffer();
+        String data = bucket.getNext();
+        while (data != null) {
+            sb.append(data);
+            data = bucket.getNext();
+        }
+        if (sb.length() > 0) {
+            emit(type, sb.toString());
+        }
+    }
+  
     private void emit(String type, String data) {
         if (data != null) {
             receivers.forEach((receiver) -> {

java/src/com/ibm/javametrics/agent/StringDataBucket.java

@@ -20,6 +20,12 @@ public class StringDataBucket implements Bucket {
     private static final int INITIAL_BUCKET_SIZE = 4 * 1024;
     private StringBuffer bucket = new StringBuffer(INITIAL_BUCKET_SIZE);
 
+    private int maxBucketSize;
+
+    public StringDataBucket(int maxBucketSize) {
+        this.maxBucketSize = maxBucketSize;
+    }
+
     @Override
     public int getSize() {
         return bucket.length();
@@ -31,14 +37,21 @@ public void empty() {
     }
 
     @Override
-    public void pushData(String data) {
+    public boolean addData(String data) {
+        if ((getSize() + data.length()) > maxBucketSize) {
+            return false;
+        }
         bucket.append(data);
+        return true;
     }
 
     @Override
     public String getNext() {
-        String data = bucket.toString();
-        bucket = new StringBuffer(INITIAL_BUCKET_SIZE);
+        String data = null;
+        if (bucket.length() > 0) {
+            data = bucket.toString();
+            bucket = new StringBuffer(INITIAL_BUCKET_SIZE);
+        }
         return data;
     }
 }

java/src/com/ibm/javametrics/impl/JavametricsImpl.java

@@ -63,7 +63,6 @@ private void initializeProviders() {
      * java.lang.String)
      */
     public void receiveData(String type, String data) {
-        System.err.println("received: " + type + " " + data);
         for (Iterator<JavametricsListener> iterator = javametricsListeners.iterator(); iterator.hasNext();) {
             JavametricsListener javametricsListener = iterator.next();
             javametricsListener.receive(type, data);

web/src/com/ibm/javametrics/web/DataHandler.java

@@ -17,7 +17,6 @@
 
 import java.io.StringReader;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;