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

added support for putAll

Browse Source
This commit is contained in:
Rob Austin
2015-03-09 18:49:49 +00:00
parent 3c7ef12798
commit ff2b20fc6d
5 changed files with 254 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/BytesChronicleMap.java
+1 -1
View File
@@ -26,7 +26,7 @@ import java.util.Set;
public class BytesChronicleMap implements AbstractChronicleMap<Bytes, Bytes> {
private final VanillaChronicleMap<?, ?, ?, ?, ?, ?> delegate;
final VanillaChronicleMap<?, ?, ?, ?, ?, ?> delegate;
MapIOBuffer output;
public BytesChronicleMap(VanillaChronicleMap<?, ?, ?, ?, ?, ?> delegate) {
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+105 -51
View File
@@ -44,10 +44,7 @@ import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -171,11 +168,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
if (len == -1 || len == 0)
return len;
// process the messages for remaining chunks,
if (in != null) {
in.run();
return len;
}
while (inWire.bytes().remaining() > 4) {
@@ -263,6 +255,40 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
Map<Long, Runnable> incompleteWork = new HashMap<>();
private void writeChunked(long transactionId,
@NotNull final Function<Map, Iterator<byte[]>> function,
@NotNull final Consumer<Iterator<byte[]>> c) {
final BytesChronicleMap m = bytesMap(channelId);
final Iterator<byte[]> iterator = function.apply(m);
// this allows us to write more data than the buffer will allow
out = () -> {
// each chunk has its own transaction-id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
write(map -> {
boolean hasNext = iterator.hasNext();
outWire.write(() -> "HAS_NEXT").bool(hasNext);
if (hasNext)
c.accept(iterator);
else
// setting out to NULL denotes that there are no more chunks
out = null;
});
};
out.run();
}
@SuppressWarnings("UnusedReturnValue")
void onEvent() {
@@ -284,7 +310,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
long markStart = outWire.bytes().position();
outWire.bytes().skip(4);
try {
@@ -303,6 +328,12 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
return;
}
if ("PUT_ALL".contentEquals(methodName)) {
putAll(transactionId);
return;
}
// write the transaction id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
@@ -410,11 +441,6 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
return;
}
if ("PUT_ALL".contentEquals(methodName)) {
putAll();
return;
}
throw new IllegalStateException("unsupported event=" + methodName);
@@ -424,69 +450,97 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
} finally {
int len = (int) (outWire.bytes().position() - markStart);
outWire.bytes().writeInt(markStart, len);
if (len == 4)
// remove the 4 byte len as nothing was written
outWire.bytes().position(markStart);
else
outWire.bytes().writeInt(markStart, len);
if (len > 4)
System.out.println("\n----------------------------\n\nserver wrote:\n" + Bytes.toDebugString(outWire.bytes(), markStart + 4, len - 4));
}
}
private byte[] toBytes(String fieldName) {
private void writeChunked(long transactionId,
@NotNull final Function<Map, Iterator<byte[]>> function,
@NotNull final Consumer<Iterator<byte[]>> c) {
Wire wire = inWire;
ValueIn read = wire.read(() -> fieldName);
final BytesChronicleMap m = bytesMap(channelId);
final Iterator<byte[]> iterator = function.apply(m);
long l = read.readLength();
if (l > Integer.MAX_VALUE)
throw new BufferOverflowException();
// this allows us to write more data than the buffer will allow
out = () -> {
int fieldLength = (int) l;
// each chunk has its own transaction-id
outWire.write(() -> "TRANSACTION_ID").int64(transactionId);
long endPos = wire.bytes().position() + fieldLength;
long limit = wire.bytes().limit();
write(map -> {
try {
byte[] bytes = new byte[fieldLength];
boolean hasNext = iterator.hasNext();
outWire.write(() -> "HAS_NEXT").bool(hasNext);
wire.bytes().read(bytes);
return bytes;
if (hasNext)
c.accept(iterator);
else
// setting out to NULL denotes that there are no more chunks
out = null;
});
};
out.run();
} finally
{
wire.bytes().position(endPos);
wire.bytes().limit(limit);
}
}
private void putAll() {
private void putAll(long transactionId) {
final BytesChronicleMap bytesMap = bytesMap(StatelessWiredConnector.this.channelId);
if (bytesMap == null)
return;
out = () -> {
if (!inWire.read(() -> "HAS_MORE_CHUNKS").bool())
// finished reading all the chunks
in = null;
Runnable runnable = incompleteWork.get(transactionId);
bytesMap.output = outMessageAdapter.outBuffer;
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
try {
bytesMap.put(reader, reader, timestamp, identifier());
} catch (Exception e) {
// todo send back exceptions
if (runnable != null) {
runnable.run();
return;
}
runnable = new Runnable() {
// we should try and collect the data and then apply it atomically as quickly possible
final Map<byte[], byte[]> collectData = new HashMap<>();
@Override
public void run() {
if (inWire.read(() -> "HAS_NEXT").bool()) {
collectData.put(toBytes("ARG_1"), toBytes("ARG_2"));
} else {
// 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);
nofityDataWritten();
}
}
}
};
incompleteWork.put(transactionId, runnable);
runnable.run();
}
private byte identifier() {
// if the client provides a remote identifier then we will use that otherwise we will use the local identifier
return remoteIdentifier == 0 ? localIdentifer : remoteIdentifier;
src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java
+92 -31
View File
@@ -42,6 +42,17 @@ import static net.openhft.chronicle.map.WiredStatelessChronicleMap.EventId.*;
public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Cloneable {
private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class);
// todo replace with an ENUM
public static final String METHOD_NAME = "METHOD_NAME";
public static final String RESULT_IS_NULL = "RESULT_IS_NULL";
public static final String HAS_NEXT = "HAS_NEXT";
public static final String RESULT = "RESULT";
public static final String ARG_1 = "ARG_1";
public static final String ARG_2 = "ARG_2";
private final WiredStatelessClientTcpConnectionHub hub;
protected Class<K> kClass;
@@ -258,6 +269,60 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key);
}
public void putAll(@NotNull Map<? extends K, ? extends V> map) {
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();
hub.outBytesLock().lock();
try {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
Wire wire = hub.outWire();
assert hub.outBytesLock().isHeldByCurrentThread();
hub.markSize(wire);
hub.startTime(startTime);
wire.write(() -> "TRANSACTION_ID").int64(transactionId);
wire.write(() -> "TIME_STAMP").int64(startTime);
wire.write(() -> "CHANNEL_ID").int16(channelID);
writeField(METHOD_NAME, "PUT_ALL");
hub.outWire().write(() -> "HAS_NEXT").bool(hasNext);
if (hasNext) {
Map.Entry<? extends K, ? extends V> e = iterator.next();
writeField(ARG_1, e.getKey());
writeField(ARG_2, e.getValue());
}
hub.writeSocket(hub.outWire());
} finally {
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
readVoid(transactionId, startTime);
}
public V put(K key, V value) {
if (key == null || value == null)
throw new NullPointerException();
@@ -296,8 +361,8 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> "HAS_NEXT").bool()) {
K k = readK("RESULT", wireIn, null);
if (wireIn.read(() -> HAS_NEXT).bool()) {
K k = readK(RESULT, wireIn, null);
result.add(k);
} else
break;
@@ -329,8 +394,8 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> "HAS_NEXT").bool()) {
V v = readV("RESULT", wireIn, null);
if (wireIn.read(() -> HAS_NEXT).bool()) {
V v = readV(RESULT, wireIn, null);
result.add(v);
} else
break;
@@ -359,7 +424,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> "HAS_NEXT").bool()) {
if (wireIn.read(() -> HAS_NEXT).bool()) {
result.put(
readK("RESULT_KEY", wireIn, null),
readV("RESULT_VALUE", wireIn, null));
@@ -380,10 +445,6 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
throw new UnsupportedOperationException();
}
public void putAll(@NotNull Map<? extends K, ? extends V> map) {
throw new UnsupportedOperationException();
}
private long readLong(long transactionId, long startTime) {
assert !hub.outBytesLock().isHeldByCurrentThread();
@@ -392,7 +453,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
// receive
hub.inBytesLock().lock();
try {
return hub.proxyReply(timeoutTime, transactionId).read(() -> "RESULT").int64();
return hub.proxyReply(timeoutTime, transactionId).read(() -> RESULT).int64();
} finally {
hub.inBytesLock().unlock();
}
@@ -427,7 +488,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
if (wireIn.read(() -> "RESULT_IS_NULL").bool())
if (wireIn.read(() -> RESULT_IS_NULL).bool())
return null;
return readV(argName, wireIn, usingValue);
@@ -500,7 +561,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
return wireIn.read(() -> "RESULT").bool();
return wireIn.read(() -> RESULT).bool();
} finally {
hub.inBytesLock().unlock();
}
@@ -515,7 +576,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, transactionId);
return wireIn.read(() -> "RESULT").int32();
return wireIn.read(() -> RESULT).int32();
} finally {
hub.inBytesLock().unlock();
}
@@ -530,9 +591,9 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField("ARG_1", key);
writeField("ARG_2", value);
writeField(METHOD_NAME, methodName);
writeField(ARG_1, key);
writeField(ARG_2, value);
hub.writeSocket(hub.outWire());
} finally {
@@ -551,9 +612,9 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField("ARG_1", value1);
writeField("ARG_2", value2);
writeField(METHOD_NAME, methodName);
writeField(ARG_1, value1);
writeField(ARG_2, value2);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -585,7 +646,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField(METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -603,7 +664,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField(METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -622,7 +683,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField(METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -641,7 +702,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField(METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -665,9 +726,9 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
assert !hub.inBytesLock().isHeldByCurrentThread();
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField("ARG_1", key);
writeField("ARG_2", value);
writeField(METHOD_NAME, methodName);
writeField(ARG_1, key);
writeField(ARG_2, value);
hub.writeSocket(hub.outWire());
} finally {
@@ -678,7 +739,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
return null;
if (resultType == vClass)
return (R) readValue("RESULT", transactionId, startTime, null);
return (R) readValue(RESULT, transactionId, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + resultType + " is not " +
@@ -697,7 +758,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
return null;
if (rClass == vClass)
return (R) readValue("RESULT", transactionId, startTime, null);
return (R) readValue(RESULT, transactionId, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + rClass + " is not " +
"supported");
@@ -708,8 +769,8 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField("ARG_1", arg1);
writeField(METHOD_NAME, methodName);
writeField(ARG_1, arg1);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
@@ -723,7 +784,7 @@ public class WiredStatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Clo
hub.outBytesLock().lock();
try {
transactionId = hub.writeHeader(startTime, channelID, hub.outWire());
writeField("METHOD_NAME", methodName);
writeField(METHOD_NAME, methodName);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
src/main/java/net/openhft/chronicle/map/WiredStatelessClientTcpConnectionHub.java
+3 -2
View File
@@ -381,7 +381,8 @@ public class WiredStatelessClientTcpConnectionHub {
intWire.bytes().skip(4);
intWire.bytes().limit(messageSize);
System.out.println("\nread:\n" + Bytes.toDebugString(intWire.bytes()));
System.out.println("\n--------------------------------\n" +
"\nclient read:\n" + Bytes.toDebugString(intWire.bytes()));
long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64();
@@ -519,7 +520,7 @@ public class WiredStatelessClientTcpConnectionHub {
outBuffer.position(4);
System.out.println("\n--------------------------------------------\n" +
"write:\n\n" + Bytes.toDebugString(outBuffer));
"client wrote:\n\n" + Bytes.toDebugString(outBuffer));
outBuffer.position(0);
src/test/java/net/openhft/chronicle/wire/TestWire.java
+53 -5
View File
@@ -24,15 +24,11 @@ import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.WiredChronicleMapStatelessClientBuilder;
import net.openhft.chronicle.map.WiredStatelessChronicleMap;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.*;
import static java.util.concurrent.TimeUnit.SECONDS;
import static net.openhft.chronicle.map.ChronicleMapBuilder.of;
@@ -319,6 +315,58 @@ public class TestWire {
}
@Test
public void testPutAll() throws IOException, InterruptedException {
{
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.of(++SERVER_PORT)
.heartBeatInterval(1, SECONDS);
hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig)
.createWithId((byte) 1);
// this is how you add maps after the custer is created
map1a = of(byte[].class, byte[].class)
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
byte identifier = (byte) 2;
final WiredChronicleMapStatelessClientBuilder<String, String> builder =
new WiredChronicleMapStatelessClientBuilder<>(
new InetSocketAddress("localhost", SERVER_PORT),
String.class,
String.class,
(short) 1);
builder.identifier(identifier);
final ChronicleMap<String, String> statelessMap = builder.create();
Map<String, String> map = new HashMap<>();
map.put("Key1", "Value1");
map.put("Key2", "Value2");
statelessMap.putAll(map);
Set<String> keys = statelessMap.keySet();
assertEquals(2, keys.size());
assertTrue(keys.contains("Key2"));
assertTrue(keys.contains("Key1"));
Collection<String> values = statelessMap.values();
assertEquals(2, values.size());
assertTrue(values.contains("Value1"));
assertTrue(values.contains("Value2"));
map1a.close();
}
}
public static class Details implements Marshallable {
Class keyClass;