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

Merge branch 'master' of github.com:OpenHFT/Chronicle-Map

Browse Source 
Conflicts:
	src/test/java/eg/BGChronicleTest.java
This commit is contained in:
leventov
2014-12-06 00:11:30 +07:00
parent f8cb1d49d9 d742aec3cf
commit 3f8386a0d5
6 changed files with 526 additions and 360 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+12 -11
View File
@@ -273,8 +273,8 @@ public class NodeDiscovery {
}
/**
* bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset
* and returns the destination
* bitwise OR's the two bit sets or put another way, merges the source bitset into the
* destination bitset and returns the destination
*
* @param source to or
* @param destination to or
@@ -326,7 +326,7 @@ public class NodeDiscovery {
"going the grid.");
}
if (possible == 128)
if (possible == Byte.MAX_VALUE)
possible = 0;
else
possible++;
@@ -336,8 +336,8 @@ public class NodeDiscovery {
}
/**
* Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote nodes
* just by name, this functionality requires UDP
* Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote
* nodes just by name, this functionality requires UDP
*
* @author Rob Austin.
*/
@@ -604,10 +604,11 @@ class AddressAndPort implements Comparable<AddressAndPort>, BytesMarshallable {
return port;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (o == null || AddressAndPort.class != o.getClass()) return false;
AddressAndPort that = (AddressAndPort) o;
@@ -728,8 +729,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable {
}
/**
* this is used to tell nodes that are connecting to us which host and ports are in our grid, along with
* all the identifiers.
* this is used to tell nodes that are connecting to us which host and ports are in our grid,
* along with all the identifiers.
*/
@Override
public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) {
@@ -887,7 +888,7 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable {
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (o == null || ProposedNodes.class != o.getClass()) return false;
ProposedNodes that = (ProposedNodes) o;
@@ -906,8 +907,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable {
}
/**
* sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the host:port
* and perhaps even proposed identifier of the node that sent it.
* sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the
* host:port and perhaps even proposed identifier of the node that sent it.
*
* @param proposedNodes
*/
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+209 -201
View File
@@ -41,7 +41,7 @@ import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static java.nio.ByteBuffer.allocateDirect;
import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE;
@@ -82,8 +82,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private final boolean putReturnsNull;
private final boolean removeReturnsNull;
private final ReentrantLock inBytesLock = new ReentrantLock(true);
private final ReentrantLock outBytesLock = new ReentrantLock(true);
private final ReentrantReadWriteLock inBytesLock = new ReentrantReadWriteLock(true);
private ExecutorService executorService;
@@ -204,8 +204,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
return executorService;
}
private SocketChannel lazyConnect(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);
@@ -252,7 +252,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
* java.net.InetSocketAddress)
*/
private void attemptConnect(final InetSocketAddress remoteAddress) throws IOException {
private synchronized void attemptConnect(final InetSocketAddress remoteAddress) throws IOException {
// ensures that the excising connection are closed
closeExisting();
@@ -285,7 +285,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
* @param clientChannel
* @throws java.io.IOException
*/
private void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException {
private synchronized void doHandShaking(@NotNull final SocketChannel clientChannel) throws IOException {
connectionByte[0] = STATELESS_CLIENT_IDENTIFIER;
this.connectionOutBuffer.clear();
@@ -428,7 +428,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
return fetchObject(String.class, TO_STRING);
}
public synchronized boolean isEmpty() {
public boolean isEmpty() {
return fetchBoolean(IS_EMPTY);
}
@@ -649,7 +649,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private void resizeBufferInBuffer(int size, long start) {
if (LOG.isDebugEnabled())
LOG.debug("resizing buffer to size=" + size);
LOG.debug("InBuffer resizing buffer to size=" + size);
if (size < inBuffer.capacity())
throw new IllegalStateException("it not possible to resize the buffer smaller");
@@ -704,7 +704,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
BytesReader<V> valueReader = valueReaderWithSize.readerForLoop(null);
for (; ; ) {
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
final Bytes in = blockingFetchReadOnly(timeoutTime, transactionId);
@@ -722,7 +722,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
break;
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
return result;
@@ -745,7 +745,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
for (; ; ) {
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
Bytes in = blockingFetchReadOnly(timeoutTime, transactionId);
@@ -765,7 +765,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
break;
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -784,7 +784,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
for (; ; ) {
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
final Bytes in = blockingFetchReadOnly(timeoutTime, transactionId);
@@ -801,7 +801,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
break;
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -813,11 +813,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long timeoutTime = startTime + this.config.timeoutMs();
// receive
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
return blockingFetchReadOnly(timeoutTime, transactionId).readLong();
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -940,6 +940,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// send out all the bytes
send(outBytes, timeoutTime);
outBytes.clear();
outBytes.buffer().clear();
break;
} catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) {
@@ -961,27 +963,32 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private Bytes blockingFetchThrowable(long sizeLocation, final long transactionId, final long timeoutTime) throws IOException, InterruptedException {
for (; ; ) {
if (clientChannel == null)
clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress());
if (clientChannel == null) {
synchronized (this) {
if (clientChannel == null) {
clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress());
}
}
try {
try {
if (LOG.isDebugEnabled())
LOG.debug("sending data with transactionId=" + transactionId);
if (LOG.isDebugEnabled())
LOG.debug("sending data with transactionId=" + transactionId);
writeSizeAndTransactionIdAt(sizeLocation, transactionId);
writeSizeAndTransactionIdAt(sizeLocation, transactionId);
// send out all the bytes
send(outBytes, timeoutTime);
// 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());
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 {
@@ -995,6 +1002,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
} catch (Exception e) {
close();
throw new RuntimeException(e);
} catch (AssertionError e) {
LOG.error("", e);
throw e;
}
}
@@ -1003,80 +1013,139 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private volatile int parkedRemainingBytes;
private volatile long parkedTransactionTimeStamp;
ByteBufferBytes sizeAndTransactionIdBuffer = new ByteBufferBytes(ByteBuffer.allocate
ByteBufferBytes buffer = new ByteBufferBytes(ByteBuffer.allocate
(SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID));
/**
* 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 receiveBuffer(int requiredNumberOfBytes, long timeoutTime) throws IOException {
private synchronized Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException,
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;
// you can't read and write from the buffer a the same time
len = clientChannel.read(buffer.buffer());
// System.out.println(("receiveBuffer = bytes read=" + len));
if (len == -1)
throw new IORuntimeException("Disconnected to remote server");
checkTimeout(timeoutTime);
}
}
buffer.clear();
// System.out.println("buffer=" + AbstractBytes.toHex(buffer));
return buffer;
}
private Bytes blockingFetchThrowable(long timeoutTime, long transactionId) throws IOException,
InterruptedException {
int remainingBytes = 0;
for (; ; ) {
// read the next item from the socket
if (parkedTransactionId == 0) {
try {
for (; ; ) {
assert parkedTransactionTimeStamp == 0;
assert parkedRemainingBytes == 0;
// read the next item from the socket
if (parkedTransactionId == 0) {
receive(SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID, timeoutTime);
final int messageSize = inBytes.readInt();
final int remainingBytes0 = messageSize - (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID);
final long transactionId0 = inBytes.readLong();
assert transactionId0 != 0;
assert parkedTransactionTimeStamp == 0;
assert parkedRemainingBytes == 0;
// if the transaction id is not for this thread, park it and read the next one
if (transactionId0 == transactionId) {
receiveBuffer(SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID, timeoutTime);
// we have the correct transaction id !
parkedTransactionId = 0;
remainingBytes = remainingBytes0;
assert remainingBytes > 0;
final int messageSize = buffer.readInt();
final int remainingBytes0 = messageSize - (SIZE_OF_SIZE + SIZE_OF_TRANSACTION_ID);
final long transactionId0 = buffer.readLong();
assert transactionId0 != 0;
// check the transaction id is reasonable
assert String.valueOf(transactionId0).startsWith("14");
// if the transaction id is not for this thread, park it and read the next one
if (transactionId0 == transactionId) {
// we have the correct transaction id !
parkedTransactionId = 0;
remainingBytes = remainingBytes0;
assert remainingBytes > 0;
clearParked();
break;
} else {
parkedTransactionTimeStamp = System.currentTimeMillis();
parkedRemainingBytes = remainingBytes0;
parkedTransactionId = transactionId0;
pause();
continue;
}
}
// the transaction id was read by another thread, but it is for this thread
if (parkedTransactionId == transactionId) {
remainingBytes = parkedRemainingBytes;
clearParked();
break;
}
// time out the old transaction id
if (System.currentTimeMillis() - timeoutTime >
parkedTransactionTimeStamp) {
LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" +
parkedTransactionTimeStamp +
", this can occur when you have another thread which has called the " +
"stateless client and terminated abruptly before the message has been " +
"returned from the server"));
// read the the next message
receive(parkedRemainingBytes, timeoutTime);
clearParked();
break;
} else {
parkedTransactionTimeStamp = System.currentTimeMillis();
parkedRemainingBytes = remainingBytes0;
parkedTransactionId = transactionId0;
pause();
continue;
}
pause();
}
// the transaction id was read by another thread, but it is for this thread
if (parkedTransactionId == transactionId) {
remainingBytes = parkedRemainingBytes;
clearParked();
break;
}
// time out the old transaction id
if (System.currentTimeMillis() - timeoutTime >
parkedTransactionTimeStamp) {
LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" +
parkedTransactionTimeStamp +
", this can occur when you have another thread which has called the " +
"stateless client and terminated abruptly before the message has been " +
"returned from the server"));
parkedTransactionTimeStamp = 0;
parkedTransactionId = 0;
parkedRemainingBytes = 0;
// read the the next message
receive(remainingBytes, timeoutTime);
continue;
}
pause();
} catch (Exception e) {
parkedTransactionTimeStamp = 0;
parkedTransactionId = 0;
parkedRemainingBytes = 0;
LOG.error("", e);
throw e;
} catch (Error e) {
parkedTransactionTimeStamp = 0;
parkedTransactionId = 0;
parkedRemainingBytes = 0;
LOG.error("", e);
throw e;
}
@@ -1089,34 +1158,34 @@ 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);
// block until we have received all the bytes in this chunk
receive(remainingBytes, timeoutTime);
final boolean isException = inBytes.readBoolean();
final boolean 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;
}
@@ -1127,8 +1196,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void pause() throws InterruptedException {
inBytesLock.unlock();
inBytesLock.lock();
inBytesLock.writeLock().unlock();
inBytesLock.writeLock().lock();
}
@@ -1142,10 +1211,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
*/
private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException {
inBytes.position(0);
inBytes.limit(requiredNumberOfBytes);
inBytes.buffer().position(0);
inBytes.buffer().limit(requiredNumberOfBytes);
@@ -1189,14 +1254,19 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long size = outBytes.position() - locationOfSize;
final long pos = outBytes.position();
outBytes.position(locationOfSize);
outBytes.writeInt((int) size - SIZE_OF_SIZE);
int size0 = (int) size - SIZE_OF_SIZE;
assert size0 > 0;
outBytes.writeInt(size0);
outBytes.writeLong(transactionId);
outBytes.position(pos);
}
private void writeSizeAt(long locationOfSize) {
final long size = outBytes.position() - locationOfSize;
outBytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE);
int size0 = (int) size - SIZE_OF_SIZE;
assert size0 > 0;
outBytes.writeInt(locationOfSize, size0);
}
private long markSizeLocation() {
@@ -1292,7 +1362,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
int size = Integer.parseInt(substring.substring(0, i));
long requiresExtra = size - outBytes.remaining();
System.out.println(size + substring);
resizeBufferOutBuffer((int) (outBytes.capacity() + requiresExtra), start);
} else
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
@@ -1304,11 +1373,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long timeoutTime = startTime + this.config.timeoutMs();
// receive
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
return readValue(copies, blockingFetchReadOnly(timeoutTime, transactionId));
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -1316,11 +1385,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long timeoutTime = startTime + this.config.timeoutMs();
// receive
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
return (O) (blockingFetchReadOnly(timeoutTime, transactionId).readObject());
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -1328,11 +1397,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long timeoutTime = startTime + this.config.timeoutMs();
// receive
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
return blockingFetchReadOnly(timeoutTime, transactionId).readBoolean();
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -1341,11 +1410,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long timeoutTime = startTime + this.config.timeoutMs();
// receive
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
return blockingFetchReadOnly(timeoutTime, transactionId).readInt();
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -1353,12 +1422,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId = nextUniqueTransaction(startTime);
// send
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = writeEventAnSkip(eventId);
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
return transactionId;
@@ -1374,8 +1440,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = writeEventAnSkip(eventId);
@@ -1385,8 +1450,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
copies = writeValue(value, copies);
}
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
@@ -1400,14 +1463,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = writeEventAnSkip(eventId);
writeValue(value, null);
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
@@ -1421,14 +1482,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = writeEventAnSkip(eventId);
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
return readLong(transactionId, startTime);
@@ -1440,14 +1496,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = writeEventAnSkip(eventId);
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
@@ -1463,11 +1514,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long timeoutTime = startTime + this.config.timeoutMs();
// receive
inBytesLock.lock();
inBytesLock.writeLock().lock();
try {
return (O) blockingFetchReadOnly(timeoutTime, transactionId).readObject(tClass);
} finally {
inBytesLock.unlock();
inBytesLock.writeLock().unlock();
}
}
@@ -1477,12 +1528,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = writeEventAnSkip(eventId);
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
@@ -1497,8 +1545,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = eventReturnsNull(eventId)
? writeEvent(eventId)
@@ -1509,10 +1556,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
copies = writeValue(value, copies);
}
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
if (eventReturnsNull(eventId))
return null;
@@ -1530,8 +1576,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = eventReturnsNull(eventId)
? writeEvent(eventId)
@@ -1539,8 +1584,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
copies = writeKey(key);
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
if (eventReturnsNull(eventId))
@@ -1553,38 +1596,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
"supported");
}
private <O> O fetchObject(final EventId eventId) {
final long startTime = System.currentTimeMillis();
long transactionId = send(eventId, startTime);
return readObject(transactionId, startTime);
}
private <O> O fetchObject(final EventId eventId, K key, V... values) {
final long startTime = System.currentTimeMillis();
long transactionId = sendAndGetTransactionId(startTime, eventId, key, values);
return (O) readObject(transactionId, startTime);
}
private long sendAndGetTransactionId(long startTime, final EventId eventId, K key, V... values) {
ThreadLocalCopies copies;
long transactionId;// send
outBytesLock.lock();
try {
final long sizeLocation = (eventReturnsNull(eventId))
? writeEvent(eventId)
: writeEventAnSkip(eventId);
copies = writeKey(key);
for (V value : values) {
copies = writeValue(value, copies);
}
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
return transactionId;
}
private boolean eventReturnsNull(EventId eventId) {
@@ -1622,14 +1633,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long startTime = System.currentTimeMillis();
long transactionId;
outBytesLock.lock();
try {
synchronized (this) {
final long sizeLocation = writeEventAnSkip(eventId);
writeKey(key);
writeObject(object);
transactionId = blockingSend(sizeLocation, startTime);
} finally {
outBytesLock.unlock();
}
if (eventReturnsNull(eventId))
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+22 -12
View File
@@ -487,15 +487,15 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
channel.socket().setSoLinger(false, 0);
final Attached attached = new Attached();
channel.register(selector, OP_READ, attached);
throttle(channel);
attached.entryReader = new TcpSocketChannelEntryReader();
attached.entryWriter = new TcpSocketChannelEntryWriter();
attached.isServer = true;
attached.entryWriter.identifierToBuffer(localIdentifier);
attached.isServer = true;
channel.register(selector, OP_READ, attached);
throttle(channel);
}
/**
@@ -622,17 +622,19 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
final long approxTime) throws InterruptedException, IOException {
final SocketChannel socketChannel = (SocketChannel) key.channel();
final Attached attached = (Attached) key.attachment();
if (attached.entryWriter.isWorkIncomplete()) {
final boolean completed = attached.entryWriter.doWork();
if (attached == null) throw new NullPointerException("No attached");
TcpSocketChannelEntryWriter entryWriter = attached.entryWriter;
if (entryWriter == null) throw new NullPointerException("No entryWriter");
if (entryWriter.isWorkIncomplete()) {
final boolean completed = entryWriter.doWork();
if (completed)
attached.entryWriter.workCompleted();
entryWriter.workCompleted();
} else if (attached.remoteModificationIterator != null)
attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
try {
final int len = attached.entryWriter.writeBufferToSocket(socketChannel,
final int len = entryWriter.writeBufferToSocket(socketChannel,
approxTime);
if (len == -1)
@@ -641,7 +643,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
if (len > 0)
contemplateThrottleWrites(len);
if (!attached.entryWriter.hasBytesToWrite() && !attached.entryWriter.isWorkIncomplete())
if (!entryWriter.hasBytesToWrite() && !entryWriter.isWorkIncomplete())
// TURN OP_WRITE_OFF
key.interestOps(key.interestOps() & ~OP_WRITE);
} catch (IOException e) {
@@ -1179,6 +1181,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
// state is used for both heartbeat and stateless
state = out.readByte();
sizeInBytes = out.readInt();
//System.out.println("received size=" + sizeInBytes);
// System.out.println("......sizeInBytes=>" + sizeInBytes);
assert sizeInBytes > 0;
// if the buffer is too small to read this payload we will have to grow the
// size of the buffer
@@ -1200,6 +1205,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
}
final long nextEntryPos = out.position() + sizeInBytes;
assert nextEntryPos > 0;
final long limit = out.limit();
out.limit(nextEntryPos);
@@ -1522,6 +1528,8 @@ class StatelessServerConnector<K, V> {
writeException(writer, e);
writeSizeAndFlags(sizeLocation + SIZE_OF_TRANSACTION_ID, true, writer);
e.printStackTrace();
return null;
}
@@ -1866,6 +1874,8 @@ class StatelessServerConnector<K, V> {
long pos = out.position();
long limit = out.limit();
// System.out.println("Sending with size=" + size);
if (LOG.isDebugEnabled()) {
out.position(locationOfSize);
out.limit(pos);