Level1Merger extends Thread { public static void mergeSort(LevelQueue lq1, LevelQueue lq2, int ways, String dir, short shard) throws IOException, ClassNotFoundException { boolean hasLevel2MapTable = lq2.size() > 0; List<AbstractMapTable> tables = new ArrayList<AbstractMapTable>(ways); lq1.getReadLock().lock(); try { Iterator<AbstractMapTable> iter = lq1.descendingIterator(); for(int i = 0; i < ways - 1; i++) { tables.add(iter.next()); } if (hasLevel2MapTable) { tables.add(lq2.get(0)); } else { tables.add(iter.next()); } } finally { lq1.getReadLock().unlock(); } long expectedInsertions = 0; for(AbstractMapTable table : tables) { expectedInsertions += table.getAppendedSize(); } if (expectedInsertions > Integer.MAX_VALUE) expectedInsertions = Integer.MAX_VALUE; AbstractSortedMapTable sortedMapTable = new FCMapTable(dir, shard, SDB.LEVEL2, System.nanoTime(), (int)expectedInsertions); PriorityQueue<QueueElement> pq = new PriorityQueue<QueueElement>(); for(AbstractMapTable table : tables) { QueueElement qe = new QueueElement(); qe.sortedMapTable = table; qe.size = qe.sortedMapTable.getAppendedSize(); qe.index = 0; qe.queue = new LinkedList<IMapEntry>(); IMapEntry me = qe.getNextMapEntry(); if (me != null) { qe.key = me.getKey(); qe.mapEntry = me; qe.keyHash = me.getKeyHash(); pq.add(qe); } } LinkedList<IMapEntry> targetCacheQueue = new LinkedList<IMapEntry>(); while(pq.size() > 0) { QueueElement qe1 = pq.poll(); while(pq.peek() != null && qe1.keyHash == pq.peek().keyHash && BytesUtil.compare(qe1.key, pq.peek().key) == 0) { QueueElement qe2 = pq.poll(); IMapEntry me = qe2.getNextMapEntry(); if (me != null) { qe2.key = me.getKey(); qe2.mapEntry = me; qe2.keyHash = me.getKeyHash(); pq.add(qe2); } } if (!qe1.mapEntry.isDeleted() && !qe1.mapEntry.isExpired()) { targetCacheQueue.add(qe1.mapEntry); } if (targetCacheQueue.size() >= CACHED_MAP_ENTRIES * DEFAULT_MERGE_WAYS) { while(targetCacheQueue.size() > 0) { IMapEntry mapEntry = targetCacheQueue.poll(); byte[] value = mapEntry.getValue(); if (mapEntry.isExpired()) { continue; } sortedMapTable.appendNew(mapEntry.getKey(), mapEntry.getKeyHash(), value, mapEntry.getTimeToLive(), mapEntry.getCreatedTime(), mapEntry.isDeleted(), mapEntry.isCompressed()); } } IMapEntry me = qe1.getNextMapEntry(); if (me != null) { qe1.key = me.getKey(); qe1.mapEntry = me; qe1.keyHash = me.getKeyHash(); pq.add(qe1); } } while(targetCacheQueue.size() > 0) { IMapEntry mapEntry = targetCacheQueue.poll(); byte[] value = mapEntry.getValue(); if (mapEntry.isExpired()) { continue; } sortedMapTable.appendNew(mapEntry.getKey(), mapEntry.getKeyHash(), value, mapEntry.getTimeToLive(), mapEntry.getCreatedTime(), mapEntry.isDeleted(), mapEntry.isCompressed()); } sortedMapTable.reMap(); sortedMapTable.saveMetadata(); lq1.getWriteLock().lock(); lq2.getWriteLock().lock(); try { for(int i = 0; i < ways - 1; i++) { lq1.removeLast(); } if (hasLevel2MapTable) { lq2.removeLast(); } else { lq1.removeLast(); } for(AbstractMapTable table : tables) { table.markUsable(false); } sortedMapTable.markUsable(true); lq2.addFirst(sortedMapTable); } finally { lq2.getWriteLock().unlock(); lq1.getWriteLock().unlock(); } for(AbstractMapTable table : tables) { table.close(); table.delete(); } } Level1Merger(SDB sdb, List<LevelQueue> levelQueueList, CountDownLatch countDownLatch, short shard,
                        SDBStats stats); @Override void run(); static void mergeSort(LevelQueue lq1, LevelQueue lq2, int ways, String dir, short shard); void setStop();  }
@Test public void testCase02() throws IOException, ClassNotFoundException { int maxSize = AbstractMapTable.INIT_INDEX_ITEMS_PER_TABLE * 4 * 4; MMFMapTable[] sourceTables = new MMFMapTable[3]; LevelQueue lq1 = new LevelQueue(); for(int i = 0; i < 3; i++) { sourceTables[i] = new MMFMapTable(testDir, SDB.LEVEL1, System.nanoTime() + i, maxSize / 4, 4); lq1.addFirst(sourceTables[i]); } MMFMapTable table3 = sourceTables[2]; int start = 0; List<String> keyList = new ArrayList<String>(); while(start < maxSize) { keyList.add(String.valueOf(start)); start = start + 4; } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(String key : keyList) { table3.appendNew(key.getBytes(), Arrays.hashCode(key.getBytes()), key.getBytes(), AbstractMapTable.NO_TIMEOUT, System.currentTimeMillis(), true, false); } MMFMapTable table2 = sourceTables[1]; start = 1; keyList = new ArrayList<String>(); while(start < maxSize) { keyList.add(String.valueOf(start)); start = start + 4; } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(String key : keyList) { table2.appendNew(key.getBytes(), Arrays.hashCode(key.getBytes()), key.getBytes(), 200, System.currentTimeMillis(), false, false); } try { Thread.sleep(2000); } catch (InterruptedException e) { } MMFMapTable table1 = sourceTables[0]; start = 2; keyList = new ArrayList<String>(); while(start < maxSize) { keyList.add(String.valueOf(start)); start = start + 4; } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(String key : keyList) { table1.appendNew(key.getBytes(), Arrays.hashCode(key.getBytes()), key.getBytes(), 600 * 1000, System.currentTimeMillis(), false, false); } FCMapTable table4 = new FCMapTable(testDir, SDB.LEVEL2, System.nanoTime() + 3, maxSize); start = 0; keyList = new ArrayList<String>(); while(start < maxSize) { keyList.add(String.valueOf(start)); start = start + 1; } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(String key : keyList) { table4.appendNew(key.getBytes(), Arrays.hashCode(key.getBytes()), key.getBytes(), 1200 * 1000, System.currentTimeMillis(), false, false); } LevelQueue lq2 = new LevelQueue(); lq2.add(table4); Level1Merger.mergeSort(lq1, lq2, 4, testDir, (short)2); assertTrue(lq1.size() == 0); assertTrue(lq2.size() == 1); FCMapTable targetTable = (FCMapTable) lq2.poll(); System.out.println(targetTable.getAppendedSize() + "==" + maxSize / 2); assertTrue(targetTable.getAppendedSize() == maxSize / 2); keyList = new ArrayList<String>(); for(long i = 0; i < maxSize; i++) { if (i % 4 == 0 || i % 4 == 1) continue; keyList.add(String.valueOf(i)); } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); int index = 0; for(int i = 0; i < maxSize; i++) { if (i % 4 == 0 || i % 4 == 1) continue; IMapEntry mapEntry = targetTable.getMapEntry(index); assertTrue(mapEntry.getIndex() == index); assertTrue(new String(mapEntry.getKey()).equals(keyList.get(index))); assertTrue(new String(mapEntry.getValue()).equals(keyList.get(index))); index++; } Random random = new Random(); for(int i = 0; i < 1024; i++) { int key = random.nextInt(maxSize); if (key % 4 == 0 || key % 4 == 1) continue; GetResult result = targetTable.get(String.valueOf(key).getBytes()); assertTrue(result.isFound()); } targetTable.close(); targetTable.delete(); }
@Test public void testCase01() throws IOException, ClassNotFoundException { int maxSize = AbstractMapTable.INIT_INDEX_ITEMS_PER_TABLE; String value = TestUtil.randomString(1024 * 3); MMFMapTable[] sourceTables = new MMFMapTable[4]; LevelQueue lq1 = new LevelQueue(); for(int i = 0; i < 4; i++) { sourceTables[i] = new MMFMapTable(testDir, SDB.LEVEL1, System.nanoTime() + i, maxSize, 4); lq1.addFirst(sourceTables[i]); } int totalCount = 0; for(int i = 0; i < 4; i++) { int start = i; MMFMapTable table = sourceTables[i]; List<String> keyList = new ArrayList<String>(); while(keyList.size() < maxSize) { keyList.add(String.valueOf(start)); totalCount++; start = start + 4; } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(String key : keyList) { table.appendNew(key.getBytes(), value.getBytes(), AbstractMapTable.NO_TIMEOUT); } } LevelQueue lq2 = new LevelQueue(); long start = System.currentTimeMillis(); Level1Merger.mergeSort(lq1, lq2, 4, testDir, (short)3); long end = System.currentTimeMillis(); System.out.println("Time spent to merge " + totalCount + " items in 4 ways is " + (end - start) / 1000 + "s"); assertTrue(lq1.size() == 0); assertTrue(lq2.size() == 1); FCMapTable targetTable = (FCMapTable) lq2.poll(); assertTrue(targetTable.getLevel() == SDB.LEVEL2); assertTrue(targetTable.getAppendedSize() == totalCount); assertTrue(totalCount == maxSize * 4); List<String> keyList = new ArrayList<String>(); for(long i = 0; i < totalCount; i++) { keyList.add(String.valueOf(i)); } Collections.sort(keyList, new Comparator<String>() { @Override public int compare(String arg0, String arg1) { int hash0 = Arrays.hashCode(arg0.getBytes()); int hash1 = Arrays.hashCode(arg1.getBytes()); if (hash0 < hash1) return -1; else if (hash0 > hash1) return 1; else return 0; } }); for(int i = 0; i < totalCount; i++) { IMapEntry mapEntry = targetTable.getMapEntry(i); assertTrue(mapEntry.getIndex() == i); assertTrue(new String(mapEntry.getKey()).equals(keyList.get(i))); assertTrue(new String(mapEntry.getValue()).equals(value)); } start = System.currentTimeMillis(); Random random = new Random(); for(int i = 0; i < 1024; i++) { long key = random.nextInt(totalCount); GetResult result = targetTable.get(String.valueOf(key).getBytes()); assertTrue(result.isFound()); } end = System.currentTimeMillis(); System.out.println("Time to lookup 1024 random key in the target table is " + (end - start) + "ms"); targetTable.close(); targetTable.delete(); }