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

Refactor entry writing to replication buffer, #62

Browse Source
This commit is contained in:
leventov
2015-12-17 13:46:26 +03:00
parent 12c8503434
commit e78e070558
6 changed files with 179 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+2 -3
View File
@@ -31,7 +31,6 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.*;
import java.security.AccessController;
import java.security.PrivilegedAction;
@@ -465,7 +464,7 @@ abstract class AbstractChannelReplicator implements Closeable {
@Override
public boolean onEntry(
@NotNull final Bytes entry, final int chronicleId, final long bootstrapTime) {
ReplicableEntry entry, Bytes payload, int chronicleId, long bootstrapTime) {
long pos0 = entryIn.writePosition();
// used to denote that this is not a heartbeat
@@ -478,7 +477,7 @@ abstract class AbstractChannelReplicator implements Closeable {
long start = entryIn.writePosition();
externalizable.writeExternalEntry(entry, entryIn, chronicleId, bootstrapTime);
externalizable.writeExternalEntry(entry, payload, entryIn, chronicleId, bootstrapTime);
if (entryIn.writePosition() == start) {
entryIn.writePosition(pos0);
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+13 -11
View File
@@ -121,14 +121,14 @@ final class ChannelProvider implements Closeable {
* @param chronicleChannel used in cluster into identify the canonical map or queue
*/
@Override
public void writeExternalEntry(@NotNull Bytes entry,
public void writeExternalEntry(ReplicableEntry entry, Bytes payload,
@NotNull Bytes destination,
int chronicleChannel, long bootstrapTime) {
channelDataReadLock();
try {
destination.writeStopBit(chronicleChannel);
channelEntryExternalizables[chronicleChannel]
.writeExternalEntry(entry, destination, chronicleChannel, bootstrapTime);
channelEntryExternalizables[chronicleChannel].writeExternalEntry(
entry, payload, destination, chronicleChannel, bootstrapTime);
} finally {
channelDataLock.readLock().unlock();
}
@@ -481,7 +481,7 @@ final class ChannelProvider implements Closeable {
@Override
public ModificationIterator acquireModificationIterator(
final byte remoteIdentifier) {
final ModificationIterator result = systemModificationIterator.get(remoteIdentifier);
ModificationIterator result = systemModificationIterator.get(remoteIdentifier);
if (result != null)
return result;
@@ -498,10 +498,10 @@ final class ChannelProvider implements Closeable {
@Override
public boolean nextEntry(@NotNull EntryCallback callback, int na) {
final Bytes bytes = payloads.poll();
if (bytes == null)
final Bytes payload = payloads.poll();
if (payload == null)
return false;
callback.onEntry(bytes, 0, System.currentTimeMillis());
callback.onEntry(null, payload, 0, 0);
return true;
}
@@ -511,7 +511,8 @@ final class ChannelProvider implements Closeable {
}
@Override
public void setModificationNotifier(@NotNull ModificationNotifier modificationNotifier) {
public void setModificationNotifier(
@NotNull ModificationNotifier modificationNotifier) {
modificationNotifier0 = modificationNotifier;
}
@@ -559,9 +560,10 @@ final class ChannelProvider implements Closeable {
}
@Override
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
int na, long bootstrapTime) {
destination.write(entry, entry.readPosition(), entry.readRemaining());
public void writeExternalEntry(
ReplicableEntry entry, @NotNull Bytes payload, @NotNull Bytes destination,
int na, long bootstrapTime) {
destination.write(payload, payload.readPosition(), payload.readRemaining());
}
@Override
src/main/java/net/openhft/chronicle/map/Replica.java
+4 -4
View File
@@ -156,7 +156,7 @@ public interface Replica extends Closeable {
* match the maps local
* @param chronicleId is the channel id used to identify the canonical map or queue
*/
void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
void writeExternalEntry(ReplicableEntry entry, Bytes payload, @NotNull Bytes destination,
int chronicleId, long bootstrapTime);
/**
@@ -189,17 +189,17 @@ public interface Replica extends Closeable {
* in the ModificationIterator the entry may have been updated.
*/
public abstract boolean onEntry(
final Bytes entry, final int chronicleId, long bootstrapTime);
ReplicableEntry entry, Bytes payload, int chronicleId, long bootstrapTime);
/**
* Called just after {@link #onEntry(Bytes, int, long)}.
* Called just after {@link #onEntry(ReplicableEntry, Bytes, int, long)}.
* No-op by default.
*/
public void onAfterEntry() {
}
/**
* Called just before {@link #onEntry(Bytes, int, long)}.
* Called just before {@link #onEntry(ReplicableEntry, Bytes, int, long)}.
* No-op by default.
*/
public void onBeforeEntry() {
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+30 -35
View File
@@ -19,6 +19,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.algo.bitset.*;
import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.core.Maths;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.VanillaGlobalMutableState;
import net.openhft.chronicle.hash.impl.TierCountersArea;
import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
@@ -29,6 +30,7 @@ import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.chronicle.map.impl.CompiledReplicatedMapIterationContext;
import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.map.replication.MapRemoteQueryContext;
import net.openhft.chronicle.values.Values;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
@@ -401,62 +403,58 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
*/
@Override
public void writeExternalEntry(
@NotNull Bytes entry, @NotNull Bytes destination, int chronicleId, long bootstrapTime) {
ReplicableEntry entry, Bytes payload, @NotNull Bytes destination,
int chronicleId, long bootstrapTime) {
final long keySize = keySizeMarshaller.readSize(entry);
final long keyPosition = entry.readPosition();
entry.readSkip(keySize);
final long timeStamp = entry.readLong();
final byte identifier = entry.readByte();
final boolean isDeleted = entry.readBoolean();
long valueSize;
if (!isDeleted) {
valueSize = valueSizeMarshaller.readSize(entry);
Data key;
Data value;
boolean isDeleted;
if (entry instanceof MapEntry) {
isDeleted = false;
MapEntry mapEntry = (MapEntry) entry;
key = mapEntry.key();
value = mapEntry.value();
} else {
valueSize = Math.max(0, valueSizeMarshaller.minStorableSize());
isDeleted = true;
MapAbsentEntry mapAbsentEntry = (MapAbsentEntry) entry;
key = mapAbsentEntry.absentKey();
value = ((MapRemoteQueryContext) mapAbsentEntry.context()).dummyZeroValue();
}
final long valuePosition = entry.readPosition();
destination.writeLong(bootstrapTime);
keySizeMarshaller.writeSize(destination, keySize);
valueSizeMarshaller.writeSize(destination, valueSize);
destination.writeStopBit(timeStamp);
keySizeMarshaller.writeSize(destination, key.size());
valueSizeMarshaller.writeSize(destination, value.size());
destination.writeStopBit(entry.originTimestamp());
if (identifier == 0)
if (entry.originIdentifier() == 0)
throw new IllegalStateException("Identifier can't be 0");
destination.writeByte(identifier);
destination.writeByte(entry.originIdentifier());
destination.writeBoolean(isDeleted);
// write the key
destination.write(entry, keyPosition, keySize);
key.writeTo(destination, destination.writePosition());
destination.writeSkip(key.size());
boolean debugEnabled = LOG.isDebugEnabled();
String message = null;
if (debugEnabled) {
if (isDeleted) {
LOG.debug("WRITING ENTRY TO DEST - into local-id={}, remove(key={})",
identifier(), entry.toString().trim());
identifier(), key);
} else {
message = String.format(
"WRITING ENTRY TO DEST - into local-id=%d, put(key=%s,",
identifier(), entry.toString().trim());
identifier(), key);
}
}
if (isDeleted)
return;
// skipping the alignment, as alignment wont work when we send the data over the wire.
long valueAddr = entry.address(valuePosition);
long skip = alignAddr(valueAddr, alignment) - valueAddr;
// writes the value
destination.write(entry, valuePosition + skip, valueSize);
value.writeTo(destination, destination.writePosition());
destination.writeSkip(value.size());
if (debugEnabled) {
LOG.debug(message + "value=" + entry.toString().trim() + ")");
LOG.debug(message + "value=" + value + ")");
}
}
@@ -750,14 +748,11 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
final long segmentPos = position & posMask;
context.readExistingEntry(segmentPos);
// it may not be successful if the buffer can not be re-sized so we will
// process it later, by NOT clearing the changes.clear(position)
context.segmentBytes()
.readLimit(context.valueOffset() + context.valueSize());
context.segmentBytes().readPosition(context.keySizeOffset());
boolean success = entryCallback.onEntry(
context.segmentBytes(), chronicleId, bootStrapTimeStamp());
(ReplicableEntry) context.entryForIteration(), null,
chronicleId, bootStrapTimeStamp());
entryCallback.onAfterEntry();
if (success)
src/test/java/net/openhft/chronicle/map/Issue62ChronicleClient.java
+58
View File
@@ -0,0 +1,58 @@
/*
* Copyright (C) 2015 higherfrequencytrading.com
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 3 of the License.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import static net.openhft.chronicle.map.Issue62ChronicleServer.MAP_FILE_B;
import static net.openhft.chronicle.map.Issue62ChronicleServer.prepare;
public class Issue62ChronicleClient {
private final static Logger LOGGER = LoggerFactory.getLogger(Issue62ChronicleClient.class);
public static void main(String[] args) throws Exception {
prepare(MAP_FILE_B);
ChronicleMapBuilder<String, Long> cityPostalCodesMapBuilder =
ChronicleMapBuilder.of(String.class, Long.class)
//.averageKeySize(100)
.averageKey(Issue62ChronicleServer.STR)
.replication((byte) 2, TcpTransportAndNetworkConfig
.of(8077, new InetSocketAddress("localhost", 8076))
.heartBeatInterval(1, TimeUnit.SECONDS))
.entries(50_000);
try (ChronicleMap<String, Long> map =
cityPostalCodesMapBuilder.createPersistedTo(MAP_FILE_B)) {
LOGGER.info("Starting");
Thread.sleep(3000);
for (Entry<String, Long> entry : map.entrySet()) {
LOGGER.info("{} : {}", entry.getKey(), entry.getValue());
}
}
}
}
src/test/java/net/openhft/chronicle/map/Issue62ChronicleServer.java
+72
View File
@@ -0,0 +1,72 @@
/*
* Copyright (C) 2015 higherfrequencytrading.com
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 3 of the License.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.net.InetSocketAddress;
import java.nio.file.Paths;
import java.util.concurrent.TimeUnit;
public class Issue62ChronicleServer {
private final static Logger LOGGER = LoggerFactory.getLogger(Issue62ChronicleServer.class);
final static String STR =
//"This is just a long string, which causes sink to fail for some reason.";
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
private final static File MAP_FILE_A =
Paths.get(System.getProperty("java.io.tmpdir"), "map.a").toFile();
final static File MAP_FILE_B =
Paths.get(System.getProperty("java.io.tmpdir"), "map.b").toFile();
static void prepare(File file) {
if (file.exists())
file.delete();
file.deleteOnExit();
}
public static void main(String[] args) throws Exception {
prepare(MAP_FILE_A);
ChronicleMapBuilder<String, Long> cityPostalCodesMapBuilder =
ChronicleMapBuilder.of(String.class, Long.class)
//.averageKeySize(100)
.averageKey(STR)
.replication((byte) 1, TcpTransportAndNetworkConfig
.of(8076, new InetSocketAddress("localhost", 8077))
.heartBeatInterval(1, TimeUnit.SECONDS))
.entries(50_000);
ChronicleMap<String, Long> cityPostalCodes =
cityPostalCodesMapBuilder.createPersistedTo(MAP_FILE_A);
for(int i = 0; i < 100; i++ ) {
cityPostalCodes.put(STR + i, (long) i);
}
LOGGER.info("Map created");
Thread.sleep(15000);
}
}