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 entrySet works for small chunks

Browse Source
This commit is contained in:
Rob Austin
2014-10-27 17:09:36 +00:00
parent a2c20e0ad6
commit b374a9bd50
4 changed files with 237 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+22 -13
View File
@@ -419,6 +419,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
private void writeEntries(Map<? extends K, ? extends V> map) {
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
@@ -605,20 +606,25 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
long sizeLocation = writeEvent(ENTRY_SET);
// get the data back from the server
Bytes in = blockingFetch(sizeLocation);
final Bytes in = blockingFetch(sizeLocation);
final Set<Map.Entry<K, V>> result = new HashSet<Map.Entry<K, V>>();
long size = in.readStopBit();
LOG.info("size="+size);
Set<Map.Entry<K, V>> result = new HashSet<Map.Entry<K, V>>();
boolean hasMoreEntries;
do {
hasMoreEntries = in.readBoolean();
for (int i = 0; i < size; i++) {
K k = keyValueSerializer.readKey(in);
V v = keyValueSerializer.readValue(in);
LOG.info("k="+k);
// number of entries in the chunk
long size = in.readUnsignedShort();
LOG.info("v="+v);
result.add(new Entry(k, v));
}
LOG.info("size=" + size);
for (int i = 0; i < size; i++) {
K k = keyValueSerializer.readKey(in);
V v = keyValueSerializer.readValue(in);
result.add(new Entry(k, v));
}
} while (hasMoreEntries);
return result;
}
@@ -730,9 +736,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
// the number of bytes in the response
int size = receive(SIZE_OF_SIZE, timeoutTime).readInt();
if (bytes.capacity() < size)
bytes = new ByteBufferBytes(allocateDirect(size));
int requiredSize = size + SIZE_OF_SIZE;
if (bytes.capacity() < requiredSize) {
bytes = new ByteBufferBytes(allocateDirect(requiredSize));
}
// block until we have received all the byte in this chunk
receive(size, timeoutTime);
boolean isException = bytes.readBoolean();
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+58 -7
View File
@@ -548,9 +548,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
if (completed)
attached.entryWriter.workCompleted();
// attached.hasRemoteHeartbeatInterval = false;
// attached.hasRemoteHeartbeatInterval = false;
} else if (attached.remoteModificationIterator != null)
} else if (attached.remoteModificationIterator != null)
attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
try {
@@ -1646,15 +1646,18 @@ class StatelessServerConnector<K, V> {
private Work entrySet(Bytes reader, Bytes writer) {
final long sizeLocation = reflectTransactionId(reader, writer);
final long transactionId = reader.readLong();
final Set<Map.Entry<K, V>> entries;
try {
entries = map.entrySet();
} catch (RuntimeException e) {
final long sizeLocation = reflectTransactionId(reader, writer);
return sendException(writer, sizeLocation, e);
}
writer.writeStopBit(entries.size());
final Iterator<Map.Entry<K, V>> iterator = entries.iterator();
@@ -1664,23 +1667,71 @@ class StatelessServerConnector<K, V> {
@Override
public boolean doWork(Bytes writer) {
final long sizeLocation = writer.position();
writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // SIZE_OF_SIZE + is
// exception
writer.writeLong(transactionId);
// hasAnotherChunk
writer.skip(1);
// count
writer.skip(2);
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (writer.remaining() <= maxEntrySizeBytes)
if (writer.remaining() <= maxEntrySizeBytes) {
writeHeader(writer, sizeLocation, count, true);
return false;
}
count++;
final Map.Entry<K, V> next = iterator.next();
writeKey(next.getKey(), writer);
writeValue(next.getValue(), writer);
}
writeSizeAndFlags(sizeLocation, false, writer);
writeHeader(writer, sizeLocation, count, false);
return true;
}
private void writeHeader(Bytes writer, long sizeLocation, int count, final boolean hasAnotherChunk) {
final long end = writer.position();
final int size = (int) (end - sizeLocation);
writer.position(sizeLocation);
// size in bytes
writer.writeInt(size);
// is exception
writer.writeBoolean(false);
//transaction id;
writer.skip(8);
writer.writeBoolean(hasAnotherChunk);
// count
writer.writeUnsignedShort(count);
writer.position(end);
}
};
}
private void writeMoreData(long location, boolean hasMoreData, Bytes out) {
out.writeBoolean(location, hasMoreData);
}
private void entryCount(long location, int count, Bytes out) {
}
private Work putIfAbsent(Bytes reader, Bytes writer) {
final K key = readKey(reader);
final V v = readValue(reader);
@@ -1726,7 +1777,7 @@ class StatelessServerConnector<K, V> {
private long reflectTransactionId(Bytes reader, Bytes writer) {
final long sizeLocation = writer.position();
writer.skip(1 + AbstractChannelReplicator.SIZE_OF_SIZE); // event size + SIZE_OF_SIZE
writer.skip(AbstractChannelReplicator.SIZE_OF_SIZE + 1); // isException + SIZE_OF_SIZE
final long transactionId = reader.readLong();
writer.writeLong(transactionId);
return sizeLocation;