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

fixed putAll() in wire stateless map

Browse Source
This commit is contained in:
Rob Austin
2015-03-13 13:23:43 +00:00
parent 1a8998d521
commit fcad7b4e80
2 changed files with 94 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+11 -16
View File
@@ -274,18 +274,13 @@ class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clon
final long startTime = System.currentTimeMillis();
long transactionId = hub.nextUniqueTransaction(startTime);
Set<? extends Map.Entry<? extends K, ? extends V>> entries = map.entrySet();
Iterator<? extends Map.Entry<? extends K, ? extends V>> iterator = entries.iterator();
boolean hasNext;
do {
hasNext = iterator.hasNext();
while (iterator.hasNext()) {
hub.outBytesLock().lock();
try {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
@@ -294,20 +289,19 @@ class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clon
assert hub.outBytesLock().isHeldByCurrentThread();
hub.markSize(wire);
hub.startTime(startTime);
wire.write(Fields.TYPE).text("MAP");
wire.write(Fields.TRANSACTION_ID).int64(transactionId);
wire.write(Fields.TIME_STAMP).int64(startTime);
wire.write(Fields.CHANNEL_ID).int16(channelID);
writeField(Fields.METHOD_NAME, "PUT_ALL");
hub.outWire().write(Fields.HAS_NEXT).bool(hasNext);
Map.Entry<? extends K, ? extends V> e = iterator.next();
hub.outWire().write(Fields.HAS_NEXT).bool(iterator.hasNext());
writeField(Fields.ARG_1, e.getKey());
writeField(Fields.ARG_2, e.getValue());
if (hasNext) {
Map.Entry<? extends K, ? extends V> e = iterator.next();
writeField(Fields.ARG_1, e.getKey());
writeField(Fields.ARG_2, e.getValue());
}
hub.writeSocket(hub.outWire());
@@ -315,7 +309,8 @@ class ClientWiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clon
hub.outBytesLock().unlock();
}
} while (hasNext);
}
// wait for the transaction id to be received, this will only be received once the
// last chunk has been processed
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+83 -97
View File
@@ -42,10 +42,7 @@ import java.io.PrintWriter;
import java.io.StreamCorruptedException;
import java.io.StringWriter;
import java.nio.BufferOverflowException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -181,34 +178,46 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
/* private void writeEntryChunked(long transactionId,
@NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator<Map.Entry<byte[], byte[]>>> c) {
final BytesChronicleMap m = bytesMap(channelId);
final Iterator<Map.Entry<byte[], byte[]>> iterator = function.apply(m);
final Map<Long, Runnable> incompleteWork = new HashMap<Long, Runnable>();
// this allows us to write more data than the buffer will allow
for (; ; ) {
private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) {
if (bytes == null || bytes.remaining() == 0)
return new byte[]{};
// each chunk has its own transaction-id
outWire.write(TRANSACTION_ID).int64(transactionId);
if (bytes.remaining() > Integer.MAX_VALUE)
throw new BufferOverflowException();
write(map -> {
byte[] result = new byte[(int) bytes.remaining()];
bytes.write(result);
return result;
}
boolean hasNext = iterator.hasNext();
outWire.write(HAS_NEXT).bool(hasNext);
private byte[] toBytes(WireKey fieldName) {
if (!hasNext)
return;
final Wire wire = inWire;
c.accept(iterator);
final ValueIn read = wire.read(fieldName);
final long l = read.readLength();
});
if (l > Integer.MAX_VALUE)
throw new BufferOverflowException();
final int fieldLength = (int) l;
final long endPos = wire.bytes().position() + fieldLength;
final long limit = wire.bytes().limit();
try {
byte[] bytes = new byte[fieldLength];
wire.bytes().read(bytes);
return bytes;
} finally {
wire.bytes().position(endPos);
wire.bytes().limit(limit);
}
}*/
}
@SuppressWarnings("UnusedReturnValue")
void onEvent() throws StreamCorruptedException {
@@ -220,41 +229,48 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
channelId = inWire.read(CHANNEL_ID).int16();
inWire.read(METHOD_NAME).text(methodName);
if ("PUT_WITHOUT_ACC".contentEquals(methodName)) {
writeVoid(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2);
// todo bytesMap.put(reader, reader, timestamp, identifier());
bytesMap.put(reader, reader);
});
return;
if (!incompleteWork.isEmpty()) {
Runnable runnable = incompleteWork.get(transactionId);
if (runnable != null) {
runnable.run();
return;
}
}
if ("KEY_SET".contentEquals(methodName)) {
writeChunked(transactionId, map -> map.keySet().iterator(), writeElement);
return;
}
if ("VALUES".contentEquals(methodName)) {
writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement);
return;
}
if ("ENTRY_SET".contentEquals(methodName)) {
writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry);
return;
}
// write the transaction id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
try {
if ("PUT_WITHOUT_ACC".contentEquals(methodName)) {
writeVoid(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2);
// todo bytesMap.put(reader, reader, timestamp, identifier());
bytesMap.put(reader, reader);
});
return;
}
if ("KEY_SET".contentEquals(methodName)) {
writeChunked(transactionId, map -> map.keySet().iterator(), writeElement);
return;
}
if ("VALUES".contentEquals(methodName)) {
writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement);
return;
}
if ("ENTRY_SET".contentEquals(methodName)) {
writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry);
return;
}
if ("PUT_ALL".contentEquals(methodName)) {
putAll(transactionId);
return;
}
// write the transaction id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
if ("CREATE_CHANNEL".contentEquals(methodName)) {
writeVoid(() -> {
short channelId1 = inWire.read(ARG_1).int16();
@@ -394,55 +410,19 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
// if (len > 4)
if (EventGroup.IS_DEBUG) {
long len = outWire.bytes().position() - SIZE_OF_SIZE;
System.out.println("--------------------------------------------\nserver wrote:\n\n" + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len));
if (len == 0)
System.out.println("--------------------------------------------\nserver wrote:\n\n<EMPTY>");
else
System.out.println("--------------------------------------------\nserver wrote:\n\n" + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len));
}
}
}
private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) {
if (bytes == null || bytes.remaining() == 0)
return new byte[]{};
if (bytes.remaining() > Integer.MAX_VALUE)
throw new BufferOverflowException();
byte[] result = new byte[(int) bytes.remaining()];
bytes.write(result);
return result;
}
private byte[] toBytes(WireKey fieldName) {
final Wire wire = inWire;
System.out.println(Bytes.toDebugString(inWire.bytes()));
final ValueIn read = wire.read(fieldName);
final long l = read.readLength();
if (l > Integer.MAX_VALUE)
throw new BufferOverflowException();
final int fieldLength = (int) l;
final long endPos = wire.bytes().position() + fieldLength;
final long limit = wire.bytes().limit();
try {
byte[] bytes = new byte[fieldLength];
wire.bytes().read(bytes);
return bytes;
} finally {
wire.bytes().position(endPos);
wire.bytes().limit(limit);
}
}
private void putAll(long transactionId) {
/*final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId);
final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId);
if (bytesMap == null)
return;
@@ -455,6 +435,9 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
return;
}
// Note : you can not assume that all the entries in a putAll will be continuous, they maybe other transactions from other threads.
// we it should be possible for a single entry to fill the Tcp buffer, so each entry should have the ability to be processed separately
// and then only applied to the map once all the entries are received.
runnable = new Runnable() {
// we should try and collect the data and then apply it atomically as quickly possible
@@ -462,18 +445,21 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
@Override
public void run() {
if (inWire.read(HAS_NEXT).bool()) {
collectData.put(toBytes(ARG_1), toBytes(ARG_2));
} else {
boolean hasNext = inWire.read(HAS_NEXT).bool();
collectData.put(toBytes(ARG_1), toBytes(ARG_2));
if (!hasNext) {
// the old code assumed that all the data would fit into a single buffer
// this assumption is invalid
if (!collectData.isEmpty()) {
bytesMap.delegate.putAll((Map) collectData);
incompleteWork.remove(transactionId);
outWire.write(TRANSACTION_ID).int64(transactionId);
// todo handle the case where there is an exception
outWire.write(IS_EXCEPTION).bool(false);
writeVoid(() -> {
bytesMap.delegate.putAll((Map) collectData);
return null;
});
}
}
@@ -481,7 +467,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
};
incompleteWork.put(transactionId, runnable);
runnable.run();*/
runnable.run();
}
@@ -715,10 +701,10 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
bytesMap.output = null;
outWire.bytes().mark();
outWire.write(IS_EXCEPTION).bool(false);
try {
r.call();
outWire.write(IS_EXCEPTION).bool(false);
} catch (Exception e) {
outWire.bytes().reset();
// the idea of wire is that is platform independent,