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

further work in progress on stateless wire chronicle map

Browse Source
This commit is contained in:
Rob Austin
2015-02-28 23:37:57 +00:00
parent 1e24e1d8f1
commit 6258dc8a3c
6 changed files with 102 additions and 34 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;
TcpReplicator.TcpSocketChannelEntryWriter output;
MapIOBuffer output;
public BytesChronicleMap(VanillaChronicleMap<?, ?, ?, ?, ?, ?> delegate) {
this.delegate = delegate;
src/main/java/net/openhft/chronicle/map/CrossBytes.java
+12
View File
@@ -0,0 +1,12 @@
package net.openhft.chronicle.map;
import net.openhft.lang.io.MultiStoreBytes;
/**
* Created by Rob Austin
*/
public interface CrossBytes {
void write(MultiStoreBytes entry, long valueOffset, long valueSize);
void writeBoolean(boolean b);
}
src/main/java/net/openhft/chronicle/map/MapIOBuffer.java
+13
View File
@@ -0,0 +1,13 @@
package net.openhft.chronicle.map;
import net.openhft.lang.io.Bytes;
/**
* Created by Rob Austin
*/
interface MapIOBuffer {
void ensureBufferSize(long l);
Bytes in();
}
src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java
+59 -15
View File
@@ -27,12 +27,16 @@ import net.openhft.chronicle.bytes.BytesMarshallable;
import net.openhft.chronicle.wire.TextWire;
import net.openhft.chronicle.wire.Wire;
import net.openhft.lang.io.DirectStore;
import net.openhft.lang.io.NativeBytes;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
@@ -44,6 +48,25 @@ import java.util.List;
*/
class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarshallable> {
private static final Logger LOG = LoggerFactory.getLogger(StatelessWiredConnector.class);
static Field ADDRESS;
static Field CAPACITY;
static {
try {
ADDRESS = ByteBuffer.class.getDeclaredField("address");
ADDRESS.setAccessible(true);
CAPACITY = ByteBuffer.class.getDeclaredField("capacity");
CAPACITY.setAccessible(true);
} catch (Exception e) {
LOG.error("", e);
}
}
private final NativeBytes langBytes = new NativeBytes(0, 0);
private boolean handshingComplete;
// private final byte identifier;
@@ -56,6 +79,34 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
private TextWire inWire = new TextWire(Bytes.elasticByteBuffer());
private TextWire outWire = new TextWire(Bytes.elasticByteBuffer());
MapIOBuffer mapIOBuffer = new MapIOBuffer() {
@Override
public void ensureBufferSize(long l) {
outWire.bytes().ensureCapacity(l);
}
/**
* maps the outWire.bytes() to lang bytes
* @return lang bytes that represent the {@code outWire.bytes()}
*/
@Override
public net.openhft.lang.io.Bytes in() {
try {
ByteBuffer buffer = (ByteBuffer) outWire.bytes().underlyingObject();
langBytes.address(ADDRESS.getLong(buffer));
langBytes.capacity(CAPACITY.getLong(buffer));
langBytes.limit(buffer.limit());
langBytes.position(buffer.position());
return langBytes;
} catch (Exception e) {
LOG.error("", e);
}
return langBytes;
}
};
private byte identifier;
private long transactionId;
private long timestamp;
@@ -187,10 +238,9 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
}
private void writeLength(@NotNull final Wire textWire,
@NotNull final TcpReplicator.TcpSocketChannelEntryWriter writer) {
private void writeLength(@NotNull final Wire textWire) {
if (writer.in().position() > Integer.MAX_VALUE)
if (textWire.bytes().position() > Integer.MAX_VALUE)
throw new IllegalStateException("position too large");
// write the size
@@ -234,26 +284,20 @@ class StatelessWiredConnector<K extends BytesMarshallable, V extends BytesMarsha
final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2");
// move the lang bytes to the same location as the chronicle bytes
// todo - in comment code below
/* writer.in().position(out.bytes().position());
final BytesChronicleMap bytesMap = bytesMap(channelId);
bytesMap.output = writer;
bytesMap.output = mapIOBuffer;
try {
bytesMap.put(reader, reader, timestamp, identifier);
outWire.bytes().position(langBytes.position());
} catch (Throwable e) {
// move back to the start
out.bytes().position(2);
return sendException(out, e);
outWire.bytes().position(2);
return sendException(outWire, e);
} finally {
bytesMap.output = null;
writeLength(out, writer);
}*/
writeLength(outWire);
}
return null;
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+7 -8
View File
@@ -527,7 +527,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
*/
boolean isValidVersionNumber(String versionNumber) {
if (versionNumber.length()<=2)
if (versionNumber.length() <= 2)
return false;
for (char c : versionNumber.toCharArray()) {
@@ -668,7 +668,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
if (!isValidVersionNumber(attached.serverVersion)) {
LOG.warn("Please check that you don't have a third party system incorrectly connecting to ChronicleMap, " +
"Closing the remote connection as Chronicle can not make sense of the remote version number received from the external connection, version="+attached.serverVersion+", Chronicle is expecting the version number to only contain '.',A-Z,a-z,0-9");
"Closing the remote connection as Chronicle can not make sense of the remote version number received from the external connection, version=" + attached.serverVersion + ", Chronicle is expecting the version number to only contain '.',A-Z,a-z,0-9");
socketChannel.close();
}
@@ -1071,8 +1071,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
/**
* @author Rob Austin.
*/
class TcpSocketChannelEntryWriter {
public class TcpSocketChannelEntryWriter implements MapIOBuffer {
@NotNull
private final EntryCallback entryCallback;
@@ -1111,7 +1110,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
in().writeByte(localIdentifier);
}
void ensureBufferSize(long size) {
public void ensureBufferSize(long size) {
if (in().remaining() < size) {
size += entryCallback.in().position();
if (size > Integer.MAX_VALUE)
@@ -1137,7 +1136,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
throw e;
}
Bytes in() {
public Bytes in() {
return entryCallback.in();
}
@@ -1377,8 +1376,8 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
/**
* reads entries from the buffer till empty
*
* @param attached * @param key the selection key
channel
* @param attached * @param key the selection key
* channel
* @param key
* @throws InterruptedException
*/
src/main/java/net/openhft/chronicle/map/VanillaContext.java
+10 -10
View File
@@ -18,7 +18,6 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.impl.ContextFactory;
import net.openhft.chronicle.hash.impl.HashContext;
import net.openhft.chronicle.hash.impl.hashlookup.EntryConsumer;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.DelegatingMetaBytesInterop;
@@ -33,10 +32,10 @@ class VanillaContext<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
extends HashContext<K, KI, MKI>
implements MapKeyContext<K, V> {
/////////////////////////////////////////////////
// Inner state & lifecycle
enum VanillaChronicleMapContextFactory implements ContextFactory<VanillaContext> {
INSTANCE;
@@ -55,9 +54,10 @@ class VanillaContext<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
return VanillaContext.class;
}
}
VanillaContext() {}
VanillaContext() {
}
VanillaContext(HashContext contextCache, int indexInContextCache) {
super(contextCache, indexInContextCache);
}
@@ -75,11 +75,11 @@ class VanillaContext<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
/////////////////////////////////////////////////
// Map
public VanillaChronicleMap<K, KI, MKI, V, VI, MVI> m() {
return (VanillaChronicleMap<K, KI, MKI, V, VI, MVI>) h;
}
@Override
public void closeHashDependants() {
super.closeHashDependants();
@@ -87,7 +87,7 @@ class VanillaContext<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
closeValue();
closeValueReader();
}
/////////////////////////////////////////////////
// Map and context locals
@@ -589,7 +589,7 @@ class VanillaContext<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
private final MultiStoreBytes valueCopy = new MultiStoreBytes();
TcpReplicator.TcpSocketChannelEntryWriter output;
MapIOBuffer output;
void initBytesValueModel0() {
valueInterop = (VI) BytesBytesInterop.INSTANCE;