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 504d08b58..5dc965fa1 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -805,7 +805,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/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java index 6e2fd77b4..397a69ee3 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java @@ -3,9 +3,14 @@ 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; import java.io.File; import java.io.IOException; +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; @@ -18,16 +23,23 @@ import static net.openhft.chronicle.map.StatelessMapClient.EventId.*; * * @author Rob Austin. */ -class StatelessMapClient implements ChronicleMap { +class StatelessMapClient { + + private static final Logger LOG = LoggerFactory.getLogger(StatelessMapClient.class); + + + private final SocketChannel clientChannel; + public static final ByteBuffer STATELESS_CLIENT_IDENTIFER = ByteBuffer.wrap(new byte[]{-127}); 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, @@ -46,23 +58,48 @@ class StatelessMapClient implements ChronicleMap { private long timeoutMs = TimeUnit.SECONDS.toMillis(20); - public StatelessMapClient(Bytes out, KeyValueSerializer keyValueSerializer) { - this.out = out; + public StatelessMapClient(final KeyValueSerializer keyValueSerializer, final InetSocketAddress remote) throws IOException { + this.keyValueSerializer = keyValueSerializer; + + clientChannel = SocketChannel.open(); + clientChannel.connect(remote); + + doHandShaking(); } - @Override + 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; } - @Override public void close() throws IOException { throw new UnsupportedOperationException("This is not supported in the " + this.getClass() .getSimpleName()); } - long nextUniqueTransaction() { long time = System.currentTimeMillis(); @@ -76,160 +113,142 @@ class StatelessMapClient implements ChronicleMap { } - private final Bytes out; + private final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocate(1024)); + private final ByteBufferBytes in = new ByteBufferBytes(ByteBuffer.allocate(1024)); private final KeyValueSerializer keyValueSerializer; - @Override - public V putIfAbsent(K key, V value) { - writeEvent(PUT_IF_ABSENT); + public synchronized V putIfAbsent(K key, V value) throws IOException { + + long sizeLocation = writeEvent(PUT_IF_ABSENT); writeKey(key); writeValue(value); - - return readKey(); + return readKey(sizeLocation); } - @Override - public boolean remove(Object key, Object value) { - writeEvent(REMOVE_WITH_VALUE); + public synchronized boolean remove(Object key, Object value) throws IOException { + long sizeLocation = writeEvent(REMOVE_WITH_VALUE); writeKey((K) key); writeValue((V) value); // get the data back from the server - return blockingFetch().readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public boolean replace(K key, V oldValue, V newValue) { - writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE); + + public synchronized boolean replace(K key, V oldValue, V newValue) throws IOException { + long sizeLocation = writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE); writeKey(key); writeValue(oldValue); writeValue(newValue); // get the data back from the server - return blockingFetch().readBoolean(); - } - - private void writeKey(K key) { - keyValueSerializer.writeKey(key, out); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public V replace(K key, V value) { - writeEvent(REPLACE); + public synchronized V replace(K key, V value) throws IOException { + long sizeLocation = writeEvent(REPLACE); writeKey(key); writeValue(value); // get the data back from the server - return readKey(); + return readKey(sizeLocation); } - @Override - public int size() { + public synchronized int size() throws IOException { - writeEvent(SIZE); + long sizeLocation = writeEvent(SIZE); // get the data back from the server - return blockingFetch().readInt(); + return blockingFetch(sizeLocation).readInt(); } - @Override - public boolean isEmpty() { - writeEvent(IS_EMPTY); + public synchronized boolean isEmpty() throws IOException { + long sizeLocation = writeEvent(IS_EMPTY); // get the data back from the server - return blockingFetch().readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public boolean containsKey(Object key) { - writeEvent(CONTAINS_KEY); + + public synchronized boolean containsKey(Object key) throws IOException { + long sizeLocation = writeEvent(CONTAINS_KEY); writeKey((K) key); // get the data back from the server - return blockingFetch().readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public boolean containsValue(Object value) { - writeEvent(CONTAINS_VALUE); + + public synchronized boolean containsValue(Object value) throws IOException { + long sizeLocation = writeEvent(CONTAINS_VALUE); writeValue((V) value); // get the data back from the server - return blockingFetch().readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } - @Override - public long longSize() { - writeEvent(LONG_SIZE); + public synchronized long longSize() throws IOException { + long sizeLocation = writeEvent(LONG_SIZE); // get the data back from the server - return blockingFetch().readLong(); + return blockingFetch(sizeLocation).readLong(); } - @Override - public V get(Object key) { - writeEvent(GET); + + public synchronized V get(Object key) throws IOException { + long sizeLocation = writeEvent(GET); writeKey((K) key); + // get the data back from the server - return readKey(); + return readKey(sizeLocation); } - @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) { - writeEvent(PUT); + public synchronized V put(K key, V value) throws IOException { + + long sizeLocation = writeEvent(PUT); writeKey(key); writeValue(value); // get the data back from the server - return readKey(); + return readKey(sizeLocation); } - private V readKey() { - return keyValueSerializer.readValue(blockingFetch()); - } - private void writeValue(V value) { - keyValueSerializer.writeValue(value, out); - } - - @Override - public V remove(Object key) { - writeEvent(REMOVE); + public synchronized V remove(Object key) throws IOException { + long sizeLocation = writeEvent(REMOVE); writeKey((K) key); + // get the data back from the server - return readKey(); + return readKey(sizeLocation); } - @Override - public void putAll(Map map) { - writeEvent(PUT_ALL); - writeEntries(map); + public synchronized void putAll(Map map) throws IOException { - // get the data back from the server - blockingFetch(); + for (Map.Entry e : map.entrySet()) { + put(e.getKey(), e.getValue()); + } } @@ -247,42 +266,32 @@ class StatelessMapClient implements ChronicleMap { } } - @Override - public void clear() { - writeEvent(CLEAR); + + public synchronized void clear() throws IOException { + long sizeLocation = writeEvent(CLEAR); // get the data back from the server - blockingFetch(); + blockingFetch(sizeLocation); } @NotNull - @Override - public Set keySet() { - writeEvent(KEY_SET); + + public synchronized Set keySet() throws IOException { + long sizeLocation = writeEvent(KEY_SET); // get the data back from the server - return readKeySet(blockingFetch()); - } - - 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; + return readKeySet(blockingFetch(sizeLocation)); } @NotNull - @Override - public Collection values() { - writeEvent(VALUES); + + public synchronized Collection values() throws IOException { + long sizeLocation = writeEvent(VALUES); // get the data back from the server - final Bytes in = blockingFetch(); + final Bytes in = blockingFetch(sizeLocation); final long size = in.readStopBit(); @@ -299,13 +308,12 @@ class StatelessMapClient implements ChronicleMap { @NotNull - @Override - public Set> entrySet() { - writeEvent(ENTRY_SET); + public synchronized Set> entrySet() throws IOException { + long sizeLocation = writeEvent(ENTRY_SET); // get the data back from the server - Bytes in = blockingFetch(); + Bytes in = blockingFetch(sizeLocation); long size = in.readStopBit(); @@ -321,8 +329,13 @@ class StatelessMapClient implements ChronicleMap { } - private void writeEvent(EventId event) { + private long writeEvent(EventId event) { + out.clear(); + out.buffer().clear(); + out.write((byte) event.ordinal()); + long sizeLocation = markSizeLocation(); + return sizeLocation; } public Buffer buffer() { @@ -337,12 +350,12 @@ class StatelessMapClient implements ChronicleMap { volatile ByteBufferBytes buffer = null; - @Override + public void set(ByteBufferBytes source) { buffer = source; } - @Override + public ByteBufferBytes get() { return buffer; } @@ -359,7 +372,7 @@ class StatelessMapClient implements ChronicleMap { * @param transactionId * @return */ - private Bytes blockingFetch(long transactionId) { + private Bytes blockingFetchWithTransaction(long transactionId) { final Buffer buffer = buffer(); buffer.set(null); @@ -404,7 +417,11 @@ class StatelessMapClient implements ChronicleMap { public final V setValue(V newValue) { V oldValue = value; - StatelessMapClient.this.put((K) getKey(), (V) newValue); + try { + StatelessMapClient.this.put((K) getKey(), (V) newValue); + } catch (IOException e) { + LOG.error("", e); + } return oldValue; } @@ -431,15 +448,80 @@ class StatelessMapClient implements ChronicleMap { public final String toString() { return getKey() + "=" + getValue(); } - } - - private Bytes blockingFetch() { + private Bytes blockingFetch(long sizeLocation) throws IOException { long transactionId = nextUniqueTransaction(); + + LOG.info("sending data with transactionId=" + transactionId); out.writeLong(transactionId); - return blockingFetch(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 + 2) { + clientChannel.read(in.buffer()); + } + + long inTransactionId = in.readLong(); + + if (inTransactionId != transactionId) { + throw new IllegalStateException("the received transaction-id=" + inTransactionId + + ", does not match the expected transaction-id=" + transactionId); + } + + 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) throws IOException { + 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; + } + } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java index 0069e56c7..ca3f949e4 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java @@ -24,13 +24,11 @@ class StatelessServerConnector { } - public void marshall(@NotNull Bytes in, @NotNull Bytes out) { + public void processStatelessEvent(byte eventId, @NotNull Bytes in, @NotNull Bytes out) { - byte b = in.readByte(); - StatelessMapClient.EventId eventId = StatelessMapClient.EventId.values()[b]; - - switch (eventId) { + StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId]; + switch (event) { case LONG_SIZE: longSize(in, out); @@ -55,9 +53,6 @@ class StatelessServerConnector { case REMOVE: remove(in, out); break; - case PUT_ALL: - putAll(in, out); - break; case CLEAR: clear(in, out); break; @@ -83,73 +78,93 @@ class StatelessServerConnector { 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)); - reflectTransactionId(in, out); out.writeBoolean(result); } private void replaceWithOldAndNew(Bytes in, Bytes out) { + final K key = readKey(in); V oldValue = readValue(in); V newValue = readValue(in); - reflectTransactionId(in, out); + + long sizeLocation = reflectTransactionId(in, out); map.replace(key, oldValue, newValue); + + writeSizeAt(sizeLocation, out); } + public void longSize(Bytes in, Bytes out) { - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); out.writeLong(map.longSize()); + writeSizeAt(sizeLocation, out); } public void size(Bytes in, Bytes out) { - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); out.writeInt(map.size()); + writeSizeAt(sizeLocation, out); } public void isEmpty(Bytes in, Bytes out) { - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); out.writeBoolean(map.isEmpty()); + writeSizeAt(sizeLocation, out); } public void containsKey(Bytes in, Bytes out) { K k = readKey(in); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); out.writeBoolean(map.containsKey(k)); + writeSizeAt(sizeLocation, out); } public void containsValue(Bytes in, Bytes out) { V v = readValue(in); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); out.writeBoolean(map.containsValue(v)); + writeSizeAt(sizeLocation, out); } public void get(Bytes in, Bytes out) { K k = readKey(in); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); writeValue(map.get(k), out); + writeSizeAt(sizeLocation, out); } public void put(Bytes in, Bytes out) { K k = readKey(in); V v = readValue(in); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); writeValue(map.put(k, v), out); + writeSizeAt(sizeLocation, out); } public void remove(Bytes in, Bytes out) { final V value = map.remove(readKey(in)); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); writeValue(value, out); + writeSizeAt(sizeLocation, out); } public void putAll(Bytes in, Bytes out) { map.putAll(readEntries(in)); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); + writeSizeAt(sizeLocation, out); } private Map readEntries(Bytes in) { @@ -166,31 +181,34 @@ class StatelessServerConnector { public void clear(Bytes in, Bytes out) { map.clear(); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); + writeSizeAt(sizeLocation, out); } public void keySet(Bytes in, Bytes out) { - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); final Set ks = map.keySet(); out.writeStopBit(ks.size()); for (K key : ks) { writeKey(key, out); } + writeSizeAt(sizeLocation, out); } public void values(Bytes in, Bytes out) { - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); final Collection values = map.values(); out.writeStopBit(values.size()); for (final V value : values) { writeValue(value, out); } + writeSizeAt(sizeLocation, out); } public void entrySet(Bytes in, Bytes out) { - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); final Set> entries = map.entrySet(); out.writeStopBit(entries.size()); @@ -203,8 +221,9 @@ class StatelessServerConnector { public void putIfAbsent(Bytes in, Bytes out) { K key = readKey(in); V v = readValue(in); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); writeValue(map.putIfAbsent(key, v), out); + writeSizeAt(sizeLocation, out); } @@ -212,8 +231,9 @@ class StatelessServerConnector { K k = readKey(in); V v = readValue(in); - reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(in, out); map.replace(k, v); + writeSizeAt(sizeLocation, out); } /** @@ -225,8 +245,12 @@ class StatelessServerConnector { keyValueSerializer.writeKey(key, out); } - private void reflectTransactionId(Bytes in, Bytes out) { - out.writeLong(in.readLong()); + private long reflectTransactionId(Bytes in, Bytes out) { + long sizeLocation = out.position(); + out.skip(2); + long transactionId = in.readLong(); + out.writeLong(transactionId); + return sizeLocation; } private void writeValue(V value, final Bytes out) { @@ -241,6 +265,10 @@ class StatelessServerConnector { return keyValueSerializer.readValue(in); } + private void writeSizeAt(long locationOfSize, Bytes out) { + long size = out.position() - locationOfSize; + out.writeUnsignedShort((int) size); + } } class KeyValueSerializer { @@ -285,4 +313,5 @@ class KeyValueSerializer { 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/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/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java new file mode 100644 index 000000000..da229954a --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -0,0 +1,49 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.io.ByteBufferBytes; +import org.junit.Test; +import org.testng.Assert; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; + +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); + + final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocateDirect(1024)); + // ByteBufferBytes in = out.slice(); + + InetSocketAddress remote = new InetSocketAddress("localhost", 8076); + StatelessMapClient map = new StatelessMapClient(keyValueSerializer, remote); + + + Assert.assertEquals("EXAMPLE-10", map.get(10)); + Assert.assertEquals("EXAMPLE-10", map.get(10)); + + Assert.assertEquals(1, map.size()); + } +} +