diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 50e2beaae..f39ab8d38 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -559,7 +559,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { assert bytes.limit() == bytes.capacity(); - } @@ -605,25 +604,38 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized Set> entrySet() { long sizeLocation = writeEvent(ENTRY_SET); + long startTime = System.currentTimeMillis(); + long transactionId = nextUniqueTransaction(startTime); + + // get the data back from the server - final Bytes in = blockingFetch(sizeLocation); + Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); + final Set> result = new HashSet>(); boolean hasMoreEntries; do { hasMoreEntries = in.readBoolean(); - + LOG.info("entrySet - hasMoreEntries=" + hasMoreEntries); // number of entries in the chunk - long size = in.readUnsignedShort(); + long size = in.readInt(); - LOG.info("size=" + size); + LOG.info("entrySet - size=" + size); for (int i = 0; i < size; i++) { K k = keyValueSerializer.readKey(in); V v = keyValueSerializer.readValue(in); + // LOG.info("v="+v); result.add(new Entry(k, v)); } + if (hasMoreEntries) + try { + in = blockingFetch(System.currentTimeMillis() + builder.timeoutMs(), transactionId); + } catch (IOException e) { + e.printStackTrace(); + } + } while (hasMoreEntries); return result; @@ -697,7 +709,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private Bytes blockingFetch(long sizeLocation) { try { - return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs()); + long startTime = System.currentTimeMillis(); + return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs(), + nextUniqueTransaction(startTime), startTime); } catch (IOException e) { close(); throw new IORuntimeException(e); @@ -708,11 +722,25 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs) throws IOException { - long startTime = System.currentTimeMillis(); + private Bytes blockingFetch0(long sizeLocation, final long transactionId, long startTime) { + + try { + return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs(), transactionId, startTime); + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } catch (Exception e) { + close(); + throw e; + } + + } + + + private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs, final long transactionId, final long startTime) throws IOException { + long timeoutTime = startTime + timeOutMs + 1000000; - long transactionId = nextUniqueTransaction(startTime); for (; ; ) { @@ -730,31 +758,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { // send out all the bytes send(bytes, timeoutTime); - bytes.clear(); - bytes.buffer().clear(); + return blockingFetch(timeoutTime, transactionId); - // the number of bytes in the response - int size = receive(SIZE_OF_SIZE, timeoutTime).readInt(); - - int requiredSize = size + SIZE_OF_SIZE; - if (bytes.capacity() < requiredSize) { - bytes = new ByteBufferBytes(allocateDirect(requiredSize)); - } - - // block until we have received all the byte in this chunk - receive(size, timeoutTime); - - boolean isException = bytes.readBoolean(); - long inTransactionId = bytes.readLong(); - - if (inTransactionId != transactionId) - throw new IllegalStateException("the received transaction-id=" + inTransactionId + - ", does not match the expected transaction-id=" + transactionId); - - if (isException) - throw (RuntimeException) bytes.readObject(); - - return bytes; } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { checkTimeout(timeoutTime); clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); @@ -762,6 +767,34 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } } + private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException { + bytes.clear(); + bytes.buffer().clear(); + + // the number of bytes in the response + int size = receive(SIZE_OF_SIZE, timeoutTime).readInt(); + + int requiredSize = size + SIZE_OF_SIZE; + if (bytes.capacity() < requiredSize) { + bytes = new ByteBufferBytes(allocateDirect(requiredSize)); + } + + // block until we have received all the byte in this chunk + receive(size, timeoutTime); + + boolean isException = bytes.readBoolean(); + long inTransactionId = bytes.readLong(); + + if (inTransactionId != transactionId) + throw new IllegalStateException("the received transaction-id=" + inTransactionId + + ", does not match the expected transaction-id=" + transactionId); + + if (isException) + throw (RuntimeException) bytes.readObject(); + + return bytes; + } + private void write(long transactionId) { bytes.writeLong(transactionId); } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 43efdeb39..01203df82 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1670,6 +1670,7 @@ class StatelessServerConnector { final long sizeLocation = writer.position(); writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is + // exception writer.writeLong(transactionId); @@ -1677,7 +1678,7 @@ class StatelessServerConnector { writer.skip(1); // count - writer.skip(2); + writer.skip(4); int count = 0; while (iterator.hasNext()) { @@ -1686,7 +1687,7 @@ class StatelessServerConnector { // some data, we don't know the max key size, we will use the entrySize instead if (writer.remaining() <= maxEntrySizeBytes) { writeHeader(writer, sizeLocation, count, true); - + LOG.info("One more chunk !"); return false; } @@ -1698,6 +1699,7 @@ class StatelessServerConnector { writeHeader(writer, sizeLocation, count, false); + LOG.info("Last chunk !"); return true; } @@ -1718,7 +1720,7 @@ class StatelessServerConnector { writer.writeBoolean(hasAnotherChunk); // count - writer.writeUnsignedShort(count); + writer.writeInt(count); writer.position(end); } }; diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 84a531b40..973d8d35f 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -36,7 +36,7 @@ import static net.openhft.chronicle.common.StatelessBuilder.remoteAddress; public class StatelessClientTest { - public static final int SIZE = 100; + public static final int SIZE = 1000; @Test