From abcf1fabc4554a23657c6f47243af5bbe274fac0 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 25 Nov 2014 13:21:46 +0000 Subject: [PATCH] HCOLL-216 refactored so that we only have one StatelessChronicleMap class --- .../map/AbstractChannelReplicator.java | 2 +- .../chronicle/map/StatelessChronicleMap.java | 1175 +++++++++++++++-- .../StatelessChronicleMapUnsynchronized.java | 1162 ---------------- .../openhft/chronicle/map/TcpReplicator.java | 53 +- 4 files changed, 1082 insertions(+), 1310 deletions(-) delete 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 41bb91813..a18173bdc 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(StatelessChronicleMapUnsynchronized.EventId.STATEFUL_UPDATE.ordinal()); + in.writeByte(StatelessChronicleMap.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 e53e32216..757c43f39 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -18,39 +18,118 @@ 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.thread.NamedThreadFactory; +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.util.Collection; -import java.util.Map; -import java.util.Set; +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.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, 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 final AbstractChronicleMapBuilder chronicleMapBuilder; - private final StatelessMapConfig builder; - private final ChronicleMap statelessClient; + + 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; - private ChronicleMap asyncStatelessClient; private ExecutorService executorService; - public StatelessChronicleMap(final StatelessMapConfig builder, - final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { - this.builder = builder; - this.chronicleMapBuilder = chronicleMapBuilder; - statelessClient = new StatelessChronicleMapUnsynchronized(builder, chronicleMapBuilder); + 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 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()); } @@ -59,7 +138,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return lazyExecutorService().submit(new Callable() { @Override public V call() throws Exception { - return lazyAsyncStatelessClient().get(key); + return StatelessChronicleMap.this.get(key); } }); } @@ -69,7 +148,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return lazyExecutorService().submit(new Callable() { @Override public V call() throws Exception { - return lazyAsyncStatelessClient().put(key, value); + + V oldValue = StatelessChronicleMap.this.put(key, value); + return oldValue; } }); } @@ -90,181 +171,1029 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - /** - * @return a lazily created instance of the StatelessClient used for the async calls. - */ - private ChronicleMap lazyAsyncStatelessClient() { + private SocketChannel lazyConnect(final long timeoutMs, + final InetSocketAddress remoteAddress) throws IOException { - if (asyncStatelessClient == null) { - synchronized (this) { - if (asyncStatelessClient == null) - try { - asyncStatelessClient = new StatelessChronicleMapUnsynchronized(builder, - chronicleMapBuilder); - } catch (IOException e) { - throw new RuntimeException(e); - } + 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 asyncStatelessClient; + return result; } /** - * 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 + * 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.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 java.io.IOException + * @see net.openhft.chronicle.map.StatelessChronicleMap#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 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(); + } + + + /** + * 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) { - return statelessClient.equals(object); + 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); + } @Override - public int hashCode() { - return statelessClient.hashCode(); + public synchronized int hashCode() { + return blockingFetch(writeEventAnSkip(HASH_CODE)).readInt(); } - @Override - public String toString() { - return statelessClient.toString(); + public synchronized String toString() { + return "name=" + name + ", " + blockingFetch(writeEventAnSkip(TO_STRING)).readObject(String + .class); } - @Override - public synchronized long longSize() { - return statelessClient.longSize(); - } - @Override - public synchronized int size() { - return statelessClient.size(); - } - - @Override public synchronized boolean isEmpty() { - return statelessClient.isEmpty(); + return blockingFetch(writeEventAnSkip(IS_EMPTY)).readBoolean(); } - @Override + public synchronized boolean containsKey(Object key) { - return statelessClient.containsKey(key); + final long sizeLocation = writeEventAnSkip(CONTAINS_KEY); + writeKey((K) key); + + // get the data back from the server + return blockingFetch(sizeLocation).readBoolean(); + } - @Override + public synchronized boolean containsValue(Object value) { - return statelessClient.containsValue(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(); + } + + public synchronized V get(Object key) { - return statelessClient.get(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); } - @Override - public synchronized V put(K key, V value) { - 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); + 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 synchronized WriteContext acquireUsingLocked(@NotNull K key, @NotNull V + usingValue) { + throw new UnsupportedOperationException(); } @NotNull @Override public synchronized ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue) { - return statelessClient.getUsingLocked(key, usingValue); + throw new UnsupportedOperationException(); } - @Override - public synchronized V acquireUsing(@NotNull K key, V usingValue) { - return statelessClient.acquireUsing(key, usingValue); + + 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); } - @NotNull - @Override - public synchronized WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { - return statelessClient.acquireUsingLocked(key, usingValue); + + 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); + } - @Override + public synchronized R mapForKey(K key, @NotNull Function function) { - return statelessClient.mapForKey(key, 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 synchronized R updateForKey(K key, @NotNull Mutator mutator) { - return statelessClient.updateForKey(key, 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 File file() { - return statelessClient.file(); + private synchronized 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 void close() { - statelessClient.close(); + + 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 V putIfAbsent(K key, V value) { - return statelessClient.putIfAbsent(key, value); + + private synchronized 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); + } } - @Override - public boolean remove(Object key, Object value) { - return statelessClient.remove(key, value); + + private synchronized 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 replace(K key, V oldValue, V newValue) { - return statelessClient.replace(key, oldValue, newValue); + 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 V replace(K key, V value) { - return statelessClient.replace(key, value); + /** + * 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 synchronized 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 synchronized 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 synchronized 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(StatelessChronicleMap.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.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/StatelessChronicleMapUnsynchronized.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMapUnsynchronized.java deleted file mode 100644 index a6da4ee41..000000000 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMapUnsynchronized.java +++ /dev/null @@ -1,1162 +0,0 @@ -/* - * 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(@NotNull K key) { - throw new UnsupportedOperationException(); - } - - @Override - public Future putLater(@NotNull K key, @NotNull V value) { - 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 956a00297..bef98ca3e 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.StatelessChronicleMapUnsynchronized.EventId.HEARTBEAT; +import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT; interface Work { @@ -54,8 +54,8 @@ interface Work { } /** - * Used with a {@link net.openhft.chronicle.map.ReplicatedChronicleMap} to send data between the maps using a - * socket connection {@link net.openhft.chronicle.map.TcpReplicator} + * Used with a {@link net.openhft.chronicle.map.ReplicatedChronicleMap} to send data between the + * maps using a socket connection {@link net.openhft.chronicle.map.TcpReplicator} * * @author Rob Austin. */ @@ -86,10 +86,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private long selectorTimeout; /** - * @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, it can not - * be smaller than the size of and entry, if it is set smaller the buffer - * will over flow, it can be larger then the entry, but setting it too - * large reduces the workable space in the buffer. + * @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, + * it can not be smaller than the size of and entry, if it is + * set smaller the buffer will over flow, it can be larger then + * the entry, but setting it too large reduces the workable + * space in the buffer. * @param statelessServerConnector set to NULL if not required * @throws IOException on an io error. */ @@ -295,7 +296,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * check to see if we have lost connection with the remote node and if we have attempts a reconnect. + * check to see if we have lost connection with the remote node and if we have attempts a + * reconnect. * * @param key the key relating to the heartbeat that we are checking * @param approxTimeOutTime the approximate time in milliseconds @@ -442,8 +444,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear all the - * old bootstrap information + * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear + * all the old bootstrap information * * @param key the nio SelectionKey */ @@ -455,7 +457,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * used to exchange identifiers and timestamps and heartbeat intervals between the server and client + * used to exchange identifiers and timestamps and heartbeat intervals between the server and + * client * * @param key the SelectionKey relating to the this cha * @param socketChannel @@ -651,10 +654,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * sets interestOps to "selector keys",The change to interestOps much be on the same thread as the - * selector. This class, allows via {@link AbstractChannelReplicator .KeyInterestUpdater#set(int)} to - * holds a pending change in interestOps ( via a bitset ), this change is processed later on the same - * thread as the selector + * sets interestOps to "selector keys",The change to interestOps much be on the same thread as + * the selector. This class, allows via {@link AbstractChannelReplicator + * .KeyInterestUpdater#set(int)} to holds a pending change in interestOps ( via a bitset ), + * this change is processed later on the same thread as the selector */ private static class KeyInterestUpdater { @@ -685,8 +688,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * @param keyIndex the index of the key that has changed, the list of keys is provided by the - * constructor {@link KeyInterestUpdater(int, SelectionKey[])} + * @param keyIndex the index of the key that has changed, the list of keys is provided by + * the constructor {@link KeyInterestUpdater(int, SelectionKey[])} */ public void set(int keyIndex) { changeOfOpWriteRequired.set(keyIndex); @@ -926,7 +929,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * writes all the entries that have changed, to the buffer which will later be written to TCP/IP + * writes all the entries that have changed, to the buffer which will later be written to + * TCP/IP * * @param modificationIterator a record of which entries have modification * @param selectionKey @@ -1004,7 +1008,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } /** - * used to send an single zero byte if we have not send any data for up to the localHeartbeatInterval + * used to send an single zero byte if we have not send any data for up to the + * localHeartbeatInterval */ private void writeHeartbeatToBuffer() { @@ -1021,8 +1026,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable /** - * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE will get - * added back in as soon as we have data to write + * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE + * will get added back in as soon as we have data to write * * @param socketChannel the socketChannel we wish to stop writing to * @param attached data associated with the socketChannels key @@ -1259,8 +1264,8 @@ class StatelessServerConnector { ("mapSupportsBytes")); private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class .getName()); - public static final StatelessChronicleMapUnsynchronized.EventId[] VALUES - = StatelessChronicleMapUnsynchronized.EventId.values(); + public static final StatelessChronicleMap.EventId[] VALUES + = StatelessChronicleMap.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; @@ -1284,7 +1289,7 @@ class StatelessServerConnector { @NotNull final ByteBufferBytes reader) { - final StatelessChronicleMapUnsynchronized.EventId event = VALUES[eventId]; + final StatelessChronicleMap.EventId event = VALUES[eventId]; // these methods don't return a result switch (event) {