diff --git a/README.md b/README.md index cdbbc3dac..9e26588df 100644 --- a/README.md +++ b/README.md @@ -260,11 +260,13 @@ You set the maximum number of entries by the builder: ``` java ConcurrentMap map = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + constantValueSizeBySample("a long sample string"), .entries(1000) // set the max number of entries here .create(); ``` In this example above we have set 1000 entries. + We have optimised chronicle, So that you can have situations where you either don't use; - all the entries you have allowed for. This works best on Unix where the disk space and memory @@ -275,6 +277,22 @@ lines (128 bytes +), only the lines you touch sit in your CPU cache and if you h (8+ Kbytes) only the pages you touch use memory or disk. The CPU cache usage matters as it can be 10000x smaller than main memory. +#### Entry size + +The size of each entry depends on the type of the Keys and Values, as some types are larger than others. For example, if an entry uses Integers for both the keys and values, both the key and value each take exactly 4 bytes. +There is also some other overhead which is internal to chronicle, such as its internal multi-map, which has an overhead of +* 6 bytes per entry for segment size of < 64k, +* 9 bytes per entry for segment size of < 16m but > 64k +* 12 bytes per entry for segment size of > 16m entries per segment. + +Also, if you create a replicated map, there is an additional 10 bytes per entry. + +We suggest you don't configure size for constant-sized keys or values, instead you can use the +builder methods .constantKeySizeBySample(sampleKey) and +.constantValueSizeBySample(sampleValue), For common types like Integer we suggest you don't use +these methods, for example ChronicleMap knows that Integer is 4 bytes long, Long is 8, etc. + + ### Size of space reserved on disk In linux, if you looked at the size of the 'file', it will report the used entry size so if you @@ -295,6 +313,26 @@ To illustrate this with an example - On Ubuntu we can create a 100 TB chronicle `ls -l` say the process virtual size / file size is 100 TB, however the resident memory via `du` says the size is 71 MB after adding 10000 entries. You can see the size actually used with du. +### How Operating Systems differ + +As a pure java library, the same chronicle map java byte code can be run on Windows, Linux and Mac OSX. +However these operating systems work with memory mapped files differently, these differences effect how +chronicle is able to map memory to a file, and hence this can impact the total number of entries that you are +able to configure. + +- Windows allocates memory and disk eagerly, Windows will fail if more than 4 GB is allocated in a single memory +mapping, ( calculated as 4GB = 2^20 * 4 KB pages). Windows doesn't fail when a memory mapped region is mapped, rather it will fail when it is used up. This limitation doesn't apply to newer or server based versions of Windows. Eager memory allocation means you can't map more than free memory, but it should reduce jitter when you use it. In the future we may support multiple mappings to avoid this limitation, but there is no immediate plan to do so. +- Linux allocates memory and disk lazily. Linux systems see a performance degradation at around 200% of main memory. +- Mac OSX allocates memory lazily and disk eagerly. + +Chronicle Map allocates head room which is a waste on Windows (Linux's sparse allocation means the head room has little +impact). + +##### For production +- on Windows we recommend you use map sizes of less than 4 GB each, and less than 50% main memory in total. +- on Linux we recommend you use small to large maps of less than double main memory. e.g. if you have a 128 GB server, we recommend you have less than 256 GB of maps on the server. +- on Mac OSX, we have no specific recommendations. + ### Chronicle Map Interface The Chronicle Map interface adds a few methods above an beyond the standard ConcurrentMap, the ChronicleMapBuilder can also be used to return the ChronicleMap, see the example below : @@ -312,18 +350,17 @@ supports the following methods : - [`ReadContext getUsingLocked(@NotNull K key, @NotNull V usingValue);`] (https://github.com/OpenHFT/Chronicle-Map#off-heap-storage-and-how-using-a-proxy-object-can-improve-performance) - [`WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue);`] (https://github.com/OpenHFT/Chronicle-Map#acquireusinglocked) -These methods let you provide the object which the data will be written to, even if the object is -immutable. For example +These methods let you provide the object which the data will be written to, but the value use to be mutable. For example ``` java -String myString = ""; -String myResult = map.getUsing("key", myString); +CharSequence using = new StringBuilder(); +CharSequence myResult = map.getUsing("key", using); // at this point the myString and myResult will both point to the same object ``` The `map.getUsing()` method is similar to `map.get()`, but because Chronicle Map stores its data off heap, if you were to call get("key"), a new object would be created each time, map.getUsing() works -by reusing the heap memory which was used by the original Object "myString". This technique provides +by reusing the heap memory which was used by the original Object "using". This technique provides you with better control over your object creation. Exactly like `map.getUsing()`, `map.acquireUsing()` will give you back a reference to an value @@ -416,7 +453,7 @@ bond.getCoupon() lets say that it is only the `coupon` field that we are interested in, then its better not to have to deserialise the whole object that implements the `BondVOInterface`. The `ChronicleMapBuilder` will look a the types of keys and values that you use, If the value type is a simple accessor/mutator interface that is exposing a non nested pojo, which uses simple types - like String and primitives with corresponding get..() and + like `CharSequence` and primitives with corresponding get..() and set..() methods, Chronicle is able to generate off heap poxies so the whole object is not desrialized each time it is accessed, The off heap poxies are able to read and write into @@ -424,7 +461,7 @@ the off heap data structures directly, this reduced serialisation can give you a Below we show you how you can work directly with the off heap entries. ``` java - ChronicleMap chm = ChronicleMapBuilder + ChronicleMap chm = ChronicleMapBuilder .of(String.class, BondVOInterface.class) .keySize(10) .create(); @@ -433,7 +470,7 @@ Below we show you how you can work directly with the off heap entries. notice that the ``` java -.of(String.class, BondVOInterface.class) +.of(CharSequence.class, BondVOInterface.class) ``` value class, in our case `BondVOInterface.class` is an `interface` rather than a `class`, now @@ -505,14 +542,13 @@ context.close() // the lock will get released when this is called ``` #### acquireUsingLocked() - -just like getUsing(), acquireUsing() will also recycle the value you pass it, the following -code is a pattern that you will often come across, acquireUsing(key,value) offers this -functionality with a single method call : +Just like getUsing(), acquireUsing() will also recycle the value you pass it, the following +code is a pattern that you will often come across, `acquireUsing(key,value)` offers this +functionality i the example, below with a single method call : ``` java V acquireUsing(key,value) { - Lock l = ...; // the segement lock of the map + Lock l = ...; // the segment lock of the map l.lock(); try { V v = map.getUsing(key,value) @@ -527,12 +563,12 @@ V acquireUsing(key,value) { } ``` -if you are only accessing ChronicleMap from a single thread and you are not doing replication -and don't care about atomic reads, then its simpler ( and faster ) to use acquireUsing() otherwise we -recommend acquireUsingLocked(key,value) +If you are only accessing ChronicleMap from a single thread. If you are not doing replication +and don't care about atomic reads. Then its simpler ( and faster ) to use acquireUsing() otherwise we +recommend you use `acquireUsingLocked(key,value)` -because acquireUsing can end up creating an entry the acquireUsingLocked(key,value) method holds -a segment write lock, this is unlike the getUsing(key,using) method that holds a segment read lock. +Since the acquireUsing() method can end up creating an entry, the acquireUsingLocked(key,value) method must hold +a segment write lock, this is unlike getUsing(key,using) which only holds a segment read lock. ``` java BondVOInterface bond = ... // create your instance @@ -547,17 +583,17 @@ try (WriteContext context = map.acquireUsingLocked("one", bo } ``` -if after you have read the 'issueDate' and 'symbol' and you wish to remove the entry based on some -business logic, it more efficient to use the 'context' to remove the entry, as the contents is +If after you have read the 'issueDate' and 'symbol' and you wish to remove the entry based on some +business logic, its more efficient to use the 'context' to remove the entry, as the contents is already aware when the entry is in memory. ## Oversized Entries Support -It is possible for the size of your entry to be twice as large as the maximum entry size, +It is possible for the size of your entry to be 64 times as large as the maximum entry size, we refer to this type of entry as an oversized entry. Oversized entries are there to cater for the case where only a small -percentage of your entries are twise as large as the others, in this case your large entry will -span across two entries. The alternative would be to increase your maximum entry size to be similar +percentage of your entries are larger than the others, in this case your large entry will +span across a number of entries. The alternative would be to increase your maximum entry size to be similar to the size of the largest entry, but this approach is wasteful of memory, especially when most entries are no where near the max entry size. @@ -592,10 +628,12 @@ void putAll(File fromFile) throws IOException; Its only the entries of your map that are exported, not the configuration of your map. So care must be taken to populate the data in to a map of the correct Key/Value type and with enough available entries. When importing data : + * entries that are in the map but not in the JSON file will remain untouched. * entries that are in the map and in the JSON file will be updated. -* entries that are not in the map but are in the JSON file wil added. -* In other words importing data into a Chronicle Map works like `map.putAll()`. +* entries that are not in the map but are in the JSON file will be added. + +In other words importing data into a Chronicle Map works like `map.putAll()`. When Importing data if you are also writing to the map at the same time, the last update will win. In other words a write lock is not held for the entire import process. @@ -610,19 +648,23 @@ VALUES, future versions will support a binary encoding of objects that are `net. ## Close Unlike ConcurrentHashMap, chronicle map stores its data off heap, often in a memory mapped file. Its recommended that you call close() once you have finished working with a Chronicle Map. + ``` java map.close() ``` -You only need to close to clean up resources deterministically. If your program is exiting, -you don't need to close the collection, as Chronicle never knows when the program might crash, -so we have designed it so you don't have to close() it. +This is especially important when working with chronicle map replication, as failure to call close may prevent +you from restarting a replicated map on the same port. In the event that your application crashes it may not +be possible to call close(). Your operating system will usually close dangling ports automatically, +so although it is recommended that you close() when you have finished with the map, +its not something that you must do, its just something that we recommend you should do. -WARNING : If you call close too early before you have finished working with the map, this can cause +###### WARNING + +If you call close too early before you have finished working with the map, this can cause your JVM to crash. Close MUST BE the last thing that you do with the map. - # TCP / UDP Replication Chronicle Map supports both TCP and UDP replication @@ -1087,13 +1129,13 @@ For in memory data-structures like a HashMap this isn’t a big problem. But in So if you don’t require the old value and don’t wish to block until your `put()` has been received by the server, then you may wish to consider using the following configuration : ``` java -.putReturnsNull +.putReturnsNull(true) ``` and also for the `remove()` method ``` java -.removeReturnsNull +.removeReturnsNull(true) ``` ``` java @@ -1114,6 +1156,55 @@ ChronicleMapBuilder.of(Integer.class, CharSequence.class) .create(); ``` +##### Performance + +The throughput and latency performance for different configurations. + +Tested using a test called BGChronicleTest. + +On one machine (i7 3.5 GHz) we have two persisted replicas run as + +-Dreplicas=2 eg.BGChronicleTest server + +On another machine (Dual Xeon 8 core 2.6 GHz) connected via a pair of Solarflare SFN5121T 10 Gig-E with onload enabled. + +-Dreplicas=2 -Dclients={see below} -DmaxRate=30000 -DreadRatio=2 eg.BGChronicleTest client + +2 clients +Throughput test +messages per seconds: 58,864 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 33 / 80 / 111 // 120 / 148 / 3,921 us + +4 clients +Throughput test +messages per seconds: 94,006 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 32 / 94 / 106 // 131 / 177 / 2,153 us + +8 clients +Throughput test +messages per seconds: 162,961 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 35 / 94 / 117 // 140 / 167 / 1,685 us + +16 clients +Throughput test +messages per seconds: 267,097 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 38 / 97 / 122 // 149 / 174 / 2,771 us + +24 clients +Throughput test +messages per seconds: 253,052 + +Latency test at 30,000 msg/sec +50% / 90% / 99% // 99.9% / 99.99% / worst latency was 40 / 99 / 121 // 151 / 243 / 3,669 us + ##### Close its always important to close `ChronicleMap`'s and `ChronicleSet` 's when you have finished with them diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 80267adbf..8f87caed2 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -53,7 +53,7 @@ abstract class AbstractChannelReplicator implements Closeable { public static final int SIZE_OF_SIZE = 4; public static final int SIZE_OF_TRANSACTION_ID = 8; - public static final int SIZE_OF_IS_EXCEPTION = 1; + public static final int SIZE_OF_TIME_SHIFT = 2; private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index e0cef2c72..c7e3764fd 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -211,68 +211,6 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { */ R updateForKey(K key, @NotNull Mutator mutator); - /** - * Returns a {@link java.util.concurrent.Future} containing the value to which the specified key - * is mapped, or {@code null} if this map contains no mapping for the key. This method behaves - * the same as {@link java.util.Map#get(java.lang.Object)} yet wraps result in a future a {@link - * java.util.concurrent.Future}. - * - * Calling this method may or may not block until the operating is complete depending on the - * implementation. - * - * @param key the key whose associated value is to be returned in the future - * @return a future containing the value to which the specified key is mapped after this method - * call, or {@code null} if no value is mapped - * @see java.util.Map#get(java.lang.Object) - * @see java.util.concurrent.Future - */ - Future getLater(@NotNull K key); - - /** - * Associates the specified value with the specified key in this map (optional operation). If - * the map previously contained a mapping for the key, the old value is replaced by the - * specified value. (A map m is said to contain a mapping for a key k if and - * only if {@link #containsKey(Object) m.containsKey(k)} would return true.) - * - * Calling this method may or may not block until the operating is complete depending on the - * implementation. - * - * @param key key with which the specified value is to be associated - * @param value value to be associated with the specified key * @return a future containing the - * previous value associated with key, or null if there was no - * mapping for key. (A null return can also indicate that the map - * previously associated null with key, if the implementation - * supports null values.) - * @return a future containing either, the previous value associated with key, or - * null if there was no mapping for key. - * @see java.util.Map#put(java.lang.Object, java.lang.Object) - * @see java.util.concurrent.Future - */ - Future putLater(@NotNull K key, @NotNull V value); - - /** - * Removes the mapping for a key from this map if it is present (optional operation). More - * formally, if this map contains a mapping from key k to value v such that - * (key==null ? k==null : key.equals(k)), that mapping is removed. (The map can - * contain at most one such mapping.) - * - *

Returns a future containing, either the value to which this map previously associated the - * key, or null if the map contained no mapping for the key. - * - * This method behaves the same as {@link java.util.Map#remove(java.lang.Object)} yet wraps - * result in a future a {@link java.util.concurrent.Future}. - * - * Calling this method may or may not block until the operating is complete depending on the - * implementation. - * - * @param key key whose mapping is to be removed from the map - * @return a future containing either, the previous value associated with key, or - * null if there was no mapping for key. - * @see java.util.Map#remove(java.lang.Object) - * @see java.util.concurrent.Future - */ - Future removeLater(@NotNull K key); - /** * exports all the entries to a {@link java.io.File} storing them in JSON format, an attempt is * made where possible to use standard java serialisation and keep the data human readable, data diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 59e2dbf3a..5a7a82458 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -40,13 +40,14 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.channels.SocketChannel; import java.util.*; -import java.util.concurrent.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; import static java.nio.ByteBuffer.allocateDirect; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID; +import static net.openhft.chronicle.map.AbstractChannelReplicator.*; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; /** @@ -125,6 +126,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon UPDATE_FOR_KEY } + + // used by the enterprise version + private int identifier; + + void identifier(int identifier) { + this.identifier = identifier; + } + + @NotNull private final AtomicLong transactionID = new AtomicLong(0); @@ -157,40 +167,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - @NotNull - @Override - public Future getLater(@NotNull final K key) { - return lazyExecutorService().submit(new Callable() { - @Nullable - @Override - public V call() throws Exception { - return StatelessChronicleMap.this.get(key); - } - }); - } - - @NotNull - @Override - public Future putLater(@NotNull final K key, @NotNull final V value) { - return lazyExecutorService().submit(new Callable() { - @Nullable - @Override - public V call() throws Exception { - return StatelessChronicleMap.this.put(key, value); - } - }); - } - - @Override - public Future removeLater(@NotNull final K key) { - return lazyExecutorService().submit(new Callable() { - @Override - public V call() throws Exception { - return StatelessChronicleMap.this.remove(key); - } - }); - } - @Override public void getAll(File toFile) { throw new UnsupportedOperationException(); @@ -238,9 +214,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon doHandShaking(result); break; } catch (IOException e) { - if (closeables!=null) closeables.closeQuietly(); + if (closeables != null) closeables.closeQuietly(); } catch (Exception e) { - if (closeables!=null) closeables.closeQuietly(); + if (closeables != null) closeables.closeQuietly(); throw e; } } @@ -271,7 +247,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon clientChannel.connect(remoteAddress); doHandShaking(clientChannel); } catch (IOException e) { - if (closeables!=null) closeables.closeQuietly(); + if (closeables != null) closeables.closeQuietly(); } } @@ -375,7 +351,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon @SuppressWarnings("NullableProblems") - public boolean remove( Object key, Object value) { + public boolean remove(Object key, Object value) { if (key == null) throw new NullPointerException(); @@ -394,7 +370,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } @SuppressWarnings("NullableProblems") - public V replace(K key, V value) { + public V replace(K key, V value) { if (key == null || value == null) throw new NullPointerException(); @@ -521,56 +497,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private synchronized void writeEntriesForPutAll(@NotNull Map map) { - final int numberOfEntries = map.size(); - int numberOfEntriesReadSoFar = 0; - - outBytes.writeStopBit(numberOfEntries); - assert outBytes.limit() == outBytes.capacity(); - - ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); - final Object keyWriter = keyWriterWithSize.writerForLoop(copies); - copies = valueWriterWithSize.getCopies(copies); - final Object valueWriter = valueWriterWithSize.writerForLoop(copies); - - for (final Map.Entry e : map.entrySet()) { - final K key = e.getKey(); - final V value = e.getValue(); - if (key == null || value == null) - throw new NullPointerException(); - - numberOfEntriesReadSoFar++; - - long start = outBytes.position(); - - // putAll if a bit more complicated than the others - // as the volume of data could cause us to have to resize our buffers - resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start); - - final Class keyClass = key.getClass(); - if (!kClass.isAssignableFrom(keyClass)) { - throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + - "and should" + - " be of type=" + kClass); - } - - writeKeyInLoop(key, keyWriter, copies); - - final Class valueClass = value.getClass(); - if (!vClass.isAssignableFrom(valueClass)) - throw new ClassCastException("value=" + value + " is of type=" + valueClass + - " and " + - "should be of type=" + vClass); - - writeValueInLoop(value, valueWriter, copies); - - final int len = (int) (outBytes.position() - start); - - if (len > maxEntrySize) - maxEntrySize = len; - } - } - private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) { final long remaining = outBytes.remaining(); @@ -755,7 +681,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public void putAll(@NotNull Map map) { - final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : + final long sizeLocation = putReturnsNull ? writeEventAnSkip(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); final long startTime = System.currentTimeMillis(); @@ -888,6 +814,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // skips for the transaction id outBytes.skip(SIZE_OF_TRANSACTION_ID); + outBytes.skip(SIZE_OF_TIME_SHIFT); + outBytes.writeByte(identifier); return sizeLocation; } @@ -966,10 +894,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon if (LOG.isDebugEnabled()) LOG.debug("sending data with transactionId=" + transactionId); - writeSizeAndTransactionIdAt(sizeLocation, transactionId); + writeSizeAndTransactionIdAt(sizeLocation, transactionId, startTime); // send out all the bytes - send(outBytes, timeoutTime); + writeBytesToSocket(outBytes, timeoutTime); outBytes.clear(); outBytes.buffer().clear(); break; @@ -1007,16 +935,57 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - // this is a transactionid and size that has been read by another thread. + // this is a transaction id and size that has been read by another thread. private volatile long parkedTransactionId; private volatile int parkedRemainingBytes; private volatile long parkedTransactionTimeStamp; private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, InterruptedException { - int remainingBytes = 0; + + int remainingBytes = nextEntry(timeoutTime, transactionId); + + if (inBytes.capacity() < remainingBytes) { + long pos = inBytes.position(); + long limit = inBytes.position(); + inBytes.position(limit); + resizeBufferInBuffer(remainingBytes, pos); + } else + inBytes.limit(inBytes.capacity()); + + // block until we have received all the bytes in this chunk + receive(remainingBytes, timeoutTime); + + final boolean isException = inBytes.readBoolean(); + + if (isException) { + Throwable throwable = (Throwable) inBytes.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(); + //noinspection ManualArrayToCollectionCopy + 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 inBytes; + } + + private int nextEntry(long timeoutTime, long transactionId) throws IOException { + int remainingBytes; for (; ; ) { // read the next item from the socket @@ -1084,47 +1053,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon pause(); } - - - final int minBufferSize = remainingBytes; - - if (inBytes.capacity() < minBufferSize) { - long pos = inBytes.position(); - long limit = inBytes.position(); - inBytes.position(limit); - resizeBufferInBuffer(minBufferSize, pos); - } else - inBytes.limit(inBytes.capacity()); - - // block until we have received all the bytes in this chunk - receive(remainingBytes, timeoutTime); - - final boolean isException = inBytes.readBoolean(); - - if (isException) { - Throwable throwable = (Throwable) inBytes.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(); - //noinspection ManualArrayToCollectionCopy - 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 inBytes; + return remainingBytes; } private void clearParked() { @@ -1172,7 +1101,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void send(@NotNull final Bytes out, long timeoutTime) throws IOException { + private void writeBytesToSocket(@NotNull final Bytes out, long timeoutTime) throws IOException { outBuffer.limit((int) out.position()); outBuffer.position(0); @@ -1191,7 +1120,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon throw new RemoteCallTimeoutException(); } - private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId) { + private void writeSizeAndTransactionIdAt(long locationOfSize, final long transactionId, long startTime) { final long size = outBytes.position() - locationOfSize; final long pos = outBytes.position(); outBytes.position(locationOfSize); @@ -1199,15 +1128,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert size0 > 0; outBytes.writeInt(size0); + assert transactionId != 0; outBytes.writeLong(transactionId); - outBytes.position(pos); - } + outBytes.writeUnsignedShort((int) (transactionId - startTime)); - private void writeSizeAt(long locationOfSize) { - final long size = outBytes.position() - locationOfSize; - int size0 = (int) size - SIZE_OF_SIZE; - assert size0 > 0; - outBytes.writeInt(locationOfSize, size0); + outBytes.position(pos); } private long markSizeLocation() { @@ -1576,9 +1501,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); copies = writeValue(value, copies); @@ -1608,9 +1531,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); copies = writeValue(value1, copies); @@ -1642,9 +1563,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytesLock.lock(); try { - final long sizeLocation = eventReturnsNull(eventId) - ? writeEvent(eventId) - : writeEventAnSkip(eventId); + final long sizeLocation = writeEventAnSkip(eventId); copies = writeKey(key); transactionId = send(sizeLocation, startTime); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 4bdad2b43..958d42ee1 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -974,24 +974,31 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // this can occur when new map are added to a channel final boolean handShakingComplete = attached.isHandShakingComplete(); - for (; ; ) { - final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0); + int entriesWritten = 0; + try { + for (; ; entriesWritten++) { + final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0); - if (!wasDataRead) { - // if we have no more data to write to the socket then we will - // un-register OP_WRITE on the selector, until more data becomes available - if (in.position() == 0 && handShakingComplete) - disableWrite(socketChannel, attached); + if (!wasDataRead) { + // if we have no more data to write to the socket then we will + // un-register OP_WRITE on the selector, until more data becomes available + if (in.position() == 0 && handShakingComplete) + disableWrite(socketChannel, attached); - return; + return; + } + + // we've filled up the buffer lets give another channel a chance to send + // some data + if (in.remaining() <= maxEntrySizeBytes) + return; + + // if we have space in the buffer to write more data and we just wrote data + // into the buffer then let try and write some more } - - // we've filled up the buffer lets give another channel a chance to send some data - if (in.remaining() <= maxEntrySizeBytes) - return; - - // if we have space in the buffer to write more data and we just wrote data into the - // buffer then let try and write some more + } finally { + if (LOG.isDebugEnabled()) + LOG.debug("Entries written: {}", entriesWritten); } } @@ -1152,82 +1159,89 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @throws InterruptedException */ void entriesFromBuffer(@NotNull Attached attached, @NotNull SelectionKey key) { - for (; ; ) { - out.limit(in.position()); + int entriesRead = 0; + try { + for (; ; entriesRead++) { + out.limit(in.position()); - // its set to MIN_VALUE when it should be read again - if (state == NOT_SET) { - if (out.remaining() < SIZE_OF_SIZE + 1) - return; + // its set to MIN_VALUE when it should be read again + if (state == NOT_SET) { + if (out.remaining() < SIZE_OF_SIZE + 1) + return; - // state is used for both heartbeat and stateless - state = out.readByte(); - sizeInBytes = out.readInt(); + // state is used for both heartbeat and stateless + state = out.readByte(); + sizeInBytes = out.readInt(); - assert sizeInBytes >= 0; + assert sizeInBytes >= 0; - // if the buffer is too small to read this payload we will have to grow the - // size of the buffer - long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1; - if (out.capacity() < requiredSize) { - attached.entryReader.resizeBuffer(requiredSize + HEADROOM); + // if the buffer is too small to read this payload we will have to grow the + // size of the buffer + long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1; + if (out.capacity() < requiredSize) { + attached.entryReader.resizeBuffer(requiredSize + HEADROOM); + } + + // this is the : + // -- heartbeat if its 0 + // -- stateful update if its 1 + // -- the id of the stateful event + if (state == NOT_SET) + continue; } - // this is the : - // -- heartbeat if its 0 - // -- stateful update if its 1 - // -- the id of the stateful event - if (state == NOT_SET) - continue; - } + if (out.remaining() < sizeInBytes) { + return; + } - if (out.remaining() < sizeInBytes) { - return; - } + final long nextEntryPos = out.position() + sizeInBytes; + assert nextEntryPos > 0; + final long limit = out.limit(); + out.limit(nextEntryPos); - final long nextEntryPos = out.position() + sizeInBytes; - assert nextEntryPos > 0; - final long limit = out.limit(); - out.limit(nextEntryPos); + boolean isStatelessClient = (state != 1); - boolean isStatelessClient = (state != 1); + if (isStatelessClient) { + if (statelessServerConnector == null) { + LOG.error("", new IllegalArgumentException("received an event " + + "from a stateless map, stateless maps are not " + + "currently supported when using Chronicle Channels")); + } else { - if (isStatelessClient) { - if (statelessServerConnector == null) { - LOG.error("", new IllegalArgumentException("received an event " + - "from a stateless map, stateless maps are not " + - "currently supported when using Chronicle Channels")); - } else { + final Work futureWork = + statelessServerConnector.processStatelessEvent(state, + attached.entryWriter.in, attached.entryReader.out); - final Work futureWork = statelessServerConnector.processStatelessEvent(state, - attached.entryWriter.in, attached.entryReader.out); + // turn the OP_WRITE on + key.interestOps(key.interestOps() | OP_WRITE); - // turn the OP_WRITE on - key.interestOps(key.interestOps() | OP_WRITE); - - // in some cases it may not be possible to send out all the data before we - // fill out the write buffer, so this data will be send when the buffer - // is no longer full, and as such is treated as future work - if (futureWork != null) { - try { // we will complete what we can for now - boolean isComplete = futureWork.doWork(attached.entryWriter.in); - if (!isComplete) - attached.entryWriter.uncompletedWork = futureWork; - } catch (Exception e) { - LOG.error("", e); + // in some cases it may not be possible to send out all the data before + // we fill out the write buffer, so this data will be send when + // the buffer is no longer full, and as such is treated as future work + if (futureWork != null) { + try { // we will complete what we can for now + boolean isComplete = futureWork.doWork(attached.entryWriter.in); + if (!isComplete) + attached.entryWriter.uncompletedWork = futureWork; + } catch (Exception e) { + LOG.error("", e); + } } } - } - } else - externalizable.readExternalEntry(copies, segmentState, out); + } else + externalizable.readExternalEntry(copies, segmentState, out); - out.limit(limit); + out.limit(limit); - // skip onto the next entry - out.position(nextEntryPos); + // skip onto the next entry + out.position(nextEntryPos); - state = NOT_SET; - sizeInBytes = 0; + state = NOT_SET; + sizeInBytes = 0; + } + } finally { + if (LOG.isDebugEnabled()) + LOG.debug("Entries read: {}", entriesRead); } } @@ -1315,28 +1329,39 @@ class StatelessServerConnector { @NotNull final ByteBufferBytes reader) { final StatelessChronicleMap.EventId event = VALUES[eventId]; - // these methods don't return a result + long transactionId = reader.readLong(); + + // the time stamp and the transaction are usually the same, or out by the shift + int timestampShift = reader.readUnsignedShort(); + long timestamp = transactionId - timestampShift; + + byte identifier = reader.readByte(); + + + // these methods don't return a result to the client or don't return a result to the + // client immediately switch (event) { case KEY_SET: - return keySet(reader, writer); + return keySet(reader, writer, transactionId); case VALUES: - return values(reader, writer); + return values(reader, writer, transactionId); case ENTRY_SET: - return entrySet(reader, writer); + return entrySet(reader, writer, transactionId); case PUT_WITHOUT_ACC: - return put(reader); + return put(reader, timestamp, identifier); case PUT_ALL_WITHOUT_ACC: - return putAll(reader); + return putAll(reader, timestamp, identifier); case REMOVE_WITHOUT_ACC: - return remove(reader); + return remove(reader, timestamp, identifier); } - final long sizeLocation = reflectTransactionId(reader, writer); + final long sizeLocation = reflectTransactionId(writer, transactionId); + // these methods return a result @@ -1354,34 +1379,33 @@ class StatelessServerConnector { return containsValue(reader, writer, sizeLocation); case GET: - return get(reader, writer, sizeLocation); + return get(reader, writer, sizeLocation, timestamp); case PUT: - return put(reader, writer, sizeLocation); + return put(reader, writer, sizeLocation, timestamp, identifier); case REMOVE: - return remove(reader, writer, sizeLocation); + return remove(reader, writer, sizeLocation, timestamp, identifier); case CLEAR: - return clear(writer, sizeLocation); + return clear(writer, sizeLocation, timestamp, identifier); case REPLACE: - return replace(reader, writer, sizeLocation); + return replace(reader, writer, sizeLocation, timestamp, identifier); case REPLACE_WITH_OLD_AND_NEW_VALUE: - return replaceWithOldAndNew(reader, writer, sizeLocation); + return replaceWithOldAndNew(reader, writer, sizeLocation, timestamp, identifier); case PUT_IF_ABSENT: - return putIfAbsent(reader, writer, sizeLocation); + return putIfAbsent(reader, writer, sizeLocation, timestamp, identifier); case REMOVE_WITH_VALUE: - return removeWithValue(reader, writer, sizeLocation); - + return removeWithValue(reader, writer, sizeLocation, timestamp, identifier); case TO_STRING: return toString(writer, sizeLocation); case PUT_ALL: - return putAll(reader, writer, sizeLocation); + return putAll(reader, writer, sizeLocation, timestamp, identifier); case HASH_CODE: return hashCode(writer, sizeLocation); @@ -1430,7 +1454,7 @@ class StatelessServerConnector { } @Nullable - private Work removeWithValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work removeWithValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { writer.writeBoolean(map.removeWithValue(reader)); @@ -1449,7 +1473,8 @@ class StatelessServerConnector { } @Nullable - private Work replaceWithOldAndNew(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work replaceWithOldAndNew(Bytes reader, @NotNull Bytes writer, final long + sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.replaceWithOldAndNew(reader, writer); @@ -1571,7 +1596,9 @@ class StatelessServerConnector { } @Nullable - private Work get(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work get(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long + transactionId) { + try { if (MAP_SUPPORTS_BYTES) { map.get(reader, writer); @@ -1589,7 +1616,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work put(Bytes reader) { + private Work put(Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.put(reader); } else { @@ -1604,7 +1631,7 @@ class StatelessServerConnector { } @Nullable - private Work put(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work put(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.put(reader, writer); @@ -1624,7 +1651,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work remove(Bytes reader) { + private Work remove(Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.removeKeyAsBytes(reader); } else { @@ -1636,7 +1663,7 @@ class StatelessServerConnector { } @Nullable - private Work remove(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work remove(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.remove(reader, writer); @@ -1655,7 +1682,7 @@ class StatelessServerConnector { } @Nullable - private Work putAll(@NotNull Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work putAll(@NotNull Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); @@ -1672,7 +1699,7 @@ class StatelessServerConnector { @SuppressWarnings("SameReturnValue") @Nullable - private Work putAll(@NotNull Bytes reader) { + private Work putAll(@NotNull Bytes reader, long timestamp, byte id) { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); } else { @@ -1683,7 +1710,7 @@ class StatelessServerConnector { } @Nullable - private Work clear(@NotNull Bytes writer, final long sizeLocation) { + private Work clear(@NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { map.clear(); } catch (Throwable e) { @@ -1695,8 +1722,7 @@ class StatelessServerConnector { } @Nullable - private Work values(@NotNull Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private Work values(@NotNull Bytes reader, @NotNull Bytes writer, final long transactionId) { Collection values; @@ -1738,8 +1764,7 @@ class StatelessServerConnector { } @Nullable - private Work keySet(@NotNull Bytes reader, @NotNull final Bytes writer) { - final long transactionId = reader.readLong(); + private Work keySet(@NotNull Bytes reader, @NotNull final Bytes writer, final long transactionId) { Set ks; @@ -1781,8 +1806,7 @@ class StatelessServerConnector { } @Nullable - private Work entrySet(@NotNull final Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private Work entrySet(@NotNull final Bytes reader, @NotNull Bytes writer, final long transactionId) { final Set> entries; @@ -1830,7 +1854,7 @@ class StatelessServerConnector { } @Nullable - private Work putIfAbsent(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work putIfAbsent(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.putIfAbsent(reader, writer); @@ -1849,7 +1873,7 @@ class StatelessServerConnector { } @Nullable - private Work replace(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { + private Work replace(Bytes reader, @NotNull Bytes writer, final long sizeLocation, long timestamp, byte id) { try { if (MAP_SUPPORTS_BYTES) { map.replaceKV(reader, writer); @@ -1867,8 +1891,7 @@ class StatelessServerConnector { return null; } - private long reflectTransactionId(@NotNull Bytes reader, @NotNull Bytes writer) { - final long transactionId = reader.readLong(); + private long reflectTransactionId(@NotNull Bytes writer, final long transactionId) { final long sizeLocation = writer.position(); writer.skip(SIZE_OF_SIZE); assert transactionId != 0; @@ -1926,7 +1949,7 @@ class StatelessServerConnector { @Nullable private Work sendException(@NotNull Bytes reader, @NotNull Bytes writer, @NotNull Throwable e) { - final long sizeLocation = reflectTransactionId(reader, writer); + final long sizeLocation = reflectTransactionId(writer, reader.readLong()); return sendException(writer, sizeLocation, e); } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 2670bec71..d83b83753 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -50,10 +50,7 @@ import java.lang.reflect.Array; import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.util.*; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import static java.lang.Class.forName; import static java.lang.Long.numberOfTrailingZeros; @@ -569,102 +566,6 @@ class VanillaChronicleMap, } } - @Override - public Future getLater(@NotNull K key) { - final V v = get(key); - return new Future() { - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return true; - } - - @Override - public V get() throws InterruptedException, ExecutionException { - return v; - } - - @Override - public V get(long timeout, @NotNull TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - return v; - } - }; - } - - @Override - public Future putLater(@NotNull K key, @NotNull V value) { - final V v = put(key, value); - return new Future() { - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return true; - } - - @Override - public V get() throws InterruptedException, ExecutionException { - return v; - } - - @Override - public V get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - return v; - } - }; - } - - @Override - public Future removeLater(@NotNull K key) { - final V v = remove(key); - return new Future() { - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return true; - } - - @Override - public V get() throws InterruptedException, ExecutionException { - return v; - } - - @Override - public V get(long timeout, TimeUnit unit) - throws InterruptedException, ExecutionException, TimeoutException { - return v; - } - }; - } - @Override public synchronized void getAll(File toFile) throws IOException { diff --git a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java index 6c98eb444..98eab7796 100644 --- a/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java +++ b/src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java @@ -36,7 +36,7 @@ public class CHMUseCasesTest { */ try (ChronicleMap map = ChronicleMapBuilder .of(String.class, String.class) - .checkSerializedValues() // for testng purposes only + .checkSerializedValues() // for testing purposes only .create()) { map.put("Hello", "World"); assertEquals("World", map.get("Hello")); @@ -65,10 +65,6 @@ public class CHMUseCasesTest { // TODO replace with the specific exception. } - assertEquals(null, map.putLater("Bye", "For now").get()); - assertEquals("For now", map.getLater("Bye").get()); - assertEquals("For now", map.removeLater("Bye").get()); - assertEquals(null, map.removeLater("Bye").get()); } } @@ -81,7 +77,7 @@ public class CHMUseCasesTest { public void testCharSequenceCharSequenceMap() throws ExecutionException, InterruptedException { try (ChronicleMap map = ChronicleMapBuilder .of(CharSequence.class, CharSequence.class) - .checkSerializedValues() // for testng purposes only + .checkSerializedValues() // for testing purposes only .create()) { map.put("Hello", "World"); StringBuilder key = new StringBuilder(); @@ -129,10 +125,6 @@ public class CHMUseCasesTest { } })); - assertEquals(null, map.putLater("Bye", "For now").get()); - assertEquals("For now", map.getLater("Bye").get().toString()); - assertEquals("For now", map.removeLater("Bye").get().toString()); - assertEquals(null, map.removeLater("Bye").get()); } } @@ -144,7 +136,7 @@ public class CHMUseCasesTest { public void testStringValueStringValueMap() { try (ChronicleMap map = ChronicleMapBuilder .of(StringValue.class, StringValue.class) - .checkSerializedValues() // for testng purposes only + .checkSerializedValues() // for testing purposes only .create()) { StringValue key1 = DataValueClasses.newDirectInstance(StringValue.class); StringValue key2 = DataValueClasses.newInstance(StringValue.class); @@ -283,10 +275,7 @@ public class CHMUseCasesTest { } catch (Exception todoMoreSpecificException) { } - assertEquals(null, map.putLater(3, 4).get()); - assertEquals((Integer) 4, map.getLater(3).get()); - assertEquals((Integer) 4, map.removeLater(3).get()); - assertEquals(null, map.removeLater(3).get()); + } } @@ -331,10 +320,7 @@ public class CHMUseCasesTest { } catch (Exception todoMoreSpecificException) { } - assertEquals(null, map.putLater(3L, 4L).get()); - assertEquals((Long) 4L, map.getLater(3L).get()); - assertEquals((Long) 4L, map.removeLater(3L).get()); - assertEquals(null, map.removeLater(3L).get()); + } } @@ -379,10 +365,7 @@ public class CHMUseCasesTest { } catch (Exception todoMoreSpecificException) { } - assertEquals(null, map.putLater(3.0, 4.0).get()); - assertEquals((Double) 4.0, map.getLater(3.0).get()); - assertEquals((Double) 4.0, map.removeLater(3.0).get()); - assertEquals(null, map.removeLater(3.0).get()); + } } @@ -430,10 +413,6 @@ public class CHMUseCasesTest { byte[] key3 = {3, 3, 3, 3}; byte[] value3 = {4, 4, 4, 4}; - assertEquals(null, map.putLater(key3, value3).get()); - assertTrue(Arrays.equals(value3, map.getLater(key3).get())); - assertTrue(Arrays.equals(value3, map.removeLater(key3).get())); - assertEquals(null, map.removeLater(key3).get()); } } @@ -521,13 +500,7 @@ public class CHMUseCasesTest { assertBBEquals(bb1, valueA); } - ByteBuffer key3 = ByteBuffer.wrap(new byte[]{3, 3, 3, 3}); - ByteBuffer value3 = ByteBuffer.wrap(new byte[]{4, 4, 4, 4}); - assertEquals(null, map.putLater(key3, value3).get()); - assertBBEquals(value3, map.getLater(key3).get()); - assertBBEquals(value3, map.removeLater(key3).get()); - assertEquals(null, map.removeLater(key3).get()); } } @@ -579,15 +552,6 @@ public class CHMUseCasesTest { assertBBEquals(ByteBuffer.wrap(new byte[]{12, 10, 11, 11}), map.get(key1)); - ByteBuffer key3 = ByteBuffer.wrap(new byte[]{3, 3, 3, 3}); - ByteBuffer value3 = ByteBuffer.allocateDirect(4); - value3.put(new byte[]{4, 4, 4, 4}); - value3.flip(); - - assertEquals(null, map.putLater(key3, value3).get()); - assertBBEquals(value3, map.getLater(key3).get()); - assertBBEquals(value3, map.removeLater(key3).get()); - assertEquals(null, map.removeLater(key3).get()); } } diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 10292c16f..e70843d33 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -26,7 +26,6 @@ import java.util.Collection; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Future; /** * @author Rob Austin. @@ -228,21 +227,6 @@ public class ReplicationCheckingMap implements ChronicleMap { return map1.updateForKey(key, mutator); } - @Override - public Future getLater(@NotNull K key) { - throw new UnsupportedOperationException(); - } - - @Override - public Future putLater(@NotNull K key, @NotNull V value) { - throw new UnsupportedOperationException(); - } - - @Override - public Future removeLater(@NotNull K key) { - throw new UnsupportedOperationException(); - } - @Override public void getAll(File toFile) { throw new UnsupportedOperationException(); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 66b664e45..c502b49ed 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -157,8 +157,9 @@ public class StatelessClientTest { } statelessMap.putAll(payload); + assertEquals(SIZE, serverMap.size()); - Set> entries = statelessMap.entrySet(); + Set> entries = statelessMap.entrySet(); Map.Entry next = entries.iterator().next(); assertEquals("some value=" + next.getKey(), next.getValue()); @@ -453,60 +454,7 @@ public class StatelessClientTest { } } - @Test(timeout = 10000) - 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"); - - assertEquals("some value", statelessMap.getLater(1).get()); - assertEquals(1, statelessMap.size()); - - statelessMap.remove(1); - - assertEquals(null, statelessMap.getLater(1).get()); - assertEquals(0, statelessMap.size()); - } - } - } - - @Test - public void testPutLater() 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()) { - CharSequence oldValue = statelessMap.putLater(1, "some value").get(); - assertEquals("some value", statelessMap.get(1)); - assertEquals(1, statelessMap.size()); - } - } - } - - @Test(timeout = 10000) - public void testRemoveLater() 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"); - statelessMap.removeLater(1).get(); - assertEquals(null, statelessMap.get(1)); - assertEquals(0, statelessMap.size()); - } - } - } @Test(timeout = 10000) public void testGetAndEntryWeDontHave() throws IOException, diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 14dd1dcfe..3d17c22d3 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -5,7 +5,10 @@ import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.DataValueClasses; import net.openhft.lang.values.IntValue; -import org.junit.*; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.IOException; @@ -100,19 +103,19 @@ public class TcpReplicationSoakTest { final int select = rnd.nextInt(2); final ChronicleMap map = select > 0 ? map1 : map2; - if (rnd.nextBoolean()) { - map.put(rnd.nextInt(Builder.SIZE), "test" + j); - } else { - map.remove(rnd.nextInt(Builder.SIZE)); + if (rnd.nextBoolean()) { + map.put(rnd.nextInt(Builder.SIZE), "test" + j); + } else { + map.remove(rnd.nextInt(Builder.SIZE)); + } } } - } - Thread.sleep(1000); + Thread.sleep(1000); System.out.println("\nwaiting till equal"); - waitTillEqual(10000); + waitTillEqual(10000); - Assert.assertEquals(new TreeMap(map1), new TreeMap(map2)); + Assert.assertEquals(map1, map2); } diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index 3aa530b2c..1c487acf1 100755 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -33,7 +33,6 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.*; -import java.util.concurrent.Future; import static org.junit.Assert.*; @@ -180,21 +179,6 @@ public class StatelessChronicleMapTest extends JSR166TestCase { return d.updateForKey(key, mutator); } - @Override - public Future getLater(@NotNull K key) { - return d.getLater(key); - } - - @Override - public Future putLater(@NotNull K key, @NotNull V value) { - return d.putLater(key, value); - } - - @Override - public Future removeLater(@NotNull K key) { - return d.removeLater(key); - } - @Override public void getAll(File toFile) throws IOException { d.getAll(toFile);