tinkerpop-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ok...@apache.org
Subject [16/50] [abbrv] incubator-tinkerpop git commit: Kryo shim refactoring and documentation
Date Wed, 08 Jun 2016 13:33:16 GMT
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraphSerializer.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraphSerializer.java
b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraphSerializer.java
index d5ba90d..431e1eb 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraphSerializer.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraphSerializer.java
@@ -27,11 +27,21 @@ import org.apache.tinkerpop.gremlin.structure.Direction;
 import org.apache.tinkerpop.gremlin.structure.Edge;
 import org.apache.tinkerpop.gremlin.structure.T;
 import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.io.kryoshim.InputShim;
-import org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShim;
-import org.apache.tinkerpop.gremlin.structure.io.kryoshim.OutputShim;
-import org.apache.tinkerpop.gremlin.structure.io.kryoshim.SerializerShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.InputShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.OutputShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.SerializerShim;
 
+/**
+ * Kryo serializer for {@link StarGraph}.  Implements an internal versioning capability for
backward compatibility.
+ * The single byte at the front of the serialization stream denotes the version.  That version
can be used to choose
+ * the correct deserialization mechanism.  The limitation is that this versioning won't help
with backward
+ * compatibility for custom serializers from providers.  Providers should be encouraged to
write their serializers
+ * with backward compatibility in mind.
+ *
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
 public class StarGraphSerializer implements SerializerShim<StarGraph> {
 
     private final Direction edgeDirectionToSerialize;

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/HadoopPoolShimService.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/HadoopPoolShimService.java
b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/HadoopPoolShimService.java
index 2053280..c19b914 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/HadoopPoolShimService.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/HadoopPoolShimService.java
@@ -18,7 +18,7 @@
  */
 package org.apache.tinkerpop.gremlin.hadoop.structure.io;
 
-import org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShimService;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService;
 import org.apache.tinkerpop.shaded.kryo.Kryo;
 import org.apache.tinkerpop.shaded.kryo.io.Input;
 import org.apache.tinkerpop.shaded.kryo.io.Output;

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritable.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritable.java
b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritable.java
index e7a38a5..88f7ee1 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritable.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritable.java
@@ -21,11 +21,9 @@ package org.apache.tinkerpop.gremlin.hadoop.structure.io;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.io.WritableUtils;
 import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
-import org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShimServiceLoader;
-import org.apache.tinkerpop.shaded.kryo.io.Output;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimServiceLoader;
 
 import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/VertexWritable.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/VertexWritable.java
b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/VertexWritable.java
index 7ac8e8c..2252ded 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/VertexWritable.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/VertexWritable.java
@@ -21,12 +21,11 @@ package org.apache.tinkerpop.gremlin.hadoop.structure.io;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
 import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShimServiceLoader;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimServiceLoader;
 import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
 import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
 
 import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService
b/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService
new file mode 100644
index 0000000..0b27e72
--- /dev/null
+++ b/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService
@@ -0,0 +1 @@
+org.apache.tinkerpop.gremlin.hadoop.structure.io.HadoopPoolShimService # HadoopPools provides/caches
instances of TinkerPop's shaded Kryo

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShimService
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShimService
b/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShimService
deleted file mode 100644
index 0b27e72..0000000
--- a/hadoop-gremlin/src/main/resources/META-INF/services/org.apache.tinkerpop.gremlin.structure.io.kryoshim.KryoShimService
+++ /dev/null
@@ -1 +0,0 @@
-org.apache.tinkerpop.gremlin.hadoop.structure.io.HadoopPoolShimService # HadoopPools provides/caches
instances of TinkerPop's shaded Kryo

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/TinkerPopKryoRegistrator.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/TinkerPopKryoRegistrator.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/TinkerPopKryoRegistrator.java
new file mode 100644
index 0000000..4c99e70
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/TinkerPopKryoRegistrator.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.tinkerpop.gremlin.spark.structure.io;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.serializers.JavaSerializer;
+import com.google.common.base.Preconditions;
+import org.apache.spark.serializer.KryoRegistrator;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.GroupStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.OrderGlobalStep;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalExplanation;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.MessagePayload;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewIncomingPayload;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewOutgoingPayload;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewPayload;
+import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.ObjectWritableSerializer;
+import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.VertexWritableSerializer;
+import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.kryoshim.unshaded.UnshadedSerializerAdapter;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoMapper;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.TypeRegistration;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.SerializerShim;
+import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A spark.kryo.registrator implementation that installs TinkerPop types.
+ * This is intended for use with spark.serializer=KryoSerializer, not GryoSerializer.
+ */
+public class TinkerPopKryoRegistrator implements KryoRegistrator {
+
+    private static final Logger log = LoggerFactory.getLogger(TinkerPopKryoRegistrator.class);
+
+    @Override
+    public void registerClasses(Kryo kryo) {
+        // TinkerPop type registrations copied from GyroSerializer's constructor
+        kryo.register(MessagePayload.class);
+        kryo.register(ViewIncomingPayload.class);
+        kryo.register(ViewOutgoingPayload.class);
+        kryo.register(ViewPayload.class);
+        kryo.register(VertexWritable.class, new UnshadedSerializerAdapter<>(new VertexWritableSerializer()));
+        kryo.register(ObjectWritable.class, new UnshadedSerializerAdapter<>(new ObjectWritableSerializer<>()));
+
+        Set<Class<?>> shimmedClasses = new HashSet<>();
+
+        Set<Class<?>> javaSerializationClasses = new HashSet<>();
+
+        // Copy GryoMapper's default registrations
+        for (TypeRegistration<?> tr : GryoMapper.build().create().getTypeRegistrations())
{
+            // Special case for JavaSerializer, which is generally implemented in terms of
TinkerPop's
+            // problematic static GryoMapper/GryoSerializer pool (these are handled below
the loop)
+            org.apache.tinkerpop.shaded.kryo.Serializer<?> shadedSerializer = tr.getShadedSerializer();
+            SerializerShim<?> serializerShim = tr.getSerializerShim();
+            if (null != shadedSerializer &&
+                    shadedSerializer.getClass().equals(org.apache.tinkerpop.shaded.kryo.serializers.JavaSerializer.class))
{
+                javaSerializationClasses.add(tr.getTargetClass());
+            } else if (null != serializerShim) {
+                log.debug("Registering class {} to serializer shim {} (serializer shim class
{})",
+                        tr.getTargetClass(), serializerShim, serializerShim.getClass());
+                kryo.register(tr.getTargetClass(), new UnshadedSerializerAdapter<>(serializerShim));
+                shimmedClasses.add(tr.getTargetClass());
+            } else {
+                // Register with the default behavior (FieldSerializer)
+                log.debug("Registering class {} with default serializer", tr.getTargetClass());
+                kryo.register(tr.getTargetClass());
+            }
+        }
+
+        Map<Class<?>, Serializer<?>> javaSerializerReplacements = new HashMap<>();
+        javaSerializerReplacements.put(GroupStep.GroupBiOperator.class, new JavaSerializer());
+        javaSerializerReplacements.put(OrderGlobalStep.OrderBiOperator.class, null);
+        javaSerializerReplacements.put(TraversalExplanation.class, null);
+
+        for (Map.Entry<Class<?>, Serializer<?>> e : javaSerializerReplacements.entrySet())
{
+            Class<?> c = e.getKey();
+            Serializer<?> s = e.getValue();
+
+            if (javaSerializationClasses.remove(c)) {
+                if (null != s) {
+                    log.debug("Registering class {} with serializer {}", c, s);
+                    kryo.register(c, s);
+                } else {
+                    log.debug("Registering class {} with default serializer", c);
+                    kryo.register(c);
+                }
+            } else {
+                log.debug("Registering class {} with JavaSerializer", c);
+                kryo.register(c, new JavaSerializer());
+            }
+        }
+
+        // We really care about StarGraph's shim serializer, so make sure we registered it
+        if (!shimmedClasses.contains(StarGraph.class)) {
+            log.warn("No SerializerShim found for StarGraph");
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/ObjectWritableSerializer.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/ObjectWritableSerializer.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/ObjectWritableSerializer.java
index 21cbc60..4ceb045 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/ObjectWritableSerializer.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/ObjectWritableSerializer.java
@@ -20,6 +20,10 @@
 package org.apache.tinkerpop.gremlin.spark.structure.io.gryo;
 
 import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.InputShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.OutputShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.SerializerShim;
 import org.apache.tinkerpop.shaded.kryo.Kryo;
 import org.apache.tinkerpop.shaded.kryo.Serializer;
 import org.apache.tinkerpop.shaded.kryo.io.Input;
@@ -28,16 +32,16 @@ import org.apache.tinkerpop.shaded.kryo.io.Output;
 /**
  * @author Marko A. Rodriguez (http://markorodriguez.com)
  */
-public final class ObjectWritableSerializer<T> extends Serializer<ObjectWritable<T>>
{
+public final class ObjectWritableSerializer<T> implements SerializerShim<ObjectWritable<T>>
{
 
     @Override
-    public void write(final Kryo kryo, final Output output, final ObjectWritable<T>
objectWritable) {
-        kryo.writeClassAndObject(output, objectWritable.get());
+    public <O extends OutputShim> void write(KryoShim<?, O> kryo, O output, ObjectWritable<T>
starGraph) {
+        kryo.writeClassAndObject(output, starGraph.get());
         output.flush();
     }
 
     @Override
-    public ObjectWritable<T> read(final Kryo kryo, final Input input, final Class<ObjectWritable<T>>
clazz) {
+    public <I extends InputShim> ObjectWritable<T> read(KryoShim<I, ?>
kryo, I input, Class<ObjectWritable<T>> clazz) {
         return new ObjectWritable(kryo.readClassAndObject(input));
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/VertexWritableSerializer.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/VertexWritableSerializer.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/VertexWritableSerializer.java
index 97891f3..f3c1b15 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/VertexWritableSerializer.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/VertexWritableSerializer.java
@@ -20,6 +20,10 @@
 package org.apache.tinkerpop.gremlin.spark.structure.io.gryo;
 
 import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.InputShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.OutputShim;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.SerializerShim;
 import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
 import org.apache.tinkerpop.shaded.kryo.Kryo;
 import org.apache.tinkerpop.shaded.kryo.Serializer;
@@ -29,14 +33,16 @@ import org.apache.tinkerpop.shaded.kryo.io.Output;
 /**
  * @author Marko A. Rodriguez (http://markorodriguez.com)
  */
-public final class VertexWritableSerializer extends Serializer<VertexWritable> {
+public final class VertexWritableSerializer implements SerializerShim<VertexWritable>
{
+
     @Override
-    public void write(final Kryo kryo, final Output output, final VertexWritable vertexWritable)
{
+    public <O extends OutputShim> void write(KryoShim<?, O> kryo, O output, VertexWritable
vertexWritable) {
         kryo.writeObject(output, vertexWritable.get().graph());
+        output.flush();
     }
 
     @Override
-    public VertexWritable read(final Kryo kryo, final Input input, final Class<VertexWritable>
aClass) {
+    public <I extends InputShim> VertexWritable read(KryoShim<I, ?> kryo, I input,
Class<VertexWritable> clazz) {
         return new VertexWritable(kryo.readObject(input, StarGraph.class).getStarVertex());
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedInputAdapter.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedInputAdapter.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedInputAdapter.java
new file mode 100644
index 0000000..c533af7
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedInputAdapter.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * Copyright DataStax, Inc.
+ *
+ * Please see the included license file for details.
+ */
+package org.apache.tinkerpop.gremlin.spark.structure.io.gryo.kryoshim.unshaded;
+
+import com.esotericsoftware.kryo.io.Input;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.InputShim;
+
+public class UnshadedInputAdapter implements InputShim
+{
+
+    private final Input unshadedInput;
+
+    public UnshadedInputAdapter(Input unshadedInput)
+    {
+        this.unshadedInput = unshadedInput;
+    }
+
+    Input getUnshadedInput()
+    {
+        return unshadedInput;
+    }
+
+    @Override
+    public byte readByte()
+    {
+        return unshadedInput.readByte();
+    }
+
+    @Override
+    public byte[] readBytes(int size) {
+        return unshadedInput.readBytes(size);
+    }
+
+    @Override
+    public String readString()
+    {
+        return unshadedInput.readString();
+    }
+
+    @Override
+    public long readLong()
+    {
+        return unshadedInput.readLong();
+    }
+
+    @Override
+    public int readInt() {
+        return unshadedInput.readInt();
+    }
+
+    @Override
+    public double readDouble()
+    {
+        return unshadedInput.readDouble();
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoAdapter.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoAdapter.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoAdapter.java
new file mode 100644
index 0000000..b14abe0
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoAdapter.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * Copyright DataStax, Inc.
+ *
+ * Please see the included license file for details.
+ */
+package org.apache.tinkerpop.gremlin.spark.structure.io.gryo.kryoshim.unshaded;
+
+import com.esotericsoftware.kryo.Kryo;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShim;
+
+public class UnshadedKryoAdapter implements KryoShim<UnshadedInputAdapter, UnshadedOutputAdapter>
+{
+    private final Kryo unshadedKryo;
+
+    public UnshadedKryoAdapter(Kryo unshadedKryo)
+    {
+        this.unshadedKryo = unshadedKryo;
+    }
+
+    @Override
+    public <T> T readObject(UnshadedInputAdapter input, Class<T> type)
+    {
+        return unshadedKryo.readObject(input.getUnshadedInput(), type);
+    }
+
+    @Override
+    public Object readClassAndObject(UnshadedInputAdapter input)
+    {
+        return unshadedKryo.readClassAndObject(input.getUnshadedInput());
+    }
+
+    @Override
+    public void writeObject(UnshadedOutputAdapter output, Object object)
+    {
+        unshadedKryo.writeObject(output.getUnshadedOutput(), object);
+    }
+
+    @Override
+    public void writeClassAndObject(UnshadedOutputAdapter output, Object object)
+    {
+        unshadedKryo.writeClassAndObject(output.getUnshadedOutput(), object);
+    }
+
+    @Override
+    public <T> T readObjectOrNull(UnshadedInputAdapter input, Class<T> type)
+    {
+        return unshadedKryo.readObjectOrNull(input.getUnshadedInput(), type);
+    }
+
+    @Override
+    public void writeObjectOrNull(UnshadedOutputAdapter output, Object object, Class type)
+    {
+        unshadedKryo.writeObjectOrNull(output.getUnshadedOutput(), object, type);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoShimService.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoShimService.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoShimService.java
new file mode 100644
index 0000000..d0411e8
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedKryoShimService.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * Copyright DataStax, Inc.
+ * <p>
+ * Please see the included license file for details.
+ */
+package org.apache.tinkerpop.gremlin.spark.structure.io.gryo.kryoshim.unshaded;
+
+import com.twitter.chill.KryoInstantiator;
+import com.twitter.chill.KryoPool;
+import com.twitter.chill.SerDeState;
+import org.apache.spark.SparkConf;
+import org.apache.spark.serializer.KryoSerializer;
+import org.apache.tinkerpop.gremlin.spark.structure.io.TinkerPopKryoRegistrator;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.KryoShimService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public class UnshadedKryoShimService implements KryoShimService {
+
+    public static final String SPARK_KRYO_POOL_SIZE_SYSTEM_PROPERTY = "tinkerpop.kryo.poolsize";
+
+    private static final Logger log = LoggerFactory.getLogger(UnshadedKryoShimService.class);
+    private static final int SPARK_KRYO_POOL_SIZE_DEFAULT = 8;
+
+    private final KryoSerializer sparkKryoSerializer;
+    private final KryoPool kryoPool;
+
+    public UnshadedKryoShimService() {
+        this(TinkerPopKryoRegistrator.class.getCanonicalName(), getDefaultKryoPoolSize());
+    }
+
+    public UnshadedKryoShimService(String sparkKryoRegistratorClassname, int kryoPoolSize)
{
+        SparkConf sparkConf = new SparkConf();
+        sparkConf.set("spark.serializer", KryoSerializer.class.getCanonicalName());
+        sparkConf.set("spark.kryo.registrator", sparkKryoRegistratorClassname);
+        sparkKryoSerializer = new KryoSerializer(sparkConf);
+        kryoPool = KryoPool.withByteArrayOutputStream(kryoPoolSize, new KryoInstantiator());
+    }
+
+    @Override
+    public Object readClassAndObject(InputStream source) {
+        SerDeState sds = null;
+        try {
+            sds = kryoPool.borrow();
+
+            sds.setInput(source);
+
+            return sds.readClassAndObject();
+        } finally {
+            kryoPool.release(sds);
+        }
+    }
+
+    @Override
+    public void writeClassAndObject(Object o, OutputStream sink) {
+        SerDeState sds = null;
+        try {
+            sds = kryoPool.borrow();
+
+            sds.writeClassAndObject(o); // this writes to an internal buffer
+
+            sds.writeOutputTo(sink); // this copies the internal buffer to sink
+
+            sink.flush();
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        } finally {
+            kryoPool.release(sds);
+        }
+    }
+
+    @Override
+    public int getPriority() {
+        return 1024;
+    }
+
+    private static int getDefaultKryoPoolSize() {
+        String raw = System.getProperty(SPARK_KRYO_POOL_SIZE_SYSTEM_PROPERTY);
+
+        int size = SPARK_KRYO_POOL_SIZE_DEFAULT;
+        try {
+            size = Integer.valueOf(raw);
+            log.info("Setting kryo pool size to {} according to system property {}", size,
+                    SPARK_KRYO_POOL_SIZE_SYSTEM_PROPERTY);
+        } catch (NumberFormatException e) {
+            log.error("System property {}={} could not be parsed as an integer, using default
value {}",
+                    SPARK_KRYO_POOL_SIZE_SYSTEM_PROPERTY, raw, size, e);
+        }
+
+        return size;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedOutputAdapter.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedOutputAdapter.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedOutputAdapter.java
new file mode 100644
index 0000000..9cc59d4
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedOutputAdapter.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * Copyright DataStax, Inc.
+ *
+ * Please see the included license file for details.
+ */
+package org.apache.tinkerpop.gremlin.spark.structure.io.gryo.kryoshim.unshaded;
+
+import com.esotericsoftware.kryo.io.Output;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.OutputShim;
+
+public class UnshadedOutputAdapter implements OutputShim
+{
+    private final Output unshadedOutput;
+
+    public UnshadedOutputAdapter(Output unshadedOutput)
+    {
+        this.unshadedOutput = unshadedOutput;
+    }
+
+    Output getUnshadedOutput()
+    {
+        return unshadedOutput;
+    }
+
+    @Override
+    public void writeByte(byte b)
+    {
+        unshadedOutput.writeByte(b);
+    }
+
+    @Override
+    public void writeBytes(byte[] array, int offset, int count) {
+        unshadedOutput.writeBytes(array, offset, count);
+    }
+
+    @Override
+    public void writeString(String s)
+    {
+        unshadedOutput.writeString(s);
+    }
+
+    @Override
+    public void writeLong(long l)
+    {
+        unshadedOutput.writeLong(l);
+    }
+
+    @Override
+    public void writeInt(int i) {
+        unshadedOutput.writeInt(i);
+    }
+
+    @Override
+    public void writeDouble(double d)
+    {
+        unshadedOutput.writeDouble(d);
+    }
+
+    @Override
+    public void flush()
+    {
+        unshadedOutput.flush();
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/218d7909/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedSerializerAdapter.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedSerializerAdapter.java
b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedSerializerAdapter.java
new file mode 100644
index 0000000..efc9a4f
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/gryo/kryoshim/unshaded/UnshadedSerializerAdapter.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+/**
+ * Copyright DataStax, Inc.
+ *
+ * Please see the included license file for details.
+ */
+package org.apache.tinkerpop.gremlin.spark.structure.io.gryo.kryoshim.unshaded;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.kryoshim.SerializerShim;
+
+public class UnshadedSerializerAdapter<T> extends Serializer<T>
+{
+
+    SerializerShim<T> serializer;
+
+    public UnshadedSerializerAdapter(SerializerShim<T> serializer) {
+        this.serializer = serializer;
+        setImmutable(this.serializer.isImmutable());
+    }
+
+    @Override
+    public void write(Kryo kryo, Output output, T t) {
+        UnshadedKryoAdapter shadedKryoAdapter = new UnshadedKryoAdapter(kryo);
+        UnshadedOutputAdapter shadedOutputAdapter = new UnshadedOutputAdapter(output);
+        serializer.write(shadedKryoAdapter, shadedOutputAdapter, t);
+    }
+
+    @Override
+    public T read(Kryo kryo, Input input, Class<T> aClass)
+    {
+        UnshadedKryoAdapter shadedKryoAdapter = new UnshadedKryoAdapter(kryo);
+        UnshadedInputAdapter shadedInputAdapter = new UnshadedInputAdapter(input);
+        return serializer.read(shadedKryoAdapter, shadedInputAdapter, aClass);
+    }
+}


Mime
View raw message