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

Merge remote-tracking branch 'origin/2.1' into 2.1

Browse Source
This commit is contained in:
Rob Austin
2015-06-15 18:46:10 +01:00
parent 66609b6cd2 87738b4f79
commit 6829aa047e
2 changed files with 90 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+73 -76
View File
@@ -351,79 +351,47 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
super.close();
}
private ThreadLocal<IByteBufferBytes> buffer = new ThreadLocal<IByteBufferBytes>() {
@Override
protected IByteBufferBytes initialValue() {
return ByteBufferBytes.wrap(ByteBuffer.allocateDirect(1024));
}
};
private Bytes threadlocalBuffer(long requiredSize) {
final IByteBufferBytes bytes = buffer.get();
bytes.clear();
bytes.buffer().clear();
if (bytes.buffer().remaining() < requiredSize) {
if (requiredSize > Integer.MAX_VALUE)
throw new IllegalStateException("entry is too large");
IByteBufferBytes result = ByteBufferBytes.wrap(ByteBuffer.allocateDirect((int) requiredSize));
buffer.set(result);
return result;
}
return bytes;
}
private static class EntryBuffer {
NativeBytes key;
NativeBytes value;
NativeBytes getKey() {
return key;
}
void setKey(final NativeBytes key) {
this.key = key;
}
NativeBytes getValue() {
return value;
}
void setValue(final NativeBytes value) {
this.value = value;
}
}
@Override
public void put(final Bytes key, final Bytes value, final byte id, final long timestamp) {
final ThreadLocalCopies copies = SegmentState.getCopies(null);
final SegmentState segmentState = SegmentState.get(copies);
long keySize = key.remaining();
long valueSize = key.remaining();
final byte remoteIdentifier;
final long requiredBufferSize = 8 + keySize + 8 + valueSize + 12;
final Bytes source = threadlocalBuffer(requiredBufferSize);
if (id != 0) {
remoteIdentifier = id;
} else {
throw new IllegalStateException("identifier can't be 0");
}
source.writeLong(timestamp);
keySizeMarshaller.writeSize(source, keySize);
valueSizeMarshaller.writeSize(source, keySize);
source.writeStopBit(timestamp);
source.writeByte(id);
source.writeBoolean(false);
if (remoteIdentifier == this.identifier()) {
// this may occur when working with UDP, as we may receive our own data
return;
}
// key.bytes().
writeTo(source, key);
writeTo(source, value);
setLastModificationTime(remoteIdentifier, timestamp);
source.flip();
readExternalEntry(copies, segmentState, source);
long hash = hash(key);
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
boolean debugEnabled = LOG.isDebugEnabled();
String message = null;
if (debugEnabled) {
message = String.format(
"READING FROM SOURCE - into local-id=%d, remote-id=%d, put(key=%s,",
localIdentifier, remoteIdentifier, key.toString().trim());
}
segment(segmentNum).remotePut(copies, segmentState, key, value,
segmentHash, remoteIdentifier, timestamp);
if (debugEnabled) {
LOG.debug(message + "value=" + value.toString().trim() + ")");
}
}
private static void writeTo(Bytes destination, Bytes source) {
@@ -438,26 +406,35 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
public void remove(final Bytes key,
final byte remoteIdentifier,
final long timestamp) {
final ThreadLocalCopies copies = SegmentState.getCopies(null);
final SegmentState segmentState = SegmentState.get(copies);
long keySize = key.remaining();
long valueSize = key.remaining();
if (remoteIdentifier == 0)
throw new IllegalStateException("identifier can't be 0");
final long requiredBufferSize = 8 + keySize + 8 + valueSize + 12;
final Bytes source = threadlocalBuffer(requiredBufferSize);
keySizeMarshaller.writeSize(source, keySize);
valueSizeMarshaller.writeSize(source, keySize);
source.writeLong(timestamp);
source.writeByte(remoteIdentifier);
source.writeBoolean(true);
if (remoteIdentifier == this.identifier()) {
// this may occur when working with UDP, as we may receive our own data
return;
}
writeTo(source, key);
setLastModificationTime(remoteIdentifier, timestamp);
readExternalEntry(copies, segmentState, source);
long hash = hash(key);
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
boolean debugEnabled = LOG.isDebugEnabled();
if (debugEnabled) {
LOG.debug("READING FROM SOURCE - into local-id={}, remote={}, remove(key={})",
localIdentifier, remoteIdentifier, key.toString().trim()
);
}
segment(segmentNum).remoteRemove(copies, segmentState,
key, segmentHash, timestamp, remoteIdentifier);
}
@@ -1037,6 +1014,26 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
}
void remotePut(@NotNull ThreadLocalCopies copies,
@NotNull SegmentState segmentState,
@NotNull final Bytes key, @NotNull final Bytes value,
long hash2, final byte identifier, final long timestamp) {
ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes;
readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller);
writeLock();
try {
putWithoutLock(copies, segmentState,
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, key, key.remaining(), keyBytesToInstance,
GetRemoteSeparateBytesInterops.INSTANCE, value, valueBytesToInstance,
hash2, true, readValueToLazyBytes, true, identifier, timestamp,
true);
} finally {
writeUnlock();
}
}
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<RV, ? super VBI>>
RV put(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState,
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+17
View File
@@ -1016,6 +1016,23 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
}
enum GetRemoteSeparateBytesInterops
implements GetValueInterops<Bytes, BytesInterop<Bytes>,
DelegatingMetaBytesInterop<Bytes, BytesInterop<Bytes>>> {
INSTANCE;
@Override
public DelegatingMetaBytesInterop<Bytes, BytesInterop<Bytes>> getMetaValueInterop(
@NotNull ThreadLocalCopies copies, BytesInterop<Bytes> valueInterop, Bytes value) {
return DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance();
}
@Override
public BytesInterop<Bytes> getValueInterop(@NotNull ThreadLocalCopies copies) {
return BytesBytesInterop.INSTANCE;
}
}
@Override
public final MVI getMetaValueInterop(
@NotNull ThreadLocalCopies copies, VI valueInterop, V value) {