Implement LimitedQueue

core/src/main/java/ru/trader/analysis/LimitedQueue.java

@@ -0,0 +1,222 @@
+package ru.trader.analysis;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.*;
+
+public class LimitedQueue<E> extends ArrayList<E> implements Queue<E> {
+    private final Comparator<? super E> comparator;
+    private final int limit;
+    private boolean sorted=false;
+
+    public LimitedQueue(int initialCapacity, int limit) {
+        this(initialCapacity, limit, null);
+    }
+
+    public LimitedQueue(int initialCapacity, int limit, Comparator<? super E> comparator) {
+        super(initialCapacity);
+        this.limit = limit;
+        this.comparator = comparator;
+    }
+
+    public LimitedQueue(int limit) {
+        this(limit, null);
+    }
+
+    public LimitedQueue(int limit, Comparator<? super E> comparator) {
+        this.limit = limit;
+        this.comparator = comparator;
+    }
+
+    public LimitedQueue(Collection<? extends E> c, int limit) {
+        this(c, limit, null);
+    }
+
+    public LimitedQueue(Collection<? extends E> c, int limit, Comparator<? super E> comparator) {
+        super(c);
+        this.limit = limit;
+        this.comparator = comparator;
+        trimToLimit();
+    }
+
+    @Override
+    public boolean offer(E e) {
+        return add(e);
+    }
+
+    @Override
+    public E remove() {
+        return remove(0);
+    }
+
+    @Override
+    public E poll() {
+        if (isEmpty()) return null;
+        return remove(0);
+    }
+
+    @Override
+    public E element() {
+        return get(0);
+    }
+
+    @Override
+    public E peek() {
+        if (isEmpty()) return null;
+        return get(0);
+    }
+
+    public E last() {
+        if (isEmpty()) return null;
+        return get(size()-1);
+    }
+
+    @Override
+    public E get(int index) {
+        sort();
+        return super.get(index);
+    }
+
+    @Override
+    public boolean add(E element) {
+        if (comparator != null){
+            return addToTop(element);
+        } else {
+            return size() < limit && super.add(element);
+        }
+    }
+
+    @Override
+    public void add(int index, E element) {
+        throw new UnsupportedOperationException();
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public boolean addAll(@NotNull Collection<? extends E> c) {
+        if (comparator != null){
+            return addAllToTop(c);
+        } else {
+            E[] a = (E[]) c.toArray();
+            int numNew = Math.min(a.length, limit - size());
+            return numNew == 0 || super.addAll(Arrays.asList(a).subList(0, numNew));
+        }
+    }
+
+    @Override
+    public boolean addAll(int index, Collection<? extends E> c) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public E set(int index, E element) {
+        throw new UnsupportedOperationException();
+    }
+
+    public void sort(){
+        if (sorted || comparator == null) return;
+        super.sort(comparator);
+        sorted = true;
+    }
+
+    @Override
+    public void sort(Comparator<? super E> c) {
+        throw new UnsupportedOperationException();
+    }
+
+
+    private boolean addToTop(E element){
+        int size = super.size();
+        if (sorted || size == limit) {
+            sort();
+            int index = indexedBinarySearch(element, comparator);
+            if (index < 0) index = -1 - index;
+            if (index == limit || element.equals(super.get(index))) return false;
+            super.add(index, element);
+            if (size == limit)
+                super.remove(limit);
+        } else {
+            super.add(element);
+        }
+        return true;
+
+    }
+
+    private boolean addAllToTop(Collection<? extends E> c){
+        if (c.isEmpty()) return true;
+        int size = super.size();
+        int len = c.size();
+        if (sorted || size + len >= limit) {
+            List<E> a = toList(c);
+            if (size == 0){
+                super.addAll(0, a.subList(0, Math.min(limit, len)));
+                sorted = true;
+            } else {
+                sort();
+                E first = super.get(0);
+                E last = super.get(size - 1);
+                E aFirst = a.get(0);
+                E aLast = a.get(len-1);
+                if (size == limit && comparator.compare(aFirst, last) >= 0) return true;
+                if (comparator.compare(first, aLast) >= 0){
+                    super.addAll(0, a.subList(0, Math.min(limit, len)));
+                    trimToLimit();
+                    return true;
+                }
+                for (int i = 0; i < len; i++) {
+                    E element = a.get(i);
+                    int index = indexedBinarySearch(element, comparator);
+                    if (index < 0) index = -1 - index;
+                    else {
+                        if (element.equals(super.get(index))) continue;
+                    }
+                    if (index >= limit) break;
+                    super.add(index, element);
+                }
+                if (super.size() > limit)
+                    super.remove(limit);
+            }
+        } else {
+            super.addAll(0, c);
+        }
+        return true;
+
+    }
+
+    private void trimToLimit(){
+        int size = super.size();
+        if (limit >= size) return;
+        super.removeRange(limit, size);
+    }
+
+    @SuppressWarnings("unchecked")
+    private List<E> toList(Collection<? extends E> c){
+        if (c instanceof LimitedQueue && comparator == ((LimitedQueue) c).comparator) {
+            ((LimitedQueue) c).sort();
+            return (List<E>) c;
+        } else {
+            E[] a = (E[]) c.toArray();
+            Arrays.sort(a, comparator);
+            return Arrays.asList(a);
+        }
+    }
+
+    private int indexedBinarySearch(E key, Comparator<? super E> c) {
+        int low = 0;
+        int high = size()-1;
+
+        while (low <= high) {
+            int mid = (low + high) >>> 1;
+            E midVal = super.get(mid);
+            int cmp = c.compare(midVal, key);
+            if (cmp < 0)
+                low = mid + 1;
+            else if (cmp > 0)
+                high = mid - 1;
+            else
+                return mid; // key found
+        }
+        return -(low + 1);  // key not found
+    }
+
+}

core/src/main/java/ru/trader/analysis/graph/Crawler.java

@@ -3,6 +3,7 @@ package ru.trader.analysis.graph;
 import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import ru.trader.analysis.LimitedQueue;
 
 import java.util.*;
 import java.util.concurrent.ForkJoinPool;
@@ -187,26 +188,34 @@ public class Crawler<T> {
         LOG.trace("UCS2 from {} to {}, deep {}, count {}", root.vertex, target, deep, count);
         int found = 0;
         double limit = Double.MAX_VALUE;
-        PriorityQueue<Double> limitQueue = new PriorityQueue<>();
+        LimitedQueue<CTEntrySupport> targetsQueue = new LimitedQueue<>(count, Comparator.<CTEntrySupport>naturalOrder());
-        PriorityQueue<CTEntrySupport> queue = new PriorityQueue<>();
+        LimitedQueue<CTEntrySupport> queue = new LimitedQueue<>(count, Comparator.<CTEntrySupport>naturalOrder());
         root.sort();
         queue.add(new CTEntrySupport(root));
-        while (!queue.isEmpty() && count > found){
+        while (!(queue.isEmpty() && targetsQueue.isEmpty()) && count > found){
-            CTEntrySupport curr = queue.poll();
+            int alreadyFound = targetsQueue.size();
+            CTEntrySupport curr = targetsQueue.peek();
+            boolean isTarget = curr != null && (queue.isEmpty() || alreadyFound + found >= count || Comparator.<CTEntrySupport>naturalOrder().compare(curr, queue.peek()) <= 0);
+            if (isTarget){
+                targetsQueue.poll();
+            } else {
+                curr = queue.poll();
+            }
             CostTraversalEntry entry = curr.entry;
             LOG.trace("Check path entry {}, weight {}", entry, entry.weight);
-            if (entry.isConnect(target)) {
+            if (isTarget) {
                 List<Edge<T>> res = entry.toEdges();
                 LOG.trace("Path found {}", res);
                 onFoundFunc.accept(res);
                 found++;
                 if (found >= count) break;
-                limit = limitQueue.poll();
+                CTEntrySupport next = targetsQueue.peek();
+                limit = next != null ? next.entry.getWeight() :  Double.MAX_VALUE;
             }
-            if (limitQueue.size() + found < count){
+            if (alreadyFound + found < count){
                 LOG.trace("Continue search, limit {}", limit);
             } else {
-                LOG.trace("Already {} found, extracting", limitQueue.size());
+                LOG.trace("Already {} found, extracting", alreadyFound);
                 continue;
             }
             if (deep >= entry.getTarget().getLevel() || entry.size() >= maxSize){
@@ -215,8 +224,8 @@ public class Crawler<T> {
             }
             DFS task = new DFS(curr, target, deep, count - found, limit);
             POOL.invoke(task);
-            limitQueue.addAll(task.getLimits());
+            targetsQueue.addAll(task.getTargets());
-            queue.addAll(task.getResult());
+            queue.addAll(task.getQueue());
         }
         return found;
     }
@@ -284,8 +293,8 @@ public class Crawler<T> {
         private final int count;
         private final int deep;
         private final T target;
-        private final Collection<CTEntrySupport> res;
+        private final Collection<CTEntrySupport> queue;
-        private final Collection<Double> limits;
+        private final Collection<CTEntrySupport> targets;
         private final ArrayList<DFS> subTasks;
         private final boolean isSubTask;
         private double limit;
@@ -300,8 +309,8 @@ public class Crawler<T> {
             this.deep = deep;
             this.count = count;
             this.limit = limit;
-            res = new ArrayList<>(count);
+            queue = new LimitedQueue<>(count, Comparator.<CTEntrySupport>naturalOrder());
-            limits = new ArrayList<>(count);
+            targets = new LimitedQueue<>(count, Comparator.<CTEntrySupport>naturalOrder());
             subTasks = new ArrayList<>(THRESHOLD);
             isSubTask = subtask;
         }
@@ -310,16 +319,16 @@ public class Crawler<T> {
             return entry.parent == null || isSubTask && entry == root;
         }
 
-        private Collection<Double> getLimits() {
+        private Collection<CTEntrySupport> getTargets() {
             if (!isDone())
                 throw new IllegalStateException();
-            return limits;
+            return targets;
         }
 
-        private Collection<CTEntrySupport> getResult() {
+        private Collection<CTEntrySupport> getQueue() {
             if (!isDone())
                 throw new IllegalStateException();
-            return res;
+            return queue;
         }
 
         private void search(){
@@ -337,15 +346,14 @@ public class Crawler<T> {
                     curr = new CTEntrySupport(curr, nextEntry);
                     if (isTarget){
                         LOG.trace("Found, add entry {} to queue", nextEntry);
-                        res.add(curr);
+                        targets.add(curr);
-                        limits.add(limit);
                         limit = nextEntry.getWeight();
                         curr = curr.parent;
                     } else {
-                        if (nextEntry.getWeight() >= limit && limits.size() > 0){
+                        if (nextEntry.getWeight() >= limit && targets.size() > 0){
-                            if (limits.size() < count){
+                            if (targets.size() < count){
                                 LOG.trace("Not found, limit {}, add entry {} to queue", limit, nextEntry);
-                                res.add(curr);
+                                queue.add(curr);
                             } else {
                                 LOG.trace("Not found, limit {}, don't add entry {} to queue", limit, nextEntry);
                             }
@@ -395,8 +403,8 @@ public class Crawler<T> {
 
         private void fill(DFS subTask){
             LOG.trace("Sub task is done");
-            limits.addAll(subTask.getLimits());
+            targets.addAll(subTask.getTargets());
-            res.addAll(subTask.getResult());
+            queue.addAll(subTask.getQueue());
             limit = Math.min(limit, subTask.limit);
         }
 

core/src/test/java/ru/trader/analysis/LimitedQueueTest.java

@@ -0,0 +1,198 @@
+package ru.trader.analysis;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Comparator;
+
+public class LimitedQueueTest  extends Assert {
+
+    @Test
+    public void testAdd() throws Exception {
+        LimitedQueue<Integer> queue = new LimitedQueue<>(5,3);
+        queue.add(3);
+        assertEquals(1, queue.size());
+        assertEquals(3, queue.get(0).intValue());
+        queue.add(2);
+        assertEquals(2, queue.size());
+        assertEquals(3, queue.get(0).intValue());
+        queue.add(1);
+        assertEquals(3, queue.size());
+        assertEquals(3, queue.get(0).intValue());
+        assertEquals(1, queue.get(2).intValue());
+        queue.add(4);
+        assertEquals(3, queue.size());
+        assertEquals(3, queue.get(0).intValue());
+        assertEquals(1, queue.get(2).intValue());
+
+        queue = new LimitedQueue<>(3, Comparator.<Integer>naturalOrder());
+        queue.add(4);
+        assertEquals(1, queue.size());
+        assertEquals(4, queue.get(0).intValue());
+        queue.add(2);
+        assertEquals(2, queue.size());
+        assertEquals(2, queue.get(0).intValue());
+        queue.add(1);
+        assertEquals(3, queue.size());
+        assertEquals(1, queue.get(0).intValue());
+        assertEquals(4, queue.get(2).intValue());
+        queue.add(3);
+        assertEquals(3, queue.size());
+        assertEquals(1, queue.get(0).intValue());
+        assertEquals(3, queue.get(2).intValue());
+
+        queue = new LimitedQueue<>(5, Comparator.<Integer>reverseOrder());
+        queue.add(2);
+        queue.add(4);
+        queue.add(1);
+        queue.add(3);
+        assertEquals(4, queue.get(0).intValue());
+        assertEquals(4, queue.size());
+        assertEquals(1, queue.get(3).intValue());
+    }
+
+    @Test
+    public void testPoolPeek() throws Exception {
+        LimitedQueue<Integer> queue = new LimitedQueue<>(3);
+        queue.add(3);
+        queue.add(2);
+        queue.add(1);
+        assertEquals(3, queue.size());
+        Integer a = queue.peek();
+        assertEquals(3, a.intValue());
+        assertEquals(3, queue.size());
+        a = queue.peek();
+        assertEquals(3, a.intValue());
+        assertEquals(3, queue.size());
+        a = queue.peek();
+        assertEquals(3, a.intValue());
+        assertEquals(3, queue.size());
+
+        a = queue.poll();
+        assertEquals(3, a.intValue());
+        assertEquals(2, queue.size());
+        a = queue.poll();
+        assertEquals(2, a.intValue());
+        assertEquals(1, queue.size());
+        a = queue.poll();
+        assertEquals(1, a.intValue());
+        assertEquals(0, queue.size());
+        a = queue.poll();
+        assertNull(a);
+
+        queue = new LimitedQueue<>(3, Comparator.<Integer>naturalOrder());
+        queue.add(4);
+        queue.add(2);
+        queue.add(1);
+        queue.add(3);
+
+        assertEquals(3, queue.size());
+        a = queue.peek();
+        assertEquals(1, a.intValue());
+        assertEquals(3, queue.size());
+        a = queue.peek();
+        assertEquals(1, a.intValue());
+        assertEquals(3, queue.size());
+        a = queue.peek();
+        assertEquals(1, a.intValue());
+        assertEquals(3, queue.size());
+
+        a = queue.poll();
+        assertEquals(1, a.intValue());
+        assertEquals(2, queue.size());
+        a = queue.poll();
+        assertEquals(2, a.intValue());
+        assertEquals(1, queue.size());
+        a = queue.poll();
+        assertEquals(3, a.intValue());
+        assertEquals(0, queue.size());
+        a = queue.poll();
+        assertNull(a);
+    }
+
+
+    @Test
+    public void testAddAll() throws Exception {
+        LimitedQueue<Integer> queue = new LimitedQueue<>(Arrays.asList(5,3,4), 5, Comparator.naturalOrder());
+        assertEquals(3, queue.size());
+        assertEquals(3, queue.peek().intValue());
+
+        queue.addAll(Arrays.asList(0,1,2));
+        assertEquals(5, queue.size());
+        assertEquals(0, queue.peek().intValue());
+        assertEquals(4, queue.last().intValue());
+
+        queue = new LimitedQueue<>(Arrays.asList(5,3,4), 6, Comparator.naturalOrder());
+        queue.addAll(Arrays.asList(1,2));
+        assertEquals(5, queue.size());
+        assertEquals(1, queue.peek().intValue());
+        assertEquals(5, queue.last().intValue());
+
+        queue = new LimitedQueue<>(Arrays.asList(5,3,4), 3, Comparator.naturalOrder());
+        queue.addAll(Arrays.asList(6,10,7));
+        assertEquals(3, queue.size());
+        assertEquals(3, queue.peek().intValue());
+        assertEquals(5, queue.last().intValue());
+
+        queue = new LimitedQueue<>(Arrays.asList(5,3,4,9), 3, Comparator.naturalOrder());
+        queue.addAll(Arrays.asList(6,1,10));
+        assertEquals(3, queue.size());
+        assertEquals(1, queue.peek().intValue());
+        assertEquals(4, queue.last().intValue());
+
+        queue = new LimitedQueue<>(4, Comparator.naturalOrder());
+        queue.addAll(Arrays.asList(6,1,10));
+        assertEquals(3, queue.size());
+        assertEquals(1, queue.peek().intValue());
+        assertEquals(10, queue.last().intValue());
+
+        queue = new LimitedQueue<>(Arrays.asList(2,1,4,9), 4, Comparator.naturalOrder());
+        queue.addAll(Arrays.asList(6,3,7));
+        assertEquals(4, queue.size());
+        assertEquals(1, queue.peek().intValue());
+        assertEquals(4, queue.last().intValue());
+
+        queue = new LimitedQueue<>(Arrays.asList(2,6,4,9,5), 10, Comparator.naturalOrder());
+        queue.addAll(Arrays.asList(1,6,3,7,11,5,10,15));
+        assertEquals(10, queue.size());
+        assertEquals(1, queue.peek().intValue());
+        assertEquals(11, queue.last().intValue());
+
+
+        queue = new LimitedQueue<>(Arrays.asList(2,6,4,9,5), 10, Comparator.naturalOrder());
+        LimitedQueue<Integer> queue2 = new LimitedQueue<>(Arrays.asList(1,6,3,7,11,5,10,15), 10, Comparator.naturalOrder());
+        queue.addAll(queue2);
+        assertEquals(10, queue.size());
+        assertEquals(1, queue.peek().intValue());
+        assertEquals(11, queue.last().intValue());
+
+        queue = new LimitedQueue<>(Arrays.asList(2,6,4,9,5), 10, Comparator.naturalOrder());
+        queue2 = new LimitedQueue<>(Arrays.asList(1,6,3,7,11,5,10,15), 10, Comparator.reverseOrder());
+        queue.addAll(queue2);
+        assertEquals(10, queue.size());
+        assertEquals(1, queue.peek().intValue());
+        assertEquals(11, queue.last().intValue());
+
+    }
+
+    @Test
+    public void testAddAll2() throws Exception {
+        LimitedQueue<Integer> queue = new LimitedQueue<>(Arrays.asList(5,3,4), 5);
+        assertEquals(3, queue.size());
+        assertEquals(5, queue.peek().intValue());
+
+        queue.addAll(Arrays.asList(0,1,2));
+        assertEquals(5, queue.size());
+        assertEquals(5, queue.peek().intValue());
+        assertEquals(1, queue.last().intValue());
+
+        queue = new LimitedQueue<>(Arrays.asList(5,3,4), 6);
+        queue.addAll(Arrays.asList(1,2));
+        assertEquals(5, queue.size());
+        assertEquals(5, queue.peek().intValue());
+        assertEquals(2, queue.last().intValue());
+    }
+
+
+}