diff --git a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java index 71289cd32..f6ffded95 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java @@ -3,10 +3,7 @@ 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; +import java.util.*; /** * @author Rob Austin. @@ -16,6 +13,7 @@ class StatelessServerConnector { private final KeyValueSerializer keyValueSerializer; private final ChronicleMap map; + private double maxEntrySizeBytes; public StatelessServerConnector(KeyValueSerializer keyValueSerializer, ChronicleMap map) { @@ -23,64 +21,59 @@ class StatelessServerConnector { this.map = map; } + public Work processStatelessEvent(byte eventId, @NotNull Bytes in, @NotNull Bytes out) { - public void processStatelessEvent(byte eventId, @NotNull Bytes in, @NotNull Bytes out) { + final StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId]; - - StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId]; switch (event) { case LONG_SIZE: - longSize(in, out); - break; + return longSize(in, out); case IS_EMPTY: - isEmpty(in, out); - break; + return isEmpty(in, out); case CONTAINS_KEY: - containsKey(in, out); - break; + return containsKey(in, out); + case CONTAINS_VALUE: - containsValue(in, out); - break; + return containsValue(in, out); + case GET: - get(in, out); - break; + return get(in, out); + case PUT: - put(in, out); - break; + return put(in, out); + case REMOVE: - remove(in, out); - break; + return remove(in, out); + case CLEAR: - clear(in, out); - break; + return clear(in, out); + case KEY_SET: - keySet(in, out); - break; + return keySet(in, out); + case VALUES: - values(in, out); - break; + return values(in, out); + case ENTRY_SET: - entrySet(in, out); - break; + return entrySet(in, out); + case REPLACE: - replace(in, out); - break; + return replace(in, out); + case REPLACE_WITH_OLD_AND_NEW_VALUE: - replaceWithOldAndNew(in, out); - break; + return replaceWithOldAndNew(in, out); + case PUT_IF_ABSENT: - putIfAbsent(in, out); - break; + return putIfAbsent(in, out); + case REMOVE_WITH_VALUE: - removeWithValue(in, out); - break; + return removeWithValue(in, out); case SIZE: - size(in, out); - break; + return size(in, out); default: throw new IllegalStateException("unsupported event=" + event); @@ -89,12 +82,13 @@ class StatelessServerConnector { } - private void removeWithValue(Bytes in, Bytes out) { + private Work removeWithValue(Bytes in, Bytes out) { boolean result = map.remove(readKey(in), readValue(in)); out.writeBoolean(result); + return null; } - private void replaceWithOldAndNew(Bytes in, Bytes out) { + private Work replaceWithOldAndNew(Bytes in, Bytes out) { final K key = readKey(in); V oldValue = readValue(in); @@ -106,45 +100,51 @@ class StatelessServerConnector { } catch (RuntimeException e) { writeException(e, out); writeSizeAndFlags(sizeLocation, true, out); - return; + return null; } writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void longSize(Bytes in, Bytes out) { + public Work longSize(Bytes in, Bytes out) { long sizeLocation = reflectTransactionId(in, out); out.writeLong(map.longSize()); writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void size(Bytes in, Bytes out) { + public Work size(Bytes in, Bytes out) { long sizeLocation = reflectTransactionId(in, out); out.writeInt(map.size()); writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void isEmpty(Bytes in, Bytes out) { + public Work isEmpty(Bytes in, Bytes out) { long sizeLocation = reflectTransactionId(in, out); out.writeBoolean(map.isEmpty()); writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void containsKey(Bytes in, Bytes out) { + public Work containsKey(Bytes in, Bytes out) { K k = readKey(in); long sizeLocation = reflectTransactionId(in, out); out.writeBoolean(map.containsKey(k)); writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void containsValue(Bytes in, Bytes out) { + public Work containsValue(Bytes in, Bytes out) { V v = readValue(in); long sizeLocation = reflectTransactionId(in, out); out.writeBoolean(map.containsValue(v)); writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void get(Bytes in, Bytes out) { + public Work get(Bytes in, Bytes out) { K k = readKey(in); long sizeLocation = reflectTransactionId(in, out); try { @@ -152,74 +152,78 @@ class StatelessServerConnector { } catch (RuntimeException e) { writeException(e, out); writeSizeAndFlags(sizeLocation, true, out); - return; + return null; } writeSizeAndFlags(sizeLocation, false, out); + return null; } - 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) { + public Work 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); + return null; } - public void remove(Bytes in, Bytes out) { + public Work remove(Bytes in, Bytes out) { final V value = map.remove(readKey(in)); long sizeLocation = reflectTransactionId(in, out); writeValue(value, out); writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void putAll(Bytes in, Bytes out) { + public Work putAll(Bytes in, Bytes out) { map.putAll(readEntries(in)); long sizeLocation = reflectTransactionId(in, out); writeSizeAndFlags(sizeLocation, false, out); + return null; } - 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) { + public Work clear(Bytes in, Bytes out) { map.clear(); long sizeLocation = reflectTransactionId(in, out); writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void keySet(Bytes in, Bytes out) { - long sizeLocation = reflectTransactionId(in, out); + public Work keySet(Bytes in, final Bytes out) { + final 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); + + final Iterator iterator = ks.iterator(); + + // this allows us to write more data than the buffer will allow + return new Work() { + + @Override + public boolean doWork(Bytes out) { + + while (iterator.hasNext()) { + + // we've filled up the buffer, so lets give another channel a chance to send + // some data, we don't know the max key size, we will use the entrySize instead + if (out.remaining() <= maxEntrySizeBytes) + return false; + + writeKey(iterator.next(), out); + } + + writeSizeAndFlags(sizeLocation, false, out); + return true; + } + }; } - public void values(Bytes in, Bytes out) { + public Work values(Bytes in, Bytes out) { long sizeLocation = reflectTransactionId(in, out); final Collection values = map.values(); out.writeStopBit(values.size()); @@ -227,9 +231,10 @@ class StatelessServerConnector { writeValue(value, out); } writeSizeAndFlags(sizeLocation, false, out); + return null; } - public void entrySet(Bytes in, Bytes out) { + public Work entrySet(Bytes in, Bytes out) { long sizeLocation = reflectTransactionId(in, out); final Set> entries = map.entrySet(); @@ -238,24 +243,27 @@ class StatelessServerConnector { writeKey(e.getKey(), out); writeValue(e.getValue(), out); } + return null; } - public void putIfAbsent(Bytes in, Bytes out) { + public Work 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); + return null; } - public void replace(Bytes in, Bytes out) { + public Work 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); + return null; } /** @@ -292,6 +300,27 @@ class StatelessServerConnector { out.writeUnsignedShort(0L, (int) size); out.writeBoolean(2L, isException); } + + 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); + } + + + 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; + } } class KeyValueSerializer { diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index be8fd7b70..ebff0667d 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -506,6 +506,17 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { final SocketChannel socketChannel = (SocketChannel) key.channel(); final Attached attached = (Attached) key.attachment(); + if (attached.entryWriter.hasIncompletWork()) { + + boolean completed = attached.entryWriter.doWork(); + + if (completed) + attached.entryWriter.workCompleted(); + + return; + } + + if (attached.remoteModificationIterator != null) attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); @@ -532,6 +543,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { final SocketChannel socketChannel = (SocketChannel) key.channel(); final Attached attached = (Attached) key.attachment(); + if (attached.entryWriter.hasIncompletWork()) + return; + try { if (attached.entryReader.readSocketToBuffer(socketChannel) <= 0) return; @@ -743,11 +757,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { public long remoteHeartbeatInterval = heartBeatIntervalMillis; public boolean handShakingComplete; + boolean isHandShakingComplete() { return handShakingComplete; } - void clearHandShaking() { handShakingComplete = false; @@ -770,7 +784,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { TcpReplicator.this.opWriteUpdater.set(remoteIdentifier); } - } /** @@ -783,12 +796,23 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { private final EntryCallback entryCallback; private long lastSentTime; + // if uncompletedWork is set ( not null ) , this must be completed before any further work + // is carried out + public Work uncompletedWork; + private TcpSocketChannelEntryWriter() { out = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes); in = new ByteBufferBytes(out); entryCallback = new EntryCallback(externalizable, in); } + public boolean hasIncompletWork() { + return uncompletedWork != null; + } + + public void workCompleted() { + uncompletedWork = null; + } /** * @return the buffer messages can be written to @@ -936,6 +960,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } + public boolean doWork() { + return uncompletedWork.doWork(in); + } } /** @@ -985,7 +1012,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // its set to MIN_VALUE when it should be read again if (state == NOT_SET) { - if (out.remaining() < SIZE_OF_SHORT + 1 ) { + if (out.remaining() < SIZE_OF_SHORT + 1) { return; } @@ -993,7 +1020,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { state = out.readByte(); - sizeInBytes = out.readUnsignedShort(); if (state == 0 && sizeInBytes != 0) { @@ -1009,15 +1035,10 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } - if (out.remaining() < sizeInBytes) { return; } - if (sizeInBytes == 0) { - int j = 1; - } - final long nextEntryPos = out.position() + sizeInBytes; final long limit = out.limit(); @@ -1031,21 +1052,32 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { LOG.error("", new IllegalArgumentException("received an event " + "from a stateless map, stateless maps are not " + "currently supported when using Chronicle Channels")); - } else + } else { - statelessServerConnector.processStatelessEvent(state, + final Work futureWork = statelessServerConnector.processStatelessEvent(state, out, attached.entryWriter.buffer()); - // to allow the sizeInBytes to be read the next time around + // in some cases it may not be possible to send out all the data before we + // fill out the write buffer, so this data will be send when the buffer + // is no longer full, and as such is treated as future work + if (futureWork != null) { + + // we will complete what we can now + boolean isComplete = futureWork.doWork(attached.entryWriter.buffer()); + + if (!isComplete) + attached.entryWriter.uncompletedWork = futureWork; + } + } } else externalizable.readExternalEntry(out); out.limit(limit); + // skip onto the next entry out.position(nextEntryPos); - state = NOT_SET; sizeInBytes = 0; } @@ -1087,7 +1119,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { else return Long.MIN_VALUE; - // return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE; + // return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE; } public long remoteHeartbeatIntervalFromBuffer() { @@ -1096,3 +1128,13 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { } } +interface Work { + + /** + * @param in the buffer that we will fill up + * @return true when all the work is complete + */ + boolean doWork(Bytes in); +} + + diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java index 95885799c..0a2c8fed1 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTests.java @@ -83,7 +83,6 @@ public class TCPSocketReplicationBootStrapTests { } - // todo we have to fix this @Test public void testBootstrapAndHeartbeat() throws IOException, InterruptedException { map1 = newTcpSocketShmIntString((byte) 1, 8079, new InetSocketAddress("localhost", 8076));