Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-230 - reverted back to previouse version

Browse Source
This commit is contained in:
Rob Austin
2014-12-05 20:28:42 +00:00
parent a10f395db7
commit 5f0ee5e6b5
1 changed files with 163 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+163 -210
View File
@@ -82,8 +82,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<Void>() {
@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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<Void>() {
@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<SocketChannel>() {
@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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<? extends K, ? extends V> map) {
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
@@ -720,7 +672,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
inBytes.position(start);
}
public void clear() {
public void clear() {
fetchVoid(CLEAR);
}
@@ -1005,26 +957,59 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<Void>() {
// 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<K, V> implements ChronicleMap<K, V>, 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<Void>() {
@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<K, V> implements ChronicleMap<K, V>, Closeable, Clon
assert transactionId0 != 0;
// check the transaction id is reasonable
assert String.valueOf(transactionId0).startsWith("14");
@@ -1207,11 +1139,18 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<StackTraceElement> 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<StackTraceElement> 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<K, V> implements ChronicleMap<K, V>, 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<Void>() {
@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<K, V> implements ChronicleMap<K, V>, 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) {