package org.mitallast.queue.queue.service.translog.cache;

import org.mitallast.queue.queue.service.translog.MemoryMappedPage;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.locks.ReentrantLock;

public class MemoryMappedPageCacheSegment implements MemoryMappedPageCache {

    private final static Comparator<MemoryMappedPage> reserveComparator = new Comparator<MemoryMappedPage>() {
        @Override
        public int compare(MemoryMappedPage o1, MemoryMappedPage o2) {
            return (int) (o2.getTimestamp() - o1.getTimestamp());
        }
    };
    private final Loader loader;
    private final int maxPages;
    private final Map<Long, MemoryMappedPage> pageMap;
    private final ArrayList<MemoryMappedPage> garbage = new ArrayList<>();
    private final ReentrantLock writeLock = new ReentrantLock();

    public MemoryMappedPageCacheSegment(Loader loader, int maxPages) {
        this.loader = loader;
        this.maxPages = maxPages;
        pageMap = new HashMap<>(maxPages);
    }

    @Override
    public MemoryMappedPage acquire(long offset) throws IOException {
        MemoryMappedPage page = pageMap.get(offset);
        if (page != null) {
            page.acquire();
            page.setTimestamp(System.currentTimeMillis());
            return page;
        }
        writeLock.lock();
        try {
            page = pageMap.get(offset);
            if (page == null) {
                page = loader.load(offset);
                page.acquire();
                page.setTimestamp(System.currentTimeMillis());
                pageMap.put(offset, page);
                garbage.add(page);
            }
            page.acquire();
            page.setTimestamp(System.currentTimeMillis());
            return page;
        } finally {
            writeLock.unlock();
        }
    }

    @Override
    public void release(MemoryMappedPage page) throws IOException {
        writeLock.lock();
        try {
            internalRelease(page);
            garbageCollect();
        } finally {
            writeLock.unlock();
        }
    }

    private boolean internalRelease(MemoryMappedPage page) throws IOException {
        if (page.release() == 0) {
            pageMap.remove(page.getOffset());
            page.close();
            return true;
        }
        return false;
    }

    @Override
    public void flush() throws IOException {
        writeLock.lock();
        try {
            for (MemoryMappedPage page : pageMap.values()) {
                page.flush();
            }
        } finally {
            writeLock.unlock();
        }
    }

    @Override
    public void close() throws IOException {
        writeLock.lock();
        try {
            for (MemoryMappedPage page : pageMap.values()) {
                page.close();
            }
            pageMap.clear();
        } finally {
            writeLock.unlock();
        }
    }

    private void garbageCollect() throws IOException {
        if (pageMap.size() <= maxPages) {
            return;
        }
        if (pageMap.size() > maxPages) {
            garbage.addAll(pageMap.values());
            Collections.sort(garbage, reserveComparator);
            for (int i = garbage.size() - 1; i > maxPages; i--) {
                if (internalRelease(garbage.get(i))) {
                    garbage.remove(i);
                }
            }
        }
    }
}

@@ -170,9 +170,145 @@
 til.
-*
+ArrayList;%0Aimport java.util.Collections;%0Aimport java.util.Comparator;%0Aimport java.util.Map;%0Aimport java.util.concurrent.ConcurrentHashMap
 ;%0Aim
@@ -1134,16 +1134,26 @@
 p = new 
+Concurrent
 HashMap%3C
@@ -1162,16 +1162,22 @@
 maxPages
+, 0.5f
 );%0A    %7D
@@ -1724,32 +1724,39 @@
                 
+assert 
 pageMap.put(offs
@@ -1768,43 +1768,16 @@
 age)
-;%0A                garbage.add(page)
+ == null
 ;%0A  

fix memory mapped page