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-167 added a working version of StatelessChronicleMapTest but requires refactoring,

Browse Source
This commit is contained in:
Rob Austin
2014-10-21 17:21:36 +01:00
parent 25911b969e
commit c5d56752a2
4 changed files with 232 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/common/StatelessBuilder.java
+1 -1
View File
@@ -8,7 +8,7 @@ import java.util.concurrent.TimeUnit;
*/
public class StatelessBuilder {
private long timeoutMs = TimeUnit.SECONDS.toMillis(5);
private long timeoutMs = TimeUnit.SECONDS.toMillis(1);
private InetSocketAddress remoteAddress;
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+66 -10
View File
@@ -28,8 +28,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMap.class);
private final SocketChannel clientChannel;
public static final ByteBuffer STATELESS_CLIENT_IDENTIFER = ByteBuffer.wrap(new byte[]{-127});
private final CloseablesManager closeables = new CloseablesManager();
public static final byte STATELESS_CLIENT_IDENTIFER = (byte) -127;
private CloseablesManager closeables;
private final StatelessBuilder builder;
static enum EventId {
@@ -48,7 +48,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
REPLACE,
REPLACE_WITH_OLD_AND_NEW_VALUE,
PUT_IF_ABSENT,
REMOVE_WITH_VALUE
REMOVE_WITH_VALUE,
TO_STRING
}
private final AtomicLong transactionID = new AtomicLong();
@@ -65,8 +66,16 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
IOException exception = null;
SocketChannel clientChannel0 = null;
for (int i = 1; i < 5; i++) {
long timeoutAt = System.currentTimeMillis() + builder.timeoutMs();
for (; ; ) {
if (System.currentTimeMillis() > timeoutAt) {
clientChannel0.close();
close();
throw new RuntimeException("unable to connect to remote server");
}
closeables = new CloseablesManager();
clientChannel0 = AbstractChannelReplicator.openSocketChannel(closeables);
try {
@@ -75,8 +84,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
exception = null;
break;
} catch (IOException e) {
closeables.closeQuietly();
Thread.yield();
clientChannel0.close();
close();
}
}
@@ -84,6 +95,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
if (exception != null)
throw exception;
if (clientChannel0 == null)
throw new IOException("unable to connect to remote server");
return clientChannel0;
}
@@ -91,11 +104,22 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
private void doHandShaking(final SocketChannel clientChannel) throws IOException {
while (STATELESS_CLIENT_IDENTIFER.remaining() > 0) {
clientChannel.write(STATELESS_CLIENT_IDENTIFER);
out.clear();
out.writeByte(STATELESS_CLIENT_IDENTIFER);
ByteBuffer buffer = out.buffer().slice();
buffer.position(0);
buffer.limit(1);
while (buffer.hasRemaining()) {
clientChannel.write(buffer);
}
in.buffer().clear();
in.clear();
out.buffer().clear();
out.clear();
while (in.buffer().position() <= 0) {
clientChannel.read(in.buffer());
@@ -115,6 +139,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
public void close() {
closeables.closeQuietly();
clearBuffers();
}
@@ -191,6 +216,15 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
}
public synchronized String toString() {
long sizeLocation = writeEvent(TO_STRING);
// get the data back from the server
return blockingFetch(sizeLocation).readObject(String.class);
}
public synchronized boolean isEmpty() {
long sizeLocation = writeEvent(IS_EMPTY);
@@ -258,6 +292,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
}
private void compact() {
in.buffer().compact();
}
public synchronized V remove(Object key) {
long sizeLocation = writeEvent(REMOVE);
@@ -422,18 +461,28 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
try {
return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs());
} catch (IOException e) {
closeables.closeQuietly();
close();
throw new IORuntimeException(e);
} catch (Exception e) {
closeables.closeQuietly();
close();
throw e;
}
}
private void clearBuffers() {
out.clear();
out.buffer().clear();
in.clear();
in.buffer().clear();
}
private Bytes blockingFetchThrowable(long sizeLocation, long timeOutMs) throws IOException {
// free up space in the buffer
// compact();
long startTime = System.currentTimeMillis();
long transactionId = nextUniqueTransaction(startTime);
long timeoutTime = startTime + timeOutMs;
@@ -453,18 +502,25 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clos
in.clear();
in.buffer().clear();
/* in.buffer().limit(in.buffer().capacity());
in.limit(in.buffer().position());
*/
// read size
while (in.buffer().position() < 2) {
clientChannel.read(in.buffer());
checkTimeout(timeoutTime);
}
in.limit(in.buffer().position());
int size = in.readUnsignedShort();
LOG.info("size=" + size);
while (in.buffer().position() < size) {
clientChannel.read(in.buffer());
checkTimeout(timeoutTime);
}
in.limit(in.buffer().position());
boolean isException = in.readBoolean();
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+138 -36
View File
@@ -60,7 +60,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
private final int maxEntrySizeBytes;
private final Replica.EntryExternalizable externalizable;
private final TcpReplicationConfig replicationConfig;
private final StatelessServerConnector statelessServerConnector;
private long selectorTimeout;
@@ -201,6 +200,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
* @param approxTime the approximate time in milliseconds
*/
void heartBeatMonitor(long approxTime) {
for (int i = activeKeys.nextSetBit(0); i >= 0; i = activeKeys.nextSetBit(i + 1)) {
try {
final SelectionKey key = selectionKeysStore[i];
@@ -208,6 +208,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
activeKeys.clear(i);
continue;
}
final Attached attachment = (Attached) key.attachment();
if (!attachment.hasRemoteHeartbeatInterval)
continue;
try {
sendHeartbeatIfRequired(approxTime, key);
} catch (Exception e) {
@@ -442,6 +447,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
if (remoteIdentifier == STATELESS_CLIENT) {
attached.handShakingComplete = true;
attached.hasRemoteHeartbeatInterval = false;
return;
}
@@ -530,13 +536,14 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
final SocketChannel socketChannel = (SocketChannel) key.channel();
final Attached attached = (Attached) key.attachment();
if (attached.entryWriter.hasIncompletWork()) {
if (attached.entryWriter.isWorkIncomplete()) {
boolean completed = attached.entryWriter.doWork();
if (completed)
attached.entryWriter.workCompleted();
attached.hasRemoteHeartbeatInterval = false;
return;
}
@@ -567,7 +574,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
final SocketChannel socketChannel = (SocketChannel) key.channel();
final Attached attached = (Attached) key.attachment();
if (attached.entryWriter.hasIncompletWork())
if (attached.entryWriter.isWorkIncomplete())
return;
try {
@@ -820,8 +827,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
remoteHeartbeatInterval = heartBeatIntervalMillis;
hasRemoteHeartbeatInterval = false;
remoteModificationIterator = null;
}
/**
@@ -856,7 +861,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
entryCallback = new EntryCallback(externalizable, in);
}
public boolean hasIncompletWork() {
public boolean isWorkIncomplete() {
return uncompletedWork != null;
}
@@ -1068,14 +1073,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
// state is used for both heartbeat and stateless
state = out.readByte();
sizeInBytes = out.readUnsignedShort();
if (state == 0 && sizeInBytes != 0) {
int i = 1;
}
// this is the :
// -- heartbeat if its 0
// -- stateful update if its 1
@@ -1211,6 +1210,8 @@ class StatelessServerConnector<K, V> {
final StatelessChronicleMap.EventId event = StatelessChronicleMap.EventId.values()[eventId];
System.out.println("got remote event to StatelessChronicleMap");
switch (event) {
case LONG_SIZE:
@@ -1261,6 +1262,9 @@ class StatelessServerConnector<K, V> {
case SIZE:
return size(in, out);
case TO_STRING:
return toString(in, out);
default:
throw new IllegalStateException("unsupported event=" + event);
@@ -1268,8 +1272,18 @@ class StatelessServerConnector<K, V> {
}
private Work removeWithValue(Bytes in, Bytes out) {
boolean result = map.remove(readKey(in), readValue(in));
K key = readKey(in);
V readValue = readValue(in);
long sizeLocation = reflectTransactionId(in, out);
boolean result;
try {
result = map.remove(key, readValue);
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
out.writeBoolean(result);
return null;
}
@@ -1279,15 +1293,14 @@ class StatelessServerConnector<K, V> {
final K key = readKey(in);
V oldValue = readValue(in);
V newValue = readValue(in);
boolean replaced;
long sizeLocation = reflectTransactionId(in, out);
try {
map.replace(key, oldValue, newValue);
replaced = map.replace(key, oldValue, newValue);
} catch (RuntimeException e) {
writeException(e, out);
writeSizeAndFlags(sizeLocation, true, out);
return null;
return sendException(out, sizeLocation, e);
}
out.writeBoolean(replaced);
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
@@ -1295,21 +1308,53 @@ class StatelessServerConnector<K, V> {
private Work longSize(Bytes in, Bytes out) {
long sizeLocation = reflectTransactionId(in, out);
out.writeLong(map.longSize());
try {
out.writeLong(map.longSize());
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
private Work size(Bytes in, Bytes out) {
long sizeLocation = reflectTransactionId(in, out);
out.writeInt(map.size());
try {
out.writeInt(map.size());
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
private Work toString(Bytes in, Bytes out) {
long sizeLocation = reflectTransactionId(in, out);
try {
out.writeObject(map.toString());
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
private Work sendException(Bytes out, long sizeLocation, RuntimeException e) {
writeException(e, out);
writeSizeAndFlags(sizeLocation, true, out);
return null;
}
private Work isEmpty(Bytes in, Bytes out) {
long sizeLocation = reflectTransactionId(in, out);
out.writeBoolean(map.isEmpty());
try {
out.writeBoolean(map.isEmpty());
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
@@ -1317,7 +1362,11 @@ class StatelessServerConnector<K, V> {
private Work containsKey(Bytes in, Bytes out) {
K k = readKey(in);
long sizeLocation = reflectTransactionId(in, out);
out.writeBoolean(map.containsKey(k));
try {
out.writeBoolean(map.containsKey(k));
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
@@ -1325,7 +1374,11 @@ class StatelessServerConnector<K, V> {
private Work containsValue(Bytes in, Bytes out) {
V v = readValue(in);
long sizeLocation = reflectTransactionId(in, out);
out.writeBoolean(map.containsValue(v));
try {
out.writeBoolean(map.containsValue(v));
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
@@ -1336,9 +1389,7 @@ class StatelessServerConnector<K, V> {
try {
writeValue(map.get(k), out);
} catch (RuntimeException e) {
writeException(e, out);
writeSizeAndFlags(sizeLocation, true, out);
return null;
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
@@ -1349,30 +1400,54 @@ class StatelessServerConnector<K, V> {
K k = readKey(in);
V v = readValue(in);
long sizeLocation = reflectTransactionId(in, out);
writeValue(map.put(k, v), out);
try {
writeValue(map.put(k, v), out);
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
private Work remove(Bytes in, Bytes out) {
final V value = map.remove(readKey(in));
K key = readKey(in);
long sizeLocation = reflectTransactionId(in, out);
writeValue(value, out);
try {
writeValue(map.remove(key), out);
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
private Work putAll(Bytes in, Bytes out) {
map.putAll(readEntries(in));
Map<K, V> m = readEntries(in);
long sizeLocation = reflectTransactionId(in, out);
try {
map.putAll(m);
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
private Work clear(Bytes in, Bytes out) {
map.clear();
long sizeLocation = reflectTransactionId(in, out);
try {
map.clear();
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
@@ -1409,7 +1484,14 @@ class StatelessServerConnector<K, V> {
private Work values(Bytes in, Bytes out) {
long sizeLocation = reflectTransactionId(in, out);
final Collection<V> values = map.values();
final Collection<V> values;
try {
values = map.values();
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
out.writeStopBit(values.size());
for (final V value : values) {
writeValue(value, out);
@@ -1421,7 +1503,13 @@ class StatelessServerConnector<K, V> {
private Work entrySet(Bytes in, Bytes out) {
long sizeLocation = reflectTransactionId(in, out);
final Set<Map.Entry<K, V>> entries = map.entrySet();
final Set<Map.Entry<K, V>> entries;
try {
entries = map.entrySet();
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
out.writeStopBit(entries.size());
for (Map.Entry<K, V> e : entries) {
writeKey(e.getKey(), out);
@@ -1434,7 +1522,15 @@ class StatelessServerConnector<K, V> {
K key = readKey(in);
V v = readValue(in);
long sizeLocation = reflectTransactionId(in, out);
writeValue(map.putIfAbsent(key, v), out);
V value;
try {
value = map.putIfAbsent(key, v);
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeValue(value, out);
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
@@ -1443,9 +1539,14 @@ class StatelessServerConnector<K, V> {
private Work replace(Bytes in, Bytes out) {
K k = readKey(in);
V v = readValue(in);
V replaced;
long sizeLocation = reflectTransactionId(in, out);
map.replace(k, v);
try {
replaced = map.replace(k, v);
} catch (RuntimeException e) {
return sendException(out, sizeLocation, e);
}
writeValue(replaced, out);
writeSizeAndFlags(sizeLocation, false, out);
return null;
}
@@ -1455,6 +1556,7 @@ class StatelessServerConnector<K, V> {
*
* @param key the key of the map
*/
private void writeKey(K key, Bytes out) {
keyValueSerializer.writeKey(key, out);
}