Merge pull request #17 from robbinspg/bckets

Improve bucket implementation

Author: tobespc

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

@@ -0,0 +1,81 @@
+/*******************************************************************************
+ * Copyright 2017 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ ******************************************************************************/
+package com.ibm.javametrics.agent;
+
+import java.util.ArrayList;
+
+public class ArrayDataBucket implements Bucket {
+
+    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() {
+        return size;
+    }
+
+    @Override
+    public void empty() {
+        synchronized (bucket) {
+            bucket.clear();
+            cursor = 0;
+            size = 0;
+        }
+    }
+
+    @Override
+    public boolean addData(String data) {
+        synchronized (bucket) {
+
+            int newSize = size + data.length();
+            /*
+             * Spill data if necessary
+             */
+            while (newSize > maxBucketSize && cursor > 0) {
+                String removed = bucket.remove(0);
+                cursor--;
+                newSize -= removed.length();
+            }
+
+            if (newSize > maxBucketSize) {
+                return false;
+            }
+
+            bucket.add(data);
+            size = newSize;
+            return true;
+        }
+    }
+
+    @Override
+    public String getNext() {
+        String data = null;
+        synchronized (bucket) {
+            if (cursor < bucket.size()) {
+                data = bucket.get(cursor);
+                cursor++;
+            }
+        }
+        return data;
+    }
+}

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

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

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;
@@ -44,33 +44,52 @@ public JavaAgent() {
 
     private void init() {
         exec = Executors.newSingleThreadScheduledExecutor();
-        exec.scheduleAtFixedRate(this::empty, collectionInterval, collectionInterval, TimeUnit.SECONDS);
+        exec.scheduleAtFixedRate(this::drain, collectionInterval, collectionInterval, TimeUnit.SECONDS);
     }
 
     public void pushData(String type, String data) {
         synchronized (buckets) {
             Bucket bucket = buckets.get(type);
             if (bucket == null) {
-                bucket = new StringDataBucket();
+                bucket = new StringDataBucket(MAX_BUCKET_SIZE);
                 buckets.put(type, bucket);
             }
-
-            if ((bucket.getSize() + data.length()) > MAX_BUCKET_SIZE) {
-                emit(type, bucket.empty());
+            if (!bucket.addData(data)) {
+                // System.err.println("Javametrics: data dropped. Bucket size="
+                // + bucket.getSize());
             }
-
-            bucket.pushData(data);
         }
     }
 
-    private void empty() {
+    private void drain() {
         synchronized (buckets) {
             buckets.forEach((name, bucket) -> {
-                emit(name, bucket.empty());
+                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,20 +20,38 @@ 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();
     }
 
     @Override
-    public String empty() {
-        String data = bucket.toString();
+    public void empty() {
         bucket = new StringBuffer(INITIAL_BUCKET_SIZE);
-        return data;
     }
 
     @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 = null;
+        if (bucket.length() > 0) {
+            data = bucket.toString();
+            bucket = new StringBuffer(INITIAL_BUCKET_SIZE);
+        }
+        return data;
     }
 }

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

@@ -51,7 +51,7 @@ public JavametricsImpl() {
 
     private void initializeProviders() {
         /*
-         * Start the mbean data providers
+         * Start the data providers
          */
         providerManager = new DataProviderManager(COLLECTION_INTERVAL);
     }

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;