diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java index 54018f82..269a164d 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelHeapMap.java @@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap; import com.zfoo.protocol.IPacket; +import java.io.Closeable; +import java.io.IOException; import java.util.concurrent.locks.ReentrantLock; import java.util.function.BiConsumer; @@ -21,7 +23,7 @@ import java.util.function.BiConsumer; * @author jaysunxiao * @version 3.0 */ -public class ConcurrentFileChannelHeapMap implements LpMap { +public class ConcurrentFileChannelHeapMap implements LpMap, Closeable { private ReentrantLock fileChannelLock = new ReentrantLock(); @@ -33,7 +35,7 @@ public class ConcurrentFileChannelHeapMap implements LpMap fileChannelMap = new FileChannelMap<>(dbPath, clazz); concurrentHeapMap = new ConcurrentHeapMap<>(); - load(); + fileChannelMap.forEach((key, v) -> concurrentHeapMap.put(key, v)); } @Override @@ -93,18 +95,14 @@ public class ConcurrentFileChannelHeapMap implements LpMap } } - private void load() { - var maxIndex = fileChannelMap.getMaxIndex(); - if (maxIndex <= 0) { - return; - } - - for (var key = 1; key <= maxIndex; key++) { - var value = fileChannelMap.get(key); - if (value == null) { - continue; - } - concurrentHeapMap.put(key, value); + @Override + public void close() throws IOException { + fileChannelLock.lock(); + try { + fileChannelMap.close(); + concurrentHeapMap.clear(); + } finally { + fileChannelLock.unlock(); } } @@ -112,4 +110,5 @@ public class ConcurrentFileChannelHeapMap implements LpMap public void forEach(BiConsumer biConsumer) { concurrentHeapMap.forEach(biConsumer); } + } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMap.java index 8d0fa36e..4792cba4 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMap.java @@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap; import com.zfoo.protocol.IPacket; +import java.io.Closeable; +import java.io.IOException; import java.util.List; import java.util.function.BiConsumer; @@ -21,7 +23,7 @@ import java.util.function.BiConsumer; * @author jaysunxiao * @version 3.0 */ -public class ConcurrentFileChannelMap implements LpMap { +public class ConcurrentFileChannelMap implements LpMap, Closeable { private FileChannelMap fileChannelMap; @@ -69,8 +71,14 @@ public class ConcurrentFileChannelMap implements LpMap { fileChannelMap.clear(); } + @Override + public synchronized void close() throws IOException { + fileChannelMap.close(); + } + @Override public synchronized void forEach(BiConsumer biConsumer) { fileChannelMap.forEach(biConsumer); } + } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java index ca276e95..ec8ebe1c 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelHeapMap.java @@ -14,13 +14,15 @@ package com.zfoo.orm.lpmap; import com.zfoo.protocol.IPacket; +import java.io.Closeable; +import java.io.IOException; import java.util.function.BiConsumer; /** * @author jaysunxiao * @version 3.0 */ -public class FileChannelHeapMap implements LpMap { +public class FileChannelHeapMap implements LpMap, Closeable { private FileChannelMap fileChannelMap; @@ -30,7 +32,7 @@ public class FileChannelHeapMap implements LpMap { fileChannelMap = new FileChannelMap<>(dbPath, clazz); heapMap = new HeapMap<>(initialCapacity); - load(); + fileChannelMap.forEach((key, v) -> heapMap.put(key, v)); } @Override @@ -66,23 +68,15 @@ public class FileChannelHeapMap implements LpMap { heapMap.clear(); } - private void load() { - var maxIndex = fileChannelMap.getMaxIndex(); - if (maxIndex <= 0) { - return; - } - - for (var key = 1; key <= maxIndex; key++) { - var value = fileChannelMap.get(key); - if (value == null) { - continue; - } - heapMap.put(key, value); - } + @Override + public void close() throws IOException { + fileChannelMap.close(); + heapMap.clear(); } @Override public void forEach(BiConsumer biConsumer) { heapMap.forEach(biConsumer); } + } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java index ddaed6da..5821f9b2 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileChannelMap.java @@ -193,6 +193,12 @@ public class FileChannelMap implements LpMap, Closeable { } + @Override + public void close() throws IOException { + IOUtils.closeIO(indexFileRandomAccess, indexFileChannel, dbFileRandomAccess, dbFileChannel); + ReferenceCountUtil.release(indexBuffer); + ReferenceCountUtil.release(dbBuffer); + } protected void setKeyValue(long key, V value) { try { @@ -236,11 +242,4 @@ public class FileChannelMap implements LpMap, Closeable { dbBuffer.clear(); } - @Override - public void close() throws IOException { - IOUtils.closeIO(indexFileRandomAccess, indexFileChannel, dbFileRandomAccess, dbFileChannel); - ReferenceCountUtil.release(indexBuffer); - ReferenceCountUtil.release(dbBuffer); - } - } diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java index 135a9b9e..3de64521 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/FileHeapMap.java @@ -43,13 +43,13 @@ public class FileHeapMap implements LpMap { private HeapMap heapMap; - public FileHeapMap(String dbPath, int initialCapacity, Class clazz) { + public FileHeapMap(String dbPath, Class clazz) { try { this.dbFile = FileUtils.getOrCreateFile(dbPath, StringUtils.format("{}.db", clazz.getSimpleName())); var protocolId = ProtocolAnalysis.getProtocolIdByClass(clazz); protocolRegistration = ProtocolManager.getProtocol(protocolId); - heapMap = new HeapMap<>(initialCapacity); + heapMap = new HeapMap<>(); load(); } catch (IOException e) { diff --git a/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java b/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java index 9c0db287..7e3f4da8 100644 --- a/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java +++ b/orm/src/main/java/com/zfoo/orm/lpmap/HeapMap.java @@ -27,6 +27,10 @@ public class HeapMap implements LpMap { protected long maxIndex = 0; + public HeapMap() { + this(128); + } + public HeapMap(int initialCapacity) { map = new LongObjectHashMap<>(initialCapacity); } diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMapTest.java new file mode 100644 index 00000000..3ea0a79c --- /dev/null +++ b/orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentFileChannelMapTest.java @@ -0,0 +1,70 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.orm.lpmap; + +import com.zfoo.event.manager.EventBus; +import com.zfoo.orm.lpmap.model.MyPacket; +import com.zfoo.protocol.ProtocolManager; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class ConcurrentFileChannelMapTest { + + @Test + public void benchmarkTest() throws IOException, InterruptedException { + ProtocolManager.initProtocol(Set.of(MyPacket.class)); + + var map = new ConcurrentFileChannelMap("db", MyPacket.class); + var atomicInt = new AtomicInteger(0); + var count = 1000_0000; + + var countdown = new CountDownLatch(EventBus.EXECUTORS_SIZE); + for (int i = 0; i < EventBus.EXECUTORS_SIZE; i++) { + EventBus.asyncExecute().execute(new Runnable() { + @Override + public void run() { + var key = atomicInt.getAndIncrement(); + while (key < count) { + var myPacket = MyPacket.valueOf(key, String.valueOf(key)); + map.put(key, myPacket); + key = atomicInt.getAndIncrement(); + } + countdown.countDown(); + } + }); + } + countdown.await(); + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = map.get(i); + Assert.assertEquals(myPacket, packet); + } + + map.close(); + var newMap = new ConcurrentFileChannelMap("db", MyPacket.class); + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = newMap.get(i); + Assert.assertEquals(myPacket, packet); + } + } +} diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentHeapMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentHeapMapTest.java index 4617c6ad..3d67f775 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentHeapMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/ConcurrentHeapMapTest.java @@ -12,12 +12,16 @@ package com.zfoo.orm.lpmap; +import com.zfoo.event.manager.EventBus; import com.zfoo.orm.lpmap.model.MyPacket; import com.zfoo.protocol.ProtocolManager; import org.junit.Assert; import org.junit.Test; +import java.io.IOException; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; /** * @author jaysunxiao @@ -42,4 +46,34 @@ public class ConcurrentHeapMapTest { Assert.assertEquals(previous3, myPacket); } + @Test + public void benchmarkTest() throws IOException, InterruptedException { + ProtocolManager.initProtocol(Set.of(MyPacket.class)); + + var map = new ConcurrentHeapMap(); + var atomicInt = new AtomicInteger(0); + var count = 1000_0000; + + var countdown = new CountDownLatch(EventBus.EXECUTORS_SIZE); + for (int i = 0; i < EventBus.EXECUTORS_SIZE; i++) { + EventBus.asyncExecute().execute(new Runnable() { + @Override + public void run() { + var key = atomicInt.getAndIncrement(); + while (key < count) { + var myPacket = MyPacket.valueOf(key, String.valueOf(key)); + map.put(key, myPacket); + key = atomicInt.getAndIncrement(); + } + countdown.countDown(); + } + }); + } + countdown.await(); + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = map.get(i); + Assert.assertEquals(myPacket, packet); + } + } } diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java index 03e17da2..d69138c2 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/FileChannelMapTest.java @@ -14,9 +14,11 @@ package com.zfoo.orm.lpmap; import com.zfoo.orm.lpmap.model.MyPacket; import com.zfoo.protocol.ProtocolManager; +import org.junit.Assert; import org.junit.Ignore; import org.junit.Test; +import java.io.IOException; import java.util.Set; /** @@ -61,4 +63,29 @@ public class FileChannelMapTest { System.out.println(map.get(2)); System.out.println(map.get(3)); } + + @Test + public void benchmarkTest() throws IOException { + ProtocolManager.initProtocol(Set.of(MyPacket.class)); + + var map = new FileChannelMap("db", MyPacket.class); + var count = 1000_0000; + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + map.put(i, myPacket); + } + + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = map.get(i); + Assert.assertEquals(myPacket, packet); + } + map.close(); + map = new FileChannelMap("db", MyPacket.class); + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = map.get(i); + Assert.assertEquals(myPacket, packet); + } + } } diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java index 1e7a9e9f..d76066e3 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/FileHeapMapTest.java @@ -31,7 +31,7 @@ public class FileHeapMapTest { public void test() { ProtocolManager.initProtocol(Set.of(MyPacket.class)); - var map = new FileHeapMap("db", 10, MyPacket.class); + var map = new FileHeapMap("db", MyPacket.class); var myPacket = new MyPacket(); myPacket.setA(9999); @@ -55,5 +55,36 @@ public class FileHeapMapTest { map.save(); } + @Test + public void benchmarkTest() { + ProtocolManager.initProtocol(Set.of(MyPacket.class)); + var map = new FileHeapMap("db", MyPacket.class); + var count = 1000_0000; + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + map.put(i, myPacket); + } + + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = map.get(i); + Assert.assertEquals(myPacket, packet); + } + map.save(); + } + + @Test + public void loadTest() { + ProtocolManager.initProtocol(Set.of(MyPacket.class)); + + var map = new FileHeapMap("db", MyPacket.class); + var count = 1000_0000; + + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = map.get(i); + Assert.assertEquals(myPacket, packet); + } + } } diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java b/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java index d1b33cb1..7ec97a7c 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/HeapMapTest.java @@ -26,7 +26,7 @@ public class HeapMapTest { @Test public void test() { - var map = new HeapMap(10); + var map = new HeapMap(); var myPacket = new MyPacket(); var packet = map.put(0, myPacket); @@ -42,4 +42,21 @@ public class HeapMapTest { Assert.assertEquals(packet, myPacket); } + @Test + public void benchmarkTest() { + var map = new HeapMap(); + + var count = 1000_0000; + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + map.put(i, myPacket); + } + + for (var i = 0; i < count; i++) { + var myPacket = MyPacket.valueOf(i, String.valueOf(i)); + var packet = map.get(i); + Assert.assertEquals(myPacket, packet); + } + } + } diff --git a/orm/src/test/java/com/zfoo/orm/lpmap/model/MyPacket.java b/orm/src/test/java/com/zfoo/orm/lpmap/model/MyPacket.java index 4d1c9be7..4f928480 100644 --- a/orm/src/test/java/com/zfoo/orm/lpmap/model/MyPacket.java +++ b/orm/src/test/java/com/zfoo/orm/lpmap/model/MyPacket.java @@ -14,6 +14,8 @@ package com.zfoo.orm.lpmap.model; import com.zfoo.protocol.IPacket; +import java.util.Objects; + /** * @author jaysunxiao * @version 3.0 @@ -24,11 +26,42 @@ public class MyPacket implements IPacket { private int a; + private String b; + + public static MyPacket valueOf(int a, String b) { + var packet = new MyPacket(); + packet.a = a; + packet.b = b; + return packet; + } + @Override public short protocolId() { return PROTOCOL_ID; } + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + MyPacket myPacket = (MyPacket) o; + return a == myPacket.a && Objects.equals(b, myPacket.b); + } + + @Override + public int hashCode() { + return Objects.hash(a, b); + } + + @Override + public String toString() { + final StringBuilder sb = new StringBuilder("MyPacket{"); + sb.append("a=").append(a); + sb.append(", b='").append(b).append('\''); + sb.append('}'); + return sb.toString(); + } + public int getA() { return a; } @@ -36,4 +69,12 @@ public class MyPacket implements IPacket { public void setA(int a) { this.a = a; } + + public String getB() { + return b; + } + + public void setB(String b) { + this.b = b; + } }