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-240 improved assertions

Browse Source
This commit is contained in:
Rob Austin
2014-12-11 20:02:28 +00:00
parent be4d32b82c
commit ed0c964bd3
3 changed files with 70 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ChronicleMap.java
+1 -1
View File
@@ -175,7 +175,7 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
*
* <p>If the {@code ChronicleMap} is off-heap updatable, i. e. created via {@link
* ChronicleMapBuilder} builder (values are {@link Byteable}), there is one more option of what
* to do if the key is absent in the map, see {@link ChronicleMapBuilder#prepareValueBytesOnAcquire(PrepareValueBytes)}.
* to do if the key is absent in the map.
* By default, value bytes are just zeroed out, no default value, either provided for key or
* constant, is put for the absent key.
*
src/main/java/net/openhft/chronicle/map/PrepareValueBytes.java
+2 -3
View File
@@ -26,8 +26,7 @@ import java.io.Serializable;
* value during {@link ChronicleMap#acquireUsing(Object, Object)
* chronicleMap.acquireUsing()} call, when the given key is absent in the map. Example: if you
* want {@code 1L} to be the off-heap updatable ChronicleMap's {@linkplain
* ChronicleMapBuilder#defaultValue(Object) default value}, the most simple way
* to achieve this is to {@linkplain ChronicleMapBuilder#prepareValueBytesOnAcquire(
* ChronicleMapBuilder#defaultValue(Object) default value}.
* PrepareValueBytes) configure} custom {@code PrepareValueBytes} strategy in {@link
* ChronicleMapBuilder}: <pre>{@code
* ChronicleMapBuilder.of(Key.class, LongValue.class)
@@ -41,7 +40,7 @@ import java.io.Serializable;
*
* @param <K> the key type of {@link ChronicleMap}, on which {@code acquireUsing()} is queried
* @param <V> the value type of {@link ChronicleMap}, on which {@code acquireUsing()} is queried
* @see ChronicleMapBuilder#prepareValueBytesOnAcquire(PrepareValueBytes)
*/
public interface PrepareValueBytes<K, V> extends Serializable {
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+67 -47
View File
@@ -193,7 +193,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final InetSocketAddress remoteAddress) {
if (LOG.isDebugEnabled())
LOG.debug("attempting to connect to " + remoteAddress+" ,name="+name);
LOG.debug("attempting to connect to " + remoteAddress + " ,name=" + name);
SocketChannel result;
@@ -296,7 +296,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
byte remoteIdentifier = connectionByte[0];
if (LOG.isDebugEnabled())
LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier+" ,name="+name);
LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier + " ,name=" + name);
}
@NotNull
@@ -487,6 +487,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
final long remaining = outBytes.remaining();
if (remaining < maxEntrySize) {
@@ -508,11 +510,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private void resizeBufferOutBuffer(int size, long start) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
if (LOG.isDebugEnabled())
LOG.debug("resizing buffer to size=" + size+" ,name="+name);
LOG.debug("resizing buffer to size=" + size + " ,name=" + name);
if (size < outBuffer.capacity())
throw new IllegalStateException("it not possible to resize the buffer smaller");
@@ -549,8 +552,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void resizeBufferInBuffer(int size, long start) {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
if (LOG.isDebugEnabled())
LOG.debug("InBuffer resizing buffer to size=" + size+" ,name="+name);
LOG.debug("InBuffer resizing buffer to size=" + size + " ,name=" + name);
if (size < inBuffer.capacity())
throw new IllegalStateException("it not possible to resize the buffer smaller");
@@ -759,8 +765,14 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
if (!putReturnsNull)
blockingFetchReadOnly(timeoutTime, transactionId);
if (!putReturnsNull) {
inBytesLock.lock();
try {
blockingFetchReadOnly(timeoutTime, transactionId);
} finally {
inBytesLock.unlock();
}
}
}
@NotNull
@@ -809,6 +821,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private long readLong(long transactionId, long startTime) {
assert !outBytesLock.isHeldByCurrentThread();
final long timeoutTime = startTime + this.config.timeoutMs();
// receive
@@ -822,6 +837,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private long writeEvent(@NotNull StatelessChronicleMap.EventId event) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
outBuffer.clear();
outBytes.clear();
@@ -841,7 +859,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private long writeEventAnSkip(@NotNull EventId event) {
assert outBytesLock.isHeldByCurrentThread();
// assert !inBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
final long sizeLocation = writeEvent(event);
@@ -934,7 +952,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
try {
if (LOG.isDebugEnabled())
LOG.debug("sending data with transactionId=" + transactionId+" ,name="+name);
LOG.debug("sending data with transactionId=" + transactionId + " ,name=" + name);
writeSizeAndTransactionIdAt(sizeLocation, transactionId);
@@ -960,6 +978,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
try {
return blockingFetchThrowable(timeoutTime, transactionId);
} catch (IOException e) {
@@ -972,7 +994,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
close();
throw new RuntimeException(e);
} catch (AssertionError e) {
LOG.error("name="+name, e);
LOG.error("name=" + name, e);
throw e;
}
}
@@ -985,6 +1007,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException,
InterruptedException {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
int remainingBytes = nextEntry(timeoutTime, transactionId);
if (inBytes.capacity() < remainingBytes) {
@@ -1027,6 +1052,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private int nextEntry(long timeoutTime, long transactionId) throws IOException {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
int remainingBytes;
for (; ; ) {
@@ -1083,7 +1111,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
if (System.currentTimeMillis() - timeoutTime >
parkedTransactionTimeStamp) {
LOG.error("name="+name, new IllegalStateException("Skipped Message with " +
LOG.error("name=" + name, new IllegalStateException("Skipped Message with " +
"transaction-id=" +
parkedTransactionTimeStamp +
", this can occur when you have another thread which has called the " +
@@ -1102,6 +1130,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void clearParked() {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
parkedTransactionId = 0;
parkedRemainingBytes = 0;
parkedTransactionTimeStamp = 0;
@@ -1109,6 +1139,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private void pause() {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
/// don't call inBytesLock.isHeldByCurrentThread() as it not atomic
inBytesLock.unlock();
inBytesLock.lock();
@@ -1126,6 +1159,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
@SuppressWarnings("UnusedReturnValue")
private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
inBytes.position(0);
inBytes.limit(requiredNumberOfBytes);
inBytes.buffer().position(0);
@@ -1150,6 +1186,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private void writeBytesToSocket(@NotNull final Bytes out, long timeoutTime) throws IOException {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
outBuffer.limit((int) out.position());
outBuffer.position(0);
@@ -1214,7 +1253,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
} catch (IndexOutOfBoundsException e) {
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
@@ -1238,7 +1277,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
} catch (IndexOutOfBoundsException e) {
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
@@ -1259,7 +1298,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
outBytes.limit(outBytes.capacity());
return valueWriterWithSize.write(outBytes, value, copies);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
@@ -1280,7 +1319,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
outBytes.limit(outBytes.capacity());
return valueWriterWithSize.writeInLoop(outBytes, value, writer, copies);
} catch (IllegalArgumentException e) {
resizeToMessage(start, e);
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
@@ -1290,7 +1329,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
}
private void resizeToMessage(long start, @NotNull IllegalArgumentException e) {
private void resizeToMessageOutBuffer(long start, @NotNull IllegalArgumentException e) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
String message = e.getMessage();
@@ -1309,6 +1348,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private V readValue(long transactionId, long startTime, ThreadLocalCopies copies) {
assert !outBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
// receive
@@ -1322,6 +1363,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
@Nullable
private <O> O readObject(long transactionId, long startTime) {
assert !outBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
// receive
@@ -1334,6 +1377,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private boolean readBoolean(long transactionId, long startTime) {
assert !outBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
// receive
@@ -1347,6 +1392,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private int readInt(long transactionId, long startTime) {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
long timeoutTime = startTime + this.config.timeoutMs();
// receive
@@ -1376,6 +1423,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private V readValue(ThreadLocalCopies copies, Bytes in) {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
return valueReaderWithSize.readNullable(in, copies);
}
@@ -1604,37 +1653,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
"supported");
}
@Nullable
private <R> R fetchObject(Class<R> rClass, @NotNull final EventId eventId, K key, V value1, V value2) {
final long startTime = System.currentTimeMillis();
ThreadLocalCopies copies;
long transactionId;
outBytesLock.lock();
try {
final long sizeLocation = writeEventAnSkip(eventId);
copies = writeKey(key);
copies = writeValue(value1, copies);
copies = writeValue(value2, copies);
transactionId = send(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
if (eventReturnsNull(eventId))
return null;
if (rClass == vClass)
return (R) readValue(transactionId, startTime, copies);
else
throw new UnsupportedOperationException("class of type class=" + rClass + " is not " +
"supported");
}
@Nullable
private <R> R fetchObject(Class<R> rClass, @NotNull final EventId eventId, K key) {
@@ -1679,8 +1697,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void writeObject(@NotNull Object function) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
long start = outBytes.position();
for (; ; ) {
try {
@@ -1690,7 +1710,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
Throwable cause = e.getCause();
if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) {
LOG.debug("resizing buffer, name="+name);
LOG.debug("resizing buffer, name=" + name);
resizeBufferOutBuffer(outBuffer.capacity() + maxEntrySize, start);
} else
throw e;