commit 0d0d253757c9726aa0d5051449cfc3025dc04379
Author: Andreas Kohlbecker <a.kohlbecker@bgbm.org>
Date:   Tue Jul 5 08:26:06 2016 +0200

    DynamicBatch: a JVM resources aware batch manager

diff --git a/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/DynamicBatch.java b/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/DynamicBatch.java
new file mode 100644
index 0000000000..21d9d81506
--- /dev/null
+++ b/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/DynamicBatch.java
@@ -0,0 +1,240 @@
+// $Id$
+/**
+* Copyright (C) 2016 EDIT
+* European Distributed Institute of Taxonomy
+* http://www.e-taxonomy.eu
+*
+* The contents of this file are subject to the Mozilla Public License Version 1.1
+* See LICENSE.TXT at the top of this package for the full license terms.
+*/
+package eu.etaxonomy.cdm.common;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.log4j.Logger;
+
+/**
+ * DynamicBatch: a JVM resources aware batch manager.
+ *
+ * @author a.kohlbecker
+ * @date Jul 4, 2016
+ *
+ */
+public class DynamicBatch {
+
+    public static final Logger logger = Logger.getLogger(DynamicBatch.class);
+
+    int batchSize;
+    int batchItemCount = -1;
+    Long batchMinFreeHeap = null;
+    List<Integer> items = null;
+
+    int gcTimeIncreaseCount = 0;
+
+    private int allowedGcIncreases = -1;
+
+    private int itemWhereLimitsTouched = 0;
+
+
+    List<Integer> unprocessedIds = new ArrayList<Integer>(batchSize);
+
+    private final JvmMonitor jvmMonitor = new JvmMonitor();
+
+    private final long intitialFreeHeap;
+
+    public DynamicBatch(int initialSize) {
+
+        this.batchSize = initialSize;
+        this.intitialFreeHeap = jvmMonitor.getFreeHeap(true);
+    }
+
+    public DynamicBatch(int initialSize, long minInitialHeap) throws JvmLimitsException {
+
+        this.batchSize = initialSize;
+        this.intitialFreeHeap = jvmMonitor.getFreeHeap(true);
+        if(this.intitialFreeHeap < minInitialHeap) {
+            throw new JvmLimitsException("At least " + minInitialHeap + " byte of free Heap space required but only " + intitialFreeHeap + " byte available.");
+        }
+    }
+
+    /**
+     * @param requiredFreeHeap
+     * @throws JvmLimitsException
+     */
+    public void setRequiredFreeHeap(double requiredFreeHeap) throws JvmLimitsException {
+
+        this.batchMinFreeHeap = (long) (intitialFreeHeap * requiredFreeHeap);
+        if(memoryLimitsExceeded()) {
+            throw new JvmLimitsException("Not enough free heap for batch");
+        }
+    }
+
+    /**
+     * @param requiredFreeHeap
+     * @throws JvmLimitsException
+     */
+    public void setRequiredFreeHeap(long requiredFreeHeap) throws JvmLimitsException {
+        this.batchMinFreeHeap = requiredFreeHeap;
+        if(memoryLimitsExceeded()) {
+            throw new JvmLimitsException("Not enough free heap for batch");
+        }
+    }
+
+    /**
+     *
+     * @param allowedGcIncreases the amount of continiously observed increases of the gc time
+     */
+    public void setMaxAllowedGcIncreases(int allowedGcIncreases) {
+        this.allowedGcIncreases = allowedGcIncreases;
+        // reset GCtime
+        jvmMonitor.getGCtimeSiceLastCheck();
+    }
+
+    public int size() {
+        return batchSize;
+    }
+
+    public  List<Integer> items(){
+        return items;
+    }
+
+    public boolean hasUnprocessedItems() {
+        return unprocessedIds.size() > 0;
+    }
+
+    /**
+     * 1. Fills all remaining items into the new batch and pads with next items from the iterator.
+     *
+     * 2. Resets the internal batchItemCount!!
+     *
+     * @param itemIterator
+     * @return
+     */
+    public List<Integer> nextItems(Iterator<Integer> itemIterator){
+
+        logger.debug("new batch of items with size of " + batchSize);
+        items = new ArrayList<Integer>(batchSize);
+        if(unprocessedIds.size() > 0) {
+            List<Integer> remainingUnprocessed = null;
+            Iterator<Integer> unprocessedIt = unprocessedIds.iterator();
+            int i = 0;
+            while(unprocessedIt.hasNext()) {
+                Integer nextUnprocessed = unprocessedIt.next();
+                if(i < batchSize) {
+                   items.add(nextUnprocessed);
+                } else {
+                    if(remainingUnprocessed == null) {
+                        remainingUnprocessed = new ArrayList<Integer>(unprocessedIds.size() - i + 1);
+                    }
+                    remainingUnprocessed.add(nextUnprocessed);
+                }
+                i++;
+            }
+            unprocessedIds.clear();
+            if(remainingUnprocessed != null) {
+                unprocessedIds = remainingUnprocessed;
+            }
+        }
+
+        while(itemIterator.hasNext() && items.size() < batchSize ) {
+            items.add(itemIterator.next());
+        }
+
+        itemWhereLimitsTouched =  0;
+        batchItemCount = 0;
+
+        return items;
+    }
+
+    public void incementCounter() {
+        batchItemCount++;
+    }
+
+    /**
+     *
+     */
+    private void reduceSize() {
+        manageUnprocessedItems();
+        batchSize = itemWhereLimitsTouched;
+        if(batchSize < 1) {
+            batchSize = 1;
+        }
+    }
+
+    public void reduceSize(double by) {
+        manageUnprocessedItems();
+        batchSize = (int) (batchSize * by);
+        if(batchSize < 1) {
+            batchSize = 1;
+        }
+    }
+
+    /**
+     *
+     */
+    protected void manageUnprocessedItems() {
+
+        if(itemWhereLimitsTouched > 0) {
+            int batchItemsUnprocessed = items.size() - itemWhereLimitsTouched;
+            logger.info("batchSize reduced to " + itemWhereLimitsTouched);
+            if(batchItemsUnprocessed > 0) {
+                unprocessedIds.addAll(items.subList(items.size() - batchItemsUnprocessed, items.size()));
+            }
+        }
+    }
+
+
+    public boolean isWithinJvmLimits() {
+        if(memoryLimitsExceeded()) {
+            logger.info("memoryLimitsExceeded ==> reducing batchSize");
+            reduceSize();
+            return false;
+        }
+        if(allowedGcIncreases > 0 && gcLimitsExceeded()) {
+            logger.info("gcIncreaseLimitExceeded ==> reducing batchSize");
+            reduceSize();
+            return false;
+        }
+
+        return true;
+    }
+
+    public boolean gcLimitsExceeded() {
+
+        long gctimeSiceLastTime = jvmMonitor.getGCtimeSiceLastCheck();
+        if(gctimeSiceLastTime > 0) {
+            if(gcTimeIncreaseCount == 0) {
+                itemWhereLimitsTouched  = batchItemCount;
+            }
+            gcTimeIncreaseCount++;
+            logger.debug("gctimeSiceLastTime: " + gctimeSiceLastTime + ", gcTimeIncreaseCount: " + gcTimeIncreaseCount);
+        } else {
+            gcTimeIncreaseCount = 0;
+            itemWhereLimitsTouched = 0;
+        }
+        return gcTimeIncreaseCount > allowedGcIncreases;
+
+    }
+
+
+    public boolean memoryLimitsExceeded() {
+
+        if(!jvmMonitor.hasFreeHeap(batchMinFreeHeap)) {
+            if(batchItemCount > -1) { // not in initial state, that it before first batch
+                itemWhereLimitsTouched  = batchItemCount;
+            }
+            logger.debug("min free heap limit (" + batchMinFreeHeap + ") exceeded ");
+            return true;
+        } else {
+            return false;
+        }
+
+    }
+
+    public JvmMonitor getJvmMonitor() {
+        return jvmMonitor;
+    }
+
+}
diff --git a/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/JvmLimitsException.java b/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/JvmLimitsException.java
new file mode 100644
index 0000000000..03da67768e
--- /dev/null
+++ b/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/JvmLimitsException.java
@@ -0,0 +1,26 @@
+// $Id$
+/**
+* Copyright (C) 2016 EDIT
+* European Distributed Institute of Taxonomy
+* http://www.e-taxonomy.eu
+*
+* The contents of this file are subject to the Mozilla Public License Version 1.1
+* See LICENSE.TXT at the top of this package for the full license terms.
+*/
+package eu.etaxonomy.cdm.common;
+
+/**
+ * @author a.kohlbecker
+ * @date Jul 4, 2016
+ *
+ */
+public class JvmLimitsException extends Exception {
+
+    /**
+     * @param string
+     */
+    public JvmLimitsException(String string) {
+        super(string);
+    }
+
+}
diff --git a/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/JvmMonitor.java b/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/JvmMonitor.java
new file mode 100644
index 0000000000..9b90ce9555
--- /dev/null
+++ b/cdmlib-commons/src/main/java/eu/etaxonomy/cdm/common/JvmMonitor.java
@@ -0,0 +1,139 @@
+// $Id$
+/**
+* Copyright (C) 2016 EDIT
+* European Distributed Institute of Taxonomy
+* http://www.e-taxonomy.eu
+*
+* The contents of this file are subject to the Mozilla Public License Version 1.1
+* See LICENSE.TXT at the top of this package for the full license terms.
+*/
+package eu.etaxonomy.cdm.common;
+
+import java.lang.management.GarbageCollectorMXBean;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.MemoryUsage;
+import java.util.List;
+
+import org.apache.log4j.Logger;
+
+/**
+ * @author a.kohlbecker
+ * @date Jul 1, 2016
+ *
+ */
+public class JvmMonitor {
+
+
+    public static final Logger logger = Logger.getLogger(JvmMonitor.class);
+
+    private long gcTimeLast = 0;
+
+    private long lastCheckTime = 0;
+
+    /**
+     * Returns the sum of approximate accumulated collection elapsed time in milliseconds
+     * as reported by all garbage collectors.
+     *
+     * This method returns -1 if the collection elapsed time is undefined.
+     *
+     * @return
+     */
+    public long gcTime() {
+        List<GarbageCollectorMXBean> gcMXBeans = ManagementFactory.getGarbageCollectorMXBeans();
+
+        //logger.setLevel(Level.DEBUG);
+
+        long gcTime = -1;
+        long collectorGcTime;
+        for(GarbageCollectorMXBean gcMXBean : gcMXBeans){
+                if(gcTime == -1) {
+                    gcTime = 0;
+                }
+                collectorGcTime = gcMXBean.getCollectionTime();
+                logger.debug("cgMxBean: " + gcMXBean.getName()
+                        + " gcTime = " + collectorGcTime
+                        + " gcCount = " + gcMXBean.getCollectionCount());
+                if(collectorGcTime != -1) {
+                    // only sum up if the time is defined
+                    gcTime = gcMXBean.getCollectionTime();
+                }
+        }
+        logger.debug("gcTimeSum = " + gcTime);
+        return gcTime;
+
+    }
+
+    public MemoryUsage getHeapMemoryUsage(){
+
+        MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean();
+        if(memoryMXBean != null){
+            logger.debug("HeapMemoryUsage: " + memoryMXBean.getHeapMemoryUsage());
+            return memoryMXBean.getHeapMemoryUsage();
+        }
+        return null;
+    }
+
+
+    public boolean hasFreeHeap(long freeHeapLimit) {
+
+        if(!_hasFreeHeap(freeHeapLimit)) {
+            Runtime.getRuntime().gc();
+            return _hasFreeHeap(freeHeapLimit);
+        }
+        return true;
+    }
+
+    /**
+     * @param maxUsedFraction
+     * @return
+     */
+    private boolean _hasFreeHeap(long freeHeapLimit) {
+        long freeHeap = getFreeHeap(false);
+        return freeHeap > freeHeapLimit;
+    }
+
+    /**
+     * @return
+     */
+    public long getFreeHeap(boolean gcBeforeMeasure) {
+        if(gcBeforeMeasure) {
+            Runtime.getRuntime().gc();
+        }
+        MemoryUsage heapUsage = getHeapMemoryUsage();
+        long freeHeap =  heapUsage.getMax() - heapUsage.getUsed();
+        return freeHeap;
+    }
+
+    /**
+     * Returns the gcTime in milliseconds as obtained through {@link #gctime()} of the
+     * time interval since this method has been called the last time and now.
+     *
+     * @return
+     */
+    public long getGCtimeSiceLastCheck() {
+        long gcTimeNow = gcTime();
+        long gcTimeSince = gcTimeNow - gcTimeLast;
+        gcTimeLast = gcTimeNow;
+        lastCheckTime  = System.currentTimeMillis();
+        return gcTimeSince;
+    }
+
+    /**
+     * Returns the time spend in gc as proportion (0.0 to 1.0) of the
+     * time interval since this method has been called the last time and now.
+     *
+     * @return
+     */
+    public double getGCRateSiceLastCheck() {
+
+        long gcTimeSince = getGCtimeSiceLastCheck();
+        long timeDiff = System.currentTimeMillis() - lastCheckTime;
+        double gcRate = gcTimeSince / (double) timeDiff;
+        return gcRate;
+    }
+
+
+
+
+}
diff --git a/cdmlib-commons/src/test/java/eu/etaxonomy/cdm/common/JvmMonitorTest.java b/cdmlib-commons/src/test/java/eu/etaxonomy/cdm/common/JvmMonitorTest.java
new file mode 100644
index 0000000000..983c9666c0
--- /dev/null
+++ b/cdmlib-commons/src/test/java/eu/etaxonomy/cdm/common/JvmMonitorTest.java
@@ -0,0 +1,70 @@
+// $Id$
+/**
+* Copyright (C) 2016 EDIT
+* European Distributed Institute of Taxonomy
+* http://www.e-taxonomy.eu
+*
+* The contents of this file are subject to the Mozilla Public License Version 1.1
+* See LICENSE.TXT at the top of this package for the full license terms.
+*/
+package eu.etaxonomy.cdm.common;
+
+import org.apache.log4j.Logger;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author a.kohlbecker
+ * @date Jul 1, 2016
+ *
+ */
+public class JvmMonitorTest extends Assert{
+
+
+    public static final Logger logger = Logger.getLogger(JvmMonitorTest.class);
+
+    @Test
+    public void testGcTime() {
+        JvmMonitor jvmMonitor = new JvmMonitor();
+        assertNotEquals(-1l, jvmMonitor.gcTime());
+
+        Runtime.getRuntime().gc();
+        Runtime.getRuntime().gc();
+        Runtime.getRuntime().gc();
+        long gcTimeLast_1 = jvmMonitor.getGCtimeSiceLastCheck();
+        assertTrue(gcTimeLast_1 > 0);
+        Runtime.getRuntime().gc();
+        Runtime.getRuntime().gc();
+        Runtime.getRuntime().gc();
+        long gcTimeLast_2 = jvmMonitor.getGCtimeSiceLastCheck();
+        assertTrue(gcTimeLast_1 > 0);
+        assertTrue(jvmMonitor.gcTime() > gcTimeLast_2);
+
+    }
+
+    @Test
+    public void testHeapUsage() {
+        int MB = 1024 * 1024;
+        int failWithMB = 300 * MB;
+        JvmMonitor jvmMonitor = new JvmMonitor();
+
+        long baseline = jvmMonitor.getHeapMemoryUsage().getUsed();
+        logger.debug("before: " + baseline);
+/*
+        assertTrue(jvmMonitor.hasFreeHeap(0.9));
+
+        logger.setLevel(Level.DEBUG);
+
+        Object[] measure = new Object[MB]; // 1MB
+        double bytePerObject = (jvmMonitor.getHeapMemoryUsage().getUsed() - baseline) / MB;
+        long maxHeap = jvmMonitor.getHeapMemoryUsage().getMax();
+        logger.debug("max: " + maxHeap);
+        Object[] heapEater = new Object[(int)Math.round((failWithMB / bytePerObject))];
+        logger.debug("after: " + jvmMonitor.getHeapMemoryUsage().getUsed());
+
+        assertFalse(jvmMonitor.hasFreeHeap((failWithMB * 2) / (double)maxHeap));
+*/
+
+    }
+
+}
