diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 6ee6aaa18..d56d34a17 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -992,9 +992,49 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, InterruptedException { - int remainingBytes = 0; + + int remainingBytes = nextEntry(timeoutTime, transactionId); + + if (inBytes.capacity() < remainingBytes) { + long pos = inBytes.position(); + long limit = inBytes.position(); + inBytes.position(limit); + resizeBufferInBuffer(remainingBytes, pos); + } else + inBytes.limit(inBytes.capacity()); + + // block until we have received all the bytes in this chunk + receive(remainingBytes, timeoutTime); + + final boolean isException = inBytes.readBoolean(); + + if (isException) { + Throwable throwable = (Throwable) inBytes.readObject(); + try { + Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); + stackTrace.setAccessible(true); + List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); + // prune the end of the stack. + for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { + stes.remove(i); + } + InetSocketAddress address = config.remoteAddress(); + stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); + StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); + for (int i = 4; i < stackTrace2.length; i++) + stes.add(stackTrace2[i]); + stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); + } catch (Exception ignore) { + } + NativeBytes.UNSAFE.throwException(throwable); + } + return inBytes; + } + + private int nextEntry(long timeoutTime, long transactionId) throws IOException, InterruptedException { + int remainingBytes; for (; ; ) { // read the next item from the socket @@ -1062,46 +1102,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon pause(); } - - - final int minBufferSize = remainingBytes; - - if (inBytes.capacity() < minBufferSize) { - long pos = inBytes.position(); - long limit = inBytes.position(); - inBytes.position(limit); - resizeBufferInBuffer(minBufferSize, pos); - } else - inBytes.limit(inBytes.capacity()); - - // block until we have received all the bytes in this chunk - receive(remainingBytes, timeoutTime); - - final boolean isException = inBytes.readBoolean(); - - if (isException) { - Throwable throwable = (Throwable) inBytes.readObject(); - try { - Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); - stackTrace.setAccessible(true); - List stes = new ArrayList<>(Arrays.asList((StackTraceElement[]) stackTrace.get(throwable))); - // prune the end of the stack. - for (int i = stes.size() - 1; i > 0 && stes.get(i).getClassName().startsWith("Thread"); i--) { - stes.remove(i); - } - InetSocketAddress address = config.remoteAddress(); - stes.add(new StackTraceElement("~ remote", "tcp ~", address.getHostName(), address.getPort())); - StackTraceElement[] stackTrace2 = Thread.currentThread().getStackTrace(); - for (int i = 4; i < stackTrace2.length; i++) - stes.add(stackTrace2[i]); - stackTrace.set(throwable, stes.toArray(new StackTraceElement[stes.size()])); - } catch (Exception ignore) { - } - NativeBytes.UNSAFE.throwException(throwable); - } - - - return inBytes; + return remainingBytes; } private void clearParked() { diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index 923c0a800..79d4ffc3c 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -32,7 +32,7 @@ import java.util.concurrent.TimeUnit; */ public class Builder { - public static final int SIZE = 10000; + public static final int SIZE = 1000; // added to ensure uniqueness static int count; static String WIN_OS = "WINDOWS"; diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index 14dd1dcfe..3d17c22d3 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -5,7 +5,10 @@ import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.DataValueClasses; import net.openhft.lang.values.IntValue; -import org.junit.*; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.IOException; @@ -100,19 +103,19 @@ public class TcpReplicationSoakTest { final int select = rnd.nextInt(2); final ChronicleMap map = select > 0 ? map1 : map2; - if (rnd.nextBoolean()) { - map.put(rnd.nextInt(Builder.SIZE), "test" + j); - } else { - map.remove(rnd.nextInt(Builder.SIZE)); + if (rnd.nextBoolean()) { + map.put(rnd.nextInt(Builder.SIZE), "test" + j); + } else { + map.remove(rnd.nextInt(Builder.SIZE)); + } } } - } - Thread.sleep(1000); + Thread.sleep(1000); System.out.println("\nwaiting till equal"); - waitTillEqual(10000); + waitTillEqual(10000); - Assert.assertEquals(new TreeMap(map1), new TreeMap(map2)); + Assert.assertEquals(map1, map2); }