@Test public void testTopOfManyIntoQueueMultithreaded() { BlockingQueue<NumericIDValue> top3 = Queues.newLinkedBlockingQueue(); float[] queueLeastValue = { Float.NEGATIVE_INFINITY }; List<NumericIDValue> candidates = makeNCandidates(20); TopN.selectTopNIntoQueueMultithreaded(top3, queueLeastValue, candidates.iterator(), 3); assertNotNull(top3); assertEquals(3, top3.size()); assertEquals(18L, top3.peek().getID()); assertEquals(18.0f, top3.peek().getValue()); }
public static void selectTopNIntoQueueMultithreaded(Queue<NumericIDValue> topN, float[] queueLeastValue, Iterator<NumericIDValue> values, int n) { float localQueueLeastValue = queueLeastValue[0]; while (values.hasNext()) { NumericIDValue value = values.next(); if (value != null) { long itemID = value.getID(); float valueScore = value.getValue(); if (valueScore >= localQueueLeastValue) { synchronized (topN) { if (topN.size() >= n) { float currentQueueLeastValue = topN.peek().getValue(); localQueueLeastValue = currentQueueLeastValue; if (valueScore > currentQueueLeastValue) { NumericIDValue recycled = topN.poll(); recycled.set(itemID, valueScore); topN.add(recycled); } } else { topN.add(new NumericIDValue(itemID, valueScore)); } } } } } queueLeastValue[0] = localQueueLeastValue; }
TopN { public static void selectTopNIntoQueueMultithreaded(Queue<NumericIDValue> topN, float[] queueLeastValue, Iterator<NumericIDValue> values, int n) { float localQueueLeastValue = queueLeastValue[0]; while (values.hasNext()) { NumericIDValue value = values.next(); if (value != null) { long itemID = value.getID(); float valueScore = value.getValue(); if (valueScore >= localQueueLeastValue) { synchronized (topN) { if (topN.size() >= n) { float currentQueueLeastValue = topN.peek().getValue(); localQueueLeastValue = currentQueueLeastValue; if (valueScore > currentQueueLeastValue) { NumericIDValue recycled = topN.poll(); recycled.set(itemID, valueScore); topN.add(recycled); } } else { topN.add(new NumericIDValue(itemID, valueScore)); } } } } } queueLeastValue[0] = localQueueLeastValue; } }
TopN { public static void selectTopNIntoQueueMultithreaded(Queue<NumericIDValue> topN, float[] queueLeastValue, Iterator<NumericIDValue> values, int n) { float localQueueLeastValue = queueLeastValue[0]; while (values.hasNext()) { NumericIDValue value = values.next(); if (value != null) { long itemID = value.getID(); float valueScore = value.getValue(); if (valueScore >= localQueueLeastValue) { synchronized (topN) { if (topN.size() >= n) { float currentQueueLeastValue = topN.peek().getValue(); localQueueLeastValue = currentQueueLeastValue; if (valueScore > currentQueueLeastValue) { NumericIDValue recycled = topN.poll(); recycled.set(itemID, valueScore); topN.add(recycled); } } else { topN.add(new NumericIDValue(itemID, valueScore)); } } } } } queueLeastValue[0] = localQueueLeastValue; } private  TopN(); }
TopN { public static void selectTopNIntoQueueMultithreaded(Queue<NumericIDValue> topN, float[] queueLeastValue, Iterator<NumericIDValue> values, int n) { float localQueueLeastValue = queueLeastValue[0]; while (values.hasNext()) { NumericIDValue value = values.next(); if (value != null) { long itemID = value.getID(); float valueScore = value.getValue(); if (valueScore >= localQueueLeastValue) { synchronized (topN) { if (topN.size() >= n) { float currentQueueLeastValue = topN.peek().getValue(); localQueueLeastValue = currentQueueLeastValue; if (valueScore > currentQueueLeastValue) { NumericIDValue recycled = topN.poll(); recycled.set(itemID, valueScore); topN.add(recycled); } } else { topN.add(new NumericIDValue(itemID, valueScore)); } } } } } queueLeastValue[0] = localQueueLeastValue; } private  TopN(); static Queue<NumericIDValue> initialQueue(int n); static void selectTopNIntoQueue(Queue<NumericIDValue> topN,
                                         Iterator<NumericIDValue> values,
                                         int n); static void selectTopNIntoQueueMultithreaded(Queue<NumericIDValue> topN,
                                                      float[] queueLeastValue,
                                                      Iterator<NumericIDValue> values,
                                                      int n); static List<NumericIDValue> selectTopNFromQueue(Queue<NumericIDValue> topN, int n); static List<NumericIDValue> selectTopN(Iterator<NumericIDValue> values, int n); }
TopN { public static void selectTopNIntoQueueMultithreaded(Queue<NumericIDValue> topN, float[] queueLeastValue, Iterator<NumericIDValue> values, int n) { float localQueueLeastValue = queueLeastValue[0]; while (values.hasNext()) { NumericIDValue value = values.next(); if (value != null) { long itemID = value.getID(); float valueScore = value.getValue(); if (valueScore >= localQueueLeastValue) { synchronized (topN) { if (topN.size() >= n) { float currentQueueLeastValue = topN.peek().getValue(); localQueueLeastValue = currentQueueLeastValue; if (valueScore > currentQueueLeastValue) { NumericIDValue recycled = topN.poll(); recycled.set(itemID, valueScore); topN.add(recycled); } } else { topN.add(new NumericIDValue(itemID, valueScore)); } } } } } queueLeastValue[0] = localQueueLeastValue; } private  TopN(); static Queue<NumericIDValue> initialQueue(int n); static void selectTopNIntoQueue(Queue<NumericIDValue> topN,
                                         Iterator<NumericIDValue> values,
                                         int n); static void selectTopNIntoQueueMultithreaded(Queue<NumericIDValue> topN,
                                                      float[] queueLeastValue,
                                                      Iterator<NumericIDValue> values,
                                                      int n); static List<NumericIDValue> selectTopNFromQueue(Queue<NumericIDValue> topN, int n); static List<NumericIDValue> selectTopN(Iterator<NumericIDValue> values, int n);  }