Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-134 improved stateless map when configured with very small entry sizes

Browse Source
This commit is contained in:
Rob Austin
2014-11-21 17:38:45 +00:00
parent 92d5a9d45e
commit 5fb243e301
2 changed files with 48 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+44 -40
View File
@@ -54,6 +54,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private final byte[] connectionByte = new byte[1];
private final ByteBuffer connectionOutBuffer = ByteBuffer.wrap(connectionByte);
private final String name;
private ByteBuffer buffer;
private ByteBufferBytes bytes;
@@ -108,9 +109,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
this.putReturnsNull = chronicleMapBuilder.putReturnsNull();
this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull();
this.maxEntrySize = chronicleMapBuilder.entrySize(true);
if (maxEntrySize < 128)
maxEntrySize = 128; // the stateless client will not work with extreemly small buffers
this.vClass = chronicleMapBuilder.valueBuilder.eClass;
this.kClass = chronicleMapBuilder.keyBuilder.eClass;
this.name = chronicleMapBuilder.name();
attemptConnect(builder.remoteAddress());
buffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder());
@@ -946,6 +951,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException {
// bytes.buffer().position(0);
// the number of bytes in the response
final int size = receive(SIZE_OF_SIZE, timeoutTime).readInt();
@@ -953,10 +959,14 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
final int requiredSize = size + SIZE_OF_SIZE;
if (bytes.capacity() < requiredSize) {
bytes = new ByteBufferBytes(allocateDirect(requiredSize));
}
bytes.limit(bytes.capacity());
long pos = bytes.position();
long limit = bytes.position();
bytes.position(limit);
resizeBuffer(requiredSize, pos);
} else
bytes.limit(bytes.capacity());
// block until we have received all the byte in this chunk
receive(size, timeoutTime);
@@ -969,7 +979,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
", does not match the expected transaction-id=" + transactionId);
if (isException) {
Throwable throwable = (Throwable) bytes.readObject();
Throwable throwable = (Throwable) bytes.readObject();
try {
Field stackTrace = Throwable.class.getDeclaredField("stackTrace");
stackTrace.setAccessible(true);
@@ -996,6 +1006,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException {
while (bytes.buffer().position() < requiredNumberOfBytes) {
assert requiredNumberOfBytes <= bytes.capacity();
int len = clientChannel.read(bytes.buffer());
if (len == -1)
@@ -1052,31 +1065,21 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private void writeKey(K key) {
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeKey(key, bytes);
return;
} catch (IllegalArgumentException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
writeKey(key, null);
}
private void writeKey(K key, ThreadLocalCopies local) {
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeKey(key, bytes, local);
return;
} catch (IllegalArgumentException e) {
} catch (IndexOutOfBoundsException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
@@ -1095,15 +1098,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
private void writeValue(V value) {
long start = bytes.position();
for (; ; ) {
try {
keyValueSerializer.writeValue(value, bytes, null);
return;
} catch (IllegalArgumentException e) {
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
}
}
writeValue(value, null);
}
private void writeValue(V value, ThreadLocalCopies local) {
@@ -1112,21 +1107,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
long start = bytes.position();
for (; ; ) {
try {
assert bytes.position() == start;
bytes.limit(bytes.capacity());
keyValueSerializer.writeValue(value, bytes, null);
keyValueSerializer.writeValue(value, bytes, local);
return;
} catch (IllegalArgumentException e) {
String message = e.getMessage();
if (message.startsWith(START_OF)) {
String substring = message.substring("Attempt to write ".length(), message.length());
int i = substring.indexOf(' ');
if (i != -1) {
int size = Integer.parseInt(substring.substring(0, i));
resizeBuffer(size + maxEntrySize, start);
} else
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
}
resizeToMessage(start, e);
} catch (IllegalStateException e) {
@@ -1141,5 +1128,22 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
private void resizeToMessage(long start, IllegalArgumentException e) {
String message = e.getMessage();
if (message.startsWith(START_OF)) {
String substring = message.substring("Attempt to write ".length(), message.length());
int i = substring.indexOf(' ');
if (i != -1) {
int size = Integer.parseInt(substring.substring(0, i));
long requiresExtra = size - bytes.remaining();
System.out.println(size + substring);
resizeBuffer((int) (bytes.capacity() + requiresExtra), start);
} else
resizeBuffer((int) (bytes.capacity() + maxEntrySize), start);
} else
throw e;
}
}
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+4 -5
View File
@@ -314,7 +314,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
void checkValue(Object value) {
if (vClass != Void.class && !vClass.isInstance(value)) {
throw new ClassCastException("Value must be a " + vClass.getName() +
" but was a " + value.getClass());
" but was a "+ value.getClass());
}
}
@@ -1107,10 +1107,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* Removes a key (or key-value pair) from the Segment.
* <p/>
* <p>The entry will only be removed if {@code expectedValue} equals to {@code null} or the value
* previously corresponding to the specified key.
* Removes a key (or key-value pair) from the Segment. <p/> <p>The entry will only be removed if
* {@code expectedValue} equals to {@code null} or the value previously corresponding to the specified
* key.
*
* @param hash2 a hash code related to the {@code keyBytes}
* @return the value of the entry that was removed if the entry corresponding to the {@code keyBytes}