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/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 d56d34a17..29fc940f8 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -27,6 +27,7 @@ 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.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,13 +54,14 @@ import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; */ class StatelessChronicleMap implements ChronicleMap, Closeable, Cloneable { - public static final String START_OF = "Attempt to write "; + private 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 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; + @NotNull private final AbstractChronicleMapBuilder chronicleMapBuilder; private volatile ByteBuffer outBuffer; @@ -68,13 +70,19 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private volatile ByteBuffer inBuffer; private volatile ByteBufferBytes inBytes; + @NotNull private final ReaderWithSize keyReaderWithSize; + @NotNull private final WriterWithSize keyWriterWithSize; + @NotNull private final ReaderWithSize valueReaderWithSize; + @NotNull private final WriterWithSize valueWriterWithSize; private volatile SocketChannel clientChannel; + @Nullable private CloseablesManager closeables; + @NotNull private final StatelessMapConfig config; private int maxEntrySize; private final Class kClass; @@ -117,10 +125,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon UPDATE_FOR_KEY } - private AtomicLong transactionID = new AtomicLong(0); + @NotNull + private final AtomicLong transactionID = new AtomicLong(0); - StatelessChronicleMap(final StatelessMapConfig config, - final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { + StatelessChronicleMap(@NotNull final StatelessMapConfig config, + @NotNull final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { this.chronicleMapBuilder = chronicleMapBuilder; this.config = config; keyReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.keyBuilder); @@ -148,36 +157,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - @Override - public Future getLater(@NotNull final K key) { - return lazyExecutorService().submit(new Callable() { - @Override - public V call() throws Exception { - return StatelessChronicleMap.this.get(key); - } - }); - } - - @Override - public Future putLater(@NotNull final K key, @NotNull final V value) { - return lazyExecutorService().submit(new Callable() { - @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(); @@ -202,7 +181,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private synchronized SocketChannel lazyConnect(final long timeoutMs, - final InetSocketAddress remoteAddress) throws IOException { + final InetSocketAddress remoteAddress) { if (LOG.isDebugEnabled()) LOG.debug("attempting to connect to " + remoteAddress); @@ -225,9 +204,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon doHandShaking(result); break; } catch (IOException e) { - closeables.closeQuietly(); + if (closeables!=null) closeables.closeQuietly(); } catch (Exception e) { - closeables.closeQuietly(); + if (closeables!=null) closeables.closeQuietly(); throw e; } } @@ -243,13 +222,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * the first map method call. * * @param remoteAddress the Inet Socket Address - * @return * @throws java.io.IOException * @see net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, * java.net.InetSocketAddress) */ - private synchronized void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { + private synchronized void attemptConnect(final InetSocketAddress remoteAddress) { // ensures that the excising connection are closed closeExisting(); @@ -259,7 +237,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon clientChannel.connect(remoteAddress); doHandShaking(clientChannel); } catch (IOException e) { - closeables.closeQuietly(); + if (closeables!=null) closeables.closeQuietly(); } } @@ -312,6 +290,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier); } + @NotNull public File file() { throw new UnsupportedOperationException(); } @@ -351,7 +330,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - public V putIfAbsent(@NotNull K key, V value) { + @SuppressWarnings("NullableProblems") + public V putIfAbsent(K key, V value) { if (key == null || value == null) throw new NullPointerException(); @@ -360,26 +340,27 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - public boolean remove(@NotNull Object key, Object value) { + @SuppressWarnings("NullableProblems") + public boolean remove( Object key, Object value) { if (key == null) throw new NullPointerException(); - if (value == null) - return false; + return value != null && fetchBoolean(REMOVE_WITH_VALUE, (K) key, (V) value); - return fetchBoolean(REMOVE_WITH_VALUE, (K) key, (V) value); } - public boolean replace(@NotNull K key, @NotNull V oldValue, @NotNull V newValue) { + @SuppressWarnings("NullableProblems") + public boolean replace(K key, V oldValue, V newValue) { if (key == null || oldValue == null || newValue == null) throw new NullPointerException(); return fetchBoolean(REPLACE_WITH_OLD_AND_NEW_VALUE, key, oldValue, newValue); } - public V replace(@NotNull K key, @NotNull V value) { + @SuppressWarnings("NullableProblems") + public V replace(K key, V value) { if (key == null || value == null) throw new NullPointerException(); @@ -400,7 +381,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @return true if the contain the same data */ @Override - public boolean equals(Object object) { + public boolean equals(@Nullable Object object) { if (this == object) return true; if (object == null || object.getClass().isAssignableFrom(Map.class)) return false; @@ -421,6 +402,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchInt(HASH_CODE); } + @NotNull public String toString() { return fetchObject(String.class, TO_STRING); } @@ -434,6 +416,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchBooleanK(CONTAINS_KEY, (K) key); } + @NotNull private NullPointerException keyNotNullNPE() { return new NullPointerException("key can not be null"); } @@ -450,11 +433,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(vClass, GET, (K) key); } + @NotNull public V getUsing(K key, V usingValue) { throw new UnsupportedOperationException("getUsing() is not supported for stateless " + "clients"); } + @NotNull public V acquireUsing(@NotNull K key, V usingValue) { throw new UnsupportedOperationException("acquireUsing() is not supported for stateless " + "clients"); @@ -485,22 +470,24 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(vClass, putReturnsNull ? PUT_WITHOUT_ACC : PUT, key, value); } - public R mapForKey(K key, @NotNull Function function) { + @Nullable + public R mapForKey(@Nullable K key, @NotNull Function function) { if (key == null) throw keyNotNullNPE(); return fetchObject(MAP_FOR_KEY, key, function); } + @Nullable @Override - public R updateForKey(K key, @NotNull Mutator mutator) { + public R updateForKey(@Nullable K key, @NotNull Mutator mutator) { if (key == null) throw keyNotNullNPE(); return fetchObject(UPDATE_FOR_KEY, key, mutator); } - private synchronized void writeEntriesForPutAll(Map map) { + private synchronized void writeEntriesForPutAll(@NotNull Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; @@ -615,7 +602,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert size < Integer.MAX_VALUE; - final ByteBuffer result = ByteBuffer.allocate((int) size).order(ByteOrder.nativeOrder()); + final ByteBuffer result = ByteBuffer.allocate(size).order(ByteOrder.nativeOrder()); final long bytesPosition = inBytes.position(); inBytes = new ByteBufferBytes(result.slice()); @@ -850,7 +837,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private long writeEvent(StatelessChronicleMap.EventId event) { + private long writeEvent(@NotNull StatelessChronicleMap.EventId event) { outBuffer.clear(); outBytes.clear(); @@ -862,7 +849,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon /** * skips for the transactionid */ - private long writeEventAnSkip(EventId event) { + private long writeEventAnSkip(@NotNull EventId event) { final long sizeLocation = writeEvent(event); // skips for the transaction id @@ -893,7 +880,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public final V setValue(V newValue) { V oldValue = value; - StatelessChronicleMap.this.put((K) getKey(), (V) newValue); + StatelessChronicleMap.this.put(getKey(), newValue); return oldValue; } @@ -917,6 +904,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon (value == null ? 0 : value.hashCode()); } + @NotNull public final String toString() { return getKey() + "=" + getValue(); } @@ -952,7 +940,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytes.buffer().clear(); break; - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + } catch (@NotNull java.nio.channels.ClosedChannelException | ClosedConnectionException e) { checkTimeout(timeoutTime); clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); } @@ -992,49 +980,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, InterruptedException { - - 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(); - 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); - } + int remainingBytes = 0; - return inBytes; - } - - private int nextEntry(long timeoutTime, long transactionId) throws IOException, InterruptedException { - int remainingBytes; for (; ; ) { // read the next item from the socket @@ -1102,7 +1050,47 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon pause(); } - return remainingBytes; + + + 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; } private void clearParked() { @@ -1111,7 +1099,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon parkedTransactionTimeStamp = 0; } - private void pause() throws InterruptedException { + private void pause() { /// don't call inBytesLock.isHeldByCurrentThread() as it not atomic inBytesLock.unlock(); @@ -1127,6 +1115,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @return bytes read from the TCP/IP socket * @throws IOException socket failed to read data */ + @SuppressWarnings("UnusedReturnValue") private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { inBytes.buffer().position(0); @@ -1149,7 +1138,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void send(final Bytes out, long timeoutTime) throws IOException { + private void send(@NotNull final Bytes out, long timeoutTime) throws IOException { outBuffer.limit((int) out.position()); outBuffer.position(0); @@ -1197,6 +1186,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return writeKey(key, null); } + @SuppressWarnings("SameParameterValue") private ThreadLocalCopies writeKey(K key, ThreadLocalCopies copies) { long start = outBytes.position(); @@ -1216,6 +1206,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } + @SuppressWarnings("UnusedReturnValue") private ThreadLocalCopies writeKeyInLoop(K key, Object writer, ThreadLocalCopies copies) { long start = outBytes.position(); @@ -1253,6 +1244,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } + @SuppressWarnings("UnusedReturnValue") private ThreadLocalCopies writeValueInLoop(V value, Object writer, ThreadLocalCopies copies) { long start = outBytes.position(); for (; ; ) { @@ -1271,7 +1263,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - private void resizeToMessage(long start, IllegalArgumentException e) { + private void resizeToMessage(long start, @NotNull IllegalArgumentException e) { String message = e.getMessage(); if (message.startsWith(START_OF)) { String substring = message.substring("Attempt to write ".length(), message.length()); @@ -1299,6 +1291,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } + @Nullable private O readObject(long transactionId, long startTime) { long timeoutTime = startTime + this.config.timeoutMs(); @@ -1336,7 +1329,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - private long send(final EventId eventId, final long startTime) { + private long send(@NotNull final EventId eventId, final long startTime) { // send outBytesLock.lock(); try { @@ -1351,7 +1344,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return valueReaderWithSize.readNullable(in, copies); } - private boolean fetchBoolean(final EventId eventId, K key, V value) { + @SuppressWarnings("SameParameterValue") + private boolean fetchBoolean(@NotNull final EventId eventId, K key, V value) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1374,7 +1368,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean fetchBoolean(final EventId eventId, K key, V value1, V value2) { + @SuppressWarnings("SameParameterValue") + private boolean fetchBoolean(@NotNull final EventId eventId, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1398,7 +1393,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean fetchBooleanV(final EventId eventId, V value) { + @SuppressWarnings("SameParameterValue") + private boolean fetchBooleanV(@NotNull final EventId eventId, V value) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1416,7 +1412,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean fetchBooleanK(final EventId eventId, K key) { + @SuppressWarnings("SameParameterValue") + private boolean fetchBooleanK(@NotNull final EventId eventId, K key) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1434,7 +1431,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private long fetchLong(final EventId eventId) { + @SuppressWarnings("SameParameterValue") + private long fetchLong(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1450,7 +1448,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return readLong(transactionId, startTime); } - private boolean fetchBoolean(final EventId eventId) { + @SuppressWarnings("SameParameterValue") + private boolean fetchBoolean(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1466,7 +1465,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return readBoolean(transactionId, startTime); } - private void fetchVoid(final EventId eventId) { + @SuppressWarnings("SameParameterValue") + private void fetchVoid(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); @@ -1496,7 +1496,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private O fetchObject(final Class tClass, final EventId eventId) { + @SuppressWarnings("SameParameterValue") + @Nullable + private O fetchObject(final Class tClass, @NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId = send(eventId, startTime); @@ -1505,14 +1507,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // receive inBytesLock.lock(); try { - return (O) blockingFetchReadOnly(timeoutTime, transactionId).readObject(tClass); + return blockingFetchReadOnly(timeoutTime, transactionId).readObject(tClass); } finally { inBytesLock.unlock(); } } - private int fetchInt(final EventId eventId) { + @SuppressWarnings("SameParameterValue") + private int fetchInt(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1529,7 +1532,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private R fetchObject(Class rClass, final EventId eventId, K key, V value) { + @Nullable + private R fetchObject(Class rClass, @NotNull final EventId eventId, K key, V value) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1560,7 +1564,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "supported"); } - private R fetchObject(Class rClass, final EventId eventId, K key, V value1, V value2) { + @Nullable + private R fetchObject(Class rClass, @NotNull final EventId eventId, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1593,7 +1598,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "supported"); } - private R fetchObject(Class rClass, final EventId eventId, K key) { + @Nullable + private R fetchObject(Class rClass, @NotNull final EventId eventId, K key) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1623,7 +1629,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean eventReturnsNull(EventId eventId) { + private boolean eventReturnsNull(@NotNull EventId eventId) { switch (eventId) { case PUT_ALL_WITHOUT_ACC: @@ -1655,7 +1661,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private R fetchObject(final EventId eventId, K key, Object object) { + @Nullable + private R fetchObject(@NotNull final EventId eventId, K key, @NotNull Object object) { final long startTime = System.currentTimeMillis(); long transactionId; diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index be124a7a6..50bafefc5 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -64,11 +64,11 @@ interface Work { */ final class TcpReplicator extends AbstractChannelReplicator implements Closeable { - public static final int STATELESS_CLIENT = -127; - public static final byte NOT_SET = (byte) HEARTBEAT.ordinal(); + private static final int STATELESS_CLIENT = -127; + private static final byte NOT_SET = (byte) HEARTBEAT.ordinal(); private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName()); private static final int BUFFER_SIZE = 0x100000; // 1MB - public static final int SPIN_LOOP_COUNT = 100000; + private static final int SPIN_LOOP_COUNT = 100000; private final SelectionKey[] selectionKeysStore = new SelectionKey[Byte.MAX_VALUE + 1]; // used to instruct the selector thread to set OP_WRITE on a key correlated by the bit index in the // bitset @@ -76,11 +76,15 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private final BitSet activeKeys = new BitSet(selectionKeysStore.length); private final long heartBeatIntervalMillis; + @NotNull private final Replica replica; private final byte localIdentifier; private final int maxEntrySizeBytes; + @NotNull private final Replica.EntryExternalizable externalizable; + @NotNull private final TcpTransportAndNetworkConfig replicationConfig; + @Nullable private final StatelessServerConnector statelessServerConnector; private final @Nullable @@ -190,16 +194,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } } } - } catch (CancelledKeyException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - } catch (ClosedSelectorException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - } catch (ClosedChannelException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - } catch (ConnectException e) { + } catch (CancelledKeyException | ConnectException | ClosedChannelException | ClosedSelectorException e) { if (LOG.isDebugEnabled()) LOG.debug("", e); } catch (Exception e) { @@ -217,7 +212,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } } - private void processKey(long approxTime, SelectionKey key) { + private void processKey(long approxTime, @NotNull SelectionKey key) { try { if (!key.isValid()) @@ -245,19 +240,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable LOG.debug("onWrite - " + name); onWrite(key, approxTime); } - } catch (BufferUnderflowException e) { - if (!isClosed) - quietClose(key, e); - } catch (CancelledKeyException e) { - if (!isClosed) - quietClose(key, e); - } catch (ClosedSelectorException e) { - if (!isClosed) - quietClose(key, e); - } catch (IOException e) { - if (!isClosed) - quietClose(key, e); - } catch (InterruptedException e) { + } catch (BufferUnderflowException | InterruptedException | IOException | ClosedSelectorException | CancelledKeyException e) { if (!isClosed) quietClose(key, e); } catch (Exception e) { @@ -346,7 +329,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } } - private void enableOpWrite(SelectionKey key) { + private void enableOpWrite(@NotNull SelectionKey key) { int ops = key.interestOps(); if ((ops & (OP_CONNECT | OP_ACCEPT)) == 0) key.interestOps(ops | OP_WRITE); @@ -361,7 +344,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @throws ConnectException */ private void heartbeatCheckHasReceived(@NotNull final SelectionKey key, - final long approxTimeOutTime) throws ConnectException { + final long approxTimeOutTime) { final Attached attached = (Attached) key.attachment(); @@ -405,7 +388,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * called when the selector receives a OP_CONNECT message */ private void onConnect(@NotNull final SelectionKey key) - throws IOException, InterruptedException { + throws IOException { SocketChannel channel = null; @@ -498,19 +481,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable throttle(channel); } - /** - * 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 - */ - private void clearHandshaking(SelectionKey key) { - final Attached attached = (Attached) key.attachment(); - activeKeys.clear(attached.remoteIdentifier); - selectionKeysStore[attached.remoteIdentifier] = null; - attached.clearHandShaking(); - } - /** * used to exchange identifiers and timestamps and heartbeat intervals between the server and * client @@ -520,8 +490,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @throws java.io.IOException * @throws InterruptedException */ - private void doHandShaking(@NotNull final SelectionKey key, SocketChannel socketChannel) - throws IOException, InterruptedException { + private void doHandShaking(@NotNull final SelectionKey key, @NotNull SocketChannel socketChannel) + throws IOException { final Attached attached = (Attached) key.attachment(); final TcpSocketChannelEntryWriter writer = attached.entryWriter; final TcpSocketChannelEntryReader reader = attached.entryReader; @@ -619,7 +589,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * called when the selector receives a OP_WRITE message */ private void onWrite(@NotNull final SelectionKey key, - final long approxTime) throws InterruptedException, IOException { + final long approxTime) throws IOException { final SocketChannel socketChannel = (SocketChannel) key.channel(); final Attached attached = (Attached) key.attachment(); if (attached == null) throw new NullPointerException("No attached"); @@ -658,7 +628,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * called when the selector receives a OP_READ message */ - private void onRead(final SelectionKey key, + private void onRead(@NotNull final SelectionKey key, final long approxTime) throws IOException, InterruptedException { final SocketChannel socketChannel = (SocketChannel) key.channel(); @@ -701,6 +671,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } } + @Nullable private ServerSocketChannel openServerSocketChannel() throws IOException { ServerSocketChannel result = null; @@ -722,11 +693,13 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private static class KeyInterestUpdater { private final AtomicBoolean wasChanged = new AtomicBoolean(); + @NotNull private final BitSet changeOfOpWriteRequired; + @NotNull private final SelectionKey[] selectionKeys; private final int op; - KeyInterestUpdater(int op, final SelectionKey[] selectionKeys) { + KeyInterestUpdater(int op, @NotNull final SelectionKey[] selectionKeys) { this.op = op; this.selectionKeys = selectionKeys; changeOfOpWriteRequired = new BitSet(selectionKeys.length); @@ -759,6 +732,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private class ServerConnector extends AbstractConnector { + @NotNull private final Details details; private ServerConnector(@NotNull Details details) { @@ -766,6 +740,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable this.details = details; } + @NotNull @Override public String toString() { return "ServerConnector{" + @@ -773,6 +748,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable '}'; } + @Nullable SelectableChannel doConnect() throws IOException, InterruptedException { @@ -815,6 +791,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private class ClientConnector extends AbstractConnector { + @NotNull private final Details details; private ClientConnector(@NotNull Details details) { @@ -822,6 +799,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable this.details = details; } + @NotNull @Override public String toString() { return "ClientConnector{" + details + '}'; @@ -896,6 +874,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable public TcpSocketChannelEntryReader entryReader; public TcpSocketChannelEntryWriter entryWriter; + @Nullable public Replica.ModificationIterator remoteModificationIterator; public AbstractConnector connector; public long remoteBootstrapTimestamp = Long.MIN_VALUE; @@ -935,11 +914,15 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable */ private class TcpSocketChannelEntryWriter { + @NotNull final ByteBufferBytes in; + @NotNull private final ByteBuffer out; + @NotNull private final EntryCallback entryCallback; // if uncompletedWork is set ( not null ) , this must be completed before any further work // is carried out + @Nullable public Work uncompletedWork; private long lastSentTime; @@ -983,8 +966,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @param selectionKey */ void entriesToBuffer(@NotNull final Replica.ModificationIterator modificationIterator, - @NotNull final SelectionKey selectionKey) - throws InterruptedException, IOException { + @NotNull final SelectionKey selectionKey) { final SocketChannel socketChannel = (SocketChannel) selectionKey.channel(); final Attached attached = (Attached) selectionKey.attachment(); @@ -992,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); } } @@ -1092,7 +1081,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } public boolean doWork() { - return uncompletedWork.doWork(in); + return uncompletedWork != null && uncompletedWork.doWork(in); } public boolean hasBytesToWrite() { @@ -1169,83 +1158,90 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * @param key * @throws InterruptedException */ - void entriesFromBuffer(Attached attached, SelectionKey key) throws InterruptedException, IOException { - for (; ; ) { - out.limit(in.position()); + void entriesFromBuffer(@NotNull Attached attached, @NotNull SelectionKey key) { + 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); } } @@ -1295,7 +1291,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable */ class StatelessServerConnector { - public static final int TRANSACTION_ID_OFFSET = SIZE_OF_SIZE + 1; public static final boolean MAP_SUPPORTS_BYTES = true; private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class .getName()); @@ -1304,12 +1299,17 @@ class StatelessServerConnector { public static final int SIZE_OF_IS_EXCEPTION = 1; public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTION_ID; + @NotNull private final ReaderWithSize keyReaderWithSize; + @NotNull private final WriterWithSize keyWriterWithSize; + @NotNull private final ReaderWithSize valueReaderWithSize; + @NotNull private final WriterWithSize valueWriterWithSize; + @NotNull private final VanillaChronicleMap map; - private double maxEntrySizeBytes; + private final double maxEntrySizeBytes; StatelessServerConnector( SerializationBuilder keySerializationBuilder, @@ -1323,6 +1323,7 @@ class StatelessServerConnector { this.maxEntrySizeBytes = maxEntrySizeBytes; } + @Nullable Work processStatelessEvent(final byte eventId, @NotNull final Bytes writer, @NotNull final ByteBufferBytes reader) { @@ -1355,10 +1356,10 @@ class StatelessServerConnector { switch (event) { case LONG_SIZE: - return longSize(reader, writer, sizeLocation); + return longSize(writer, sizeLocation); case IS_EMPTY: - return isEmpty(reader, writer, sizeLocation); + return isEmpty(writer, sizeLocation); case CONTAINS_KEY: return containsKey(reader, writer, sizeLocation); @@ -1376,7 +1377,7 @@ class StatelessServerConnector { return remove(reader, writer, sizeLocation); case CLEAR: - return clear(reader, writer, sizeLocation); + return clear(writer, sizeLocation); case REPLACE: return replace(reader, writer, sizeLocation); @@ -1391,13 +1392,13 @@ class StatelessServerConnector { return removeWithValue(reader, writer, sizeLocation); case TO_STRING: - return toString(reader, writer, sizeLocation); + return toString(writer, sizeLocation); case PUT_ALL: return putAll(reader, writer, sizeLocation); case HASH_CODE: - return hashCode(reader, writer, sizeLocation); + return hashCode(writer, sizeLocation); case MAP_FOR_KEY: return mapForKey(reader, writer, sizeLocation); @@ -1410,7 +1411,8 @@ class StatelessServerConnector { } } - public Work mapForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) { + @Nullable + public Work mapForKey(@NotNull ByteBufferBytes reader, @NotNull Bytes writer, long sizeLocation) { final K key = keyReaderWithSize.read(reader, null); final Function function = (Function) reader.readObject(); try { @@ -1425,7 +1427,8 @@ class StatelessServerConnector { return null; } - public Work updateForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) { + @Nullable + public Work updateForKey(@NotNull ByteBufferBytes reader, @NotNull Bytes writer, long sizeLocation) { final K key = keyReaderWithSize.read(reader, null); final Mutator mutator = (Mutator) reader.readObject(); try { @@ -1440,7 +1443,8 @@ class StatelessServerConnector { return null; } - private Work removeWithValue(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work removeWithValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { writer.writeBoolean(map.removeWithValue(reader)); @@ -1458,7 +1462,8 @@ class StatelessServerConnector { return null; } - private Work replaceWithOldAndNew(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work replaceWithOldAndNew(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { map.replaceWithOldAndNew(reader, writer); @@ -1477,7 +1482,8 @@ class StatelessServerConnector { return null; } - private Work longSize(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work longSize(@NotNull Bytes writer, final long sizeLocation) { try { writer.writeLong(map.longSize()); } catch (Throwable e) { @@ -1487,7 +1493,8 @@ class StatelessServerConnector { return null; } - private Work hashCode(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work hashCode(@NotNull Bytes writer, final long sizeLocation) { try { writer.writeInt(map.hashCode()); } catch (Throwable e) { @@ -1498,7 +1505,8 @@ class StatelessServerConnector { return null; } - private Work toString(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work toString(@NotNull Bytes writer, final long sizeLocation) { final String str; final long remaining = writer.remaining(); @@ -1520,7 +1528,9 @@ class StatelessServerConnector { return null; } - private Work sendException(Bytes writer, long sizeLocation, Throwable e) { + @SuppressWarnings("SameReturnValue") + @Nullable + private Work sendException(@NotNull Bytes writer, long sizeLocation, @NotNull Throwable e) { // move the position to ignore any bytes written so far writer.position(sizeLocation + HEADER_SIZE); @@ -1532,7 +1542,8 @@ class StatelessServerConnector { return null; } - private Work isEmpty(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work isEmpty(@NotNull Bytes writer, final long sizeLocation) { try { writer.writeBoolean(map.isEmpty()); } catch (Throwable e) { @@ -1543,7 +1554,8 @@ class StatelessServerConnector { return null; } - private Work containsKey(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work containsKey(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { writer.writeBoolean(map.containsKey(reader)); @@ -1558,7 +1570,8 @@ class StatelessServerConnector { return null; } - private Work containsValue(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work containsValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { // todo optimize -- eliminate final V v = valueReaderWithSize.read(reader, null); @@ -1571,7 +1584,8 @@ class StatelessServerConnector { return null; } - private Work get(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work get(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { map.get(reader, writer); @@ -1587,6 +1601,8 @@ class StatelessServerConnector { return null; } + @SuppressWarnings("SameReturnValue") + @Nullable private Work put(Bytes reader) { if (MAP_SUPPORTS_BYTES) { map.put(reader); @@ -1601,7 +1617,8 @@ class StatelessServerConnector { return null; } - private Work put(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work put(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { map.put(reader, writer); @@ -1619,6 +1636,8 @@ class StatelessServerConnector { return null; } + @SuppressWarnings("SameReturnValue") + @Nullable private Work remove(Bytes reader) { if (MAP_SUPPORTS_BYTES) { map.removeKeyAsBytes(reader); @@ -1630,7 +1649,8 @@ class StatelessServerConnector { return null; } - private Work remove(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work remove(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { map.remove(reader, writer); @@ -1648,7 +1668,8 @@ class StatelessServerConnector { return null; } - private Work putAll(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work putAll(@NotNull Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); @@ -1663,7 +1684,9 @@ class StatelessServerConnector { return null; } - private Work putAll(Bytes reader) { + @SuppressWarnings("SameReturnValue") + @Nullable + private Work putAll(@NotNull Bytes reader) { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); } else { @@ -1673,7 +1696,8 @@ class StatelessServerConnector { return null; } - private Work clear(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work clear(@NotNull Bytes writer, final long sizeLocation) { try { map.clear(); } catch (Throwable e) { @@ -1684,7 +1708,8 @@ class StatelessServerConnector { return null; } - private Work values(Bytes reader, Bytes writer) { + @Nullable + private Work values(@NotNull Bytes reader, @NotNull Bytes writer) { final long transactionId = reader.readLong(); Collection values; @@ -1700,7 +1725,7 @@ class StatelessServerConnector { // this allows us to write more data than the buffer will allow return new Work() { @Override - public boolean doWork(Bytes out) { + public boolean doWork(@NotNull Bytes out) { final long sizeLocation = header(out, transactionId); ThreadLocalCopies copies = valueWriterWithSize.getCopies(null); @@ -1726,7 +1751,8 @@ class StatelessServerConnector { }; } - private Work keySet(Bytes reader, final Bytes writer) { + @Nullable + private Work keySet(@NotNull Bytes reader, @NotNull final Bytes writer) { final long transactionId = reader.readLong(); Set ks; @@ -1742,7 +1768,7 @@ class StatelessServerConnector { // this allows us to write more data than the buffer will allow return new Work() { @Override - public boolean doWork(Bytes out) { + public boolean doWork(@NotNull Bytes out) { final long sizeLocation = header(out, transactionId); ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); @@ -1768,7 +1794,8 @@ class StatelessServerConnector { }; } - private Work entrySet(final Bytes reader, Bytes writer) { + @Nullable + private Work entrySet(@NotNull final Bytes reader, @NotNull Bytes writer) { final long transactionId = reader.readLong(); final Set> entries; @@ -1784,7 +1811,7 @@ class StatelessServerConnector { // this allows us to write more data than the buffer will allow return new Work() { @Override - public boolean doWork(Bytes out) { + public boolean doWork(@NotNull Bytes out) { if (out.remaining() <= maxEntrySizeBytes) return false; @@ -1816,7 +1843,8 @@ class StatelessServerConnector { }; } - private Work putIfAbsent(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work putIfAbsent(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { map.putIfAbsent(reader, writer); @@ -1834,7 +1862,8 @@ class StatelessServerConnector { return null; } - private Work replace(Bytes reader, Bytes writer, final long sizeLocation) { + @Nullable + private Work replace(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { try { if (MAP_SUPPORTS_BYTES) { map.replaceKV(reader, writer); @@ -1852,7 +1881,7 @@ class StatelessServerConnector { return null; } - private long reflectTransactionId(Bytes reader, Bytes writer) { + private long reflectTransactionId(@NotNull Bytes reader, @NotNull Bytes writer) { final long transactionId = reader.readLong(); final long sizeLocation = writer.position(); writer.skip(SIZE_OF_SIZE); @@ -1862,7 +1891,7 @@ class StatelessServerConnector { return sizeLocation; } - private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { + private void writeSizeAndFlags(long locationOfSize, boolean isException, @NotNull Bytes out) { final long size = out.position() - locationOfSize; out.writeInt(locationOfSize, (int) size); // size @@ -1887,11 +1916,12 @@ class StatelessServerConnector { } - private void writeException(Bytes out, Throwable e) { + private void writeException(@NotNull Bytes out, Throwable e) { out.writeObject(e); } - private Map readEntries(Bytes reader) { + @NotNull + private Map readEntries(@NotNull Bytes reader) { final long numberOfEntries = reader.readStopBit(); final Map result = new HashMap(); @@ -1908,12 +1938,13 @@ class StatelessServerConnector { return result; } - private Work sendException(Bytes reader, Bytes writer, Throwable e) { + @Nullable + private Work sendException(@NotNull Bytes reader, @NotNull Bytes writer, @NotNull Throwable e) { final long sizeLocation = reflectTransactionId(reader, writer); return sendException(writer, sizeLocation, e); } - private long header(Bytes writer, final long transactionId) { + private long header(@NotNull Bytes writer, final long transactionId) { final long sizeLocation = writer.position(); writer.skip(SIZE_OF_SIZE); @@ -1930,7 +1961,7 @@ class StatelessServerConnector { return sizeLocation; } - private void writeHeader(Bytes writer, long sizeLocation, int count, + private void writeHeader(@NotNull Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) { final long end = writer.position(); final int size = (int) (end - sizeLocation); 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/eg/BGChronicleTest.java b/src/test/java/eg/BGChronicleTest.java index e156dce6a..f61421742 100644 --- a/src/test/java/eg/BGChronicleTest.java +++ b/src/test/java/eg/BGChronicleTest.java @@ -440,7 +440,7 @@ public class BGChronicleTest { return port; } - @Test + public void test() throws IOException, InterruptedException { main(); } 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..1d4288b61 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -453,60 +453,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/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);