diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 9465b3d76..80399efc0 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -602,41 +602,36 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { @NotNull public synchronized Set> entrySet() { - long sizeLocation = writeEvent(ENTRY_SET); - - long startTime = System.currentTimeMillis(); - long transactionId = nextUniqueTransaction(startTime); + final long sizeLocation = writeEvent(ENTRY_SET); + final long startTime = System.currentTimeMillis(); + final long transactionId = nextUniqueTransaction(startTime); + final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs(); // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); final Set> result = new HashSet>(); - boolean hasMoreEntries; - do { - hasMoreEntries = in.readBoolean(); + for (; ; ) { + + boolean hasMoreEntries = in.readBoolean(); LOG.info("entrySet - hasMoreEntries=" + hasMoreEntries); + // number of entries in the chunk long size = in.readInt(); - 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(); - } + if (!hasMoreEntries) + break; - } while (hasMoreEntries); + in = blockingFetchReadOnly(timeoutTime, transactionId); + } return result; } @@ -767,6 +762,21 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } } + private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { + + try { + return blockingFetch(timeoutTime, transactionId); + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } catch (Exception e) { + close(); + throw e; + } + + } + + private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException { bytes.clear(); bytes.buffer().clear(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 01203df82..71df8e960 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1667,18 +1667,7 @@ class StatelessServerConnector { @Override public boolean doWork(Bytes writer) { - final long sizeLocation = writer.position(); - - writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is - - // exception - writer.writeLong(transactionId); - - // hasAnotherChunk - writer.skip(1); - - // count - writer.skip(4); + final long sizeLocation = header(writer, transactionId); int count = 0; while (iterator.hasNext()) { @@ -1703,26 +1692,7 @@ class StatelessServerConnector { return true; } - private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) { - final long end = writer.position(); - final int size = (int) (end - sizeLocation); - writer.position(sizeLocation); - // size in bytes - writer.writeInt(size); - - // is exception - writer.writeBoolean(false); - - //transaction id; - writer.skip(8); - - writer.writeBoolean(hasAnotherChunk); - - // count - writer.writeInt(count); - writer.position(end); - } }; } @@ -1734,6 +1704,43 @@ class StatelessServerConnector { } + private long header(Bytes writer, final long transactionId) { + final long sizeLocation = writer.position(); + + writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is + + // exception + writer.writeLong(transactionId); + + // hasAnotherChunk + writer.skip(1); + + // count + writer.skip(4); + return sizeLocation; + } + + private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) { + final long end = writer.position(); + final int size = (int) (end - sizeLocation); + writer.position(sizeLocation); + + // size in bytes + writer.writeInt(size); + + // is exception + writer.writeBoolean(false); + + //transaction id; + writer.skip(8); + + writer.writeBoolean(hasAnotherChunk); + + // count + writer.writeInt(count); + writer.position(end); + } + private Work putIfAbsent(Bytes reader, Bytes writer) { final K key = readKey(reader); final V v = readValue(reader); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 973d8d35f..cbf02d834 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 = 1000; + public static final int SIZE = 100000; @Test @@ -78,11 +78,50 @@ public class StatelessClientTest { serverMap.close(); statelessMap.close(); - - // System.out.print("" + entries); - } + + @Test + public void testBufferOverFlowPutAllAndKeySet() throws IOException, InterruptedException { + + final ChronicleMap serverMap; + final ChronicleMap statelessMap; + + + // stateless client + { + statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8076))).create(); + } + + // server + { + serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replicators((byte) 2, TcpReplicationConfig.of(8076)).create(); + } + + + Map payload = new HashMap(); + + for (int i = 0; i < SIZE; i++) { + payload.put(i, "some value=" + i); + } + + + statelessMap.putAll(payload); + + + final Set keys = statelessMap.keySet(); + + Assert.assertEquals(keys.size(), payload.size()); + + + serverMap.close(); + statelessMap.close(); + } + + @Test public void test() throws IOException, InterruptedException { @@ -187,9 +226,6 @@ public class StatelessClientTest { } - - - @Test public void testEquals() throws IOException, InterruptedException {