Optimized generalized data value generation ser/deser, when Byteable (i. e. direct impl) is given

This commit is contained in:
leventov
2014-12-27 17:50:21 +07:00
parent b6ec1598e8
commit cf5ae63e84
7 changed files with 319 additions and 137 deletions
@@ -0,0 +1,70 @@
/*
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.hash.serialization.internal;
import net.openhft.chronicle.hash.hashing.Hasher;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
public abstract class BasicCopyingMetaBytesInterop<E, W> implements MetaBytesInterop<E, W> {
private static final long serialVersionUID = 0L;
static final Provider<DirectBytesBuffer> provider =
Provider.of(DirectBytesBuffer.class);
static abstract class BasicCopyingMetaBytesInteropProvider<E, I,
MI extends MetaBytesInterop<E, I>> implements MetaProvider<E, I, MI> {
private static final long serialVersionUID = 0L;
@Override
public ThreadLocalCopies getCopies(ThreadLocalCopies copies) {
return provider.getCopies(copies);
}
}
final DirectBytesBuffer buffer;
transient long size;
transient long hash;
protected BasicCopyingMetaBytesInterop(DirectBytesBuffer buffer) {
this.buffer = buffer;
}
@Override
public long size(W writer, E e) {
return size;
}
@Override
public boolean startsWith(W writer, Bytes bytes, E e) {
return bytes.startsWith(buffer.buffer);
}
@Override
public long hash(W writer, E e) {
long h;
if ((h = hash) == 0L)
return hash = Hasher.hash(buffer.buffer);
return h;
}
@Override
public void write(W writer, Bytes bytes, E e) {
bytes.write(buffer.buffer);
}
}
@@ -28,7 +28,7 @@ import net.openhft.lang.model.Byteable;
import org.jetbrains.annotations.NotNull;
public abstract class ByteableMarshaller<E extends Byteable>
implements BytesInterop<E>,
implements BytesInterop<E>, MetaBytesInterop<E, Object>,
DeserializationFactoryConfigurableBytesReader<E, ByteableMarshaller<E>>,
SizeMarshaller {
private static final long serialVersionUID = 0L;
@@ -160,6 +160,26 @@ public abstract class ByteableMarshaller<E extends Byteable>
return (E) NativeBytes.UNSAFE.allocateInstance(tClass);
}
@Override
public boolean startsWith(Object interop, Bytes bytes, E e) {
return startsWith(bytes, e);
}
@Override
public long hash(Object interop, E e) {
return hash(e);
}
@Override
public long size(Object writer, E e) {
return size(e);
}
@Override
public void write(Object writer, Bytes bytes, E e) {
write(bytes, e);
}
private static class Default<E extends Byteable> extends ByteableMarshaller<E> {
private static final long serialVersionUID = 0L;
@@ -19,23 +19,15 @@
package net.openhft.chronicle.hash.serialization.internal;
import net.openhft.chronicle.hash.serialization.BytesWriter;
import net.openhft.chronicle.hash.hashing.Hasher;
import net.openhft.lang.MemoryUnit;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.DirectBytes;
import net.openhft.lang.io.DirectStore;
import net.openhft.lang.io.serialization.BytesMarshaller;
import net.openhft.lang.io.serialization.JDKObjectSerializer;
import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.StatefulCopyable;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<E, W> {
private static final long serialVersionUID = 0L;
public abstract class CopyingMetaBytesInterop<E, W> extends BasicCopyingMetaBytesInterop<E, W> {
private static final long serialVersionUID = 1L;
private static final long MAX_REASONABLE_SERIALIZED_SIZE = MemoryUnit.MEGABYTES.toBytes(512L);
@@ -50,14 +42,11 @@ public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<
}
}
final DirectBytesBuffer buffer;
transient long size;
transient long hash;
transient W writer;
transient E cur;
private CopyingMetaBytesInterop(DirectBytesBuffer buffer) {
this.buffer = buffer;
protected CopyingMetaBytesInterop(DirectBytesBuffer buffer) {
super(buffer);
}
void init(W writer, E e, boolean mutable, long maxSize) {
@@ -82,29 +71,6 @@ public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<
abstract void innerWrite(W writer, Bytes bytes, E e);
@Override
public long size(W writer, E e) {
return size;
}
@Override
public boolean startsWith(W writer, Bytes bytes, E e) {
return bytes.startsWith(buffer.buffer);
}
@Override
public long hash(W writer, E e) {
long h;
if ((h = hash) == 0L)
return hash = Hasher.hash(buffer.buffer);
return h;
}
@Override
public void write(W writer, Bytes bytes, E e) {
bytes.write(buffer.buffer);
}
DirectBytesBuffer buffer() {
return buffer;
}
@@ -119,20 +85,6 @@ public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<
return new DirectBytesBuffer(bufferIdentity).forBytesWriter;
}
private static final Provider<DirectBytesBuffer> provider =
Provider.of(DirectBytesBuffer.class);
private static abstract class AbstractCopyingMetaBytesInteropProvider<E, I,
MI extends CopyingMetaBytesInterop<E, I>>
implements MetaProvider<E, I, MI> {
private static final long serialVersionUID = 0L;
@Override
public ThreadLocalCopies getCopies(ThreadLocalCopies copies) {
return provider.getCopies(copies);
}
}
public static <E, M extends BytesMarshaller<E>>
MetaProvider<E, M, CopyingMetaBytesInterop<E, M>> providerForBytesMarshaller(boolean mutable,
long maxSize) {
@@ -141,7 +93,7 @@ public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<
private static class BytesMarshallerCopyingMetaBytesInteropProvider<E,
M extends BytesMarshaller<E>>
extends AbstractCopyingMetaBytesInteropProvider<E, M, CopyingMetaBytesInterop<E, M>> {
extends BasicCopyingMetaBytesInteropProvider<E, M, CopyingMetaBytesInterop<E, M>> {
private static final long serialVersionUID = 0L;
private final boolean mutable;
private final long maxSize;
@@ -169,7 +121,7 @@ public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<
}
private static class BytesWriterCopyingMetaBytesInteropProvider<E, W extends BytesWriter<E>>
extends AbstractCopyingMetaBytesInteropProvider<E, W, CopyingMetaBytesInterop<E, W>> {
extends BasicCopyingMetaBytesInteropProvider<E, W, CopyingMetaBytesInterop<E, W>> {
private static final long serialVersionUID = 0L;
private final boolean mutable;
private final long maxSize;
@@ -189,83 +141,4 @@ public abstract class CopyingMetaBytesInterop<E, W> implements MetaBytesInterop<
return forBytesWriter;
}
}
private static class DirectBytesBuffer
implements StatefulCopyable<DirectBytesBuffer>, Serializable {
private static final long serialVersionUID = 0L;
private final Serializable identity;
private transient DirectBytes buffer;
private transient ForBytesMarshaller forBytesMarshaller;
private transient ForBytesWriter forBytesWriter;
DirectBytesBuffer(Serializable identity) {
this.identity = identity;
initTransients();
}
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
initTransients();
}
private void initTransients() {
forBytesMarshaller = new ForBytesMarshaller();
forBytesWriter = new ForBytesWriter();
}
Bytes obtain(long maxSize) {
DirectBytes buf;
if ((buf = buffer) != null) {
if (maxSize <= buf.capacity()) {
return buf.clear();
} else {
DirectStore store = (DirectStore) buf.store();
store.resize(maxSize, false);
return buffer = store.bytes();
}
} else {
buffer = new DirectStore(JDKObjectSerializer.INSTANCE, maxSize, false).bytes();
return buffer;
}
}
@Override
public Object stateIdentity() {
return identity;
}
@Override
public DirectBytesBuffer copy() {
return new DirectBytesBuffer(identity);
}
private class ForBytesMarshaller<E, M extends BytesMarshaller<E>>
extends CopyingMetaBytesInterop<E, M> {
private static final long serialVersionUID = 0L;
private ForBytesMarshaller() {
super(DirectBytesBuffer.this);
}
@Override
void innerWrite(M writer, Bytes bytes, E e) {
writer.write(bytes, e);
}
}
private class ForBytesWriter<E, W extends BytesWriter<E>>
extends CopyingMetaBytesInterop<E, W> {
private static final long serialVersionUID = 0L;
private ForBytesWriter() {
super(DirectBytesBuffer.this);
}
@Override
void innerWrite(W writer, Bytes bytes, E e) {
writer.write(bytes, e);
}
}
}
}
@@ -0,0 +1,92 @@
/*
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.hash.serialization.internal;
import net.openhft.chronicle.hash.serialization.BytesWriter;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.model.*;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
public abstract class DataValueMetaBytesInterop<E>
extends BasicCopyingMetaBytesInterop<E, BytesWriter<E>> {
private static final long serialVersionUID = 0L;
protected DataValueMetaBytesInterop(DirectBytesBuffer buffer) {
super(buffer);
}
void init(BytesWriter<E> writer, E e, long size) {
Bytes buffer = this.buffer.obtain(size);
writer.write(buffer, e);
buffer.flip();
this.size = size;
hash = 0L;
}
public static <E> MetaBytesInterop<E, BytesWriter<E>> forIdentity(Serializable bufferIdentity) {
return new DirectBytesBuffer(bufferIdentity).forDataValueWriter;
}
public static <E> MetaProvider<E, BytesWriter<E>, MetaBytesInterop<E, BytesWriter<E>>>
interopProvider(Class<E> eClass) {
return new InteropProvider<>(eClass);
}
private static class InteropProvider<E> extends BasicCopyingMetaBytesInteropProvider<
E, BytesWriter<E>, MetaBytesInterop<E, BytesWriter<E>>> {
private static final long serialVersionUID = 0L;
private final Class<E> eClass;
transient long size;
transient MetaBytesInterop metaByteableInterop;
private InteropProvider(Class<E> eClass) {
this.eClass = eClass;
initTransients();
}
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
initTransients();
}
private void initTransients() {
DataValueModel<E> model = DataValueModels.acquireModel(eClass);
size = DataValueGenerator.computeNonScalarOffset(model, eClass);
// TODO make this pick up configured deserializationFactory somehow
metaByteableInterop = ByteableMarshaller.of(
(Class) DataValueClasses.directClassFor(eClass));
}
@Override
public MetaBytesInterop<E, BytesWriter<E>> get(
ThreadLocalCopies copies, MetaBytesInterop<E, BytesWriter<E>> originalMetaWriter,
BytesWriter<E> writer, E e) {
if (e instanceof Byteable)
return metaByteableInterop;
DirectBytesBuffer.ForDataValueWriter forDataValueWriter =
provider.get(copies, ((DataValueMetaBytesInterop)originalMetaWriter).buffer)
.forDataValueWriter;
forDataValueWriter.init(writer, e, size);
return forDataValueWriter;
}
}
}
@@ -0,0 +1,117 @@
/*
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.hash.serialization.internal;
import net.openhft.chronicle.hash.serialization.BytesWriter;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.DirectBytes;
import net.openhft.lang.io.DirectStore;
import net.openhft.lang.io.serialization.BytesMarshaller;
import net.openhft.lang.io.serialization.JDKObjectSerializer;
import net.openhft.lang.threadlocal.StatefulCopyable;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
class DirectBytesBuffer
implements StatefulCopyable<DirectBytesBuffer>, Serializable {
private static final long serialVersionUID = 0L;
private final Serializable identity;
transient DirectBytes buffer;
transient ForBytesMarshaller forBytesMarshaller;
transient ForBytesWriter forBytesWriter;
transient ForDataValueWriter forDataValueWriter;
DirectBytesBuffer(Serializable identity) {
this.identity = identity;
initTransients();
}
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
initTransients();
}
private void initTransients() {
forBytesMarshaller = new ForBytesMarshaller();
forBytesWriter = new ForBytesWriter();
forDataValueWriter = new ForDataValueWriter();
}
Bytes obtain(long maxSize) {
DirectBytes buf;
if ((buf = buffer) != null) {
if (maxSize <= buf.capacity()) {
return buf.clear();
} else {
DirectStore store = (DirectStore) buf.store();
store.resize(maxSize, false);
return buffer = store.bytes();
}
} else {
buffer = new DirectStore(JDKObjectSerializer.INSTANCE, maxSize, false).bytes();
return buffer;
}
}
@Override
public Object stateIdentity() {
return identity;
}
@Override
public DirectBytesBuffer copy() {
return new DirectBytesBuffer(identity);
}
class ForBytesMarshaller<E, M extends BytesMarshaller<E>>
extends CopyingMetaBytesInterop<E, M> {
private static final long serialVersionUID = 0L;
private ForBytesMarshaller() {
super(DirectBytesBuffer.this);
}
@Override
void innerWrite(M writer, Bytes bytes, E e) {
writer.write(bytes, e);
}
}
class ForBytesWriter<E, W extends BytesWriter<E>>
extends CopyingMetaBytesInterop<E, W> {
private static final long serialVersionUID = 0L;
private ForBytesWriter() {
super(DirectBytesBuffer.this);
}
@Override
void innerWrite(W writer, Bytes bytes, E e) {
writer.write(bytes, e);
}
}
class ForDataValueWriter<E> extends DataValueMetaBytesInterop<E> {
private ForDataValueWriter() {
super(DirectBytesBuffer.this);
}
}
}
@@ -108,7 +108,10 @@ final class SerializationBuilder<E> implements Cloneable, Serializable {
DataValueModel<E> model = DataValueModels.acquireModel(eClass);
int size = DataValueGenerator.computeNonScalarOffset(model, eClass);
reader(reader);
writer(writer);
copyingInterop(null);
setInterop(writer);
metaInterop((MetaBytesInterop<E, ?>) DataValueMetaBytesInterop.forIdentity(role));
metaInteropProvider(DataValueMetaBytesInterop.interopProvider(eClass));
sizeMarshaller(constant((long) size));
return;
} catch (Exception e) {