From 1e87d6ba8af9a3a3d966b4be967415322b3c4660 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 24 Nov 2014 14:53:57 +0000 Subject: [PATCH] HCOLL-216 added a test for getLater --- .../map/AbstractChannelReplicator.java | 2 +- .../chronicle/map/StatelessChronicleMap.java | 1203 +++-------------- .../StatelessChronicleMapUnsynchronized.java | 1157 ++++++++++++++++ .../openhft/chronicle/map/TcpReplicator.java | 7 +- .../chronicle/map/VanillaChronicleMap.java | 2 +- .../chronicle/map/StatelessClientTest.java | 32 + 6 files changed, 1351 insertions(+), 1052 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/StatelessChronicleMapUnsynchronized.java diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index a18173bdc..41bb91813 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -307,7 +307,7 @@ abstract class AbstractChannelReplicator implements Closeable { long pos0 = in.position(); // used to denote that this is not a stateless map event - in.writeByte(StatelessChronicleMap.EventId.STATEFUL_UPDATE.ordinal()); + in.writeByte(StatelessChronicleMapUnsynchronized.EventId.STATEFUL_UPDATE.ordinal()); long sizeLocation = in.position(); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 54b9af5ab..19e5f0425 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -18,311 +18,87 @@ package net.openhft.chronicle.map; -import com.sun.jdi.connect.spi.ClosedConnectionException; -import net.openhft.chronicle.hash.RemoteCallTimeoutException; -import net.openhft.lang.io.ByteBufferBytes; -import net.openhft.lang.io.Bytes; -import net.openhft.lang.io.NativeBytes; -import net.openhft.lang.threadlocal.ThreadLocalCopies; +import net.openhft.lang.thread.NamedThreadFactory; import org.jetbrains.annotations.NotNull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.File; import java.io.IOException; -import java.lang.reflect.Field; -import java.net.InetSocketAddress; -import java.nio.ByteBuffer; -import java.nio.ByteOrder; -import java.nio.channels.SocketChannel; -import java.util.*; +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; -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.*; - /** * @author Rob Austin. */ -class StatelessChronicleMap implements ChronicleMap, Closeable { +class StatelessChronicleMap implements ChronicleMap, Closeable, Cloneable { - public static final String START_OF = "Attempt to write "; - private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMap.class); - public static final byte STATELESS_CLIENT_IDENTIFIER = (byte) -127; - private final byte[] connectionByte = new byte[1]; - private final ByteBuffer connectionOutBuffer = ByteBuffer.wrap(connectionByte); - private final String name; - - private ByteBuffer buffer; - private ByteBufferBytes bytes; - - private final KeyValueSerializer keyValueSerializer; - private volatile SocketChannel clientChannel; - - private CloseablesManager closeables; + private final AbstractChronicleMapBuilder chronicleMapBuilder; private final StatelessMapConfig builder; - private int maxEntrySize; - private final Class kClass; - private final Class vClass; - private final boolean putReturnsNull; - private final boolean removeReturnsNull; + ChronicleMap statelessClient; - static enum EventId { - HEARTBEAT, - STATEFUL_UPDATE, - LONG_SIZE, SIZE, - IS_EMPTY, - CONTAINS_KEY, - CONTAINS_VALUE, - GET, - PUT, - PUT_WITHOUT_ACC, - REMOVE, - REMOVE_WITHOUT_ACC, - CLEAR, - KEY_SET, - VALUES, - ENTRY_SET, - REPLACE, - REPLACE_WITH_OLD_AND_NEW_VALUE, - PUT_IF_ABSENT, - REMOVE_WITH_VALUE, - TO_STRING, - PUT_ALL, - PUT_ALL_WITHOUT_ACC, - HASH_CODE, - MAP_FOR_KEY, - UPDATE_FOR_KEY - } - - private long transactionID; - - StatelessChronicleMap(final StatelessMapConfig builder, - final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { + public StatelessChronicleMap(final StatelessMapConfig builder, + final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { this.builder = builder; - this.keyValueSerializer = new KeyValueSerializer(chronicleMapBuilder.keyBuilder, - chronicleMapBuilder.valueBuilder); - this.putReturnsNull = chronicleMapBuilder.putReturnsNull(); - this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull(); - this.maxEntrySize = chronicleMapBuilder.entrySize(true); - - if (maxEntrySize < 128) - maxEntrySize = 128; // the stateless client will not work with extreemly small buffers - - this.vClass = chronicleMapBuilder.valueBuilder.eClass; - this.kClass = chronicleMapBuilder.keyBuilder.eClass; - this.name = chronicleMapBuilder.name(); - attemptConnect(builder.remoteAddress()); - - buffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); - bytes = new ByteBufferBytes(buffer.slice()); + this.chronicleMapBuilder = chronicleMapBuilder; + statelessClient = new StatelessChronicleMapUnsynchronized(builder, chronicleMapBuilder); } - private SocketChannel lazyConnect(final long timeoutMs, - final InetSocketAddress remoteAddress) throws IOException { - - if (LOG.isDebugEnabled()) - LOG.debug("attempting to connect to " + remoteAddress); - - SocketChannel result = null; - - long timeoutAt = System.currentTimeMillis() + timeoutMs; - - for (; ; ) { - - checkTimeout(timeoutAt); - - // ensures that the excising connection are closed - closeExisting(); - - try { - result = AbstractChannelReplicator.openSocketChannel(closeables); - result.connect(builder.remoteAddress()); - result.socket().setTcpNoDelay(true); - - - doHandShaking(result); - break; - } catch (IOException e) { - closeables.closeQuietly(); - } catch (Exception e) { - closeables.closeQuietly(); - throw e; + @Override + public synchronized Future getLater(final K key) { + return lazyExecutorService().submit(new Callable() { + @Override + public V call() throws Exception { + return lazyAsyncStatelessClient().get(key); } - - } - - return result; + }); } + private ExecutorService executorService = null; + + private ExecutorService lazyExecutorService() { + + if (executorService == null) { + synchronized (this) { + if (executorService == null) + executorService = Executors.newSingleThreadExecutor(new NamedThreadFactory(chronicleMapBuilder.name() + + "-stateless-client-async", + true)); + } + } + + return executorService; + } + + + private ChronicleMap asyncStatelessClient; /** - * attempts a single connect without a timeout and eat any IOException used typically in the constructor, - * its possible the server is not running with this instance is created, {@link - * StatelessChronicleMap#lazyConnect(long, java.net.InetSocketAddress)} will attempt to establish the - * connection when the client make the first map method call. - * - * @param remoteAddress - * @return - * @throws IOException - * @see StatelessChronicleMap#lazyConnect(long, java.net.InetSocketAddress) + * @return a lazily created instance of the StatelessClient used for the async calls. */ - private void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { + private ChronicleMap lazyAsyncStatelessClient() { - // ensures that the excising connection are closed - closeExisting(); - - try { - clientChannel = AbstractChannelReplicator.openSocketChannel(closeables); - clientChannel.connect(remoteAddress); - doHandShaking(clientChannel); - - } catch (IOException e) { - closeables.closeQuietly(); + if (asyncStatelessClient == null) { + synchronized (this) { + if (asyncStatelessClient == null) + try { + asyncStatelessClient = new StatelessChronicleMapUnsynchronized(builder, + chronicleMapBuilder); + } catch (IOException e) { + throw new RuntimeException(e); + } + } } - } - - - /** - * closes the existing connections and establishes a new closeables - */ - private void closeExisting() { - - // ensure that any excising connection are first closed - if (closeables != null) - closeables.closeQuietly(); - - closeables = new CloseablesManager(); - } - - - /** - * initiates a very simple level of handshaking with the remote server, we send a special ID of -127 ( - * when the server receives this it knows its dealing with a stateless client, receive back an identifier - * from the server - * - * @param clientChannel - * @throws IOException - */ - private void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { - - connectionByte[0] = STATELESS_CLIENT_IDENTIFIER; - this.connectionOutBuffer.clear(); - - long timeoutTime = System.currentTimeMillis() + builder.timeoutMs(); - - // write a single byte - while (connectionOutBuffer.hasRemaining()) { - clientChannel.write(connectionOutBuffer); - checkTimeout(timeoutTime); - } - - this.connectionOutBuffer.clear(); - - // read a single byte back - while (this.connectionOutBuffer.position() <= 0) { - clientChannel.read(this.connectionOutBuffer); - checkTimeout(timeoutTime); - } - - byte remoteIdentifier = connectionByte[0]; - - if (LOG.isDebugEnabled()) - LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier); - - } - - public File file() { - throw new UnsupportedOperationException(); - } - - public synchronized void close() { - if (closeables != null) - closeables.closeQuietly(); - closeables = null; - try { - Thread.sleep(100); - } catch (InterruptedException e) { - // - } - - } - - /** - * the transaction id are generated as unique timestamps - * - * @param time in milliseconds - * @return a unique transactionId - */ - private long nextUniqueTransaction(long time) { - transactionID = (time == transactionID) ? time + 1 : time; - return transactionID; - } - - - public synchronized V putIfAbsent(K key, V value) { - final long sizeLocation = writeEventAnSkip(PUT_IF_ABSENT); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey(key, local); - writeValue(value, local); - return readValue(sizeLocation, local); - } - - - public synchronized boolean remove(Object key, Object value) { - - if (key == null) - throw new NullPointerException("key can not be null"); - - final long sizeLocation = writeEventAnSkip(REMOVE_WITH_VALUE); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey((K) key, local); - writeValue((V) value, local); - - // get the data back from the server - return blockingFetch(sizeLocation).readBoolean(); - } - - - public synchronized boolean replace(K key, V oldValue, V newValue) { - if (key == null) - throw new NullPointerException("key can not be null"); - - final long sizeLocation = writeEventAnSkip(REPLACE_WITH_OLD_AND_NEW_VALUE); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey(key, local); - writeValue(oldValue, local); - writeValue(newValue, local); - - // get the data back from the server - return blockingFetch(sizeLocation).readBoolean(); - } - - - public synchronized V replace(K key, V value) { - if (key == null) - throw new NullPointerException("key can not be null"); - - final long sizeLocation = writeEventAnSkip(REPLACE); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey(key, local); - writeValue(value, local); - - // get the data back from the server - return readValue(sizeLocation, local); - } - - - public synchronized int size() { - return (int) longSize(); + return asyncStatelessClient; } @@ -335,818 +111,151 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { * @return true if the contain the same data */ @Override - public synchronized boolean equals(Object object) { - - if (this == object) return true; - if (object == null || object.getClass().isAssignableFrom(Map.class)) - return false; - - final Map that = (Map) object; - - final int size = size(); - - if (that.size() != size) - return false; - - final Set> entries = entrySet(); - - return that.entrySet().equals(entries); - + public boolean equals(Object object) { + return statelessClient.equals(object); } @Override - public synchronized int hashCode() { - return blockingFetch(writeEventAnSkip(HASH_CODE)).readInt(); + public int hashCode() { + return statelessClient.hashCode(); } - public synchronized String toString() { - return blockingFetch(writeEventAnSkip(TO_STRING)).readObject(String.class); + @Override + public String toString() { + return statelessClient.toString(); } - - public synchronized boolean isEmpty() { - return blockingFetch(writeEventAnSkip(IS_EMPTY)).readBoolean(); - } - - - public synchronized boolean containsKey(Object key) { - final long sizeLocation = writeEventAnSkip(CONTAINS_KEY); - writeKey((K) key); - - // get the data back from the server - return blockingFetch(sizeLocation).readBoolean(); - - } - - - public synchronized boolean containsValue(Object value) { - final long sizeLocation = writeEventAnSkip(CONTAINS_VALUE); - writeValue((V) value); - - // get the data back from the server - return blockingFetch(sizeLocation).readBoolean(); - } - - + @Override public synchronized long longSize() { - return blockingFetch(writeEventAnSkip(LONG_SIZE)).readLong(); + return statelessClient.longSize(); } - public synchronized Future getLater(K key) { - throw new UnsupportedOperationException(); + @Override + public synchronized int size() { + return statelessClient.size(); } + @Override + public synchronized boolean isEmpty() { + return statelessClient.isEmpty(); + } + @Override + public synchronized boolean containsKey(Object key) { + return statelessClient.containsKey(key); + } + + @Override + public synchronized boolean containsValue(Object value) { + return statelessClient.containsValue(value); + } + + @Override public synchronized V get(Object key) { - final long sizeLocation = writeEventAnSkip(GET); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey((K) key, local); - - // get the data back from the server - return readValue(sizeLocation, local); + return statelessClient.get(key); } - public synchronized V getUsing(K key, V usingValue) { - throw new UnsupportedOperationException("getUsing() is not supported for stateless " + - "clients"); - } - - - public synchronized V acquireUsing(@NotNull K key, V usingValue) { - throw new UnsupportedOperationException("acquireUsing() is not supported for stateless " + - "clients"); - } - - @NotNull @Override - public WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { - throw new UnsupportedOperationException(); - } - - @NotNull - @Override - public ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue) { - throw new UnsupportedOperationException(); - } - - - public synchronized V remove(Object key) { - - if (key == null) - throw new NullPointerException("key can not be null"); - - final long sizeLocation = (removeReturnsNull) ? writeEvent(REMOVE_WITHOUT_ACC) : writeEventAnSkip(REMOVE); - - - writeKey((K) key); - - if (removeReturnsNull) { - sendWithoutAcc(bytes, sizeLocation); - return null; - } else - // get the data back from the server - return readValue(sizeLocation); - } - - public synchronized V put(K key, V value) { - - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - - final long sizeLocation = (putReturnsNull) ? writeEvent(PUT_WITHOUT_ACC) : writeEventAnSkip(PUT); - - writeKey(key, local); - writeValue(value, local); - - - if (putReturnsNull) { - sendWithoutAcc(bytes, sizeLocation); - return null; - } else - // get the data back from the server - return readValue(sizeLocation, local); - + return statelessClient.put(key, value); } + @Override + public synchronized V remove(Object key) { + return statelessClient.remove(key); + } + @Override + public synchronized void putAll(Map m) { + statelessClient.putAll(m); + } + + @Override + public synchronized void clear() { + statelessClient.clear(); + } + + @NotNull + @Override + public synchronized Set keySet() { + return statelessClient.keySet(); + } + + @NotNull + @Override + public synchronized Collection values() { + return statelessClient.values(); + } + + @NotNull + @Override + public synchronized Set> entrySet() { + return statelessClient.entrySet(); + } + + @Override + public synchronized V getUsing(K key, V usingValue) { + return statelessClient.getUsing(key, usingValue); + } + + @NotNull + @Override + public synchronized ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue) { + return statelessClient.getUsingLocked(key, usingValue); + } + + @Override + public synchronized V acquireUsing(@NotNull K key, V usingValue) { + return statelessClient.acquireUsing(key, usingValue); + } + + @NotNull + @Override + public synchronized WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { + return statelessClient.acquireUsingLocked(key, usingValue); + } + + @Override public synchronized R mapForKey(K key, @NotNull Function function) { - - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final long sizeLocation = writeEventAnSkip(MAP_FOR_KEY); - - writeKey(key, local); - writeObject(function); - - final Bytes reader = blockingFetch(sizeLocation); - return (R) keyValueSerializer.readObject(reader); + return statelessClient.mapForKey(key, function); } @Override public synchronized R updateForKey(K key, @NotNull Mutator mutator) { - - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final long sizeLocation = writeEventAnSkip(UPDATE_FOR_KEY); - - writeKey(key, local); - writeObject(mutator); - - final Bytes reader = blockingFetch(sizeLocation); - return (R) keyValueSerializer.readObject(reader); + return statelessClient.updateForKey(key, mutator); } - private void writeObject(@NotNull Object function) { - long start = bytes.position(); - for (; ; ) { - try { - bytes.writeObject(function); - return; - } catch (IllegalStateException e) { - - Throwable cause = e.getCause(); - - if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) { - LOG.debug("resizing buffer"); - resizeBuffer(buffer.capacity() + maxEntrySize, start); - - } else - throw e; - - } - } + @Override + public File file() { + return statelessClient.file(); } - - public synchronized void putAll(Map map) { - - final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); - - writeEntries(map); - - if (putReturnsNull) - sendWithoutAcc(bytes, sizeLocation); - else - blockingFetch(sizeLocation); - + @Override + public void close() { + statelessClient.close(); } - - private void sendWithoutAcc(final Bytes bytes, long sizeLocation) { - writeSizeAt(sizeLocation); - long timeoutTime = builder.timeoutMs() + System.currentTimeMillis(); - try { - for (; ; ) { - - if (clientChannel == null) - clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); - - try { - - send(bytes, timeoutTime); - return; - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { - checkTimeout(timeoutTime); - clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); - } - } - } catch (IOException e) { - close(); - throw new IORuntimeException(e); - } + @Override + public V putIfAbsent(K key, V value) { + return statelessClient.putIfAbsent(key, value); } - - private void writeEntries(Map map) { - - final int numberOfEntries = map.size(); - int numberOfEntriesReadSoFar = 0; - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - - bytes.writeStopBit(numberOfEntries); - assert bytes.limit() == bytes.capacity(); - for (final Map.Entry e : map.entrySet()) { - - numberOfEntriesReadSoFar++; - - long start = bytes.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 K key = e.getKey(); - - final Class keyClass = key.getClass(); - if (!kClass.isAssignableFrom(keyClass)) - throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + - "and should" + - " be of type=" + kClass); - - - writeKey(key, local); - - final V value = e.getValue(); - - final Class valueClass = value.getClass(); - if (!vClass.isAssignableFrom(valueClass)) - throw new ClassCastException("value=" + value + " is of type=" + valueClass + - " and " + - "should be of type=" + vClass); - - - writeValue(value, local); - - final int len = (int) (bytes.position() - start); - - if (len > maxEntrySize) - maxEntrySize = len; - } + @Override + public boolean remove(Object key, Object value) { + return statelessClient.remove(key, value); } - private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) { - - final long remaining = bytes.remaining(); - - if (remaining < maxEntrySize) { - long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar); - resizeBuffer((int) (estimatedRequiredSize + maxEntrySize), start); - } + @Override + public boolean replace(K key, V oldValue, V newValue) { + return statelessClient.replace(key, oldValue, newValue); } - /** - * estimates the size based on what been completed so far - * - * @param numberOfEntries - * @param numberOfEntriesReadSoFar - */ - private long estimateSize(int numberOfEntries, int numberOfEntriesReadSoFar) { - // we will back the size estimate on what we have so far - final double percentageComplete = (double) numberOfEntriesReadSoFar / (double) numberOfEntries; - return (long) ((double) bytes.position() / percentageComplete); + @Override + public V replace(K key, V value) { + return statelessClient.replace(key, value); } - - - private void resizeBuffer(int size, long start) { - - if (LOG.isDebugEnabled()) - LOG.debug("resizing buffer to size=" + size); - - if (size < buffer.capacity()) - throw new IllegalStateException("it not possible to resize the buffer smaller"); - - assert size < Integer.MAX_VALUE; - - final ByteBuffer result = ByteBuffer.allocate((int) size).order(ByteOrder.nativeOrder()); - - final long bytesPosition = bytes.position(); - - bytes = new ByteBufferBytes(result.slice()); - - buffer.position(0); - buffer.limit((int) bytesPosition); - - int numberOfLongs = (int) bytesPosition / 8; - - // chunk in longs first - for (int i = 0; i < numberOfLongs; i++) { - bytes.writeLong(buffer.getLong()); - } - - for (int i = numberOfLongs * 8; i < bytesPosition; i++) { - bytes.writeByte(buffer.get()); - } - - buffer = result; - - assert buffer.capacity() == bytes.capacity(); - - assert buffer.capacity() == size; - assert buffer.capacity() == bytes.capacity(); - assert bytes.limit() == bytes.capacity(); - bytes.position(start); - } - - - public synchronized void clear() { - - // get the data back from the server - blockingFetch(writeEventAnSkip(CLEAR)); - } - - - @NotNull - public synchronized Collection values() { - final long sizeLocation = writeEventAnSkip(VALUES); - - final long startTime = System.currentTimeMillis(); - final long transactionId = nextUniqueTransaction(startTime); - final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs(); - - // get the data back from the server - Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final Collection result = new ArrayList(); - - for (; ; ) { - - final boolean hasMoreEntries = in.readBoolean(); - - // number of entries in this chunk - final long size = in.readInt(); - - for (int i = 0; i < size; i++) { - result.add(keyValueSerializer.readValue(in, local)); - } - - if (!hasMoreEntries) - break; - - compact(in); - - in = blockingFetchReadOnly(timeoutTime, transactionId); - } - return result; - } - - - @NotNull - public synchronized Set> entrySet() { - final long sizeLocation = writeEventAnSkip(ENTRY_SET); - - final long startTime = System.currentTimeMillis(); - final long transactionId = nextUniqueTransaction(startTime); - final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs(); - - // get the data back from the server - Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final Map result = new HashMap(); - - for (; ; ) { - - final boolean hasMoreEntries = in.readBoolean(); - - // number of entries in this chunk - final long size = in.readInt(); - - for (int i = 0; i < size; i++) { - final K k = keyValueSerializer.readKey(in, local); - final V v = keyValueSerializer.readValue(in, local); - result.put(k, v); - } - - if (!hasMoreEntries) - break; - - compact(in); - - in = blockingFetchReadOnly(timeoutTime, transactionId); - } - - - return result.entrySet(); - } - - private void compact(Bytes in) { - if (in.remaining() == 0) { - bytes.clear(); - bytes.buffer().clear(); - } else { - buffer.compact(); - } - } - - - @NotNull - public synchronized Set keySet() { - final long sizeLocation = writeEventAnSkip(KEY_SET); - - final long startTime = System.currentTimeMillis(); - final long transactionId = nextUniqueTransaction(startTime); - final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs(); - - // get the data back from the server - Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final Set result = new HashSet<>(); - - for (; ; ) { - - boolean hasMoreEntries = in.readBoolean(); - - // number of entries in the chunk - long size = in.readInt(); - - for (int i = 0; i < size; i++) { - result.add(keyValueSerializer.readKey(in, local)); - } - - if (!hasMoreEntries) - break; - - compact(in); - - in = blockingFetchReadOnly(timeoutTime, transactionId); - } - - return result; - } - - private long writeEvent(EventId event) { - buffer.clear(); - bytes.clear(); - - - bytes.write((byte) event.ordinal()); - long sizeLocation = markSizeLocation(); - - - return sizeLocation; - } - - - /** - * skips for the transactionid - * - * @param event - * @return - */ - private long writeEventAnSkip(EventId event) { - - long sizeLocation = writeEvent(event); - - // skips for the transaction id - bytes.skip(SIZE_OF_TRANSACTIONID); - return sizeLocation; - } - - - class Entry implements Map.Entry { - - final K key; - final V value; - - /** - * Creates new entry. - */ - Entry(K k1, V v) { - value = v; - key = k1; - } - - public final K getKey() { - return key; - } - - public final V getValue() { - return value; - } - - public final V setValue(V newValue) { - V oldValue = value; - StatelessChronicleMap.this.put((K) getKey(), (V) newValue); - return oldValue; - } - - public final boolean equals(Object o) { - if (!(o instanceof Map.Entry)) - return false; - final Map.Entry e = (Map.Entry) o; - final Object k1 = getKey(); - final Object k2 = e.getKey(); - if (k1 == k2 || (k1 != null && k1.equals(k2))) { - Object v1 = getValue(); - Object v2 = e.getValue(); - if (v1 == v2 || (v1 != null && v1.equals(v2))) - return true; - } - return false; - } - - public final int hashCode() { - return (key == null ? 0 : key.hashCode()) ^ - (value == null ? 0 : value.hashCode()); - } - - public final String toString() { - return getKey() + "=" + getValue(); - } - - } - - private Bytes blockingFetch(long sizeLocation) { - - try { - long startTime = System.currentTimeMillis(); - return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs(), - nextUniqueTransaction(startTime), startTime); - } catch (IOException e) { - close(); - throw new IORuntimeException(e); - } catch (Exception e) { - close(); - throw e; - } - - } - - - private Bytes blockingFetch0(long sizeLocation, final long transactionId, long startTime) { - - try { - return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs(), transactionId, startTime); - } catch (IOException e) { - close(); - throw new IORuntimeException(e); - } catch (Exception e) { - close(); - throw e; - } - - } - - - private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs, final long transactionId, final long startTime) throws IOException { - - final long timeoutTime = startTime + timeOutMs; - - for (; ; ) { - - if (clientChannel == null) - clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); - - try { - - if (LOG.isDebugEnabled()) - LOG.debug("sending data with transactionId=" + transactionId); - - writeSizeAndTransactionIdAt(sizeLocation, transactionId); - - // send out all the bytes - send(bytes, timeoutTime); - - bytes.clear(); - bytes.buffer().clear(); - - return blockingFetch(timeoutTime, transactionId); - - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { - checkTimeout(timeoutTime); - clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); - } - } - } - - private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { - - try { - return blockingFetch(timeoutTime, transactionId); - } catch (IOException e) { - close(); - throw new IORuntimeException(e); - } catch (Exception e) { - close(); - throw e; - } - - } - - - private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException { - - // the number of bytes in the response - final int size = receive(SIZE_OF_SIZE, timeoutTime).readInt(); - - final int requiredSize = size + SIZE_OF_SIZE; - - if (bytes.capacity() < requiredSize) { - - long pos = bytes.position(); - long limit = bytes.position(); - bytes.position(limit); - resizeBuffer(requiredSize, pos); - - } else - bytes.limit(bytes.capacity()); - - // block until we have received all the byte in this chunk - receive(size, timeoutTime); - - final boolean isException = bytes.readBoolean(); - final long inTransactionId = bytes.readLong(); - - if (inTransactionId != transactionId) - throw new IllegalStateException("the received transaction-id=" + inTransactionId + - ", does not match the expected transaction-id=" + transactionId); - - if (isException) { - Throwable throwable = (Throwable) bytes.readObject(); - try { - Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); - stackTrace.setAccessible(true); - List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); - // prune the end of the stack. - for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { - stes.remove(i); - } - InetSocketAddress address = builder.remoteAddress(); - stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); - StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); - for (int i = 4; i < stackTrace2.length; i++) - stes.add(stackTrace2[i]); - stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); - } catch (Exception ignore) { - } - NativeBytes.UNSAFE.throwException(throwable); - } - - return bytes; - } - - - private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { - - while (bytes.buffer().position() < requiredNumberOfBytes) { - - assert requiredNumberOfBytes <= bytes.capacity(); - - int len = clientChannel.read(bytes.buffer()); - - if (len == -1) - throw new IORuntimeException("Disconnected to remote server"); - - checkTimeout(timeoutTime); - } - - bytes.limit(bytes.buffer().position()); - return bytes; - } - - private void send(final Bytes out, long timeoutTime) throws IOException { - - buffer.limit((int) out.position()); - buffer.position(0); - - while (buffer.remaining() > 0) { - clientChannel.write(buffer); - checkTimeout(timeoutTime); - } - - out.clear(); - buffer.clear(); - } - - - private void checkTimeout(long timeoutTime) { - if (timeoutTime < System.currentTimeMillis()) - throw new RemoteCallTimeoutException(); - } - - private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { - final long size = bytes.position() - locationOfSize; - final long pos = bytes.position(); - bytes.position(locationOfSize); - bytes.writeInt((int) size - SIZE_OF_SIZE); - bytes.writeLong(transactionId); - bytes.position(pos); - } - - - private void writeSizeAt(long locationOfSize) { - final long size = bytes.position() - locationOfSize; - bytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); - } - - - private long markSizeLocation() { - final long position = bytes.position(); - bytes.skip(SIZE_OF_SIZE); - return position; - } - - - private void writeKey(K key) { - writeKey(key, null); - } - - - private void writeKey(K key, ThreadLocalCopies local) { - long start = bytes.position(); - - for (; ; ) { - try { - keyValueSerializer.writeKey(key, bytes, local); - return; - } catch (IndexOutOfBoundsException e) { - resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); - } catch (IllegalArgumentException e) { - resizeToMessage(start, e); - } catch (IllegalStateException e) { - if (e.getMessage().contains("Not enough available space")) - resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); - else - throw e; - } - } - } - - private V readValue(final long sizeLocation) { - return keyValueSerializer.readValue(blockingFetch(sizeLocation), null); - } - - private V readValue(final long sizeLocation, ThreadLocalCopies local) { - return keyValueSerializer.readValue(blockingFetch(sizeLocation), local); - } - - private void writeValue(V value) { - writeValue(value, null); - } - - private void writeValue(V value, ThreadLocalCopies local) { - - - long start = bytes.position(); - for (; ; ) { - try { - assert bytes.position() == start; - bytes.limit(bytes.capacity()); - keyValueSerializer.writeValue(value, bytes, local); - - return; - } catch (IllegalArgumentException e) { - resizeToMessage(start, e); - - - } catch (IllegalStateException e) { - if (e.getMessage().contains("Not enough available space")) - resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); - else - throw e; - - } - } - - - } - - private void resizeToMessage(long start, IllegalArgumentException e) { - String message = e.getMessage(); - if (message.startsWith(START_OF)) { - String substring = message.substring("Attempt to write ".length(), message.length()); - int i = substring.indexOf(' '); - if (i != -1) { - int size = Integer.parseInt(substring.substring(0, i)); - - long requiresExtra = size - bytes.remaining(); - System.out.println(size + substring); - resizeBuffer((int) (bytes.capacity() + requiresExtra), start); - } else - resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); - } else - throw e; - } - } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMapUnsynchronized.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMapUnsynchronized.java new file mode 100644 index 000000000..ec777eaab --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMapUnsynchronized.java @@ -0,0 +1,1157 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import com.sun.jdi.connect.spi.ClosedConnectionException; +import net.openhft.chronicle.hash.RemoteCallTimeoutException; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.NativeBytes; +import net.openhft.lang.threadlocal.ThreadLocalCopies; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.File; +import java.io.IOException; +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.channels.SocketChannel; +import java.util.*; +import java.util.concurrent.Future; + +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.StatelessChronicleMapUnsynchronized.EventId.*; + +/** + * @author Rob Austin. + */ +class StatelessChronicleMapUnsynchronized implements ChronicleMap, Closeable { + + public static final String START_OF = "Attempt to write "; + private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMapUnsynchronized.class); + public static final byte STATELESS_CLIENT_IDENTIFIER = (byte) -127; + + private final byte[] connectionByte = new byte[1]; + private final ByteBuffer connectionOutBuffer = ByteBuffer.wrap(connectionByte); + private final String name; + private final AbstractChronicleMapBuilder chronicleMapBuilder; + + private ByteBuffer buffer; + private ByteBufferBytes bytes; + + private final KeyValueSerializer keyValueSerializer; + private volatile SocketChannel clientChannel; + + private CloseablesManager closeables; + private final StatelessMapConfig config; + private int maxEntrySize; + private final Class kClass; + private final Class vClass; + private final boolean putReturnsNull; + private final boolean removeReturnsNull; + + + static enum EventId { + HEARTBEAT, + STATEFUL_UPDATE, + LONG_SIZE, SIZE, + IS_EMPTY, + CONTAINS_KEY, + CONTAINS_VALUE, + GET, + PUT, + PUT_WITHOUT_ACC, + REMOVE, + REMOVE_WITHOUT_ACC, + CLEAR, + KEY_SET, + VALUES, + ENTRY_SET, + REPLACE, + REPLACE_WITH_OLD_AND_NEW_VALUE, + PUT_IF_ABSENT, + REMOVE_WITH_VALUE, + TO_STRING, + PUT_ALL, + PUT_ALL_WITHOUT_ACC, + HASH_CODE, + MAP_FOR_KEY, + UPDATE_FOR_KEY + } + + private long transactionID; + + StatelessChronicleMapUnsynchronized(final StatelessMapConfig config, + final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { + this.chronicleMapBuilder = chronicleMapBuilder; + this.config = config; + this.keyValueSerializer = new KeyValueSerializer(chronicleMapBuilder.keyBuilder, + chronicleMapBuilder.valueBuilder); + this.putReturnsNull = chronicleMapBuilder.putReturnsNull(); + this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull(); + this.maxEntrySize = chronicleMapBuilder.entrySize(true); + + if (maxEntrySize < 128) + maxEntrySize = 128; // the stateless client will not work with extreemly small buffers + + this.vClass = chronicleMapBuilder.valueBuilder.eClass; + this.kClass = chronicleMapBuilder.keyBuilder.eClass; + this.name = chronicleMapBuilder.name(); + attemptConnect(config.remoteAddress()); + + buffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); + bytes = new ByteBufferBytes(buffer.slice()); + } + + + private SocketChannel lazyConnect(final long timeoutMs, + final InetSocketAddress remoteAddress) throws IOException { + + if (LOG.isDebugEnabled()) + LOG.debug("attempting to connect to " + remoteAddress); + + SocketChannel result = null; + + long timeoutAt = System.currentTimeMillis() + timeoutMs; + + for (; ; ) { + + checkTimeout(timeoutAt); + + // ensures that the excising connection are closed + closeExisting(); + + try { + result = AbstractChannelReplicator.openSocketChannel(closeables); + result.connect(config.remoteAddress()); + result.socket().setTcpNoDelay(true); + + + doHandShaking(result); + break; + } catch (IOException e) { + closeables.closeQuietly(); + } catch (Exception e) { + closeables.closeQuietly(); + throw e; + } + + } + + return result; + } + + + /** + * attempts a single connect without a timeout and eat any IOException used typically in the constructor, + * its possible the server is not running with this instance is created, {@link + * net.openhft.chronicle.map.StatelessChronicleMapUnsynchronized#lazyConnect(long, + * java.net.InetSocketAddress)} will attempt to establish the connection when the client make the first + * map method call. + * + * @param remoteAddress + * @return + * @throws java.io.IOException + * @see net.openhft.chronicle.map.StatelessChronicleMapUnsynchronized#lazyConnect(long, + * java.net.InetSocketAddress) + */ + + private void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { + + // ensures that the excising connection are closed + closeExisting(); + + try { + clientChannel = AbstractChannelReplicator.openSocketChannel(closeables); + clientChannel.connect(remoteAddress); + doHandShaking(clientChannel); + + } catch (IOException e) { + closeables.closeQuietly(); + } + + } + + + /** + * closes the existing connections and establishes a new closeables + */ + private void closeExisting() { + + // ensure that any excising connection are first closed + if (closeables != null) + closeables.closeQuietly(); + + closeables = new CloseablesManager(); + } + + + /** + * initiates a very simple level of handshaking with the remote server, we send a special ID of -127 ( + * when the server receives this it knows its dealing with a stateless client, receive back an identifier + * from the server + * + * @param clientChannel + * @throws java.io.IOException + */ + private void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { + + connectionByte[0] = STATELESS_CLIENT_IDENTIFIER; + this.connectionOutBuffer.clear(); + + long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + + // write a single byte + while (connectionOutBuffer.hasRemaining()) { + clientChannel.write(connectionOutBuffer); + checkTimeout(timeoutTime); + } + + this.connectionOutBuffer.clear(); + + // read a single byte back + while (this.connectionOutBuffer.position() <= 0) { + clientChannel.read(this.connectionOutBuffer); + checkTimeout(timeoutTime); + } + + byte remoteIdentifier = connectionByte[0]; + + if (LOG.isDebugEnabled()) + LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier); + + } + + public File file() { + throw new UnsupportedOperationException(); + } + + public void close() { + if (closeables != null) + closeables.closeQuietly(); + closeables = null; + try { + Thread.sleep(100); + } catch (InterruptedException e) { + // + } + + } + + /** + * the transaction id are generated as unique timestamps + * + * @param time in milliseconds + * @return a unique transactionId + */ + private long nextUniqueTransaction(long time) { + transactionID = (time == transactionID) ? time + 1 : time; + return transactionID; + } + + + public V putIfAbsent(K key, V value) { + final long sizeLocation = writeEventAnSkip(PUT_IF_ABSENT); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + writeKey(key, local); + writeValue(value, local); + return readValue(sizeLocation, local); + } + + + public boolean remove(Object key, Object value) { + + if (key == null) + throw new NullPointerException("key can not be null"); + + final long sizeLocation = writeEventAnSkip(REMOVE_WITH_VALUE); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + writeKey((K) key, local); + writeValue((V) value, local); + + // get the data back from the server + return blockingFetch(sizeLocation).readBoolean(); + } + + + public boolean replace(K key, V oldValue, V newValue) { + if (key == null) + throw new NullPointerException("key can not be null"); + + final long sizeLocation = writeEventAnSkip(REPLACE_WITH_OLD_AND_NEW_VALUE); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + writeKey(key, local); + writeValue(oldValue, local); + writeValue(newValue, local); + + // get the data back from the server + return blockingFetch(sizeLocation).readBoolean(); + } + + + public V replace(K key, V value) { + if (key == null) + throw new NullPointerException("key can not be null"); + + final long sizeLocation = writeEventAnSkip(REPLACE); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + writeKey(key, local); + writeValue(value, local); + + // get the data back from the server + return readValue(sizeLocation, local); + } + + + public int size() { + return (int) longSize(); + } + + + /** + * calling this method should be avoided at all cost, as the entire {@code object} is serialized. This + * equals can be used to compare map that extends ChronicleMap. So two Chronicle Maps that contain the + * same data are considered equal, even if the instances of the chronicle maps were of different types + * + * @param object the object that you are comparing against + * @return true if the contain the same data + */ + @Override + public boolean equals(Object object) { + + if (this == object) return true; + if (object == null || object.getClass().isAssignableFrom(Map.class)) + return false; + + final Map that = (Map) object; + + final int size = size(); + + if (that.size() != size) + return false; + + final Set> entries = entrySet(); + + return that.entrySet().equals(entries); + + } + + + @Override + public int hashCode() { + return blockingFetch(writeEventAnSkip(HASH_CODE)).readInt(); + } + + public String toString() { + return blockingFetch(writeEventAnSkip(TO_STRING)).readObject(String.class); + } + + + public boolean isEmpty() { + return blockingFetch(writeEventAnSkip(IS_EMPTY)).readBoolean(); + } + + + public boolean containsKey(Object key) { + final long sizeLocation = writeEventAnSkip(CONTAINS_KEY); + writeKey((K) key); + + // get the data back from the server + return blockingFetch(sizeLocation).readBoolean(); + + } + + + public boolean containsValue(Object value) { + final long sizeLocation = writeEventAnSkip(CONTAINS_VALUE); + writeValue((V) value); + + // get the data back from the server + return blockingFetch(sizeLocation).readBoolean(); + } + + + public long longSize() { + return blockingFetch(writeEventAnSkip(LONG_SIZE)).readLong(); + } + + + public V get(Object key) { + final long sizeLocation = writeEventAnSkip(GET); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + writeKey((K) key, local); + + // get the data back from the server + return readValue(sizeLocation, local); + + } + + public V getUsing(K key, V usingValue) { + throw new UnsupportedOperationException("getUsing() is not supported for stateless " + + "clients"); + } + + + public V acquireUsing(@NotNull K key, V usingValue) { + throw new UnsupportedOperationException("acquireUsing() is not supported for stateless " + + "clients"); + } + + @NotNull + @Override + public WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { + throw new UnsupportedOperationException(); + } + + @NotNull + @Override + public ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue) { + throw new UnsupportedOperationException(); + } + + + public V remove(Object key) { + + if (key == null) + throw new NullPointerException("key can not be null"); + + final long sizeLocation = (removeReturnsNull) ? writeEvent(REMOVE_WITHOUT_ACC) : writeEventAnSkip(REMOVE); + + + writeKey((K) key); + + if (removeReturnsNull) { + sendWithoutAcc(bytes, sizeLocation); + return null; + } else + // get the data back from the server + return readValue(sizeLocation); + } + + + public V put(K key, V value) { + + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + + final long sizeLocation = (putReturnsNull) ? writeEvent(PUT_WITHOUT_ACC) : writeEventAnSkip(PUT); + + writeKey(key, local); + writeValue(value, local); + + + if (putReturnsNull) { + sendWithoutAcc(bytes, sizeLocation); + return null; + } else + // get the data back from the server + return readValue(sizeLocation, local); + + } + + + public R mapForKey(K key, @NotNull Function function) { + + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final long sizeLocation = writeEventAnSkip(MAP_FOR_KEY); + + writeKey(key, local); + writeObject(function); + + final Bytes reader = blockingFetch(sizeLocation); + return (R) keyValueSerializer.readObject(reader); + } + + @Override + public R updateForKey(K key, @NotNull Mutator mutator) { + + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final long sizeLocation = writeEventAnSkip(UPDATE_FOR_KEY); + + writeKey(key, local); + writeObject(mutator); + + final Bytes reader = blockingFetch(sizeLocation); + return (R) keyValueSerializer.readObject(reader); + } + + @Override + public Future getLater(K key) { + throw new UnsupportedOperationException(); + } + + + private void writeObject(@NotNull Object function) { + long start = bytes.position(); + for (; ; ) { + try { + bytes.writeObject(function); + return; + } catch (IllegalStateException e) { + + Throwable cause = e.getCause(); + + if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) { + LOG.debug("resizing buffer"); + resizeBuffer(buffer.capacity() + maxEntrySize, start); + + } else + throw e; + + } + } + } + + + public void putAll(Map map) { + + final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); + + writeEntries(map); + + if (putReturnsNull) + sendWithoutAcc(bytes, sizeLocation); + else + blockingFetch(sizeLocation); + + } + + + private void sendWithoutAcc(final Bytes bytes, long sizeLocation) { + writeSizeAt(sizeLocation); + long timeoutTime = config.timeoutMs() + System.currentTimeMillis(); + try { + for (; ; ) { + + if (clientChannel == null) + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + + try { + + send(bytes, timeoutTime); + return; + } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + checkTimeout(timeoutTime); + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + } + } + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } + } + + + private void writeEntries(Map map) { + + final int numberOfEntries = map.size(); + int numberOfEntriesReadSoFar = 0; + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + + bytes.writeStopBit(numberOfEntries); + assert bytes.limit() == bytes.capacity(); + for (final Map.Entry e : map.entrySet()) { + + numberOfEntriesReadSoFar++; + + long start = bytes.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 K key = e.getKey(); + + final Class keyClass = key.getClass(); + if (!kClass.isAssignableFrom(keyClass)) + throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + + "and should" + + " be of type=" + kClass); + + + writeKey(key, local); + + final V value = e.getValue(); + + final Class valueClass = value.getClass(); + if (!vClass.isAssignableFrom(valueClass)) + throw new ClassCastException("value=" + value + " is of type=" + valueClass + + " and " + + "should be of type=" + vClass); + + + writeValue(value, local); + + final int len = (int) (bytes.position() - start); + + if (len > maxEntrySize) + maxEntrySize = len; + } + } + + private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) { + + final long remaining = bytes.remaining(); + + if (remaining < maxEntrySize) { + long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar); + resizeBuffer((int) (estimatedRequiredSize + maxEntrySize), start); + } + } + + /** + * estimates the size based on what been completed so far + * + * @param numberOfEntries + * @param numberOfEntriesReadSoFar + */ + private long estimateSize(int numberOfEntries, int numberOfEntriesReadSoFar) { + // we will back the size estimate on what we have so far + final double percentageComplete = (double) numberOfEntriesReadSoFar / (double) numberOfEntries; + return (long) ((double) bytes.position() / percentageComplete); + } + + + private void resizeBuffer(int size, long start) { + + if (LOG.isDebugEnabled()) + LOG.debug("resizing buffer to size=" + size); + + if (size < buffer.capacity()) + throw new IllegalStateException("it not possible to resize the buffer smaller"); + + assert size < Integer.MAX_VALUE; + + final ByteBuffer result = ByteBuffer.allocate((int) size).order(ByteOrder.nativeOrder()); + + final long bytesPosition = bytes.position(); + + bytes = new ByteBufferBytes(result.slice()); + + buffer.position(0); + buffer.limit((int) bytesPosition); + + int numberOfLongs = (int) bytesPosition / 8; + + // chunk in longs first + for (int i = 0; i < numberOfLongs; i++) { + bytes.writeLong(buffer.getLong()); + } + + for (int i = numberOfLongs * 8; i < bytesPosition; i++) { + bytes.writeByte(buffer.get()); + } + + buffer = result; + + assert buffer.capacity() == bytes.capacity(); + + assert buffer.capacity() == size; + assert buffer.capacity() == bytes.capacity(); + assert bytes.limit() == bytes.capacity(); + bytes.position(start); + } + + + public void clear() { + + // get the data back from the server + blockingFetch(writeEventAnSkip(CLEAR)); + } + + + @NotNull + public Collection values() { + final long sizeLocation = writeEventAnSkip(VALUES); + + final long startTime = System.currentTimeMillis(); + final long transactionId = nextUniqueTransaction(startTime); + final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + + // get the data back from the server + Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final Collection result = new ArrayList(); + + for (; ; ) { + + final boolean hasMoreEntries = in.readBoolean(); + + // number of entries in this chunk + final long size = in.readInt(); + + for (int i = 0; i < size; i++) { + result.add(keyValueSerializer.readValue(in, local)); + } + + if (!hasMoreEntries) + break; + + compact(in); + + in = blockingFetchReadOnly(timeoutTime, transactionId); + } + return result; + } + + + @NotNull + public Set> entrySet() { + final long sizeLocation = writeEventAnSkip(ENTRY_SET); + + final long startTime = System.currentTimeMillis(); + final long transactionId = nextUniqueTransaction(startTime); + final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + + // get the data back from the server + Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final Map result = new HashMap(); + + for (; ; ) { + + final boolean hasMoreEntries = in.readBoolean(); + + // number of entries in this chunk + final long size = in.readInt(); + + for (int i = 0; i < size; i++) { + final K k = keyValueSerializer.readKey(in, local); + final V v = keyValueSerializer.readValue(in, local); + result.put(k, v); + } + + if (!hasMoreEntries) + break; + + compact(in); + + in = blockingFetchReadOnly(timeoutTime, transactionId); + } + + + return result.entrySet(); + } + + private void compact(Bytes in) { + if (in.remaining() == 0) { + bytes.clear(); + bytes.buffer().clear(); + } else { + buffer.compact(); + } + } + + + @NotNull + public Set keySet() { + final long sizeLocation = writeEventAnSkip(KEY_SET); + + final long startTime = System.currentTimeMillis(); + final long transactionId = nextUniqueTransaction(startTime); + final long timeoutTime = System.currentTimeMillis() + config.timeoutMs(); + + // get the data back from the server + Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final Set result = new HashSet<>(); + + for (; ; ) { + + boolean hasMoreEntries = in.readBoolean(); + + // number of entries in the chunk + long size = in.readInt(); + + for (int i = 0; i < size; i++) { + result.add(keyValueSerializer.readKey(in, local)); + } + + if (!hasMoreEntries) + break; + + compact(in); + + in = blockingFetchReadOnly(timeoutTime, transactionId); + } + + return result; + } + + private long writeEvent(StatelessChronicleMapUnsynchronized.EventId event) { + buffer.clear(); + bytes.clear(); + + + bytes.write((byte) event.ordinal()); + long sizeLocation = markSizeLocation(); + + + return sizeLocation; + } + + + /** + * skips for the transactionid + * + * @param event + * @return + */ + private long writeEventAnSkip(EventId event) { + + long sizeLocation = writeEvent(event); + + // skips for the transaction id + bytes.skip(SIZE_OF_TRANSACTIONID); + return sizeLocation; + } + + + class Entry implements Map.Entry { + + final K key; + final V value; + + /** + * Creates new entry. + */ + Entry(K k1, V v) { + value = v; + key = k1; + } + + public final K getKey() { + return key; + } + + public final V getValue() { + return value; + } + + public final V setValue(V newValue) { + V oldValue = value; + StatelessChronicleMapUnsynchronized.this.put((K) getKey(), (V) newValue); + return oldValue; + } + + public final boolean equals(Object o) { + if (!(o instanceof Map.Entry)) + return false; + final Map.Entry e = (Map.Entry) o; + final Object k1 = getKey(); + final Object k2 = e.getKey(); + if (k1 == k2 || (k1 != null && k1.equals(k2))) { + Object v1 = getValue(); + Object v2 = e.getValue(); + if (v1 == v2 || (v1 != null && v1.equals(v2))) + return true; + } + return false; + } + + public final int hashCode() { + return (key == null ? 0 : key.hashCode()) ^ + (value == null ? 0 : value.hashCode()); + } + + public final String toString() { + return getKey() + "=" + getValue(); + } + + } + + private Bytes blockingFetch(long sizeLocation) { + + try { + long startTime = System.currentTimeMillis(); + return blockingFetchThrowable(sizeLocation, this.config.timeoutMs(), + nextUniqueTransaction(startTime), startTime); + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } catch (Exception e) { + close(); + throw e; + } + + } + + + private Bytes blockingFetch0(long sizeLocation, final long transactionId, long startTime) { + + try { + return blockingFetchThrowable(sizeLocation, this.config.timeoutMs(), transactionId, startTime); + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } catch (Exception e) { + close(); + throw e; + } + + } + + + private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs, final long transactionId, final long startTime) throws IOException { + + final long timeoutTime = startTime + timeOutMs; + + for (; ; ) { + + if (clientChannel == null) + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + + try { + + if (LOG.isDebugEnabled()) + LOG.debug("sending data with transactionId=" + transactionId); + + writeSizeAndTransactionIdAt(sizeLocation, transactionId); + + // send out all the bytes + send(bytes, timeoutTime); + + bytes.clear(); + bytes.buffer().clear(); + + return blockingFetch(timeoutTime, transactionId); + + } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + checkTimeout(timeoutTime); + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + } + } + } + + private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { + + try { + return blockingFetch(timeoutTime, transactionId); + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } catch (Exception e) { + close(); + throw e; + } + + } + + + private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException { + + // the number of bytes in the response + final int size = receive(SIZE_OF_SIZE, timeoutTime).readInt(); + + final int requiredSize = size + SIZE_OF_SIZE; + + if (bytes.capacity() < requiredSize) { + + long pos = bytes.position(); + long limit = bytes.position(); + bytes.position(limit); + resizeBuffer(requiredSize, pos); + + } else + bytes.limit(bytes.capacity()); + + // block until we have received all the byte in this chunk + receive(size, timeoutTime); + + final boolean isException = bytes.readBoolean(); + final long inTransactionId = bytes.readLong(); + + if (inTransactionId != transactionId) + throw new IllegalStateException("the received transaction-id=" + inTransactionId + + ", does not match the expected transaction-id=" + transactionId); + + if (isException) { + Throwable throwable = (Throwable) bytes.readObject(); + try { + Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); + stackTrace.setAccessible(true); + List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); + // prune the end of the stack. + for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { + stes.remove(i); + } + InetSocketAddress address = config.remoteAddress(); + stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); + StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); + for (int i = 4; i < stackTrace2.length; i++) + stes.add(stackTrace2[i]); + stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); + } catch (Exception ignore) { + } + NativeBytes.UNSAFE.throwException(throwable); + } + + return bytes; + } + + + private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { + + while (bytes.buffer().position() < requiredNumberOfBytes) { + + assert requiredNumberOfBytes <= bytes.capacity(); + + int len = clientChannel.read(bytes.buffer()); + + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); + + checkTimeout(timeoutTime); + } + + bytes.limit(bytes.buffer().position()); + return bytes; + } + + private void send(final Bytes out, long timeoutTime) throws IOException { + + buffer.limit((int) out.position()); + buffer.position(0); + + while (buffer.remaining() > 0) { + clientChannel.write(buffer); + checkTimeout(timeoutTime); + } + + out.clear(); + buffer.clear(); + } + + + private void checkTimeout(long timeoutTime) { + if (timeoutTime < System.currentTimeMillis()) + throw new RemoteCallTimeoutException(); + } + + private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { + final long size = bytes.position() - locationOfSize; + final long pos = bytes.position(); + bytes.position(locationOfSize); + bytes.writeInt((int) size - SIZE_OF_SIZE); + bytes.writeLong(transactionId); + bytes.position(pos); + } + + + private void writeSizeAt(long locationOfSize) { + final long size = bytes.position() - locationOfSize; + bytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); + } + + + private long markSizeLocation() { + final long position = bytes.position(); + bytes.skip(SIZE_OF_SIZE); + return position; + } + + + private void writeKey(K key) { + writeKey(key, null); + } + + + private void writeKey(K key, ThreadLocalCopies local) { + long start = bytes.position(); + + for (; ; ) { + try { + keyValueSerializer.writeKey(key, bytes, local); + return; + } catch (IndexOutOfBoundsException e) { + resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); + } catch (IllegalArgumentException e) { + resizeToMessage(start, e); + } catch (IllegalStateException e) { + if (e.getMessage().contains("Not enough available space")) + resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); + else + throw e; + } + } + } + + private V readValue(final long sizeLocation) { + return keyValueSerializer.readValue(blockingFetch(sizeLocation), null); + } + + private V readValue(final long sizeLocation, ThreadLocalCopies local) { + return keyValueSerializer.readValue(blockingFetch(sizeLocation), local); + } + + private void writeValue(V value) { + writeValue(value, null); + } + + private void writeValue(V value, ThreadLocalCopies local) { + + + long start = bytes.position(); + for (; ; ) { + try { + assert bytes.position() == start; + bytes.limit(bytes.capacity()); + keyValueSerializer.writeValue(value, bytes, local); + + return; + } catch (IllegalArgumentException e) { + resizeToMessage(start, e); + + + } catch (IllegalStateException e) { + if (e.getMessage().contains("Not enough available space")) + resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); + else + throw e; + + } + } + + + } + + private void resizeToMessage(long start, IllegalArgumentException e) { + String message = e.getMessage(); + if (message.startsWith(START_OF)) { + String substring = message.substring("Attempt to write ".length(), message.length()); + int i = substring.indexOf(' '); + if (i != -1) { + int size = Integer.parseInt(substring.substring(0, i)); + + long requiresExtra = size - bytes.remaining(); + System.out.println(size + substring); + resizeBuffer((int) (bytes.capacity() + requiresExtra), start); + } else + resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); + } else + throw e; + } + +} + diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 88471c00c..956a00297 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -42,7 +42,7 @@ import static java.nio.channels.SelectionKey.*; import static java.util.concurrent.TimeUnit.MILLISECONDS; 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.HEARTBEAT; +import static net.openhft.chronicle.map.StatelessChronicleMapUnsynchronized.EventId.HEARTBEAT; interface Work { @@ -1259,7 +1259,8 @@ class StatelessServerConnector { ("mapSupportsBytes")); private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class .getName()); - public static final StatelessChronicleMap.EventId[] VALUES = StatelessChronicleMap.EventId.values(); + public static final StatelessChronicleMapUnsynchronized.EventId[] VALUES + = StatelessChronicleMapUnsynchronized.EventId.values(); public static final int SIZE_OF_IS_EXCEPTION = 1; public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTIONID; @@ -1283,7 +1284,7 @@ class StatelessServerConnector { @NotNull final ByteBufferBytes reader) { - final StatelessChronicleMap.EventId event = VALUES[eventId]; + final StatelessChronicleMapUnsynchronized.EventId event = VALUES[eventId]; // these methods don't return a result switch (event) { diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 31219660d..8e3430210 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -303,7 +303,7 @@ class VanillaChronicleMap, } MapEventListener> eventListener() { - return eventListener; + return eventListener == null ? MapEventListeners.NOP : eventListener; } void checkKey(Object key) { diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 16fd70559..b0e40e03b 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -25,6 +25,7 @@ import org.junit.Test; import java.io.IOException; import java.net.InetSocketAddress; import java.util.*; +import java.util.concurrent.ExecutionException; /** @@ -398,6 +399,37 @@ public class StatelessClientTest { } + + @Test + public void testGetLater() throws IOException, + InterruptedException, ExecutionException { + + try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { + + try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { + + statelessMap.put(1, "some value"); + + Assert.assertEquals("some value", statelessMap.getLater(1).get()); + Assert.assertEquals(1, statelessMap.size()); + + statelessMap.remove(1); + + + Assert.assertEquals(null, statelessMap.getLater(1).get()); + + + Assert.assertEquals(0, statelessMap.size()); + + } + } + + } + + @Test public void testEquals() throws IOException, InterruptedException {