mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
HCOLL-167 StatelessServerConnector added support for the case where the response is bigger than the buffer
This commit is contained in:
Rob Austin
@@ -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<K, V> {
|
||||
|
||||
private final KeyValueSerializer<K, V> keyValueSerializer;
|
||||
private final ChronicleMap<K, V> map;
|
||||
private double maxEntrySizeBytes;
|
||||
|
||||
|
||||
public StatelessServerConnector(KeyValueSerializer<K, V> keyValueSerializer, ChronicleMap<K, V> map) {
|
||||
@@ -23,64 +21,59 @@ class StatelessServerConnector<K, V> {
|
||||
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<K, V> {
|
||||
|
||||
}
|
||||
|
||||
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<K, V> {
|
||||
} 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<K, V> {
|
||||
} 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<K, V> readEntries(Bytes in) {
|
||||
|
||||
long size = in.readStopBit();
|
||||
final HashMap<K, V> result = new HashMap<K, V>();
|
||||
|
||||
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<K> ks = map.keySet();
|
||||
out.writeStopBit(ks.size());
|
||||
for (K key : ks) {
|
||||
writeKey(key, out);
|
||||
}
|
||||
writeSizeAndFlags(sizeLocation, false, out);
|
||||
|
||||
final Iterator<K> 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<V> values = map.values();
|
||||
out.writeStopBit(values.size());
|
||||
@@ -227,9 +231,10 @@ class StatelessServerConnector<K, V> {
|
||||
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<Map.Entry<K, V>> entries = map.entrySet();
|
||||
@@ -238,24 +243,27 @@ class StatelessServerConnector<K, V> {
|
||||
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<K, V> {
|
||||
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<K, V> readEntries(Bytes in) {
|
||||
|
||||
long size = in.readStopBit();
|
||||
final HashMap<K, V> result = new HashMap<K, V>();
|
||||
|
||||
for (long i = 0; i < size; i++) {
|
||||
result.put(readKey(in), readValue(in));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
class KeyValueSerializer<K, V> {
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user