diff --git a/src/main/java/net/openhft/chronicle/common/StatelessBuilder.java b/src/main/java/net/openhft/chronicle/common/StatelessBuilder.java index ddc3c8a13..d359e2411 100644 --- a/src/main/java/net/openhft/chronicle/common/StatelessBuilder.java +++ b/src/main/java/net/openhft/chronicle/common/StatelessBuilder.java @@ -26,7 +26,7 @@ import java.util.concurrent.TimeUnit; */ public class StatelessBuilder { - private long timeoutMs = TimeUnit.SECONDS.toMillis(1); + private long timeoutMs = TimeUnit.SECONDS.toMillis(10); private InetSocketAddress remoteAddress; diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 82e35f25b..95db26acd 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -605,15 +605,20 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { final Set> result = new HashSet>(); + int count = 0; + long total = 0; for (; ; ) { boolean hasMoreEntries = in.readBoolean(); - LOG.info("entrySet - hasMoreEntries=" + hasMoreEntries); + + + count++; // number of entries in the chunk long size = in.readInt(); for (int i = 0; i < size; i++) { + total++; K k = keyValueSerializer.readKey(in); V v = keyValueSerializer.readValue(in); result.add(new Entry(k, v)); @@ -622,9 +627,17 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { if (!hasMoreEntries) break; + if (in.remaining() == 0) { + bytes.clear(); + bytes.buffer().clear(); + } else { + buffer.compact(); + } + in = blockingFetchReadOnly(timeoutTime, transactionId); } + LOG.info("total=" + total); return result; } @@ -763,7 +776,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs, final long transactionId, final long startTime) throws IOException { - long timeoutTime = startTime + timeOutMs + 1000000; + long timeoutTime = startTime + timeOutMs; for (; ; ) { @@ -781,6 +794,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { // send out all the bytes send(bytes, timeoutTime); + bytes.clear(); + bytes.buffer().clear(); + return blockingFetch(timeoutTime, transactionId); } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { @@ -806,8 +822,7 @@ 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(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 0b7504b3f..3dccdf906 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -35,6 +35,7 @@ import java.nio.channels.*; import java.util.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static java.nio.channels.SelectionKey.*; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT; @@ -557,7 +558,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { int bytesJustWritten = attached.entryWriter.writeBufferToSocket(socketChannel, approxTime); - contemplateThrottleWrites(bytesJustWritten); + if (bytesJustWritten > 0) + contemplateThrottleWrites(bytesJustWritten); } catch (IOException e) { quietClose(key, e); @@ -946,14 +948,17 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // if we still have some unwritten writer from last time lastSentTime = approxTime; - out.limit((int) in.position()); + int size = (int) in.position(); + out.limit(size); + final int len = socketChannel.write(out); if (LOG.isDebugEnabled()) LOG.debug("bytes-written=" + len); - if (out.remaining() == 0) { + if (len == size) { + // assert len == size; out.clear(); in.clear(); } else { @@ -1169,11 +1174,14 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // is no longer full, and as such is treated as future work if (futureWork != null) { - // we will complete what we can now - boolean isComplete = futureWork.doWork(attached.entryWriter.in); + 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); + } - if (!isComplete) - attached.entryWriter.uncompletedWork = futureWork; } } @@ -1639,7 +1647,7 @@ class StatelessServerConnector { // we've filled up the buffer, so lets give another channel a chance to send // 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); + writeHeader(writer, sizeLocation, count, true, 0); return false; } @@ -1648,17 +1656,15 @@ class StatelessServerConnector { writeKey(key, writer); } - writeHeader(writer, sizeLocation, count, false); + writeHeader(writer, sizeLocation, count, false, 0); return true; } }; } - private Work entrySet(Bytes reader, Bytes writer) { + private Work entrySet(final Bytes reader, Bytes writer) { final long transactionId = reader.readLong(); - - final Set> entries; try { @@ -1667,7 +1673,8 @@ class StatelessServerConnector { return sendException(reader, writer, e); } - + LOG.info("MAP SIZE=" + map.size()); + final AtomicInteger i = new AtomicInteger(0); final Iterator> iterator = entries.iterator(); // this allows us to write more data than the buffer will allow @@ -1676,6 +1683,9 @@ class StatelessServerConnector { @Override public boolean doWork(Bytes writer) { + if (writer.remaining() <= maxEntrySizeBytes) + return false; + final long sizeLocation = header(writer, transactionId); int count = 0; @@ -1684,7 +1694,8 @@ class StatelessServerConnector { // we've filled up the buffer, so lets give another channel a chance to send // 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); + + writeHeader(writer, sizeLocation, count, true, i.incrementAndGet()); return false; } @@ -1694,7 +1705,8 @@ class StatelessServerConnector { writeValue(next.getValue(), writer); } - writeHeader(writer, sizeLocation, count, false); + + writeHeader(writer, sizeLocation, count, false, i.incrementAndGet()); return true; } @@ -1707,13 +1719,6 @@ class StatelessServerConnector { return sendException(writer, sizeLocation, e); } - private void writeMoreData(long location, boolean hasMoreData, Bytes out) { - out.writeBoolean(location, hasMoreData); - } - - private void entryCount(long location, int count, Bytes out) { - - } private long header(Bytes writer, final long transactionId) { final long sizeLocation = writer.position(); @@ -1731,7 +1736,8 @@ class StatelessServerConnector { return sizeLocation; } - private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) { + private void writeHeader(Bytes writer, long sizeLocation, int count, + final boolean hasAnotherChunk, int seq) { final long end = writer.position(); final int size = (int) (end - sizeLocation); writer.position(sizeLocation); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index cbf02d834..8e99f6f29 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -75,6 +75,7 @@ public class StatelessClientTest { Map.Entry next = entries.iterator().next(); Assert.assertEquals("some value=" + next.getKey(), next.getValue()); + Assert.assertEquals(entries.size(), SIZE); serverMap.close(); statelessMap.close();