diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleSetBuilder.java index 91bccbd0e..633b12082 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleSetBuilder.java @@ -34,18 +34,16 @@ import static net.openhft.chronicle.map.Objects.builderEquals; public class ChronicleSetBuilder implements Cloneable { - private ChronicleMapBuilder chronicleMapBuilder; + private ChronicleMapBuilder chronicleMapBuilder; ChronicleSetBuilder(Class keyClass) { - chronicleMapBuilder = new ChronicleMapBuilder(keyClass, Boolean.class); - chronicleMapBuilder.constantValueSizeBySample(Boolean.TRUE); + chronicleMapBuilder = new ChronicleMapBuilder(keyClass, Void.class); } public static ChronicleSetBuilder of(Class keyClass) { return new ChronicleSetBuilder(keyClass); } - @Override public ChronicleSetBuilder clone() { try { @@ -58,7 +56,6 @@ public class ChronicleSetBuilder implements Cloneable { } } - public ChronicleSetBuilder actualSegments(int actualSegments) { chronicleMapBuilder.actualSegments(actualSegments); return this; @@ -83,7 +80,6 @@ public class ChronicleSetBuilder implements Cloneable { return chronicleMapBuilder.minSegments(); } - public ChronicleSetBuilder actualElementsPerSegment(int actualEntriesPerSegment) { chronicleMapBuilder.actualEntriesPerSegment(actualEntriesPerSegment); return this; @@ -338,22 +334,22 @@ public class ChronicleSetBuilder implements Cloneable { public ChronicleSet create(File file) throws IOException { - final ChronicleMap map = chronicleMapBuilder.create(file); + final ChronicleMap map = chronicleMapBuilder.create(file); return new SetFromMap(map); } public ChronicleSet create() throws IOException { - final ChronicleMap map = chronicleMapBuilder.create(); + final ChronicleMap map = chronicleMapBuilder.create(); return new SetFromMap(map); } private static class SetFromMap extends AbstractSet implements ChronicleSet, Serializable { - private final ChronicleMap m; // The backing map + private final ChronicleMap m; // The backing map private transient Set s; // Its keySet - SetFromMap(ChronicleMap map) { + SetFromMap(ChronicleMap map) { if (!map.isEmpty()) throw new IllegalArgumentException("Map is non-empty"); m = map; @@ -381,7 +377,7 @@ public class ChronicleSetBuilder implements Cloneable { } public boolean add(E e) { - return m.put(e, Boolean.TRUE) == null; + return m.put(e, (Void)null) == null; } public Iterator iterator() { diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 5679c2242..55d982d99 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -381,7 +381,6 @@ class ReplicatedChronicleMap, long valueSize; if (!isDeleted) { valueSize = valueSizeMarshaller.readSize(entry); - assert valueSize > 0; } else { valueSize = 0; } diff --git a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java index 3451c2c0a..be6eaa22f 100644 --- a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java @@ -21,7 +21,10 @@ import net.openhft.chronicle.map.serialization.impl.*; import net.openhft.chronicle.map.threadlocal.Provider; import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.io.Bytes; -import net.openhft.lang.io.serialization.*; +import net.openhft.lang.io.serialization.BytesMarshallable; +import net.openhft.lang.io.serialization.BytesMarshaller; +import net.openhft.lang.io.serialization.ObjectFactory; +import net.openhft.lang.io.serialization.ObjectSerializer; import net.openhft.lang.io.serialization.impl.*; import net.openhft.lang.model.Byteable; import org.jetbrains.annotations.NotNull; @@ -40,7 +43,7 @@ final class SerializationBuilder implements Cloneable { private static boolean concreteClass(Class c) { return !c.isInterface() && !Modifier.isAbstract(c.getModifiers()); } - + private static boolean marshallerUseFactory(Class c) { return Byteable.class.isAssignableFrom(c) || BytesMarshallable.class.isAssignableFrom(c) || @@ -52,6 +55,7 @@ final class SerializationBuilder implements Cloneable { } enum Role {KEY, VALUE} + private enum CopyingInterop {FROM_MARSHALLER, FROM_WRITER} private final Serializable bufferIdentity; @@ -84,6 +88,8 @@ final class SerializationBuilder implements Cloneable { } else if (eClass == CharSequence.class || eClass == String.class) { reader((BytesReader) CharSequenceReader.of(), factory); writer((BytesWriter) CharSequenceWriter.instance()); + } else if (eClass == Void.class) { + agileMarshaller((AgileBytesMarshaller) VoidMarshaller.INSTANCE, factory); } else if (eClass == Long.class) { agileMarshaller((AgileBytesMarshaller) LongMarshaller.INSTANCE, factory); } else if (eClass == byte[].class) { diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 23f166078..ec60352ba 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -17,15 +17,16 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.map.serialization.*; +import net.openhft.chronicle.map.threadlocal.Provider; +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.Maths; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.collection.SingleThreadedDirectBitSet; import net.openhft.lang.io.*; -import net.openhft.lang.io.serialization.*; +import net.openhft.lang.io.serialization.JDKObjectSerializer; +import net.openhft.lang.io.serialization.ObjectFactory; import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.model.Byteable; -import net.openhft.chronicle.map.threadlocal.Provider; -import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -312,7 +313,7 @@ class VanillaChronicleMap, } void checkValue(Object value) { - if (!vClass.isInstance(value)) { + if (vClass != Void.class && !vClass.isInstance(value)) { throw new ClassCastException("Value must be a " + vClass.getName() + " but was a " + value.getClass()); } @@ -692,7 +693,7 @@ class VanillaChronicleMap, * true}, {@code usingValue} or a newly created instance of value class, if {@code usingValue == * null}, is put into this Segment for the key. * - * @param hash2 a hash code related to the {@code keyBytes} + * @param hash2 a hash code related to the {@code keyBytes} * @return the value which is finally associated with the given key in this Segment after execution of * this method, or {@code null}. */ @@ -827,13 +828,12 @@ class VanillaChronicleMap, } /** - * Puts entry. If {@code value} implements {@link Byteable} interface and - * {@code usingValue} is {@code true}, the value is backed with the bytes of this entry. + * Puts entry. If {@code value} implements {@link Byteable} interface and {@code usingValue} is {@code + * true}, the value is backed with the bytes of this entry. * * @param value the value to put - * @param usingValue {@code true} if the value should be backed with the bytes of the entry, - * if it implements {@link Byteable} interface, {@code false} if it - * should put itself + * @param usingValue {@code true} if the value should be backed with the bytes of the entry, if it + * implements {@link Byteable} interface, {@code false} if it should put itself * @return offset of the written entry in the Segment bytes */ private long putEntry(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, @@ -958,7 +958,7 @@ class VanillaChronicleMap, * Removes a key (or key-value pair) from the Segment.

The entry will only be removed if {@code * expectedValue} equals to {@code null} or the value previously corresponding to the specified key. * - * @param hash2 a hash code related to the {@code keyBytes} + * @param hash2 a hash code related to the {@code keyBytes} * @return the value of the entry that was removed if the entry corresponding to the {@code keyBytes} * exists and {@link #removeReturnsNull} is {@code false}, {@code null} otherwise */ @@ -1095,12 +1095,12 @@ class VanillaChronicleMap, * Replaces value in existing entry. May cause entry relocation, because there may be not enough space * for new value in location already allocated for this entry. * - * @param pos index of the first block occupied by the entry - * @param offset relative offset of the entry in Segment bytes (before, i. e. including - * metaData) - * @param entry relative pointer in Segment bytes - * @param valueSizePos relative position of value size in entry - * @param entryEndAddr absolute address of the entry end + * @param pos index of the first block occupied by the entry + * @param offset relative offset of the entry in Segment bytes (before, i. e. including + * metaData) + * @param entry relative pointer in Segment bytes + * @param valueSizePos relative position of value size in entry + * @param entryEndAddr absolute address of the entry end * @return relative offset of the entry in Segment bytes after putting value (that may cause entry * relocation) */ diff --git a/src/main/java/net/openhft/chronicle/map/serialization/impl/VoidMarshaller.java b/src/main/java/net/openhft/chronicle/map/serialization/impl/VoidMarshaller.java new file mode 100644 index 000000000..257588f25 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/serialization/impl/VoidMarshaller.java @@ -0,0 +1,71 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com + * + * 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 net.openhft.chronicle.map.serialization.impl; + +import net.openhft.chronicle.map.serialization.AgileBytesMarshaller; +import net.openhft.lang.io.Bytes; + +public enum VoidMarshaller implements AgileBytesMarshaller { + INSTANCE; + + @Override + public long size(Void e) { + return 0L; + } + + @Override + public int sizeEncodingSize(long size) { + return 0; + } + + @Override + public void writeSize(Bytes bytes, long size) { + // do nothing + } + + @Override + public boolean startsWith(Bytes bytes, Void e) { + return false; + } + + @Override + public long hash(Void e) { + return 0; + } + + @Override + public void write(Bytes bytes, Void e) { + // do nothing; + } + + @Override + public long readSize(Bytes bytes) { + return 0L; + } + + @Override + public Void read(Bytes bytes, long size) { + // Void nothing; + return null; + } + + @Override + public Void read(Bytes bytes, long size, Void aVoid) { + return null; + } + +} diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index 8b1a153dd..f0adb7e24 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -21,8 +21,12 @@ import java.io.Closeable; import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; +import java.net.InetSocketAddress; import java.util.concurrent.ArrayBlockingQueue; +import static java.util.concurrent.TimeUnit.SECONDS; +import static net.openhft.chronicle.map.Replicators.tcp; + /** * @author Rob Austin. */ @@ -103,6 +107,26 @@ public class Builder { }; } + public static > T newTcpSocketShmIntString( + final byte identifier, + final int serverPort, + final InetSocketAddress... endpoints) throws IOException { + return (T) newTcpSocketShmBuilder(Integer.class, Void.class, + identifier, serverPort, endpoints).create(getPersistenceFile()); + } + + public static ChronicleMapBuilder newTcpSocketShmBuilder( + Class kClass, Class vClass, + final byte identifier, + final int serverPort, + final InetSocketAddress... endpoints) throws IOException { + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints) + .heartBeatInterval(1L, SECONDS); + return ChronicleMapBuilder.of(kClass, vClass) + .entries(20000L) + .addReplicator(tcp(identifier, tcpConfig)); + } + interface MapProvider { T getMap(); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java index 19f07350d..5671d9c3d 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java @@ -26,9 +26,8 @@ import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; -import static java.util.concurrent.TimeUnit.SECONDS; import static net.openhft.chronicle.map.Builder.getPersistenceFile; -import static net.openhft.chronicle.map.Replicators.tcp; +import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -44,18 +43,6 @@ public class TCPSocketReplication4WayMapTest { private ChronicleMap map3; private ChronicleMap map4; - public static ChronicleMapBuilder newTcpSocketShmBuilder( - Class kClass, Class vClass, - final byte identifier, - final int serverPort, - final InetSocketAddress... endpoints) throws IOException { - TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints) - .heartBeatInterval(1L, SECONDS); - return ChronicleMapBuilder.of(kClass, vClass) - .entries(20000L) - .addReplicator(tcp(identifier, tcpConfig)); - } - public static > T newTcpSocketShmIntString( final byte identifier, final int serverPort, @@ -64,6 +51,8 @@ public class TCPSocketReplication4WayMapTest { identifier, serverPort, endpoints).create(getPersistenceFile()); } + + static ChronicleMap newTcpSocketShmIntValueString( final byte identifier, final int serverPort, diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java index b538f0764..95885799c 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java @@ -25,7 +25,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import static net.openhft.chronicle.map.Builder.getPersistenceFile; -import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder; +import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntString; import static org.junit.Assert.assertEquals; diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java index a197e51c2..78b342987 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java @@ -30,7 +30,7 @@ import java.nio.ByteBuffer; import java.util.Random; import static net.openhft.chronicle.map.Builder.getPersistenceFile; -import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder; +import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntValueString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java index 022c29f7e..e8ac1f7f7 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest.java @@ -26,7 +26,7 @@ import java.net.InetSocketAddress; import java.util.Random; import static net.openhft.chronicle.map.Builder.getPersistenceFile; -import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder; +import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmIntString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3VoidValueTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3VoidValueTest.java new file mode 100644 index 000000000..00515d98c --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3VoidValueTest.java @@ -0,0 +1,138 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com + * + * 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 net.openhft.chronicle.map; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetSocketAddress; + +import static net.openhft.chronicle.map.Builder.newTcpSocketShmIntString; +import static org.junit.Assert.*; + +/** + * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket + * + * @author Rob Austin. + */ + +public class TCPSocketReplicationTest3VoidValueTest { + + + private ChronicleMap map1; + private ChronicleMap map2; + private ChronicleMap map3; + + + @Before + public void setup() throws IOException { + map1 = newTcpSocketShmIntString((byte) 1, 8076, new InetSocketAddress("localhost", 8077), + new InetSocketAddress("localhost", 8079)); + map2 = newTcpSocketShmIntString((byte) 2, 8077, new InetSocketAddress("localhost", 8079)); + map3 = newTcpSocketShmIntString((byte) 3, 8079); + } + + @After + public void tearDown() throws InterruptedException { + Thread.sleep(5); + for (final Closeable closeable : new Closeable[]{map1, map2, map3}) { + try { + closeable.close(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + + + @Test + public void test3() throws IOException, InterruptedException { + + assertEquals(null, map3.put(5, null)); + + // allow time for the recompilation to resolve + waitTillEqual(15000); + + assertEquals(map1, map2); + assertEquals(map3, map2); + assertTrue(!map1.isEmpty()); + + } + + @Test + public void test() throws IOException, InterruptedException { + + assertEquals(null, map1.put(1, null)); + assertEquals(null, map1.put(2, null)); + + assertEquals(null, map2.put(5, null)); + assertEquals(null, map2.put(6,null)); + + map1.remove(2); + map2.remove(3); + map1.remove(3); + map2.put(5,null); + + // allow time for the recompilation to resolve + waitTillEqual(5000); + + assertEquals(map1, map2); + assertEquals(map3, map3); + assertTrue(!map1.isEmpty()); + + } + + + @Test + public void testClear() throws IOException, InterruptedException { + + assertEquals(null, map1.put(1, null)); + assertEquals(null, map1.put(2,null)); + + assertEquals(null, map2.put(5, null)); + assertEquals(null, map2.put(6, null)); + + map1.clear(); + + map2.put(5, null); + + // allow time for the recompilation to resolve + waitTillEqual(5000); + + assertEquals(map1, map2); + assertEquals(map3, map3); + assertTrue(!map1.isEmpty()); + + } + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + int t = 0; + for (; t < timeOutMs; t++) { + if (map1.equals(map2) && + map1.equals(map3)) + break; + Thread.sleep(1); + } + + } +} + + + diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java index 9b6dc69eb..b39ab79b1 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTestPostConnection.java @@ -23,7 +23,7 @@ import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; -import static net.openhft.chronicle.map.TCPSocketReplication4WayMapTest.newTcpSocketShmBuilder; +import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue;