From 5f0ee5e6b51de5aeb958dfce989cf622d5f2f6a6 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Dec 2014 20:28:42 +0000 Subject: [PATCH] HCOLL-230 - reverted back to previouse version --- .../chronicle/map/StatelessChronicleMap.java | 373 ++++++++---------- 1 file changed, 163 insertions(+), 210 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 06a1a5381..82d9450f0 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -82,8 +82,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final boolean putReturnsNull; private final boolean removeReturnsNull; - private ExecutorService ioExecutor = Executors.newSingleThreadExecutor(); - private final ReentrantReadWriteLock inBytesLock = new ReentrantReadWriteLock(true); @@ -206,8 +204,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return executorService; } - private synchronized SocketChannel lazyConnect0(final long timeoutMs, - final InetSocketAddress remoteAddress) throws IOException { + private synchronized SocketChannel lazyConnect(final long timeoutMs, + final InetSocketAddress remoteAddress) throws IOException { if (LOG.isDebugEnabled()) LOG.debug("attempting to connect to " + remoteAddress); @@ -231,9 +229,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon break; } catch (IOException e) { closeables.closeQuietly(); - } catch (Throwable e) { + } catch (Exception e) { closeables.closeQuietly(); - NativeBytes.UNSAFE.throwException(e); + throw e; } } @@ -241,7 +239,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } /** - * attempts a single connect without a timeout and eats any IOException used typically in the + * attempts a single connect without a timeout and eat any IOException used typically in the * constructor, its possible the server is not running with this instance is created, {@link * net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, * java.net.InetSocketAddress)} will attempt to establish the connection when the client make @@ -253,33 +251,19 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @see net.openhft.chronicle.map.StatelessChronicleMap#lazyConnect(long, * java.net.InetSocketAddress) */ - private synchronized void attemptConnect(@NotNull final InetSocketAddress remoteAddress) throws - IOException { + + private synchronized void attemptConnect(final InetSocketAddress remoteAddress) throws IOException { // ensures that the excising connection are closed closeExisting(); try { - ioExecutor.submit(new Callable() { - - @Override - public Void call() throws Exception { - StatelessChronicleMap.this.clientChannel = AbstractChannelReplicator - .openSocketChannel(closeables); - clientChannel.connect(remoteAddress); - doHandShaking0(clientChannel); - return null; - } - - }).get(); - - } catch (ExecutionException e) { - NativeBytes.UNSAFE.throwException(e.getCause()); - } catch (Throwable e) { + clientChannel = AbstractChannelReplicator.openSocketChannel(closeables); + clientChannel.connect(remoteAddress); + doHandShaking(clientChannel); + } catch (IOException e) { closeables.closeQuietly(); - NativeBytes.UNSAFE.throwException(e); } - } /** @@ -301,9 +285,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @param clientChannel * @throws java.io.IOException */ - private void doHandShaking0(@NotNull final SocketChannel clientChannel) throws - IOException { - + private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException { connectionByte[0] = STATELESS_CLIENT_IDENTIFIER; this.connectionOutBuffer.clear(); @@ -333,53 +315,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier); } - private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws - Throwable { - - try { - ioExecutor.submit(new Callable() { - - @Override - public Void call() throws Exception { - doHandShaking0(clientChannel); - return null; - } - - }).get(); - - } catch (ExecutionException e) { - throw e.getCause(); - } - - } - - private synchronized SocketChannel lazyConnect(final long timeoutMs, - final InetSocketAddress remoteAddress) throws Throwable { - - try { - return ioExecutor.submit(new Callable() { - - @Override - public SocketChannel call() throws Exception { - return lazyConnect0(timeoutMs, remoteAddress); - } - - }).get(); - - } catch (ExecutionException e) { - throw e.getCause(); - } - - } - - public File file() { throw new UnsupportedOperationException(); } public void close() { - if (closeables != null) closeables.closeQuietly(); closeables = null; @@ -395,17 +336,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - - if (ioExecutor != null) { - ioExecutor.shutdown(); - try { - if (!ioExecutor.awaitTermination(20, TimeUnit.SECONDS)) { - ioExecutor.shutdownNow(); - } - } catch (InterruptedException e) { - LOG.error("", e); - } - } } /** @@ -573,6 +503,28 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } + private synchronized void sendWithoutAcc(final Bytes bytes, long sizeLocation) { + writeSizeAt(sizeLocation); + long timeoutTime = config.timeoutMs() + System.currentTimeMillis(); + try { + for (; ; ) { + if (clientChannel == null) + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + + try { + send(bytes, timeoutTime); + return; + } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + checkTimeout(timeoutTime); + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + } + } + } catch (IOException e) { + close(); + throw new IORuntimeException(e); + } + } + private synchronized void writeEntriesForPutAll(Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; @@ -720,7 +672,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon inBytes.position(start); } - public void clear() { + public void clear() { fetchVoid(CLEAR); } @@ -1005,26 +957,59 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } catch (IOException e) { close(); throw new IORuntimeException(e); - - } catch (Throwable e) { + } catch (Exception e) { close(); - NativeBytes.UNSAFE.throwException(e); + throw e; } return transactionId; } +/* private Bytes blockingFetchThrowable(long sizeLocation, final long transactionId, final long timeoutTime) throws IOException, InterruptedException { + + for (; ; ) { + if (clientChannel == null) { + synchronized (this) { + if (clientChannel == null) { + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + } + } + + try { + + if (LOG.isDebugEnabled()) + LOG.debug("sending data with transactionId=" + transactionId); + + writeSizeAndTransactionIdAt(sizeLocation, transactionId); + + // send out all the bytes + send(outBytes, timeoutTime); + + return blockingFetchThrowable(timeoutTime, transactionId); + } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + checkTimeout(timeoutTime); + clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); + } + } + } + + }*/ + private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) { try { return blockingFetchThrowable(timeoutTime, transactionId); } catch (IOException e) { close(); throw new IORuntimeException(e); - } catch (Throwable e) { + } catch (RuntimeException e) { close(); + throw e; + } catch (Exception e) { + close(); + throw new RuntimeException(e); + } catch (AssertionError e) { LOG.error("", e); - NativeBytes.UNSAFE.throwException(e); - return null; + throw e; } } @@ -1044,43 +1029,32 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @return bytes read from the TCP/IP socket * @throws IOException socket failed to read data */ - private Bytes receiveBuffer(int requiredNumberOfBytes, long timeoutTime) throws Throwable { + private Bytes receiveBuffer(int requiredNumberOfBytes, long timeoutTime) throws IOException { - assert requiredNumberOfBytes <= buffer.capacity(); buffer.clear(); buffer.buffer().clear(); + /*System.out.println(buffer.position()); + System.out.println(buffer.limit()); + */ + synchronized (this) { + while (buffer.buffer().remaining() > 0) { + assert requiredNumberOfBytes <= buffer.capacity(); + int len; - try { - ioExecutor.submit(new Callable() { + // you can't read and write from the buffer a the same time - @Override - public Void call() throws Exception { - while (buffer.buffer().remaining() > 0) { - - int len; - - // you can't read and write from the buffer a the same time - - len = clientChannel.read(buffer.buffer()); + len = clientChannel.read(buffer.buffer()); - // System.out.println(("receiveBuffer = bytes read=" + len)); + // System.out.println(("receiveBuffer = bytes read=" + len)); - if (len == -1) - throw new IORuntimeException("Disconnected to remote server"); - } - return null; - } + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); - }).get(timeoutTime, TimeUnit.MILLISECONDS); + checkTimeout(timeoutTime); - - } catch (TimeoutException e) { - throw new RemoteCallTimeoutException("timed out waiting to send data."); - } catch (ExecutionException e) { - throw e.getCause(); + } } - buffer.clear(); // System.out.println("buffer=" + AbstractBytes.toHex(buffer)); @@ -1088,51 +1062,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - /** - * reads up to the number of byte in {@code requiredNumberOfBytes} - * - * @param requiredNumberOfBytes the number of bytes to read - * @param timeoutTime timeout in milliseconds - * @return bytes read from the TCP/IP socket - * @throws IOException socket failed to read data - */ - private synchronized Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws - Throwable { - - inBytes.buffer().position(0); - inBytes.buffer().limit(requiredNumberOfBytes); - - assert requiredNumberOfBytes <= inBytes.capacity(); - - try { - ioExecutor.submit(new Callable() { - - @Override - public Void call() throws Exception { - while (inBytes.buffer().remaining() > 0) { - - int len = clientChannel.read(inBytes.buffer()); - if (len == -1) - throw new IORuntimeException("Disconnected to remote server"); - } - return null; - } - - }).get(timeoutTime, TimeUnit.MILLISECONDS); - - - } catch (TimeoutException e) { - throw new RemoteCallTimeoutException("timed out waiting to send data."); - } catch (ExecutionException e) { - throw e.getCause(); - } - - inBytes.position(0); - inBytes.limit(requiredNumberOfBytes); - return inBytes; - } - - private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws Throwable { + private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException, + InterruptedException { int remainingBytes = 0; try { @@ -1153,6 +1084,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon assert transactionId0 != 0; + // check the transaction id is reasonable assert String.valueOf(transactionId0).startsWith("14"); @@ -1207,11 +1139,18 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon pause(); } - } catch (Throwable e) { + } catch (Exception e) { parkedTransactionTimeStamp = 0; parkedTransactionId = 0; parkedRemainingBytes = 0; - NativeBytes.UNSAFE.throwException(e); + LOG.error("", e); + throw e; + } catch (Error e) { + parkedTransactionTimeStamp = 0; + parkedTransactionId = 0; + parkedRemainingBytes = 0; + LOG.error("", e); + throw e; } @@ -1224,39 +1163,37 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon resizeBufferInBuffer(minBufferSize, pos); } else inBytes.limit(inBytes.capacity()); + synchronized (this) { + // block until we have received all the bytes in this chunk + receive(remainingBytes, timeoutTime); - boolean isException; + final boolean isException = inBytes.readBoolean(); - - // block until we have received all the bytes in this chunk - receive(remainingBytes, timeoutTime); - 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); + 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) { } - 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); } - NativeBytes.UNSAFE.throwException(throwable); - } + } return inBytes; } - private void clearParked() { parkedTransactionId = 0; parkedRemainingBytes = 0; @@ -1269,32 +1206,44 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void send(final Bytes out, final long timeoutTime) throws Throwable { + /** + * reads up to the number of byte in {@code requiredNumberOfBytes} + * + * @param requiredNumberOfBytes the number of bytes to read + * @param timeoutTime timeout in milliseconds + * @return bytes read from the TCP/IP socket + * @throws IOException socket failed to read data + */ + private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException { + + inBytes.buffer().position(0); + inBytes.buffer().limit(requiredNumberOfBytes); + + while (inBytes.buffer().remaining() > 0) { + assert requiredNumberOfBytes <= inBytes.capacity(); + + int len = clientChannel.read(inBytes.buffer()); + + if (len == -1) + throw new IORuntimeException("Disconnected to remote server"); + + checkTimeout(timeoutTime); + } + + inBytes.position(0); + inBytes.limit(requiredNumberOfBytes); + return inBytes; + } + + + private synchronized void send(final Bytes out, long timeoutTime) throws IOException { outBuffer.limit((int) out.position()); outBuffer.position(0); - try { - ioExecutor.submit(new Callable() { - - @Override - public Void call() throws Exception { - - while (outBuffer.remaining() > 0) { - int len = clientChannel.write(outBuffer); - if (len == -1) - throw new IORuntimeException("Disconnected to remote server"); - } - - return null; - } - - }).get(timeoutTime, TimeUnit.MILLISECONDS); - - } catch (TimeoutException e) { - throw new RemoteCallTimeoutException("timed out waiting to send data."); - } catch (ExecutionException e) { - throw e.getCause(); + while (outBuffer.remaining() > 0) { + clientChannel.write(outBuffer); + checkTimeout(timeoutTime); } out.clear(); @@ -1475,11 +1424,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } private long send(final EventId eventId, final long startTime) { + long transactionId = nextUniqueTransaction(startTime); // send synchronized (this) { - return send(writeEventAnSkip(eventId), startTime); + final long sizeLocation = writeEventAnSkip(eventId); + transactionId = send(sizeLocation, startTime); } + + return transactionId; } private V readValue(ThreadLocalCopies copies, Bytes in) {