diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 80267adbf..8f87caed2 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -53,7 +53,7 @@ abstract class AbstractChannelReplicator implements Closeable { public static final int SIZE_OF_SIZE = 4; public static final int SIZE_OF_TRANSACTION_ID = 8; - public static final int SIZE_OF_IS_EXCEPTION = 1; + public static final int SIZE_OF_TIME_SHIFT = 2; private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 3f470699f..5a7a82458 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -47,8 +47,7 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; import static java.nio.ByteBuffer.allocateDirect; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID; +import static net.openhft.chronicle.map.AbstractChannelReplicator.*; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; /** @@ -127,6 +126,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon UPDATE_FOR_KEY } + + // used by the enterprise version + private int identifier; + + void identifier(int identifier) { + this.identifier = identifier; + } + + @NotNull private final AtomicLong transactionID = new AtomicLong(0); @@ -489,56 +497,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private synchronized void writeEntriesForPutAll(@NotNull Map map) { - final int numberOfEntries = map.size(); - int numberOfEntriesReadSoFar = 0; - - outBytes.writeStopBit(numberOfEntries); - assert outBytes.limit() == outBytes.capacity(); - - ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); - final Object keyWriter = keyWriterWithSize.writerForLoop(copies); - copies = valueWriterWithSize.getCopies(copies); - final Object valueWriter = valueWriterWithSize.writerForLoop(copies); - - for (final Map.Entry e : map.entrySet()) { - final K key = e.getKey(); - final V value = e.getValue(); - if (key == null || value == null) - throw new NullPointerException(); - - numberOfEntriesReadSoFar++; - - long start = outBytes.position(); - - // putAll if a bit more complicated than the others - // as the volume of data could cause us to have to resize our buffers - resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start); - - final Class keyClass = key.getClass(); - if (!kClass.isAssignableFrom(keyClass)) { - throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + - "and should" + - " be of type=" + kClass); - } - - writeKeyInLoop(key, keyWriter, copies); - - final Class valueClass = value.getClass(); - if (!vClass.isAssignableFrom(valueClass)) - throw new ClassCastException("value=" + value + " is of type=" + valueClass + - " and " + - "should be of type=" + vClass); - - writeValueInLoop(value, valueWriter, copies); - - final int len = (int) (outBytes.position() - start); - - if (len > maxEntrySize) - maxEntrySize = len; - } - } - private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) { final long remaining = outBytes.remaining(); @@ -723,7 +681,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public void putAll(@NotNull Map map) { - final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : + final long sizeLocation = putReturnsNull ? writeEventAnSkip(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); final long startTime = System.currentTimeMillis(); @@ -856,6 +814,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // skips for the transaction id outBytes.skip(SIZE_OF_TRANSACTION_ID); + outBytes.skip(SIZE_OF_TIME_SHIFT); + outBytes.writeByte(identifier); return sizeLocation; } @@ -934,10 +894,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon if (LOG.isDebugEnabled()) LOG.debug("sending data with transactionId=" + transactionId); - writeSizeAndTransactionIdAt(sizeLocation, transactionId); + writeSizeAndTransactionIdAt(sizeLocation, transactionId, startTime); // send out all the bytes - send(outBytes, timeoutTime); + writeBytesToSocket(outBytes, timeoutTime); outBytes.clear(); outBytes.buffer().clear(); break; @@ -1141,7 +1101,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void send(@NotNull final Bytes out, long timeoutTime) throws IOException { + private void writeBytesToSocket(@NotNull final Bytes out, long timeoutTime) throws IOException { outBuffer.limit((int) out.position()); outBuffer.position(0); @@ -1160,7 +1120,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon throw new RemoteCallTimeoutException(); } - private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { + private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId, long startTime) { final long size = outBytes.position() - locationOfSize; final long pos = outBytes.position(); outBytes.position(locationOfSize); @@ -1168,15 +1128,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert size0 > 0; outBytes.writeInt(size0); + assert transactionId != 0; outBytes.writeLong(transactionId); - outBytes.position(pos); - } + outBytes.writeUnsignedShort((int) (transactionId - startTime)); - private void writeSizeAt(long locationOfSize) { - final long size = outBytes.position() - locationOfSize; - int size0 = (int) size - SIZE_OF_SIZE; - assert size0 > 0; - outBytes.writeInt(locationOfSize, size0); + outBytes.position(pos); } private long markSizeLocation() { @@ -1545,9 +1501,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); copies = writeValue(value, copies); @@ -1577,9 +1531,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); copies = writeValue(value1, copies); @@ -1611,9 +1563,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); transactionId = send(sizeLocation, startTime); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 50bafefc5..958d42ee1 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1329,28 +1329,39 @@ class StatelessServerConnector { @NotNull final ByteBufferBytes reader) { final StatelessChronicleMap.EventId event = VALUES[eventId]; - // these methods don't return a result + long transactionId = reader.readLong(); + + // the time stamp and the transaction are usually the same, or out by the shift + int timestampShift = reader.readUnsignedShort(); + long timestamp = transactionId - timestampShift; + + byte identifier = reader.readByte(); + + + // these methods don't return a result to the client or don't return a result to the + // client immediately switch (event) { case KEY_SET: - return keySet(reader, writer); + return keySet(reader, writer, transactionId); case VALUES: - return values(reader, writer); + return values(reader, writer, transactionId); case ENTRY_SET: - return entrySet(reader, writer); + return entrySet(reader, writer, transactionId); case PUT_WITHOUT_ACC: - return put(reader); + return put(reader, timestamp, identifier); case PUT_ALL_WITHOUT_ACC: - return putAll(reader); + return putAll(reader, timestamp, identifier); case REMOVE_WITHOUT_ACC: - return remove(reader); + return remove(reader, timestamp, identifier); } - final long sizeLocation = reflectTransactionId(reader, writer); + final long sizeLocation = reflectTransactionId(writer, transactionId); + // these methods return a result @@ -1368,34 +1379,33 @@ class StatelessServerConnector { return containsValue(reader, writer, sizeLocation); case GET: - return get(reader, writer, sizeLocation); + return get(reader, writer, sizeLocation, timestamp); case PUT: - return put(reader, writer, sizeLocation); + return put(reader, writer, sizeLocation, timestamp, identifier); case REMOVE: - return remove(reader, writer, sizeLocation); + return remove(reader, writer, sizeLocation, timestamp, identifier); case CLEAR: - return clear(writer, sizeLocation); + return clear(writer, sizeLocation, timestamp, identifier); case REPLACE: - return replace(reader, writer, sizeLocation); + return replace(reader, writer, sizeLocation, timestamp, identifier); case REPLACE_WITH_OLD_AND_NEW_VALUE: - return replaceWithOldAndNew(reader, writer, sizeLocation); + return replaceWithOldAndNew(reader, writer, sizeLocation, timestamp, identifier); case PUT_IF_ABSENT: - return putIfAbsent(reader, writer, sizeLocation); + return putIfAbsent(reader, writer, sizeLocation, timestamp, identifier); case REMOVE_WITH_VALUE: - return removeWithValue(reader, writer, sizeLocation); - + return removeWithValue(reader, writer, sizeLocation, timestamp, identifier); case TO_STRING: return toString(writer, sizeLocation); case PUT_ALL: - return putAll(reader, writer, sizeLocation); + return putAll(reader, writer, sizeLocation, timestamp, identifier); case HASH_CODE: return hashCode(writer, sizeLocation); @@ -1444,7 +1454,7 @@ class StatelessServerConnector { } @Nullable - private Work removeWithValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work removeWithValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { writer.writeBoolean(map.removeWithValue(reader)); @@ -1463,7 +1473,8 @@ class StatelessServerConnector { } @Nullable - private Work replaceWithOldAndNew(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work replaceWithOldAndNew(Bytes reader, @NotNull Bytes writer, final long + sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.replaceWithOldAndNew(reader, writer); @@ -1585,7 +1596,9 @@ class StatelessServerConnector { } @Nullable - private Work get(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work get(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long + transactionId) { + try { if (MAP_SUPPORTS_BYTES) { map.get(reader, writer); @@ -1603,7 +1616,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work put(Bytes reader) { + private Work put(Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.put(reader); } else { @@ -1618,7 +1631,7 @@ class StatelessServerConnector { } @Nullable - private Work put(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work put(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.put(reader, writer); @@ -1638,7 +1651,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work remove(Bytes reader) { + private Work remove(Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.removeKeyAsBytes(reader); } else { @@ -1650,7 +1663,7 @@ class StatelessServerConnector { } @Nullable - private Work remove(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work remove(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.remove(reader, writer); @@ -1669,7 +1682,7 @@ class StatelessServerConnector { } @Nullable - private Work putAll(@NotNull Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work putAll(@NotNull Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); @@ -1686,7 +1699,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work putAll(@NotNull Bytes reader) { + private Work putAll(@NotNull Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); } else { @@ -1697,7 +1710,7 @@ class StatelessServerConnector { } @Nullable - private Work clear(@NotNull Bytes writer, final long sizeLocation) { + private Work clear(@NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { map.clear(); } catch (Throwable e) { @@ -1709,8 +1722,7 @@ class StatelessServerConnector { } @Nullable - private Work values(@NotNull Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private Work values(@NotNull Bytes reader, @NotNull Bytes writer, final long transactionId) { Collection values; @@ -1752,8 +1764,7 @@ class StatelessServerConnector { } @Nullable - private Work keySet(@NotNull Bytes reader, @NotNull final Bytes writer) { - final long transactionId = reader.readLong(); + private Work keySet(@NotNull Bytes reader, @NotNull final Bytes writer, final long transactionId) { Set ks; @@ -1795,8 +1806,7 @@ class StatelessServerConnector { } @Nullable - private Work entrySet(@NotNull final Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private Work entrySet(@NotNull final Bytes reader, @NotNull Bytes writer, final long transactionId) { final Set> entries; @@ -1844,7 +1854,7 @@ class StatelessServerConnector { } @Nullable - private Work putIfAbsent(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work putIfAbsent(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.putIfAbsent(reader, writer); @@ -1863,7 +1873,7 @@ class StatelessServerConnector { } @Nullable - private Work replace(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work replace(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.replaceKV(reader, writer); @@ -1881,8 +1891,7 @@ class StatelessServerConnector { return null; } - private long reflectTransactionId(@NotNull Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private long reflectTransactionId(@NotNull Bytes writer, final long transactionId) { final long sizeLocation = writer.position(); writer.skip(SIZE_OF_SIZE); assert transactionId != 0; @@ -1940,7 +1949,7 @@ class StatelessServerConnector { @Nullable private Work sendException(@NotNull Bytes reader, @NotNull Bytes writer, @NotNull Throwable e) { - final long sizeLocation = reflectTransactionId(reader, writer); + final long sizeLocation = reflectTransactionId(writer, reader.readLong()); return sendException(writer, sizeLocation, e); } diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 1d4288b61..c502b49ed 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -157,8 +157,9 @@ public class StatelessClientTest { } statelessMap.putAll(payload); + assertEquals(SIZE, serverMap.size()); - Set> entries = statelessMap.entrySet(); + Set> entries = statelessMap.entrySet(); Map.Entry next = entries.iterator().next(); assertEquals("some value=" + next.getKey(), next.getValue());