From f4f6365eae1f8dc4db827c1e6bce908aabb00fdc Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 28 Oct 2014 19:59:02 +0000 Subject: [PATCH] HCOLL-167 moved the transaction id to the start of the payload to may refactoring easier --- .../map/AbstractChannelReplicator.java | 2 + .../chronicle/map/StatelessChronicleMap.java | 25 ++- .../openhft/chronicle/map/TcpReplicator.java | 172 +++++++++--------- 3 files changed, 112 insertions(+), 87 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index b4a71c254..f0d541071 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -47,6 +47,8 @@ abstract class AbstractChannelReplicator implements Closeable { public static final int BITS_IN_A_BYTE = 8; public static final int SIZE_OF_SIZE = 4; + public static final int SIZE_OF_TRANSACTIONID =8; + private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); final Selector selector; final CloseablesManager closeables = new CloseablesManager(); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 63a0b8470..809dceaf7 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -39,6 +39,7 @@ import java.util.*; 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_TRANSACTIONID; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; /** @@ -250,6 +251,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized boolean remove(Object key, Object value) { + + if (key == null) + throw new NullPointerException("key can not be null"); + long sizeLocation = writeEvent(REMOVE_WITH_VALUE); writeKey((K) key); writeValue((V) value); @@ -260,6 +265,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized boolean replace(K key, V oldValue, V newValue) { + if (key == null) + throw new NullPointerException("key can not be null"); + long sizeLocation = writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE); writeKey(key); writeValue(oldValue); @@ -271,6 +279,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized V replace(K key, V value) { + if (key == null) + throw new NullPointerException("key can not be null"); + long sizeLocation = writeEvent(REPLACE); writeKey(key); writeValue(value); @@ -793,8 +804,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { if (LOG.isDebugEnabled()) LOG.debug("sending data with transactionId=" + transactionId); - write(transactionId); - writeSizeAt(sizeLocation); + writeSizeAndTransactionIdAt(sizeLocation, transactionId); // send out all the bytes send(bytes, timeoutTime); @@ -888,15 +898,20 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { throw new TimeoutRuntimeException(); } - private void writeSizeAt(long locationOfSize) { + private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { long size = bytes.position() - locationOfSize; -// LOG.info("writing size = " + size); - bytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); + long pos = bytes.position(); + bytes.position(locationOfSize); + bytes.writeInt((int) size - SIZE_OF_SIZE); + bytes.writeLong(transactionId); + bytes.position(pos); + } private long markSizeLocation() { long position = bytes.position(); bytes.skip(SIZE_OF_SIZE); + bytes.skip(SIZE_OF_TRANSACTIONID); return position; } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 144e9745b..4b3640c8f 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1238,6 +1238,9 @@ class StatelessServerConnector { private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class .getName()); + public static final int SIZE_OF_SIZE = 4; + public static final int TRANSACTION_ID_OFFSET = AbstractChannelReplicator.SIZE_OF_SIZE + 1; + public static final int TRANSACTIONID_OFFSET = AbstractChannelReplicator.SIZE_OF_SIZE + 1; ; private final KeyValueSerializer keyValueSerializer; private final ChronicleMap map; @@ -1256,44 +1259,11 @@ class StatelessServerConnector { @NotNull final ByteBufferBytes reader) { -/* LOG.info("processStatelessEvent - writer position=" + reader.position() + - ", " + - "limit=" + reader.limit() + ",capacity=" + reader.capacity()); - - - LOG.info("StatelessServerConnector- reader position=" + writer.position() + - ", " + - "limit=" + writer.limit() + ",capacity=" + writer.capacity());*/ - - final StatelessChronicleMap.EventId event = StatelessChronicleMap.EventId.values()[eventId]; + switch (event) { - case LONG_SIZE: - return longSize(reader, writer); - - case IS_EMPTY: - return isEmpty(reader, writer); - - case CONTAINS_KEY: - return containsKey(reader, writer); - - case CONTAINS_VALUE: - return containsValue(reader, writer); - - case GET: - return get(reader, writer); - - case PUT: - return put(reader, writer); - - case REMOVE: - return remove(reader, writer); - - case CLEAR: - return clear(reader, writer); - case KEY_SET: return keySet(reader, writer); @@ -1303,32 +1273,64 @@ class StatelessServerConnector { case ENTRY_SET: return entrySet(reader, writer); + } + + + final long sizeLocation = reflectTransactionId(reader, writer); + + switch (event) { + + case LONG_SIZE: + return longSize(reader, writer, sizeLocation); + + case IS_EMPTY: + return isEmpty(reader, writer, sizeLocation); + + case CONTAINS_KEY: + return containsKey(reader, writer, sizeLocation); + + case CONTAINS_VALUE: + return containsValue(reader, writer, sizeLocation); + + case GET: + return get(reader, writer, sizeLocation); + + case PUT: + return put(reader, writer, sizeLocation); + + case REMOVE: + return remove(reader, writer, sizeLocation); + + case CLEAR: + return clear(reader, writer, sizeLocation); + + case REPLACE: - return replace(reader, writer); + return replace(reader, writer, sizeLocation); case REPLACE_WITH_OLD_AND_NEW_VALUE: - return replaceWithOldAndNew(reader, writer); + return replaceWithOldAndNew(reader, writer, sizeLocation); case PUT_IF_ABSENT: - return putIfAbsent(reader, writer); + return putIfAbsent(reader, writer, sizeLocation); case REMOVE_WITH_VALUE: - return removeWithValue(reader, writer); + return removeWithValue(reader, writer, sizeLocation); case SIZE: - return size(reader, writer); + return size(reader, writer, sizeLocation); case TO_STRING: - return toString(reader, writer); + return toString(reader, writer, sizeLocation); case PUT_ALL: - return putAll(reader, writer); + return putAll(reader, writer, sizeLocation); case HASH_CODE: - return hashCode(reader, writer); + return hashCode(reader, writer, sizeLocation); case EQUALS: - return equals(reader, writer); + return equals(reader, writer, sizeLocation); default: @@ -1339,11 +1341,11 @@ class StatelessServerConnector { } - private Work removeWithValue(Bytes reader, Bytes writer) { + private Work removeWithValue(Bytes reader, Bytes writer, final long sizeLocation) { final K key = readKey(reader); final V readValue = readValue(reader); - final long sizeLocation = reflectTransactionId(reader, writer); + final boolean result; try { result = map.remove(key, readValue); @@ -1354,13 +1356,13 @@ class StatelessServerConnector { return null; } - private Work replaceWithOldAndNew(Bytes reader, Bytes writer) { + private Work replaceWithOldAndNew(Bytes reader, Bytes writer, final long sizeLocation) { final K key = readKey(reader); final V oldValue = readValue(reader); final V newValue = readValue(reader); boolean replaced; - long sizeLocation = reflectTransactionId(reader, writer); + try { replaced = map.replace(key, oldValue, newValue); } catch (RuntimeException e) { @@ -1372,8 +1374,8 @@ class StatelessServerConnector { } - private Work longSize(Bytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); + private Work longSize(Bytes reader, Bytes writer, final long sizeLocation) { + try { writer.writeLong(map.longSize()); } catch (RuntimeException e) { @@ -1383,8 +1385,8 @@ class StatelessServerConnector { return null; } - private Work size(Bytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); + private Work size(Bytes reader, Bytes writer, final long sizeLocation) { + try { writer.writeInt(map.size()); } catch (RuntimeException e) { @@ -1395,8 +1397,8 @@ class StatelessServerConnector { } - private Work hashCode(ByteBufferBytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); + private Work hashCode(ByteBufferBytes reader, Bytes writer, final long sizeLocation) { + try { writer.writeInt(map.hashCode()); } catch (RuntimeException e) { @@ -1407,8 +1409,8 @@ class StatelessServerConnector { } - private Work toString(Bytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); + private Work toString(Bytes reader, Bytes writer, final long sizeLocation) { + final String str; final long remaining = writer.remaining(); @@ -1438,8 +1440,8 @@ class StatelessServerConnector { } - private Work isEmpty(Bytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); + private Work isEmpty(Bytes reader, Bytes writer, final long sizeLocation) { + try { writer.writeBoolean(map.isEmpty()); } catch (RuntimeException e) { @@ -1449,9 +1451,10 @@ class StatelessServerConnector { return null; } - private Work containsKey(Bytes reader, Bytes writer) { + private Work containsKey(Bytes reader, Bytes writer, final long sizeLocation) { + final K k = readKey(reader); - final long sizeLocation = reflectTransactionId(reader, writer); + try { writer.writeBoolean(map.containsKey(k)); } catch (RuntimeException e) { @@ -1461,9 +1464,11 @@ class StatelessServerConnector { return null; } - private Work containsValue(Bytes reader, Bytes writer) { + private Work containsValue(Bytes reader, Bytes writer, final long sizeLocation) { + + final V v = readValue(reader); - final long sizeLocation = reflectTransactionId(reader, writer); + try { writer.writeBoolean(map.containsValue(v)); } catch (RuntimeException e) { @@ -1473,9 +1478,11 @@ class StatelessServerConnector { return null; } - private Work get(Bytes reader, Bytes writer) { + private Work get(Bytes reader, Bytes writer, final long sizeLocation) { + + final K k = readKey(reader); - final long sizeLocation = reflectTransactionId(reader, writer); + try { writeValue(map.get(k), writer); } catch (RuntimeException e) { @@ -1486,10 +1493,11 @@ class StatelessServerConnector { } - private Work put(Bytes reader, Bytes writer) { + private Work put(Bytes reader, Bytes writer, final long sizeLocation) { + final K k = readKey(reader); final V v = readValue(reader); - long sizeLocation = reflectTransactionId(reader, writer); + try { writeValue(map.put(k, v), writer); } catch (RuntimeException e) { @@ -1499,9 +1507,8 @@ class StatelessServerConnector { return null; } - private Work remove(Bytes reader, Bytes writer) { + private Work remove(Bytes reader, Bytes writer, final long sizeLocation) { final K key = readKey(reader); - final long sizeLocation = reflectTransactionId(reader, writer); try { writeValue(map.remove(key), writer); @@ -1515,9 +1522,10 @@ class StatelessServerConnector { } - private Work putAll(Bytes reader, Bytes writer) { + private Work putAll(Bytes reader, Bytes writer, final long sizeLocation) { + + final Map m = readEntries(reader); - final long sizeLocation = reflectTransactionId(reader, writer); try { map.putAll(m); @@ -1530,9 +1538,9 @@ class StatelessServerConnector { } - private Work equals(Bytes reader, Bytes writer) { + private Work equals(Bytes reader, Bytes writer, final long sizeLocation) { + final Map m = readEntries(reader); - final long sizeLocation = reflectTransactionId(reader, writer); boolean result; try { @@ -1547,9 +1555,7 @@ class StatelessServerConnector { } - private Work clear(Bytes reader, Bytes writer) { - - final long sizeLocation = reflectTransactionId(reader, writer); + private Work clear(Bytes reader, Bytes writer, final long sizeLocation) { try { map.clear(); } catch (RuntimeException e) { @@ -1561,6 +1567,7 @@ class StatelessServerConnector { private Work values(Bytes reader, Bytes writer) { + final long transactionId = reader.readLong(); Collection values; @@ -1571,7 +1578,6 @@ class StatelessServerConnector { return sendException(reader, writer, e); } - final Iterator iterator = values.iterator(); // this allows us to write more data than the buffer will allow @@ -1649,6 +1655,7 @@ class StatelessServerConnector { private Work entrySet(final Bytes reader, Bytes writer) { final long transactionId = reader.readLong(); + final Set> entries; try { @@ -1694,7 +1701,6 @@ class StatelessServerConnector { return true; } - }; } @@ -1707,7 +1713,7 @@ class StatelessServerConnector { private long header(Bytes writer, final long transactionId) { final long sizeLocation = writer.position(); - writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is + writer.skip(TRANSACTION_ID_OFFSET); // SIZE_OF_SIZE + is // exception writer.writeLong(transactionId); @@ -1742,10 +1748,12 @@ class StatelessServerConnector { writer.position(end); } - private Work putIfAbsent(Bytes reader, Bytes writer) { + private Work putIfAbsent(Bytes reader, Bytes writer, final long sizeLocation) { + + final K key = readKey(reader); final V v = readValue(reader); - final long sizeLocation = reflectTransactionId(reader, writer); + final V value; try { @@ -1760,11 +1768,11 @@ class StatelessServerConnector { } - private Work replace(Bytes reader, Bytes writer) { + private Work replace(Bytes reader, Bytes writer, final long sizeLocation) { final K k = readKey(reader); final V v = readValue(reader); final V replaced; - long sizeLocation = reflectTransactionId(reader, writer); + try { replaced = map.replace(k, v); } catch (RuntimeException e) { @@ -1787,7 +1795,7 @@ class StatelessServerConnector { private long reflectTransactionId(Bytes reader, Bytes writer) { final long sizeLocation = writer.position(); - writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // isException + SIZE_OF_SIZE + writer.skip(TRANSACTION_ID_OFFSET); // isException + SIZE_OF_SIZE final long transactionId = reader.readLong(); writer.writeLong(transactionId); return sizeLocation;