diff --git a/README.md b/README.md index 4ef414560..bfb51bbc2 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,10 @@ Click here to get the [Latest Version Number](http://search.maven.org/#search%7C * [JavaDoc](http://openhft.github.io/Chronicle-Map/apidocs) * [Getting Started Guide](https://github.com/OpenHFT/Chronicle-Map#getting-started) * [Simple Construction](https://github.com/OpenHFT/Chronicle-Map#simple-construction) - * [Sharing Data Between Two or More Maps](https://github.com/OpenHFT/Chronicle-Map#sharing-data-between-two-or-more-maps) + * [Maven Download](https://github.com/OpenHFT/Chronicle-Map#maven-artifact-download-1) + * [Snapshot Download](https://github.com/OpenHFT/Chronicle-Map#maven-snapshot-download) + * [Key Value Object Types](https://github.com/OpenHFT/Chronicle-Map#key-value-object-types) + * [Sharing Data Between Two or More Maps](https://github.com/OpenHFT/Chronicle-Map#sharing-data-between-two-or-more-maps) * [Entries](https://github.com/OpenHFT/Chronicle-Map#entries) * [Size of Space Reserved on Disk](https://github.com/OpenHFT/Chronicle-Map#size-of-space-reserved-on-disk) * [Chronicle Map Interface](https://github.com/OpenHFT/Chronicle-Map#chronicle-map-interface) @@ -119,7 +122,7 @@ product suite. In addition, The original Chronicle has been renamed to Chronicle ## Getting Started #### Tutorial 1 - Creating an instance of Chronicle Map -[![ScreenShot](http://openhft.net/wp-content/uploads/2014/09/Screen-Shot-2014-10-14-at-17.49.36.png)](https://www.youtube.com/watch?v=pFVv8r8ZTZ8) +[![ScreenShot](http://openhft.net/wp-content/uploads/2014/09/Screen-Shot-2014-10-14-at-17.49.36.png)](http://openhft.net/chronicle-map-video-tutorial-1/) ### Simple Construction @@ -168,6 +171,30 @@ and define the snapshot version in your pom.xml, for example: 1.0.1-SNAPSHOT ``` + +#### Key Value Object Types + +Unlike HashMap which will support any heap object, Chronicle Map only works with objects that it +can store off heap, so the objects have to be : (one of the following ) + +- AutoBoxed primitives - for good performance. +- Strings - for good performance. +- implements Serializable +- implements Externalizable ( with a public default constructor ) +- implements our custom interface BytesMarshallable ( with a public default constructor ) - use +this for best performance. + +or value objects that are created through, a directClass interface, for example : +``` java + ChronicleMap chm = ChronicleMapBuilder + .of(String.class, directClassFor(BondVOInterface.class)) + .create(file); + +``` + +Object graphs can also be included as long as the outer object supports Serializable, Externalizable or BytesMarshallable. + + #### Java Class Construction Creating an instance of Chronicle Map is a little more complexed than just calling a constructor. @@ -199,9 +226,10 @@ the instances of Chronicle Map on the same server. The name and location of the up to you. For the best performance on many unix systems we recommend using [tmpfs](http://en.wikipedia.org/wiki/Tmpfs). -If instead, you do not wish to replicate between process on the same server or if you are only +If instead, you do not wish to replicate between processes on the same server or if you are only using TCP replication to replicate between servers, you do not have to provide the "file", -so you can call `create()` method on ChronicleMapBuilder without file parameter: +so you can call the `create()` method on you ChronicleMapBuilder without providing the file +parameter: ``` ConcurrentMap map = builder.create(); ``` diff --git a/pom.xml b/pom.xml index 162069cb3..8862cc7c4 100755 --- a/pom.xml +++ b/pom.xml @@ -26,7 +26,7 @@ 4.0.0 chronicle-map - 2.0.3a-SNAPSHOT + 2.0.5a-SNAPSHOT OpenHFT/Chronicle-Map Chronicle-Map @@ -50,7 +50,7 @@ net.openhft affinity - 2.1.0 + 2.1.2 diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 89e741a0b..1f9cfba3e 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -197,8 +197,9 @@ abstract class AbstractChannelReplicator implements Closeable { /** - * checks the number of bytes written in this interval, if this number of bytes exceeds a threshold, - * the selected will de-register the socket that is being written to, until the interval is finished. + * checks the number of bytes written in this interval, if this number of bytes exceeds a + * threshold, the selected will de-register the socket that is being written to, until the + * interval is finished. * * @throws ClosedChannelException */ @@ -261,12 +262,22 @@ abstract class AbstractChannelReplicator implements Closeable { @Override public boolean onEntry(final Bytes entry, final int chronicleId) { + + long pos0 = in.position(); + + // used to denote that this is not a stateless map event + in.writeByte(StatelessMapClient.EventId.STATEFUL_UPDATE.ordinal()); + + long sizeLocation = in.position(); + + // this is where we will store the size of the entry in.skip(SIZE_OF_SHORT); + final long start = in.position(); externalizable.writeExternalEntry(entry, in, chronicleId); if (in.position() == start) { - in.position(in.position() - SIZE_OF_SHORT); + in.position(pos0); return false; } @@ -276,7 +287,10 @@ abstract class AbstractChannelReplicator implements Closeable { if (entrySize > MAX_UNSIGNED_SHORT) throw new IllegalStateException("entry too large, the entry size=" + entrySize + ", " + "entries are limited to a size of " + MAX_UNSIGNED_SHORT); - in.writeUnsignedShort(start - SIZE_OF_SHORT, entrySize); + + if (LOG.isDebugEnabled()) + LOG.debug("sending entry of entrySize=" + entrySize); + in.writeUnsignedShort(sizeLocation, entrySize); return true; } diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index c51a56e8c..a59519c97 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -407,7 +407,7 @@ public final class ChannelProvider implements Closeable { @Override protected Closeable applyTo(ChronicleMapBuilder builder, - Replica map, EntryExternalizable entryExternalizable) { + Replica map, EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) { if (builder.entrySize() > maxEntrySize) { throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " + "maxEntrySize=" + maxEntrySize + " was specified, but map with " + diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java index 0bb05dab4..38726cbff 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java @@ -68,7 +68,7 @@ public final class ChannelProviderBuilder { replicatingCluster.asReplica, replicatingCluster.asEntryExternalizable, tcpReplicationConfig, - maxEntrySize); + maxEntrySize, null); replicatingCluster.add(tcpReplicator); } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 725815757..c400f0f43 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -386,8 +386,13 @@ public class ChronicleMapBuilder implements Cloneable, if (actualEntriesPerSegment > 0) return actualEntriesPerSegment; int as = actualSegments(); + long actualEntries = + as == 1 ? entries + 1 : // The extra 1 might not be needed. + as <= 4 ? entries * 10L / 9 : // 10%, assumes you are trying to be tight. + entries * 5L / 4; // 25% + // round up to the next multiple of 64. - return (int) (Math.max(1, entries * 2L / as) + 63) & ~63; + return (int) (Math.max(1, actualEntries / as) + 63) & ~63; } @Override @@ -786,7 +791,7 @@ public class ChronicleMapBuilder implements Cloneable, return identifier != -1; } - private void preMapConstruction() { + void preMapConstruction() { keyBuilder.objectSerializer(objectSerializer()); valueBuilder.objectSerializer(objectSerializer()); @@ -805,7 +810,7 @@ public class ChronicleMapBuilder implements Cloneable, if (map instanceof ReplicatedChronicleMap) { ReplicatedChronicleMap result = (ReplicatedChronicleMap) map; for (Replicator replicator : replicators.values()) { - Closeable token = replicator.applyTo(this, result, result); + Closeable token = replicator.applyTo(this, result, result, map); if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) { LOG.warn( "MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " + diff --git a/src/main/java/net/openhft/chronicle/map/Replicator.java b/src/main/java/net/openhft/chronicle/map/Replicator.java index 86ef2ed82..3dede414d 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicator.java +++ b/src/main/java/net/openhft/chronicle/map/Replicator.java @@ -41,6 +41,7 @@ public abstract class Replicator { * map configurations, not accessible via the map instance. * @param map a replicated map instance. Provides basic tools for replication implementation. * @param entryExternalizable the callback for ser/deser implementation in the replicator + * @param chronicleMap * @return a {@code Closeable} token to control replication resources. * It should be closed on closing the replicated map. * @throws java.io.IOException if an io error occurred during the replicator setup @@ -49,6 +50,6 @@ public abstract class Replicator { * applied to a map (or the specified number of maps) */ protected abstract Closeable applyTo(ChronicleMapBuilder builder, - Replica map, Replica.EntryExternalizable entryExternalizable) + Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) throws IOException; } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index 023e8a7b2..aeb5118d3 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -30,10 +30,18 @@ final class Replicators { @Override protected Closeable applyTo(ChronicleMapBuilder builder, - Replica map, Replica.EntryExternalizable entryExternalizable) + Replica replica, Replica.EntryExternalizable entryExternalizable, + final ChronicleMap chronicleMap) throws IOException { - return new TcpReplicator(map, entryExternalizable, replicationConfig, - builder.entrySize()); + + final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder + .keyBuilder, builder.valueBuilder); + + StatelessServerConnector statelessServerConnector = new StatelessServerConnector + (keyValueSerializer, chronicleMap); + + return new TcpReplicator(replica, entryExternalizable, replicationConfig, + builder.entrySize(), statelessServerConnector); } }; } @@ -44,7 +52,7 @@ final class Replicators { @Override protected Closeable applyTo(ChronicleMapBuilder builder, - Replica map, Replica.EntryExternalizable entryExternalizable) + Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) throws IOException { return new UdpReplicator(map, entryExternalizable, replicationConfig, builder.entrySize()); diff --git a/src/main/java/net/openhft/chronicle/map/Serializer.java b/src/main/java/net/openhft/chronicle/map/Serializer.java new file mode 100644 index 000000000..d30bb388a --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/Serializer.java @@ -0,0 +1,81 @@ +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.io.Bytes; +import net.openhft.lang.model.Byteable; +import net.openhft.lang.model.constraints.NotNull; + +/** + * uses a {@code SerializationBuilder} to read and write object(s) as bytes + * + * @author Rob Austin. + */ +class Serializer { + + final SizeMarshaller sizeMarshaller; + final BytesReader originalReader; + transient Provider> readerProvider; + final Object originalWriter; + transient Provider writerProvider; + final MetaBytesInterop originalMetaValueWriter; + final MetaProvider metaValueWriterProvider; + + Serializer(final SerializationBuilder serializationBuilder) { + + sizeMarshaller = serializationBuilder.sizeMarshaller(); + originalMetaValueWriter = serializationBuilder.metaInterop(); + metaValueWriterProvider = serializationBuilder.metaInteropProvider(); + + originalReader = serializationBuilder.reader(); + originalWriter = serializationBuilder.interop(); + + readerProvider = Provider.of((Class) originalReader.getClass()); + writerProvider = Provider.of((Class) originalWriter.getClass()); + } + + public O readMarshallable(@NotNull Bytes in) throws IllegalStateException { + + final ThreadLocalCopies copies = readerProvider.getCopies(null); + final BytesReader valueReader = readerProvider.get(copies, originalReader); + + try { + long valueSize = sizeMarshaller.readSize(in); + return valueReader.read(in, valueSize, null); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + public void writeMarshallable(O value, @NotNull Bytes out) { + + ThreadLocalCopies copies = readerProvider.getCopies(null); + Object valueWriter = writerProvider.get(copies, originalWriter); + copies = writerProvider.getCopies(copies); + + final long valueSize; + + MetaBytesWriter metaValueWriter = null; + + if ((value instanceof Byteable)) { + valueSize = ((Byteable) value).maxSize(); + } else { + copies = writerProvider.getCopies(copies); + valueWriter = writerProvider.get(copies, originalWriter); + copies = metaValueWriterProvider.getCopies(copies); + metaValueWriter = metaValueWriterProvider.get( + copies, originalMetaValueWriter, valueWriter, value); + valueSize = metaValueWriter.size(valueWriter, value); + } + + sizeMarshaller.writeSize(out, valueSize); + + if (metaValueWriter != null) + metaValueWriter.write(valueWriter, out, value); + else + throw new UnsupportedOperationException(""); + + } + +} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java index 73cae54ba..993441d5c 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java @@ -1,18 +1,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.io.ByteBufferBytes; -import net.openhft.lang.io.BytesStore; -import net.openhft.lang.io.serialization.ObjectFactory; -import net.openhft.lang.model.Byteable; +import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -import javax.swing.text.Segment; import java.io.File; import java.io.IOException; -import java.io.Serializable; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; import java.util.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -21,73 +19,27 @@ import java.util.concurrent.atomic.AtomicLong; import static net.openhft.chronicle.map.StatelessMapClient.EventId.*; /** - * * **** THIS IS VERY MUCH IN DRAFT *** * * @author Rob Austin. */ -public class StatelessMapClient, - V, VW, MVW extends MetaBytesWriter> extends AbstractMap - implements ChronicleMap, Serializable { +class StatelessMapClient implements ChronicleMap { - final Class kClass; - private ThreadLocal sourceBuffer = new ThreadLocal(); - - final BytesReader originalKeyReader; - transient Provider> keyReaderProvider; - final KI originalKeyInterop; - transient Provider keyInteropProvider; - final MKI originalMetaKeyInterop; - final MetaProvider metaKeyInteropProvider; - - final Class vClass; - final SizeMarshaller valueSizeMarshaller; - final BytesReader originalValueReader; - transient Provider> valueReaderProvider; - final VW originalValueWriter; - transient Provider valueWriterProvider; - final MVW originalMetaValueWriter; - final MetaProvider metaValueWriterProvider; - final ObjectFactory valueFactory; - final DefaultValueProvider defaultValueProvider; - - Map transactionIDs = new HashMap(); + private static final Logger LOG = LoggerFactory.getLogger(StatelessMapClient.class); - AtomicLong transactionID = new AtomicLong(); + private final SocketChannel clientChannel; + public static final ByteBuffer STATELESS_CLIENT_IDENTIFER = ByteBuffer.wrap(new byte[]{-127}); - private final long lockTimeOutNS; - - transient Segment[] segments; // non-final for close() - // non-final for close() and because it is initialized out of constructor - transient BytesStore ms; - transient long headerSize; - transient Set> entrySet; - - private int bits; - private int mask; - private long timeoutMs = TimeUnit.SECONDS.toMillis(20); - - - @Override - public File file() { - return null; - } - - @Override - public void close() throws IOException { - throw new UnsupportedOperationException("This is not supported in the " + this.getClass() - .getSimpleName()); - } - - enum EventId { + public static enum EventId { + HEARTBEAT, + STATEFUL_UPDATE, LONG_SIZE, SIZE, IS_EMPTY, CONTAINS_KEY, CONTAINS_VALUE, GET, PUT, REMOVE, - PUT_ALL, CLEAR, KEY_SET, VALUES, @@ -98,6 +50,55 @@ public class StatelessMapClient, REMOVE_WITH_VALUE } + private final ThreadLocal sourceBuffer = new ThreadLocal(); + private final Map transactionIDs = new HashMap(); + + private final AtomicLong transactionID = new AtomicLong(); + + private long timeoutMs = TimeUnit.SECONDS.toMillis(20); + + + public StatelessMapClient(final KeyValueSerializer keyValueSerializer, final InetSocketAddress remote) throws IOException { + + this.keyValueSerializer = keyValueSerializer; + + clientChannel = SocketChannel.open(); + clientChannel.connect(remote); + + doHandShaking(); + } + + private void doHandShaking() throws IOException { + + + while (STATELESS_CLIENT_IDENTIFER.remaining() > 0) { + clientChannel.write(STATELESS_CLIENT_IDENTIFER); + } + + in.buffer().clear(); + + while (in.buffer().position() <= 0) { + clientChannel.read(in.buffer()); + } + + in.limit(in.buffer().position()); + in.position(0); + byte remoteIdentifier = in.readByte(); + + LOG.info("Attached to a map with a remote identifier=" + remoteIdentifier); + + } + + + public File file() { + return null; + } + + public void close() { + throw new UnsupportedOperationException("This is not supported in the " + this.getClass() + .getSimpleName()); + } long nextUniqueTransaction() { long time = System.currentTimeMillis(); @@ -112,400 +113,235 @@ public class StatelessMapClient, } - private final ByteBufferBytes target; - private final SizeMarshaller keySizeMarshaller; - - public StatelessMapClient(ByteBufferBytes target, ChronicleMapBuilder builder) { - this.target = target; - - SerializationBuilder keyBuilder = builder.keyBuilder; - kClass = keyBuilder.eClass; - keySizeMarshaller = keyBuilder.sizeMarshaller(); - originalKeyReader = keyBuilder.reader(); - - originalKeyInterop = (KI) keyBuilder.interop(); - - originalMetaKeyInterop = (MKI) keyBuilder.metaInterop(); - metaKeyInteropProvider = (MetaProvider) keyBuilder.metaInteropProvider(); - - SerializationBuilder valueBuilder = builder.valueBuilder; - vClass = valueBuilder.eClass; - valueSizeMarshaller = valueBuilder.sizeMarshaller(); - originalValueReader = valueBuilder.reader(); - - originalValueWriter = (VW) valueBuilder.interop(); - originalMetaValueWriter = (MVW) valueBuilder.metaInterop(); - metaValueWriterProvider = (MetaProvider) valueBuilder.metaInteropProvider(); - valueFactory = valueBuilder.factory(); - defaultValueProvider = builder.defaultValueProvider(); - - lockTimeOutNS = builder.lockTimeOut(TimeUnit.NANOSECONDS); + private final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocate(1024)); + private final ByteBufferBytes in = new ByteBufferBytes(ByteBuffer.allocate(1024)); + private final KeyValueSerializer keyValueSerializer; - } + public synchronized V putIfAbsent(K key, V value) { - @Override - public V putIfAbsent(K key, V value) { - target.writeByte(PUT_IF_ABSENT.ordinal()); - - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - - writeKey(key, copies); - writeValue(value, copies); - - return null; + long sizeLocation = writeEvent(PUT_IF_ABSENT); + writeKey(key); + writeValue(value); + return readKey(sizeLocation); } - @Override - public boolean remove(Object key, Object value) { - target.writeByte(REMOVE_WITH_VALUE.ordinal()); - - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); - writeValue((V) value, copies); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized boolean remove(Object key, Object value) { + long sizeLocation = writeEvent(REMOVE_WITH_VALUE); + writeKey((K) key); + writeValue((V) value); // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public boolean replace(K key, V oldValue, V newValue) { - target.writeByte(REPLACE_WITH_OLD_AND_NEW_VALUE.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey(key, copies); - writeValue(oldValue, copies); - writeValue(newValue, copies); - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized boolean replace(K key, V oldValue, V newValue) { + long sizeLocation = writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE); + writeKey(key); + writeValue(oldValue); + writeValue(newValue); // get the data back from the server - return blockingFetch(transactionId).readBoolean(); - + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public V replace(K key, V value) { - target.writeByte(REPLACE.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey(key, copies); - writeValue(value, copies); - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized V replace(K key, V value) { + long sizeLocation = writeEvent(REPLACE); + writeKey(key); + writeValue(value); // get the data back from the server - return readValue(blockingFetch(transactionId)); + return readKey(sizeLocation); } - @Override - public int size() { + public synchronized int size() { - target.writeByte(SIZE.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + long sizeLocation = writeEvent(SIZE); // get the data back from the server - return blockingFetch(transactionId).readInt(); + return blockingFetch(sizeLocation).readInt(); } - @Override - public boolean isEmpty() { - target.writeByte(IS_EMPTY.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized boolean isEmpty() { + long sizeLocation = writeEvent(IS_EMPTY); // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public boolean containsKey(Object key) { - target.writeByte(CONTAINS_KEY.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized boolean containsKey(Object key) { + long sizeLocation = writeEvent(CONTAINS_KEY); + writeKey((K) key); // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public boolean containsValue(Object value) { - target.writeByte(CONTAINS_VALUE.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeValue((V) value, copies); - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized boolean containsValue(Object value) { + long sizeLocation = writeEvent(CONTAINS_VALUE); + writeValue((V) value); // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public long longSize() { - target.writeByte(LONG_SIZE.ordinal()); - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized long longSize() { + long sizeLocation = writeEvent(LONG_SIZE); + // get the data back from the server + return blockingFetch(sizeLocation).readLong(); + } + + + public synchronized V get(Object key) { + long sizeLocation = writeEvent(GET); + writeKey((K) key); // get the data back from the server - return blockingFetch(transactionId).readLong(); - } - - @Override - public V get(Object key) { - target.writeByte(GET.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); - - // get the data back from the server - return readValue(blockingFetch(transactionId)); + return readKey(sizeLocation); } - private V readValue(ByteBufferBytes byteBufferBytes) { - throw new UnsupportedOperationException("todo"); - } - @Override - public V getUsing(K key, V value) { + public synchronized V getUsing(K key, V value) { throw new UnsupportedOperationException("getUsing is not supported for stateless clients"); } - @Override - public V acquireUsing(K key, V value) { + + public synchronized V acquireUsing(K key, V value) { throw new UnsupportedOperationException("getUsing is not supported for stateless clients"); } - @Override - public V put(K key, V value) { - target.writeByte(PUT.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); - writeValue((V) value, copies); + public synchronized V put(K key, V value) { - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + long sizeLocation = writeEvent(PUT); + writeKey(key); + writeValue(value); // get the data back from the server - return readValue(blockingFetch(transactionId)); - } - - @Override - public V remove(Object key) { - target.writeByte(REMOVE.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); - - // get the data back from the server - return readValue(blockingFetch(transactionId)); - - } - - @Override - public void putAll(Map map) { - target.writeByte(PUT_ALL.ordinal()); - - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - - HashMap safeCopy = new HashMap(map); - - writeEntries(copies, safeCopy); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); - - // get the data back from the server - blockingFetch(transactionId); - - return; + return readKey(sizeLocation); } - private void writeEntries(ThreadLocalCopies copies, HashMap safeCopy) { - target.writeLong(safeCopy.size()); - Set entries = (Set) safeCopy.entrySet(); + public synchronized V remove(Object key) { + long sizeLocation = writeEvent(REMOVE); + writeKey((K) key); + + // get the data back from the server + return readKey(sizeLocation); + + } + + + public synchronized void putAll(Map map) { + + for (Map.Entry e : map.entrySet()) { + put(e.getKey(), e.getValue()); + } + + } + + private void writeEntries(Map map) { + + final HashMap safeCopy = new HashMap(map); + + out.writeStopBit(safeCopy.size()); + + final Set entries = (Set) safeCopy.entrySet(); for (Entry e : entries) { - writeKey(e.getKey(), copies); - writeValue(e.getValue(), copies); + writeKey(e.getKey()); + writeValue(e.getValue()); } } - @Override - public void clear() { - target.writeByte(CLEAR.ordinal()); - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized void clear() { + long sizeLocation = writeEvent(CLEAR); // get the data back from the server - blockingFetch(transactionId); - return; + blockingFetch(sizeLocation); + + } @NotNull - @Override - public Set keySet() { - target.writeByte(KEY_SET.ordinal()); - // send the transaction id - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized Set keySet() { + long sizeLocation = writeEvent(KEY_SET); // get the data back from the server - return readKeySet(blockingFetch(transactionId)); + return readKeySet(blockingFetch(sizeLocation)); } - private Set readKeySet(ByteBufferBytes bufferBytes) { - int size = bufferBytes.readInt(); - HashSet result = new HashSet<>(); + @NotNull + + public synchronized Collection values() { + long sizeLocation = writeEvent(VALUES); + + // get the data back from the server + final Bytes in = blockingFetch(sizeLocation); + + final long size = in.readStopBit(); + + if (size > Integer.MAX_VALUE) + throw new IllegalStateException("size=" + size + " is too large."); + + final ArrayList result = new ArrayList((int) size); for (int i = 0; i < size; i++) { - result.add(readKey(bufferBytes)); + result.add(keyValueSerializer.readValue(in)); } return result; } - private K readKey(ByteBufferBytes bufferBytes) { - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); - - // get the data back from the server - return readKey(blockingFetch(transactionId)); - - } @NotNull - @Override - public Collection values() { - target.writeByte(VALUES.ordinal()); - long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + public synchronized Set> entrySet() { + long sizeLocation = writeEvent(ENTRY_SET); // get the data back from the server - ByteBufferBytes bufferBytes = blockingFetch(transactionId); + Bytes in = blockingFetch(sizeLocation); - int size = bufferBytes.readInt(); - - ArrayList result = new ArrayList(); - - for (int i = 0; i < size; i++) { - result.add(readValue(bufferBytes)); - } - return result; - } - - @NotNull - @Override - public Set> entrySet() { - target.writeByte(ENTRY_SET.ordinal()); - - long transactionId = nextUniqueTransaction(); - - // get the data back from the server - ByteBufferBytes bufferBytes = blockingFetch(transactionId); - - int size = bufferBytes.readInt(); + long size = in.readStopBit(); Set> result = new HashSet>(); for (int i = 0; i < size; i++) { - K k = readKey(bufferBytes); - V v = readValue(bufferBytes); + K k = keyValueSerializer.readKey(in); + V v = keyValueSerializer.readValue(in); result.add(new Entry(k, v)); - } - return result; } - private void writeValue(V value, ThreadLocalCopies copies) { - - VW valueWriter = valueWriterProvider.get(copies, originalValueWriter); - copies = valueWriterProvider.getCopies(copies); - long valueSize; - MetaBytesWriter metaValueWriter = null; - Byteable valueAsByteable = null; - - if (value instanceof Byteable) { - valueAsByteable = (Byteable) value; - valueSize = valueAsByteable.maxSize(); - - target.writeLong(valueSize); - target.write(valueAsByteable.bytes()); - } else { - copies = valueWriterProvider.getCopies(copies); - valueWriter = valueWriterProvider.get(copies, originalValueWriter); - copies = metaValueWriterProvider.getCopies(copies); - metaValueWriter = metaValueWriterProvider.get( - copies, originalMetaValueWriter, valueWriter, value); - valueSize = metaValueWriter.size(valueWriter, value); - target.writeLong(valueSize); - metaValueWriter.write(valueWriter, target, value); - } + private long writeEvent(EventId event) { + out.clear(); + out.buffer().clear(); + out.write((byte) event.ordinal()); + long sizeLocation = markSizeLocation(); + return sizeLocation; } - /** - * write the keysize and the key to the the {@code target} buffer - * - * @param key the key of the map - * @param copies - */ - private void writeKey(K key, final ThreadLocalCopies copies) { - KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop); - - MKI metaKeyWriter = - metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); - - long keySize = metaKeyWriter.size(keyWriter, key); - - keySizeMarshaller.writeSize(target, keySize); - metaKeyWriter.write(keyWriter, target, key); - } - - public Buffer buffer() { Buffer result = sourceBuffer.get(); @@ -518,12 +354,12 @@ public class StatelessMapClient, volatile ByteBufferBytes buffer = null; - @Override + public void set(ByteBufferBytes source) { buffer = source; } - @Override + public ByteBufferBytes get() { return buffer; } @@ -540,7 +376,7 @@ public class StatelessMapClient, * @param transactionId * @return */ - private ByteBufferBytes blockingFetch(long transactionId) { + private Bytes blockingFetchWithTransaction(long transactionId) { final Buffer buffer = buffer(); buffer.set(null); @@ -585,7 +421,9 @@ public class StatelessMapClient, public final V setValue(V newValue) { V oldValue = value; + StatelessMapClient.this.put((K) getKey(), (V) newValue); + return oldValue; } @@ -612,9 +450,103 @@ public class StatelessMapClient, public final String toString() { return getKey() + "=" + getValue(); } + } + private Bytes blockingFetch(long sizeLocation) { + try { + return blockingFetchThrowable(sizeLocation); + } catch (IOException e) { + throw new RuntimeIOException(e); + } } + + + private Bytes blockingFetchThrowable(long sizeLocation) throws IOException { + long transactionId = nextUniqueTransaction(); + + LOG.info("sending data with transactionId=" + transactionId); + out.writeLong(transactionId); + writeSizeAt(sizeLocation); + + + out.buffer().limit((int) out.position()); + while (out.buffer().remaining() > 0) { + int write = clientChannel.write(out.buffer()); + System.out.println(write); + } + + in.clear(); + in.buffer().clear(); + + // read size + while (in.buffer().position() < 2) { + clientChannel.read(in.buffer()); + } + + int size = in.readUnsignedShort(); + + while (in.buffer().position() < size) { + clientChannel.read(in.buffer()); + } + + boolean isException = in.readBoolean(); + + long inTransactionId = in.readLong(); + + if (inTransactionId != transactionId) { + throw new IllegalStateException("the received transaction-id=" + inTransactionId + + ", does not match the expected transaction-id=" + transactionId); + } + + if (isException) { + int len = in.readUnsignedShort(); + + in.limit(in.position() + len); + + Object o = in.readObject(); + RuntimeException runtimeException = (RuntimeException) o; + throw runtimeException; + } + + + return in; + } + + private void writeSizeAt(long locationOfSize) { + long size = out.position() - locationOfSize; + out.writeUnsignedShort(locationOfSize, (int) size - 2); + } + + private long markSizeLocation() { + long position = out.position(); + out.skip(2); + return position; + } + + private void writeKey(K key) { + keyValueSerializer.writeKey(key, out); + } + + private V readKey(final long sizeLocation) { + return keyValueSerializer.readValue(blockingFetch(sizeLocation)); + } + + private void writeValue(V value) { + keyValueSerializer.writeValue(value, out); + } + + + private Set readKeySet(Bytes in) { + long size = in.readStopBit(); + final HashSet result = new HashSet<>(); + + for (long i = 0; i < size; i++) { + result.add(keyValueSerializer.readKey(in)); + } + return result; + } + } @@ -624,3 +556,10 @@ interface Buffer { ByteBufferBytes get(); } + +class RuntimeIOException extends RuntimeException { + + public RuntimeIOException(IOException e) { + super(e); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java new file mode 100644 index 000000000..71289cd32 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java @@ -0,0 +1,340 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.model.constraints.NotNull; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +/** + * @author Rob Austin. + */ +class StatelessServerConnector { + + + private final KeyValueSerializer keyValueSerializer; + private final ChronicleMap map; + + + public StatelessServerConnector(KeyValueSerializer keyValueSerializer, ChronicleMap map) { + this.keyValueSerializer = keyValueSerializer; + this.map = map; + } + + + public void processStatelessEvent(byte eventId, @NotNull Bytes in, @NotNull Bytes out) { + + + StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId]; + switch (event) { + + case LONG_SIZE: + longSize(in, out); + break; + + case IS_EMPTY: + isEmpty(in, out); + break; + + case CONTAINS_KEY: + containsKey(in, out); + break; + case CONTAINS_VALUE: + containsValue(in, out); + break; + case GET: + get(in, out); + break; + case PUT: + put(in, out); + break; + case REMOVE: + remove(in, out); + break; + case CLEAR: + clear(in, out); + break; + case KEY_SET: + keySet(in, out); + break; + case VALUES: + values(in, out); + break; + case ENTRY_SET: + entrySet(in, out); + break; + case REPLACE: + replace(in, out); + break; + case REPLACE_WITH_OLD_AND_NEW_VALUE: + replaceWithOldAndNew(in, out); + break; + case PUT_IF_ABSENT: + putIfAbsent(in, out); + break; + case REMOVE_WITH_VALUE: + removeWithValue(in, out); + break; + + case SIZE: + size(in, out); + break; + + default: + throw new IllegalStateException("unsupported event=" + event); + + } + + } + + private void removeWithValue(Bytes in, Bytes out) { + boolean result = map.remove(readKey(in), readValue(in)); + out.writeBoolean(result); + } + + private void replaceWithOldAndNew(Bytes in, Bytes out) { + + final K key = readKey(in); + V oldValue = readValue(in); + V newValue = readValue(in); + + long sizeLocation = reflectTransactionId(in, out); + try { + map.replace(key, oldValue, newValue); + } catch (RuntimeException e) { + writeException(e, out); + writeSizeAndFlags(sizeLocation, true, out); + return; + } + writeSizeAndFlags(sizeLocation, false, out); + } + + + public void longSize(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + out.writeLong(map.longSize()); + writeSizeAndFlags(sizeLocation, false, out); + } + + public void size(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + out.writeInt(map.size()); + writeSizeAndFlags(sizeLocation, false, out); + } + + public void isEmpty(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + out.writeBoolean(map.isEmpty()); + writeSizeAndFlags(sizeLocation, false, out); + } + + public void containsKey(Bytes in, Bytes out) { + K k = readKey(in); + long sizeLocation = reflectTransactionId(in, out); + out.writeBoolean(map.containsKey(k)); + writeSizeAndFlags(sizeLocation, false, out); + } + + public void containsValue(Bytes in, Bytes out) { + V v = readValue(in); + long sizeLocation = reflectTransactionId(in, out); + out.writeBoolean(map.containsValue(v)); + writeSizeAndFlags(sizeLocation, false, out); + } + + public void get(Bytes in, Bytes out) { + K k = readKey(in); + long sizeLocation = reflectTransactionId(in, out); + try { + writeValue(map.get(k), out); + } catch (RuntimeException e) { + writeException(e, out); + writeSizeAndFlags(sizeLocation, true, out); + return; + } + writeSizeAndFlags(sizeLocation, false, out); + + } + + private void writeException(RuntimeException e, Bytes out) { + + long start = out.position(); + out.skip(2); + out.writeObject(e); + long len = out.position() - (start + 2L); + out.writeUnsignedShort(start,(int) len); + + } + + public void put(Bytes in, Bytes out) { + K k = readKey(in); + V v = readValue(in); + long sizeLocation = reflectTransactionId(in, out); + writeValue(map.put(k, v), out); + writeSizeAndFlags(sizeLocation, false, out); + } + + public void remove(Bytes in, Bytes out) { + final V value = map.remove(readKey(in)); + long sizeLocation = reflectTransactionId(in, out); + writeValue(value, out); + writeSizeAndFlags(sizeLocation, false, out); + } + + public void putAll(Bytes in, Bytes out) { + map.putAll(readEntries(in)); + long sizeLocation = reflectTransactionId(in, out); + writeSizeAndFlags(sizeLocation, false, out); + } + + private Map readEntries(Bytes in) { + + long size = in.readStopBit(); + final HashMap result = new HashMap(); + + for (long i = 0; i < size; i++) { + result.put(readKey(in), readValue(in)); + } + return result; + + } + + public void clear(Bytes in, Bytes out) { + map.clear(); + long sizeLocation = reflectTransactionId(in, out); + writeSizeAndFlags(sizeLocation, false, out); + } + + + public void keySet(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + + final Set ks = map.keySet(); + out.writeStopBit(ks.size()); + for (K key : ks) { + writeKey(key, out); + } + writeSizeAndFlags(sizeLocation, false, out); + } + + public void values(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + final Collection values = map.values(); + out.writeStopBit(values.size()); + for (final V value : values) { + writeValue(value, out); + } + writeSizeAndFlags(sizeLocation, false, out); + } + + public void entrySet(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + + final Set> entries = map.entrySet(); + out.writeStopBit(entries.size()); + for (Map.Entry e : entries) { + writeKey(e.getKey(), out); + writeValue(e.getValue(), out); + } + } + + public void putIfAbsent(Bytes in, Bytes out) { + K key = readKey(in); + V v = readValue(in); + long sizeLocation = reflectTransactionId(in, out); + writeValue(map.putIfAbsent(key, v), out); + writeSizeAndFlags(sizeLocation, false, out); + } + + + public void replace(Bytes in, Bytes out) { + K k = readKey(in); + V v = readValue(in); + + long sizeLocation = reflectTransactionId(in, out); + map.replace(k, v); + writeSizeAndFlags(sizeLocation, false, out); + } + + /** + * write the keysize and the key to the the {@code target} buffer + * + * @param key the key of the map + */ + private void writeKey(K key, Bytes out) { + keyValueSerializer.writeKey(key, out); + } + + private long reflectTransactionId(Bytes in, Bytes out) { + long sizeLocation = out.position(); + out.skip(3); + long transactionId = in.readLong(); + out.writeLong(transactionId); + return sizeLocation; + } + + private void writeValue(V value, final Bytes out) { + keyValueSerializer.writeValue(value, out); + } + + private K readKey(Bytes in) { + return keyValueSerializer.readKey(in); + } + + private V readValue(Bytes in) { + return keyValueSerializer.readValue(in); + } + + private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { + long size = out.position() - locationOfSize; + out.writeUnsignedShort(0L, (int) size); + out.writeBoolean(2L, isException); + } +} + +class KeyValueSerializer { + + private final Serializer valueSerializer; + private final Serializer keySerializer; + + KeyValueSerializer(SerializationBuilder keySerializer, + SerializationBuilder valueSerializer) { + this.keySerializer = new Serializer(keySerializer); + this.valueSerializer = new Serializer(valueSerializer); + } + + V readValue(Bytes in) { + if (in.readBoolean()) + return null; + return valueSerializer.readMarshallable(in); + } + + K readKey(Bytes in) { + if (in.readBoolean()) + return null; + + return keySerializer.readMarshallable(in); + } + + /** + * write the keysize and the key to the the {@code target} buffer + * + * @param key the key of the map + */ + void writeKey(K key, Bytes out) { + + out.writeBoolean(key == null); + if (key != null) + keySerializer.writeMarshallable(key, out); + } + + void writeValue(V value, Bytes out) { + out.writeBoolean(value == null); + if (value != null) + valueSerializer.writeMarshallable(value, out); + } + + +} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 61a013ce3..be8fd7b70 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map; import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,10 +33,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import static java.nio.channels.SelectionKey.*; +import static net.openhft.chronicle.map.StatelessMapClient.EventId.HEARTBEAT; /** - * Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a socket - * connection

{@see net.openhft.map.OutSocketReplicator} + * Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a + * socket connection

{@see net.openhft.map.OutSocketReplicator} * * @author Rob Austin. */ @@ -43,6 +45,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName()); private static final int BUFFER_SIZE = 0x100000; // 1MB + public static final int STATELESS_CLIENT = -127; + public static final byte NOT_SET = (byte) HEARTBEAT.ordinal(); private final SelectionKey[] selectionKeysStore = new SelectionKey[Byte.MAX_VALUE + 1]; // used to instruct the selector thread to set OP_WRITE on a key correlated by the bit index in the @@ -57,22 +61,28 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { private final Replica.EntryExternalizable externalizable; private final TcpReplicationConfig replicationConfig; + private final StatelessServerConnector statelessServerConnector; + private long selectorTimeout; /** - * @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, it can not be - * smaller than the size of and entry, if it is set smaller the buffer will over - * flow, it can be larger then the entry, but setting it too large reduces the - * workable space in the buffer. + * @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, + * it can not be smaller than the size of and entry, if it is + * set smaller the buffer will over flow, it can be larger then + * the entry, but setting it too large reduces the workable + * space in the buffer. + * @param statelessServerConnector * @throws IOException */ TcpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable externalizable, @NotNull final TcpReplicationConfig replicationConfig, - final int maxEntrySizeBytes) throws IOException { + final int maxEntrySizeBytes, + StatelessServerConnector statelessServerConnector) throws IOException { super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(), maxEntrySizeBytes); + this.statelessServerConnector = statelessServerConnector; final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig(); long throttleBucketInterval = TimeUnit.MILLISECONDS.convert(throttlingConfig @@ -251,7 +261,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } /** - * check to see if we have lost connection with the remote node and if we have attempts a reconnect. + * check to see if we have lost connection with the remote node and if we have attempts a + * reconnect. * * @param key the key relating to the heartbeat that we are checking * @param approxTimeOutTime the approximate time in milliseconds @@ -374,8 +385,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } /** - * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear all the - * old bootstrap information + * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear + * all the old bootstrap information * * @param key the nio SelectionKey */ @@ -387,7 +398,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } /** - * used to exchange identifiers and timestamps and heartbeat intervals between the server and client + * used to exchange identifiers and timestamps and heartbeat intervals between the server and + * client * * @param key the SelectionKey relating to the this cha * @param socketChannel @@ -404,6 +416,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { final byte remoteIdentifier = reader.identifierFromBuffer(); + if (remoteIdentifier == STATELESS_CLIENT) { + attached.handShakingComplete = true; + return; + } + if (remoteIdentifier == Byte.MIN_VALUE) return; @@ -425,7 +442,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { final SocketAddress remoteAddress = socketChannel.getRemoteAddress(); - if ( (identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier, + if ((identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier, remoteAddress)) || remoteIdentifier == localIdentifier) // throwing this exception will cause us to disconnect, both the client and server @@ -477,7 +494,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // now we're finished we can get on with reading the entries attached.handShakingComplete = true; attached.remoteModificationIterator.dirtyEntries(attached.remoteBootstrapTimestamp); - reader.entriesFromBuffer(); + reader.entriesFromBuffer(attached); } } @@ -531,17 +548,17 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { attached.entryReader.lastHeartBeatReceived = approxTime; if (attached.isHandShakingComplete()) { - attached.entryReader.entriesFromBuffer(); + attached.entryReader.entriesFromBuffer(attached); } else { doHandShaking(key, socketChannel); } } /** - * sets interestOps to "selector keys",The change to interestOps much be on the same thread as the - * selector. This class, allows via {@link AbstractChannelReplicator .KeyInterestUpdater#set(int)} to - * holds a pending change in interestOps ( via a bitset ), this change is processed later on the same - * thread as the selector + * sets interestOps to "selector keys",The change to interestOps much be on the same thread as + * the selector. This class, allows via {@link AbstractChannelReplicator + * .KeyInterestUpdater#set(int)} to holds a pending change in interestOps ( via a bitset ), + * this change is processed later on the same thread as the selector */ private static class KeyInterestUpdater { @@ -572,8 +589,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } /** - * @param keyIndex the index of the key that has changed, the list of keys is provided by the - * constructor {@link KeyInterestUpdater(int, SelectionKey[])} + * @param keyIndex the index of the key that has changed, the list of keys is provided by + * the constructor {@link KeyInterestUpdater(int, SelectionKey[])} */ public void set(int keyIndex) { changeOfOpWriteRequired.set(keyIndex); @@ -773,12 +790,19 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } + /** + * @return the buffer messages can be written to + */ + Bytes buffer() { + return in; + } + /** * writes the timestamp into the buffer * * @param localIdentifier the current nodes identifier */ - void identifierToBuffer(final int localIdentifier) { + void identifierToBuffer(final byte localIdentifier) { in.writeByte(localIdentifier); } @@ -792,7 +816,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } /** - * writes all the entries that have changed, to the buffer which will later be written to TCP/IP + * writes all the entries that have changed, to the buffer which will later be written to + * TCP/IP * * @param modificationIterator a record of which entries have modification * @param selectionKey @@ -804,7 +829,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { final SocketChannel socketChannel = (SocketChannel) selectionKey.channel(); final Attached attached = (Attached) selectionKey.attachment(); - // this can occur when new CHM's are added to a cluster + // this can occur when new map are added to a channel final boolean handShakingComplete = attached.isHandShakingComplete(); for (; ; ) { @@ -867,9 +892,15 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } /** - * used to send an single zero byte if we have not send any data for up to the localHeartbeatInterval + * used to send an single zero byte if we have not send any data for up to the + * localHeartbeatInterval */ private void writeHeartbeatToBuffer() { + + // denotes the state - 0 for a heartbeat + in.writeByte(HEARTBEAT.ordinal()); + + // denotes the size in bytes in.writeUnsignedShort(0); } @@ -879,8 +910,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { /** - * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE will get - * added back in as soon as we have data to write + * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE + * will get added back in as soon as we have data to write * * @param socketChannel the socketChannel we wish to stop writing to * @param attached data associated with the socketChannels key @@ -914,8 +945,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { private final ByteBuffer in; private final ByteBufferBytes out; public long lastHeartBeatReceived = System.currentTimeMillis(); - // we use Integer.MIN_VALUE as N/A - private int sizeOfNextEntry = Integer.MIN_VALUE; + + private int sizeInBytes; + private byte state; private TcpSocketChannelEntryReader() { in = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes); @@ -943,45 +975,79 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { /** * reads entries from the buffer till empty * + * @param attached * @throws InterruptedException */ - private void entriesFromBuffer() throws InterruptedException, IOException { - + private void entriesFromBuffer(Attached attached) throws InterruptedException, IOException { for (; ; ) { out.limit(in.position()); // its set to MIN_VALUE when it should be read again - if (sizeOfNextEntry == Integer.MIN_VALUE) { - if (out.remaining() < SIZE_OF_SHORT) { + if (state == NOT_SET) { + if (out.remaining() < SIZE_OF_SHORT + 1 ) { return; } - int value = out.readUnsignedShort(); + // state is used for both heartbeat and stateless + state = out.readByte(); - // this is the heartbeat - if (value == 0) + + + sizeInBytes = out.readUnsignedShort(); + + if (state == 0 && sizeInBytes != 0) { + int i = 1; + } + + // this is the : + // -- heartbeat if its 0 + // -- stateful update if its 1 + // -- the id of the stateful event + if (state == NOT_SET) continue; - - sizeOfNextEntry = value; } - if (out.remaining() < sizeOfNextEntry) { + + if (out.remaining() < sizeInBytes) { return; } - final long nextEntryPos = out.position() + sizeOfNextEntry; + if (sizeInBytes == 0) { + int j = 1; + } + + + final long nextEntryPos = out.position() + sizeInBytes; final long limit = out.limit(); out.limit(nextEntryPos); - externalizable.readExternalEntry(out); + + boolean isStateless = (state != 1); + + if (isStateless) { + if (statelessServerConnector == null) { + + LOG.error("", new IllegalArgumentException("received an event " + + "from a stateless map, stateless maps are not " + + "currently supported when using Chronicle Channels")); + } else + + statelessServerConnector.processStatelessEvent(state, + out, attached.entryWriter.buffer()); + // to allow the sizeInBytes to be read the next time around + + + } else + externalizable.readExternalEntry(out); out.limit(limit); // skip onto the next entry out.position(nextEntryPos); - // to allow the sizeOfNextEntry to be read the next time around - sizeOfNextEntry = Integer.MIN_VALUE; + + state = NOT_SET; + sizeInBytes = 0; } } @@ -1016,7 +1082,12 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { * @return the timestamp or -1 if unsuccessful */ long remoteBootstrapTimestamp() { - return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE; + if (out.remaining() >= 8) + return out.readLong(); + else + return Long.MIN_VALUE; + + // return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE; } public long remoteHeartbeatIntervalFromBuffer() { diff --git a/src/main/java/net/openhft/chronicle/map/serialization/CopyingMetaBytesInterop.java b/src/main/java/net/openhft/chronicle/map/serialization/CopyingMetaBytesInterop.java index a8fc3d368..79e6111e3 100644 --- a/src/main/java/net/openhft/chronicle/map/serialization/CopyingMetaBytesInterop.java +++ b/src/main/java/net/openhft/chronicle/map/serialization/CopyingMetaBytesInterop.java @@ -16,14 +16,14 @@ package net.openhft.chronicle.map.serialization; +import net.openhft.chronicle.map.threadlocal.Provider; +import net.openhft.chronicle.map.threadlocal.StatefulCopyable; +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.DirectBytes; import net.openhft.lang.io.DirectStore; import net.openhft.lang.io.serialization.BytesMarshaller; import net.openhft.lang.io.serialization.JDKObjectSerializer; -import net.openhft.chronicle.map.threadlocal.Provider; -import net.openhft.chronicle.map.threadlocal.StatefulCopyable; -import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import java.io.IOException; import java.io.ObjectInputStream; @@ -190,7 +190,7 @@ public abstract class CopyingMetaBytesInterop implements MetaBytesInterop< forBytesWriter = new ForBytesWriter(); } - private Bytes obtain(long maxSize) { + Bytes obtain(long maxSize) { DirectBytes buf; if ((buf = buffer) != null) { if (maxSize <= buf.capacity()) { diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index 4c1abddc9..ae60b1915 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -21,8 +21,7 @@ import java.io.File; import java.io.IOException; import java.io.RandomAccessFile; import java.net.InetSocketAddress; - -import static java.util.concurrent.TimeUnit.SECONDS; +import java.util.concurrent.TimeUnit; /** * @author Rob Austin. @@ -69,7 +68,7 @@ public class Builder { final int serverPort, final InetSocketAddress... endpoints) throws IOException { TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints) - .heartBeatInterval(1L, SECONDS); + .heartBeatInterval(1L, TimeUnit.SECONDS); return ChronicleMapBuilder.of(kClass, vClass) .entries(20000L) .replicators(identifier, tcpConfig); diff --git a/src/test/java/net/openhft/chronicle/map/OSResizesMain.java b/src/test/java/net/openhft/chronicle/map/OSResizesMain.java index 3792462b5..34f8907a7 100644 --- a/src/test/java/net/openhft/chronicle/map/OSResizesMain.java +++ b/src/test/java/net/openhft/chronicle/map/OSResizesMain.java @@ -34,9 +34,8 @@ public class OSResizesMain { public static void main(String[] args) throws IOException, InterruptedException { File file = File.createTempFile("over-sized", "deleteme"); ChronicleMap map = ChronicleMapBuilder.of(String.class, String.class) - - .entrySize(1024 * 1024) - .entries(1024 * 1024) + .entrySize(100 * 1024 * 1024) + .entries(1000000) .create(file); for (int i = 0; i < 10000; i++) { char[] chars = new char[i]; @@ -48,6 +47,10 @@ public class OSResizesMain { file.length() / 1e9, run("du", "-h", file.getAbsolutePath()).split("\\s")[0], run("grep", "over-sized", "/proc/" + Jvm.getProcessId() + "/maps").split("\\s")[0]); + // show up in top. + long time = System.currentTimeMillis(); + while (time + 20000 > System.currentTimeMillis()) + Thread.yield(); map.close(); file.delete(); } diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java new file mode 100644 index 000000000..f6f1a00df --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -0,0 +1,240 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshallable; +import net.openhft.lang.model.constraints.NotNull; +import org.junit.Assert; +import org.junit.Test; + +import java.io.*; +import java.nio.ByteBuffer; +import java.util.Collections; + +public class SerializerTest { + + @Test + public void testValueMarshallable() throws Exception { + testReadWriteValue("Test"); + testReadWriteValue(1); + testReadWriteValue(1L); + testReadWriteValue(1.0); + testReadWriteValue(1.0f); + testReadWriteValue(Collections.singleton("Test")); + testReadWriteValue(Collections.EMPTY_MAP); + testReadWriteValue(new MyTestClass(3)); + testReadWriteValue(new MyTestClassMarshallable(3)); + + testReadWriteValue(new MyTestClassExternalizable(3)); + testReadWriteValue(new MyTestClassObjectGraph(3)); + } + + + @Test + public void testKeyMarshallable() throws Exception { + testReadWriteKey("Test"); + testReadWriteKey(1); + testReadWriteKey(1L); + testReadWriteKey(1.0); + testReadWriteKey(1.0f); + testReadWriteKey(Collections.singleton("Test")); + testReadWriteKey(Collections.EMPTY_MAP); + testReadWriteKey(new MyTestClass(3)); + testReadWriteKey(new MyTestClassMarshallable(3)); + + testReadWriteKey(new MyTestClassExternalizable(3)); + testReadWriteKey(new MyTestClassObjectGraph(3)); + } + + + public void testReadWriteValue(Object value) throws Exception { + + Class valueClass = value.getClass(); + + final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocateDirect(1024)); + ByteBufferBytes in = out.slice(); + + ChronicleMapBuilder builder = ChronicleMapBuilder.of(Integer.class, valueClass); + builder.preMapConstruction(); + + Serializer v = new Serializer(builder.valueBuilder); + + v.writeMarshallable(value, out); + + long position = out.position(); + in.limit(position); + + + Object actual = v.readMarshallable(in); + Assert.assertEquals(actual, value); + } + + + public void testReadWriteKey(Object key) throws Exception { + + Class clazz = key.getClass(); + + final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocateDirect(1024)); + ByteBufferBytes in = out.slice(); + + ChronicleMapBuilder builder = ChronicleMapBuilder.of(clazz, Integer.class); + builder.preMapConstruction(); + { + Serializer v = new Serializer(builder.keyBuilder); + + + v.writeMarshallable(key, out); + + long position = out.position(); + in.limit(position); + + Object actual = v.readMarshallable(in); + Assert.assertEquals(actual, key); + + } + + + } + + public static class MyTestClassExternalizable implements Externalizable { + int a; + + public MyTestClassExternalizable() { + } + + + MyTestClassExternalizable(int a) { + this.a = a; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + MyTestClassExternalizable that = (MyTestClassExternalizable) o; + + if (a != that.a) return false; + + return true; + } + + @Override + public int hashCode() { + return a; + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException { + out.write(a); + } + + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + a = in.readInt(); + } + } + + public static class MyTestClass implements Serializable { + int a; + + MyTestClass(int a) { + this.a = a; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + MyTestClass myTestClass = (MyTestClass) o; + + if (a != myTestClass.a) return false; + + return true; + } + + @Override + public int hashCode() { + return a; + } + } + + public static class MyTestClassMarshallable implements BytesMarshallable { + int a; + + public MyTestClassMarshallable() { + } + + public MyTestClassMarshallable(int a) { + this.a = a; + } + + public void setA(int a) { + this.a = a; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + MyTestClassMarshallable that = (MyTestClassMarshallable) o; + + if (a != that.a) return false; + + return true; + } + + @Override + public int hashCode() { + return a; + } + + public void readMarshallable(@NotNull Bytes in) throws IllegalStateException { + a = in.readInt(); + } + + /** + * write an object to bytes + * + * @param out to write to + */ + public void writeMarshallable(@NotNull Bytes out) { + out.writeInt(a); + } + } + + public static class MyTestClassObjectGraph implements Serializable { + + MyTestClass delegate; + + public MyTestClassObjectGraph() { + + } + + MyTestClassObjectGraph(int a) { + delegate = new MyTestClass(a); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + MyTestClassObjectGraph that = (MyTestClassObjectGraph) o; + + if (delegate != null ? !delegate.equals(that.delegate) : that.delegate != null) + return false; + + return true; + } + + @Override + public int hashCode() { + return delegate != null ? delegate.hashCode() : 0; + } + } + + +} \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java new file mode 100644 index 000000000..0622ab6ae --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -0,0 +1,41 @@ +package net.openhft.chronicle.map; + +import org.junit.Test; +import org.testng.Assert; + +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.TcpReplicationConfig.of; + +/** + * @author Rob Austin. + */ +public class StatelessClientTest { + + + @Test + public void test() throws IOException, InterruptedException { + + ChronicleMapBuilder builder = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 2, of(8076).heartBeatInterval(1L, SECONDS)); + { + final ChronicleMap map2a = + builder.create(getPersistenceFile()); + map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2 + } + + KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder.keyBuilder, builder.valueBuilder); + + InetSocketAddress remote = new InetSocketAddress("localhost", 8076); + StatelessMapClient map = new StatelessMapClient(keyValueSerializer, remote); + + Assert.assertEquals("EXAMPLE-10", map.get(10)); + Assert.assertEquals(1, map.size()); + } +} +