diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index ab3675d17..b765b6f1a 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -351,79 +351,47 @@ final class ReplicatedChronicleMap buffer = new ThreadLocal() { - @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>instance(), + BytesBytesInterop.INSTANCE, key, key.remaining(), keyBytesToInstance, + GetRemoteSeparateBytesInterops.INSTANCE, value, valueBytesToInstance, + hash2, true, readValueToLazyBytes, true, identifier, timestamp, + true); + } finally { + writeUnlock(); + } + } + , RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> RV put(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 8870b2ced..bac785813 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -1016,6 +1016,23 @@ class VanillaChronicleMap, } } + enum GetRemoteSeparateBytesInterops + implements GetValueInterops, + DelegatingMetaBytesInterop>> { + INSTANCE; + + @Override + public DelegatingMetaBytesInterop> getMetaValueInterop( + @NotNull ThreadLocalCopies copies, BytesInterop valueInterop, Bytes value) { + return DelegatingMetaBytesInterop.>instance(); + } + + @Override + public BytesInterop getValueInterop(@NotNull ThreadLocalCopies copies) { + return BytesBytesInterop.INSTANCE; + } + } + @Override public final MVI getMetaValueInterop( @NotNull ThreadLocalCopies copies, VI valueInterop, V value) {