From b0ca345be0db5d59220a5852d594b1f35c778000 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 13 Mar 2015 10:06:54 +0000 Subject: [PATCH] fixed keyset, entryset and values --- .../map/ClientWiredStatelessChronicleMap.java | 192 +++++++----------- .../openhft/chronicle/map/MapWireHandler.java | 92 +++++---- 2 files changed, 120 insertions(+), 164 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index f33279606..298ca3732 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -199,38 +199,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon @NotNull public String toString() { - final Map result = toMap(); - return result.toString(); - } - - private Map toMap() { - final long startTime = System.currentTimeMillis(); - final Map result = new HashMap(); - - // send - final long transactionId = proxySend("ENTRY_SET", startTime); - assert !hub.outBytesLock().isHeldByCurrentThread(); - final long timeoutTime = startTime + hub.timeoutMs; - - for (; ; ) { - - // receive - hub.inBytesLock().lock(); - try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - - if (wireIn.read(Fields.HAS_NEXT).bool()) { - result.put( - readK(Fields.RESULT_KEY, wireIn, null), - readV(Fields.RESULT_VALUE, wireIn, null)); - } else - break; - - } finally { - hub.inBytesLock().unlock(); - } - } - return result; + return toMap().toString(); } @NotNull @@ -376,44 +345,29 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } + @NotNull public Set keySet() { - final long startTime = System.currentTimeMillis(); - final Set result = new HashSet(); - // send - final long transactionId = proxySend("KEY_SET", startTime); - assert !hub.outBytesLock().isHeldByCurrentThread(); - final long timeoutTime = startTime + hub.timeoutMs; - - for (; ; ) { - - // receive - hub.inBytesLock().lock(); - try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - - - if (wireIn.read(Fields.HAS_NEXT).bool()) { - K k = readK(Fields.RESULT, wireIn, null); - result.add(k); - } else - break; - - } finally { - hub.inBytesLock().unlock(); - } - } - return result; + final Set usingCollection = new HashSet(); + readChunked(kClass, "KEY_SET", usingCollection); + return usingCollection; } + @NotNull public Collection values() { + final List usingCollection = new ArrayList<>(); + return readChunked(vClass, "VALUES", usingCollection); + } + + private > Collection readChunked(Class vClass1, String values, A usingCollection) { final long startTime = System.currentTimeMillis(); - final Set result = new HashSet(); + // send - final long transactionId = proxySend("VALUES", startTime); + final long transactionId = proxySend(values, startTime); assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; + OUTER: for (; ; ) { // receive @@ -421,30 +375,38 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + while (wireIn.bytes().remaining() > 0) { + boolean bool = wireIn.read(Fields.HAS_NEXT).bool(); + if (bool) { + usingCollection.add(readObject(Fields.RESULT, wireIn, null, vClass1)); + } else + break OUTER; - if (wireIn.read(Fields.HAS_NEXT).bool()) { - V v = readV(Fields.RESULT, wireIn, null); - result.add(v); - } else - break; + // todo process the exception + boolean isException = wireIn.read(Fields.IS_EXCEPTION).bool(); + } } finally { hub.inBytesLock().unlock(); } } - return result; + return usingCollection; } @NotNull public Set> entrySet() { - final long startTime = System.currentTimeMillis(); - final Map result = new HashMap(); + return toMap().entrySet(); + } + private Map toMap() { + final Map result = new HashMap(); + final long startTime = System.currentTimeMillis(); // send final long transactionId = proxySend("ENTRY_SET", startTime); assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; + OUTER: for (; ; ) { // receive @@ -452,18 +414,24 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(Fields.HAS_NEXT).bool()) { - result.put( - readK(Fields.RESULT_KEY, wireIn, null), - readV(Fields.RESULT_VALUE, wireIn, null)); - } else - break; + while (wireIn.bytes().remaining() > 0) { + boolean bool = wireIn.read(Fields.HAS_NEXT).bool(); + if (bool) { + result.put( + readK(Fields.RESULT_KEY, wireIn, null), + readV(Fields.RESULT_VALUE, wireIn, null)); + } else + break OUTER; + + // todo process the exception + boolean isException = wireIn.read(Fields.IS_EXCEPTION).bool(); + } } finally { hub.inBytesLock().unlock(); } } - return result.entrySet(); + return result; } /** @@ -500,7 +468,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon if (value instanceof Byte) wire.write(fieldName).int8((Byte) value); else if (value instanceof Character) - wire.write(fieldName).text(((Character) value).toString()); + wire.write(fieldName).text(value.toString()); else if (value instanceof Short) wire.write(fieldName).int16((Short) value); else if (value instanceof Integer) @@ -537,14 +505,22 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } private V readV(Fields argName, Wire wireIn, V usingValue) { - if (StringBuilder.class.isAssignableFrom(vClass)) { + return readObject(argName, wireIn, usingValue, vClass); + } + + private K readK(Fields argName, Wire wireIn, K usingValue) { + return readObject(argName, wireIn, usingValue, kClass); + } + + private E readObject(Fields argName, Wire wireIn, E usingValue, Class clazz) { + if (StringBuilder.class.isAssignableFrom(clazz)) { wireIn.read(argName).text((StringBuilder) usingValue); return usingValue; - } else if (Marshallable.class.isAssignableFrom(vClass)) { + } else if (Marshallable.class.isAssignableFrom(clazz)) { if (usingValue == null) try { - V v = vClass.newInstance(); + E v = clazz.newInstance(); wireIn.read(argName).marshallable((Marshallable) v); return v; } catch (Exception e) { @@ -554,39 +530,39 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon wireIn.read(argName).marshallable((Marshallable) usingValue); return usingValue; - } else if (String.class.isAssignableFrom(vClass)) { + } else if (String.class.isAssignableFrom(clazz)) { //noinspection unchecked - return (V) wireIn.read(argName).text(); + return (E) wireIn.read(argName).text(); - } else if (Long.class.isAssignableFrom(vClass)) { + } else if (Long.class.isAssignableFrom(clazz)) { //noinspection unchecked - return (V) (Long) wireIn.read(argName).int64(); - } else if (Double.class.isAssignableFrom(vClass)) { + return (E) (Long) wireIn.read(argName).int64(); + } else if (Double.class.isAssignableFrom(clazz)) { //noinspection unchecked - return (V) (Double) wireIn.read(argName).float64(); + return (E) (Double) wireIn.read(argName).float64(); - } else if (Integer.class.isAssignableFrom(vClass)) { + } else if (Integer.class.isAssignableFrom(clazz)) { //noinspection unchecked - return (V) (Integer) wireIn.read(argName).int32(); + return (E) (Integer) wireIn.read(argName).int32(); - } else if (Float.class.isAssignableFrom(vClass)) { + } else if (Float.class.isAssignableFrom(clazz)) { //noinspection unchecked - return (V) (Float) wireIn.read(argName).float32(); + return (E) (Float) wireIn.read(argName).float32(); - } else if (Short.class.isAssignableFrom(vClass)) { + } else if (Short.class.isAssignableFrom(clazz)) { //noinspection unchecked - return (V) (Short) wireIn.read(argName).int16(); + return (E) (Short) wireIn.read(argName).int16(); - } else if (Character.class.isAssignableFrom(vClass)) { + } else if (Character.class.isAssignableFrom(clazz)) { //noinspection unchecked final String text = wireIn.read(argName).text(); if (text == null || text.length() == 0) return null; - return (V) (Character) text.charAt(0); + return (E) (Character) text.charAt(0); - } else if (Byte.class.isAssignableFrom(vClass)) { + } else if (Byte.class.isAssignableFrom(clazz)) { //noinspection unchecked - return (V) (Byte) wireIn.read(argName).int8(); + return (E) (Byte) wireIn.read(argName).int8(); } else { @@ -594,32 +570,6 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } } - private K readK(Fields argName, Wire wireIn, K usingValue) { - if (StringBuilder.class.isAssignableFrom(kClass)) { - wireIn.read(argName).text((StringBuilder) usingValue); - return usingValue; - } else if (Marshallable.class.isAssignableFrom(kClass)) { - - if (usingValue == null) - try { - K v = kClass.newInstance(); - wireIn.read(argName).marshallable((Marshallable) v); - return v; - } catch (Exception e) { - throw new RuntimeException(e); - } - - wireIn.read(argName).marshallable((Marshallable) usingValue); - return usingValue; - - } else if (String.class.isAssignableFrom(vClass)) { - //noinspection unchecked - return (K) wireIn.read(argName).text(); - - } else { - throw new IllegalStateException("unsupported type"); - } - } private boolean readBoolean(long transactionId, long startTime) { assert !hub.outBytesLock().isHeldByCurrentThread(); diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index ee34e817a..cf1b2a2f5 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -42,8 +42,13 @@ import java.io.StreamCorruptedException; import java.io.StringWriter; import java.nio.BufferOverflowException; import java.nio.channels.SelectionKey; -import java.util.*; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Map; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; @@ -55,6 +60,9 @@ import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; */ class MapWireHandler implements WireHandler { + + ExecutorService work = Executors.newSingleThreadExecutor(); + private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); public static final int SIZE_OF_SIZE = 2; @@ -93,9 +101,9 @@ class MapWireHandler implements WireHandler { private List channelList; private ReplicationHub hub; private byte remoteIdentifier; - private Runnable out = null; + private SelectionKey key = null; - private Map incompleteWork = new HashMap<>(); + public MapWireHandler(@NotNull final Supplier>> chronicleHashInstanceBuilder, @NotNull final ReplicationHub hub, @@ -104,6 +112,8 @@ class MapWireHandler implements WireHandler { this(chronicleHashInstanceBuilder, hub); this.channelList = channelList; this.localIdentifier = localIdentifier; + + } public MapWireHandler(Supplier>> chronicleHashInstanceBuilder, ReplicationHub hub) { @@ -122,40 +132,42 @@ class MapWireHandler implements WireHandler { this.inWire = in; this.outWire = out; onEvent(); + + } private void writeChunked(long transactionId, @NotNull final Function function, - @NotNull final Consumer> c) { + @NotNull final Consumer c) { final BytesChronicleMap m = bytesMap(channelId); final Iterator iterator = function.apply(m); // this allows us to write more data than the buffer will allow - out = () -> { + for (; ; ) { // each chunk has its own transaction-id - outWire.write(TRANSACTION_ID).int64(transactionId); + + final boolean hasNext = iterator.hasNext(); write(map -> { - boolean hasNext = iterator.hasNext(); outWire.write(HAS_NEXT).bool(hasNext); if (hasNext) c.accept(iterator); - else - // setting out to NULL denotes that there are no more chunks - out = null; + }); - }; + if (!hasNext) + break; + + } - out.run(); } - private void writeEntryChunked(long transactionId, + /* private void writeEntryChunked(long transactionId, @NotNull final Function function, @NotNull final Consumer>> c) { @@ -163,7 +175,7 @@ class MapWireHandler implements WireHandler { final Iterator> iterator = function.apply(m); // this allows us to write more data than the buffer will allow - out = () -> { + for (; ; ) { // each chunk has its own transaction-id outWire.write(TRANSACTION_ID).int64(transactionId); @@ -173,18 +185,16 @@ class MapWireHandler implements WireHandler { boolean hasNext = iterator.hasNext(); outWire.write(HAS_NEXT).bool(hasNext); - if (hasNext) - c.accept(iterator); - else - // setting out to NULL denotes that there are no more chunks - out = null; + if (!hasNext) + return; + + c.accept(iterator); + }); - }; + } - out.run(); - - } + }*/ @SuppressWarnings("UnusedReturnValue") void onEvent() { @@ -205,6 +215,9 @@ class MapWireHandler implements WireHandler { return; } + // write the transaction id + outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + try { if ("KEY_SET".contentEquals(methodName)) { @@ -218,7 +231,7 @@ class MapWireHandler implements WireHandler { } if ("ENTRY_SET".contentEquals(methodName)) { - writeEntryChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); + writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); return; } @@ -227,8 +240,7 @@ class MapWireHandler implements WireHandler { return; } - // write the transaction id - outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + if ("CREATE_CHANNEL".contentEquals(methodName)) { writeVoid(() -> { @@ -256,30 +268,23 @@ class MapWireHandler implements WireHandler { } if ("CONTAINS_KEY".contentEquals(methodName)) { - write(b -> outWire.write(RESULT). - bool(b.containsKey(toReader(inWire, ARG_1)))); + write(b -> outWire.write(RESULT).bool(b.delegate.containsKey(toByteArray(inWire, ARG_1)))); return; } if ("CONTAINS_VALUE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT). - bool(b.containsKey(toReader(inWire, ARG_1)))); + write(b -> outWire.write(RESULT).bool(b.delegate.containsValue(toByteArray(inWire, ARG_1)))); return; } if ("GET".contentEquals(methodName)) { - - writeValueUsingDelegate(map -> { - byte[] key1 = this.toByteArray(inWire, ARG_1); - return map.get(key1); - }); - + writeValueUsingDelegate(map -> map.get(toByteArray(inWire, ARG_1))); return; } if ("PUT".contentEquals(methodName)) { writeValue(b -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); + final net.openhft.lang.io.Bytes reader = MapWireHandler.this.toReader(inWire, ARG_1, ARG_2); // todo call return b.put(reader, reader, timestamp, identifier()); return b.put(reader, reader); @@ -308,9 +313,10 @@ class MapWireHandler implements WireHandler { if ("REPLACE_WITH_OLD_AND_NEW_VALUE".contentEquals(methodName)) { - writeValue(bytesMap -> { + write(bytesMap -> { final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2, ARG_3); - return bytesMap.replace(reader, reader); + boolean result = bytesMap.replace(reader, reader, reader); + outWire.write(RESULT).bool(result); }); return; @@ -406,11 +412,12 @@ class MapWireHandler implements WireHandler { private void putAll(long transactionId) { - final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); + /*final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); if (bytesMap == null) return; + // note: a number of client threads can be using the same socket Runnable runnable = incompleteWork.get(transactionId); if (runnable != null) { @@ -444,7 +451,7 @@ class MapWireHandler implements WireHandler { }; incompleteWork.put(transactionId, runnable); - runnable.run(); + runnable.run();*/ } @@ -612,10 +619,9 @@ class MapWireHandler implements WireHandler { outWire.bytes().write(fromBytes); }); - - } + @SuppressWarnings("SameReturnValue") private void writeValueUsingDelegate(final Function, byte[]> f) {