From ce0013b9fa2984a0800f3b304f69b6f2cd487318 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 4 Mar 2015 21:38:12 +0000 Subject: [PATCH] wire stateless client - still work in progress --- .../map/StatelessWiredConnector.java | 199 +++++++++++++----- .../map/WiredStatelessChronicleMap.java | 36 +++- .../net/openhft/chronicle/wire/TestWire.java | 63 ++---- 3 files changed, 201 insertions(+), 97 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 7997bbc6b..15349c15f 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -27,7 +27,9 @@ import net.openhft.chronicle.bytes.BytesMarshallable; import net.openhft.chronicle.core.OS; import net.openhft.chronicle.wire.TextWire; import net.openhft.chronicle.wire.Wire; +import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.DirectStore; +import net.openhft.lang.io.IByteBufferBytes; import net.openhft.lang.io.NativeBytes; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -40,6 +42,7 @@ import java.io.StringWriter; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.nio.Buffer; +import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; @@ -88,7 +91,50 @@ class StatelessWiredConnector= l) + return; + + long size = outLangBytes.capacity() + l; + if (size > Integer.MAX_VALUE) + throw new BufferOverflowException(); + + // record the current possition and limit + long position = outLangBytes.position(); + + // create a new buffer and copy the data into it + outLangBytes.clear(); + final IByteBufferBytes newOutLangBytes = ByteBufferBytes.wrap(ByteBuffer.allocate((int) size)); + newOutLangBytes.write(outLangBytes); + + outChronBytes = Bytes.wrap(outLangBytes.buffer()); + + newOutLangBytes.limit(newOutLangBytes.capacity()); + newOutLangBytes.position(position); + + outChronBytes.limit(outChronBytes.capacity()); + outChronBytes.position(position); + } + + @Override + public net.openhft.lang.io.Bytes in() { + return outLangBytes; + } + }; + + private final TextWire inWire = new TextWire(Bytes.elasticByteBuffer()); private boolean handshingComplete; @@ -97,7 +143,7 @@ class StatelessWiredConnector bytesChronicleMaps = new ArrayList<>(); private final byte localIdentifier; private final StringBuilder methodName = new StringBuilder(); - private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64); + //private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64); private ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64); private final MapIOBuffer mapIOBuffer = new MapIOBuffer() { @@ -139,7 +185,11 @@ class StatelessWiredConnector channelList; private byte remoteIdentifier; - private Bytes resultBytes = Bytes.wrap(resultBuffer); + // private Bytes resultBytes = Bytes.wrap(resultBuffer); + + // todo improve later, dont make fixed sie + private Bytes resultLangBytes = Bytes.wrap(ByteBuffer.allocate(1024)); + private SelectionKey key; private long transactionId; @@ -151,6 +201,7 @@ class StatelessWiredConnector "CHANNEL_ID").int16(); inWire.read(() -> "METHOD_NAME").text(methodName); + + if ("PUT_WITHOUT_ACC".contentEquals(methodName)) + return putWithOutAcc(); + // for the length long markStart = outWire.bytes().position(); outWire.bytes().skip(4); try { + // write the transaction id outWire.write(() -> "TRANSACTION_ID").int64(transactionId); if ("PUT".contentEquals(methodName)) return put(); + + else if ("GET".contentEquals(methodName)) + return get(); + else throw new IllegalStateException("unsupported event=" + methodName); } finally { @@ -322,6 +387,8 @@ class StatelessWiredConnector field).bytes(source -> { + + System.out.println("field=" + field + ", source='" + new String(source) + "'"); bytes.writeStopBit(source.length); bytes.write(source); }); @@ -337,6 +404,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(true); wire.write(() -> "EXCEPTION").text(sw.toString()); + nofityDataWritten(); return null; } @@ -367,7 +435,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); - if (result != null && result.remaining() > 0) { - BB_ADDRESS.set(resultBuffer, result.address()); - BB_CAPACITY.set(resultBuffer, result.capacity()); - resultBytes.limit(result.capacity()); - outWire.write(() -> "RESULT_IS_NULL").bool(false); - outWire.write(() -> "RESULT").bytes(resultBytes); + bytesMap.output = outBuffer; + bytesMap.put(reader, reader, timestamp, remoteIdentifier); + + outLangBytes.flip(); + + + outChronBytes.position(outLangBytes.position()); + outChronBytes.limit(outLangBytes.limit()); + + //todo not sure why we have to skip 1, what is this field + outChronBytes.skip(1); + + boolean isException = false; + outWire.write(() -> "IS_EXCEPTION").bool(isException); + + if (!isException) { + boolean isNull = outChronBytes.readBoolean(); + outWire.write(() -> "RESULT_IS_NULL").bool(isNull); + + // the size of the result in bytes + long len = outChronBytes.readStopBit(); + + outWire.write(() -> "RESULT").bytes(outChronBytes); } else { - outWire.write(() -> "RESULT_IS_NULL").bool(true); + outWire.write(() -> "EXCEPTION").bytes(outChronBytes); } + nofityDataWritten(); } catch (Throwable e) { - + e.printStackTrace(); // move back to just after where the size - outWire.bytes().position(locationOfSize + 4); + outWire.bytes().position(start); return sendException(outWire, e); - } finally { - bytesMap.output = null; - writeLength(outWire, locationOfSize); } return null; } + @Nullable + private Work putWithOutAcc() { + + final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2"); + + // skip 4 bytes where we will write the size + + final BytesChronicleMap bytesMap = bytesMap(channelId); + + inLanByteBuffer.clear(); + inLangBytes.clear(); + + bytesMap.output = outBuffer; + bytesMap.put(reader, reader, timestamp, remoteIdentifier); + + return null; + } + + @Nullable private Work get() { final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1"); - // skip 4 bytes where we will write the size - long locationOfSize = outWire.bytes().position(); - outWire.bytes().skip(4); - + final long start = outWire.bytes().position(); final BytesChronicleMap bytesMap = bytesMap(channelId); try { @@ -446,29 +540,40 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); - if (result != null && result.remaining() > 0) { - BB_ADDRESS.set(resultBuffer, result.address()); - BB_CAPACITY.set(resultBuffer, result.capacity()); - resultBytes.limit(result.capacity()); - outWire.write(() -> "RESULT_IS_NULL").bool(false); - outWire.write(() -> "RESULT").bytes(resultBytes); + bytesMap.output = outBuffer; + bytesMap.get(reader); + + // the result of the method above is written here + outLangBytes.flip(); + outChronBytes.position(outLangBytes.position()); + outChronBytes.limit(outLangBytes.limit()); + + //todo not sure why we have to skip 1, what is this field + outChronBytes.skip(1); + + boolean isException = false; + outWire.write(() -> "IS_EXCEPTION").bool(isException); + + if (!isException) { + boolean isNull = outChronBytes.readBoolean(); + outWire.write(() -> "RESULT_IS_NULL").bool(isNull); + + // the size of the result in bytes + long len = outChronBytes.readStopBit(); + + outWire.write(() -> "RESULT").bytes(outChronBytes); } else { - outWire.write(() -> "RESULT_IS_NULL").bool(true); + outWire.write(() -> "EXCEPTION").bytes(outChronBytes); } + nofityDataWritten(); } catch (Throwable e) { - + e.printStackTrace(); // move back to just after where the size - outWire.bytes().position(locationOfSize + 4); + outWire.bytes().position(start); return sendException(outWire, e); - } finally { - bytesMap.output = null; - writeLength(outWire, locationOfSize); } return null; diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index 2a1c64c3d..32cf229ed 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -18,6 +18,9 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.bytes.EscapingStopCharTester; +import net.openhft.chronicle.bytes.StopCharTesters; import net.openhft.chronicle.hash.function.SerializableFunction; import net.openhft.chronicle.wire.Marshallable; import net.openhft.chronicle.wire.Wire; @@ -339,7 +342,7 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable } - private V readValue(long transactionId, long startTime, V usingValue) { + private V readValue(long transactionId, long startTime, final V usingValue) { assert !hub.outBytesLock().isHeldByCurrentThread(); long timeoutTime = startTime + hub.timeoutMs; @@ -348,22 +351,40 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (usingValue == null) - usingValue = newValueInstance(); - if (wireIn.read(() -> "IS_EXCEPTION").bool()) throw new RuntimeException(wireIn.read(() -> "EXCEPTION").text()); - if (wireIn.read(() -> "RESULT_IS_NULL").bool()) + if (wireIn.read(() -> "RESULT_IS_NULL").bool()) return null; if (StringBuilder.class.isAssignableFrom(vClass)) { wireIn.read(() -> "RESULT").text((StringBuilder) usingValue); return usingValue; - } else if (usingValue instanceof Marshallable) { + } else if (Marshallable.class.isAssignableFrom(vClass)) { + + if (usingValue == null) + try { + V v = vClass.newInstance(); + wireIn.read(() -> "RESULT").marshallable((Marshallable) v); + return v; + } catch (Exception e) { + throw new RuntimeException(e); + } + wireIn.read(() -> "RESULT").marshallable((Marshallable) usingValue); + return usingValue; + } else if (usingValue instanceof StringBuilder) { + + // todo optomize + + StringBuilder sb = new StringBuilder(); + Bytes b = Bytes.elasticByteBuffer(); + wireIn.read(() -> "RESULT").bytes(b); + b.parseUTF(sb, EscapingStopCharTester.escaping(StopCharTesters.COMMA_STOP)); + return (V) sb.toString(); + } else if (CharSequence.class.isAssignableFrom(vClass)) { - return (V) wireIn.read(() -> "RESULT").text(); + return (V) new String(wireIn.read(() -> "RESULT").bytes()); } else { throw new IllegalStateException("unsupported type"); } @@ -586,6 +607,7 @@ class WiredStatelessChronicleMap implements ChronicleMap, Cloneable if (resultType == vClass) return (R) readValue(transactionId, startTime, null); + else throw new UnsupportedOperationException("class of type class=" + resultType + " is not " + "supported"); diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/wire/TestWire.java index 8a05975be..986ce82fe 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/wire/TestWire.java @@ -19,18 +19,17 @@ package net.openhft.chronicle.wire; import junit.framework.Assert; -import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.hash.replication.ReplicationHub; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.WiredChronicleMapStatelessClientBuilder; import org.jetbrains.annotations.NotNull; +import org.junit.Ignore; import org.junit.Test; import java.io.IOException; import java.net.InetSocketAddress; -import static java.nio.ByteBuffer.allocate; import static java.util.concurrent.TimeUnit.SECONDS; import static net.openhft.chronicle.map.ChronicleMapBuilder.of; import static org.junit.Assert.assertEquals; @@ -62,7 +61,7 @@ public class TestWire { @Test - public void testPutStringWithChannels() throws IOException { + public void testSimplePutStringWithChannels() throws IOException, InterruptedException { { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig @@ -76,48 +75,28 @@ public class TestWire { map1a = of(byte[].class, byte[].class) .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); - map2a = of(byte[].class, byte[].class) - .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 2)).create(); - - map3a = of(byte[].class, byte[].class) - .entries(1000) - .instance().replicatedViaChannel(hubA.createChannel((short) 3)).create(); - - - byte[] key = new byte[4]; - short channelID = (short) 1; byte identifier = (byte) 2; - try (ChronicleMap statelessMap = localClient(8086, identifier, String.class, String.class, channelID)) { - for (int i = 0; i < 10; i++) { - try { - statelessMap.put("Hello" + i, "World"); - } catch (Throwable t) { - t.printStackTrace(); - } - } + final WiredChronicleMapStatelessClientBuilder builder = new WiredChronicleMapStatelessClientBuilder(new InetSocketAddress("localhost", 8086), String.class, String.class, channelID); + builder.identifier(identifier); + builder.putReturnsNull(true); - assertEquals(10, map1a.size()); - - // check that the bytes maps is correct - assertEquals("World", new String(map1a.get("Hello1".getBytes()))); + try (ChronicleMap statelessMap = builder.create()) { + statelessMap.put("Hello", "World"); + assertEquals("World", statelessMap.get("Hello")); } - net.openhft.chronicle.bytes.Bytes bytes = BytesStore.wrap(allocate(128)).bytes(); - TextWire textWire = new TextWire(bytes); - textWire.write(() -> "TYPE").text("TextWire"); - } } + @Ignore @Test - public void testPutMarshanleWithChannels() throws IOException { + public void testPutMarshanleWithChannels() throws IOException, InterruptedException { { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig @@ -139,17 +118,20 @@ public class TestWire { .entries(1000) .instance().replicatedViaChannel(hubA.createChannel((short) 3)).create(); - - byte[] key = new byte[4]; - - short channelID = (short) 1; byte identifier = (byte) 2; + final ChronicleMap statelessMap = localClient(8086, identifier, String.class, Details.class, (byte) 1); - ChronicleMap statelessMap = localClient(8086, identifier, String.class, Details.class, (byte) 1); - statelessMap.put("FirstMap", new Details(String.class, String.class, (byte) 1)); + final Details details = new Details(); + details.keyClass = String.class; + details.valueClass = String.class; + + statelessMap.put("FirstMap", details); final Details firstMap = statelessMap.get("FirstMap"); + + Thread.sleep(100); + Assert.assertNotNull(map1a.get("FirstMap".getBytes())); Assert.assertEquals(1, firstMap.channelID); } @@ -161,7 +143,7 @@ public class TestWire { TextWire wire = new TextWire(net.openhft.chronicle.bytes.Bytes.elasticByteBuffer()); - WireOut world = wire.write(() -> "hello").text("world"); + wire.write(() -> "hello").text("world"); System.out.println(wire.bytes()); @@ -173,11 +155,6 @@ public class TestWire { Class valueClass; short channelID; - public Details(Class keyClass, Class valueClass, short channelID) { - this.keyClass = keyClass; - this.valueClass = valueClass; - this.channelID = channelID; - } @Override public void writeMarshallable(WireOut wire) {