package com.jivesoftware.os.lab;

import com.jivesoftware.os.lab.api.LABIndexClosedException;
import com.jivesoftware.os.lab.api.LABIndexCorruptedException;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 *
 * @author jonathan.colt
 */
public class LabHeapPressure {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private final long maxHeapPressureInBytes;
    private final AtomicLong globalHeapCostInBytes;
    private final Map<LAB, Boolean> labs = new ConcurrentHashMap<>();

    public LabHeapPressure(long maxHeapPressureInBytes, AtomicLong globalHeapCostInBytes) {
        this.maxHeapPressureInBytes = maxHeapPressureInBytes;
        this.globalHeapCostInBytes = globalHeapCostInBytes;
    }

    public AtomicLong globalHeapCostInBytes() {
        return globalHeapCostInBytes;
    }

    void commitIfNecessary(LAB lab, long labMaxHeapPressureInBytes, boolean fsyncOnFlush) throws Exception {
        if (lab.approximateHeapPressureInBytes() > labMaxHeapPressureInBytes) {
            labs.remove(lab);
            lab.commit(fsyncOnFlush);
        } else {
            labs.compute(lab, (LAB t, Boolean u) -> {
                return u == null ? fsyncOnFlush : (boolean) u || fsyncOnFlush;
            });
        }

        if (globalHeapCostInBytes.get() > maxHeapPressureInBytes) {
            LAB[] keys = labs.keySet().toArray(new LAB[0]);
            Arrays.sort(keys, (LAB o1, LAB o2) -> {
                return -Long.compare(o1.approximateHeapPressureInBytes(), o2.approximateHeapPressureInBytes());
            });
            int i = 0;
            while (i < keys.length && globalHeapCostInBytes.get() > maxHeapPressureInBytes) {
                Boolean efsyncOnFlush = labs.remove(keys[i]);
                if (efsyncOnFlush != null) {
                    try {
                        LOG.info("Forcing flush due to heap pressure. lab:{}", lab);
                        keys[i].commit(efsyncOnFlush);
                    } catch (LABIndexCorruptedException | LABIndexClosedException x) {
                    } catch (Exception x) {
                        labs.compute(keys[i], (LAB t, Boolean u) -> {
                            return u == null ? efsyncOnFlush : (boolean) u || efsyncOnFlush;
                        });
                        throw x;
                    }
                }
                i++;
            }
        }
    }

    void close(LAB lab) {
        labs.remove(lab);
    }

}

@@ -265,16 +265,67 @@
 actory;%0A
+import com.jivesoftware.os.mlogger.core.ValueType;%0A
 import j
@@ -341,16 +341,16 @@
 Arrays;%0A
-
 import j
@@ -1269,32 +1269,92 @@
 ssureInBytes) %7B%0A
+            LOG.inc(%22lab%3Ecommit%3Einstance%3E%22 + fsyncOnFlush);%0A
             labs
@@ -1581,29 +1581,42 @@
       %7D%0A
-%0A
+ 
        
- if (
+long globalHeap = 
 globalHe
@@ -1626,32 +1626,188 @@
 ostInBytes.get()
+;%0A        LOG.set(ValueType.VALUE, %22lab%3Eheap%3Epressure%22, globalHeap);%0A        LOG.set(ValueType.VALUE, %22lab%3Ecommitable%22, labs.size());%0A        if (globalHeap
  %3E maxHeapPressu
@@ -2279,24 +2279,24 @@
  != null) %7B%0A
-
             
@@ -2309,16 +2309,168 @@
   try %7B%0A
+                        LOG.inc(%22lab%3Ecommit%3Eglobal%3E%22 + efsyncOnFlush);%0A                        LOG.set(ValueType.VALUE, %22lab%3Ecommitable%22, labs.size());%0A
         

add metrics around lah heap pressure