package com.xqbase.manager;

import com.xqbase.block.DefaultStorageBlock;
import com.xqbase.block.IBlock;

import java.io.IOException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.xqbase.stroage.StorageConfig.StorageMode;

public class StorageManager {

    private static final long DEFAULT_BLOCK_CAPACITY = 128 * 1024 * 1024L;

    private static final int DEFAULT_INITIAL_NUMBER_OF_BLOCKS = 8;

    private final AtomicInteger blockCount = new AtomicInteger(0);

    private final Lock activeBlockChangeLock = new ReentrantLock();

    private volatile IBlock activeBlock;

    private final Queue<IBlock> usedBlocks = new ConcurrentLinkedDeque<IBlock>();

    private final Queue<IBlock> freeBlocks = new PriorityBlockingQueue<IBlock>();

    private final StorageMode storageMode;

    private final String dir;

    private final int capacityPerBlock;

    private final double dirtyRatioThreshold;

    private int allowedMaxOffHeapBlockCount;

    public StorageManager(String dir, int capacityPerBlock, int initNumberOfBlocks, StorageMode storageMode, double dirtyRatioThreshold,
                long maxOffHeapMemorySize) throws IOException {
        this.dir = dir;
        this.capacityPerBlock = capacityPerBlock;
        this.dirtyRatioThreshold = dirtyRatioThreshold;
        this.storageMode = storageMode;
        if (this.storageMode != StorageMode.File) {
            this.allowedMaxOffHeapBlockCount = (int) (maxOffHeapMemorySize / capacityPerBlock);
        } else {
            this.allowedMaxOffHeapBlockCount = 0;
        }

        initBlocks(dir, initNumberOfBlocks);
    }

    private void initBlocks(String dir, int initNumberOfBlocks) throws IOException {
        for (int i = 0; i < initNumberOfBlocks; i++) {
            freeBlocks.add(createBlock(i));
            blockCount.incrementAndGet();
        }

        activeBlock = freeBlocks.poll();
        if (activeBlock == null) {
            activeBlock = new DefaultStorageBlock(dir, blockCount.incrementAndGet(), capacityPerBlock, storageMode);
        }
    }

    private IBlock createBlock(int index) throws IOException {
        if (storageMode != StorageMode.File) {
            if (allowedMaxOffHeapBlockCount < 0) {
                return new DefaultStorageBlock(dir, index, capacityPerBlock, StorageMode.File);
            }

            allowedMaxOffHeapBlockCount --;
        }

        return new DefaultStorageBlock(dir, index, capacityPerBlock, storageMode);
    }
}

@@ -486,16 +486,50 @@
 ager %7B%0A%0A
+    // Default per Block capacity%0A
     priv
@@ -596,16 +596,56 @@
 1024L;%0A%0A
+    // Default initial number of blocks%0A
     priv
@@ -704,16 +704,38 @@
 S = 8;%0A%0A
+    // Block counter.%0A
     priv
@@ -790,24 +790,60 @@
 nteger(0);%0A%0A
+    // Lock for active block change%0A
     private 
@@ -895,24 +895,69 @@
 antLock();%0A%0A
+    // Reference to the current active block%0A
     private 
@@ -978,32 +978,57 @@
 k activeBlock;%0A%0A
+    // Used block queue.%0A
     private fina
@@ -1086,32 +1086,56 @@
 que%3CIBlock%3E();%0A%0A
+    // Free block queue%0A
     private fina
@@ -1197,24 +1197,48 @@
 IBlock%3E();%0A%0A
+    // The storage mode%0A
     private 
@@ -1269,16 +1269,45 @@
 eMode;%0A%0A
+    // The storage directory%0A
     priv
@@ -1329,16 +1329,47 @@
 g dir;%0A%0A
+    // Block capacity variable%0A
     priv
@@ -1397,24 +1397,74 @@
 yPerBlock;%0A%0A
+    // Ration threshold control the block recycle%0A
     private 
@@ -1498,16 +1498,47 @@
 shold;%0A%0A
+    // Max OffHeap block count%0A
     priv
@@ -2391,11 +2391,13 @@
 cks.
-add
+offer
 (cre

update StorageManager...