ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vkuliche...@apache.org
Subject [05/10] incubator-ignite git commit: GG-9900 - Portable marshaller refactoring
Date Thu, 12 Mar 2015 05:04:27 GMT
GG-9900 - Portable marshaller refactoring


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/15b4ebb1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/15b4ebb1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/15b4ebb1

Branch: refs/heads/ignite-471
Commit: 15b4ebb19875cba90bfcebe8a2f62fbe8c3f8213
Parents: d838a9d
Author: Valentin Kulichenko <vkulichenko@gridgain.com>
Authored: Wed Mar 11 18:30:04 2015 -0700
Committer: Valentin Kulichenko <vkulichenko@gridgain.com>
Committed: Wed Mar 11 18:30:04 2015 -0700

----------------------------------------------------------------------
 .../GridClientOptimizedMarshaller.java          |   7 +-
 .../optimized/OptimizedMarshaller.java          |  45 ++--
 .../OptimizedObjectStreamRegistry.java          | 224 -------------------
 .../spi/discovery/tcp/TcpDiscoverySpi.java      |   2 +-
 .../OptimizedObjectStreamSelfTest.java          |  47 +---
 5 files changed, 27 insertions(+), 298 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/15b4ebb1/modules/core/src/main/java/org/apache/ignite/internal/client/marshaller/optimized/GridClientOptimizedMarshaller.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/marshaller/optimized/GridClientOptimizedMarshaller.java
b/modules/core/src/main/java/org/apache/ignite/internal/client/marshaller/optimized/GridClientOptimizedMarshaller.java
index 00f96b8..ff77177 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/marshaller/optimized/GridClientOptimizedMarshaller.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/marshaller/optimized/GridClientOptimizedMarshaller.java
@@ -50,17 +50,14 @@ public class GridClientOptimizedMarshaller implements GridClientMarshaller
{
      * Constructs optimized marshaller with specific parameters.
      *
      * @param requireSer Require serializable flag.
-     * @param poolSize Object streams pool size.
      * @throws IOException If an I/O error occurs while writing stream header.
      * @throws IgniteException If this marshaller is not supported on the current JVM.
      * @see OptimizedMarshaller
      */
-    public GridClientOptimizedMarshaller(boolean requireSer, int poolSize) throws IOException
{
-        opMarsh = new OptimizedMarshaller();
+    public GridClientOptimizedMarshaller(boolean requireSer) throws IOException {
+        opMarsh = new OptimizedMarshaller(requireSer);
 
         opMarsh.setContext(new ClientMarshallerContext());
-        opMarsh.setRequireSerializable(requireSer);
-        opMarsh.setPoolSize(poolSize);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/15b4ebb1/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedMarshaller.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedMarshaller.java
b/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedMarshaller.java
index bb63fdd..8eaf23b 100644
--- a/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedMarshaller.java
+++ b/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedMarshaller.java
@@ -19,6 +19,8 @@ package org.apache.ignite.marshaller.optimized;
 
 import org.apache.ignite.*;
 import org.apache.ignite.internal.util.*;
+import org.apache.ignite.internal.util.io.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.marshaller.*;
 import org.jetbrains.annotations.*;
 import sun.misc.*;
@@ -74,6 +76,9 @@ import java.nio.*;
  * For information about Spring framework visit <a href="http://www.springframework.org/">www.springframework.org</a>
  */
 public class OptimizedMarshaller extends AbstractMarshaller {
+    /** Initial buffer size. */
+    private static final int INIT_BUF_SIZE = 4 * 1024;
+
     /** Default class loader. */
     private final ClassLoader dfltClsLdr = getClass().getClassLoader();
 
@@ -101,6 +106,8 @@ public class OptimizedMarshaller extends AbstractMarshaller {
      * @param requireSer Whether to require {@link Serializable}.
      */
     public OptimizedMarshaller(boolean requireSer) {
+        this();
+
         this.requireSer = requireSer;
     }
 
@@ -122,25 +129,6 @@ public class OptimizedMarshaller extends AbstractMarshaller {
         this.mapper = mapper;
     }
 
-    /**
-     * Specifies size of cached object streams used by marshaller. Object streams are cached
for
-     * performance reason to avoid costly recreation for every serialization routine. If
{@code 0} (default),
-     * pool is not used and each thread has its own cached object stream which it keeps reusing.
-     * <p>
-     * Since each stream has an internal buffer, creating a stream for each thread can lead
to
-     * high memory consumption if many large messages are marshalled or unmarshalled concurrently.
-     * Consider using pool in this case. This will limit number of streams that can be created
and,
-     * therefore, decrease memory consumption.
-     * <p>
-     * NOTE: Using streams pool can decrease performance since streams will be shared between
-     * different threads which will lead to more frequent context switching.
-     *
-     * @param poolSize Streams pool size. If {@code 0}, pool is not used.
-     */
-    public void setPoolSize(int poolSize) {
-        OptimizedObjectStreamRegistry.poolSize(poolSize);
-    }
-
     /** {@inheritDoc} */
     @Override public void marshal(@Nullable Object obj, OutputStream out) throws IgniteCheckedException
{
         assert out != null;
@@ -148,7 +136,8 @@ public class OptimizedMarshaller extends AbstractMarshaller {
         OptimizedObjectOutputStream objOut = null;
 
         try {
-            objOut = OptimizedObjectStreamRegistry.out();
+            // TODO: IGNITE-471 - Need adaptive initial size.
+            objOut = new OptimizedObjectOutputStream(new GridUnsafeDataOutput(INIT_BUF_SIZE));
 
             objOut.context(ctx, mapper, requireSer);
 
@@ -160,7 +149,7 @@ public class OptimizedMarshaller extends AbstractMarshaller {
             throw new IgniteCheckedException("Failed to serialize object: " + obj, e);
         }
         finally {
-            OptimizedObjectStreamRegistry.closeOut(objOut);
+            U.closeQuiet(objOut);
         }
     }
 
@@ -169,7 +158,8 @@ public class OptimizedMarshaller extends AbstractMarshaller {
         OptimizedObjectOutputStream objOut = null;
 
         try {
-            objOut = OptimizedObjectStreamRegistry.out();
+            // TODO: IGNITE-471 - Need adaptive initial size.
+            objOut = new OptimizedObjectOutputStream(new GridUnsafeDataOutput(INIT_BUF_SIZE));
 
             objOut.context(ctx, mapper, requireSer);
 
@@ -181,7 +171,7 @@ public class OptimizedMarshaller extends AbstractMarshaller {
             throw new IgniteCheckedException("Failed to serialize object: " + obj, e);
         }
         finally {
-            OptimizedObjectStreamRegistry.closeOut(objOut);
+            U.closeQuiet(objOut);
         }
     }
 
@@ -193,7 +183,7 @@ public class OptimizedMarshaller extends AbstractMarshaller {
         OptimizedObjectInputStream objIn = null;
 
         try {
-            objIn = OptimizedObjectStreamRegistry.in();
+            objIn = new OptimizedObjectInputStream(new GridUnsafeDataInput());
 
             objIn.context(ctx, mapper, clsLdr != null ? clsLdr : dfltClsLdr);
 
@@ -210,7 +200,7 @@ public class OptimizedMarshaller extends AbstractMarshaller {
                 clsLdr, e);
         }
         finally {
-            OptimizedObjectStreamRegistry.closeIn(objIn);
+            U.closeQuiet(objIn);
         }
     }
 
@@ -223,10 +213,11 @@ public class OptimizedMarshaller extends AbstractMarshaller {
         OptimizedObjectInputStream objIn = null;
 
         try {
-            objIn = OptimizedObjectStreamRegistry.in();
+            objIn = new OptimizedObjectInputStream(new GridUnsafeDataInput());
 
             objIn.context(ctx, mapper, clsLdr != null ? clsLdr : dfltClsLdr);
 
+            // TODO: IGNITE-471 - Position is not moved.
             objIn.in().bytes(buf.array(), buf.position(), buf.remaining());
 
             return (T)objIn.readObject();
@@ -240,7 +231,7 @@ public class OptimizedMarshaller extends AbstractMarshaller {
                 clsLdr, e);
         }
         finally {
-            OptimizedObjectStreamRegistry.closeIn(objIn);
+            U.closeQuiet(objIn);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/15b4ebb1/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamRegistry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamRegistry.java
b/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamRegistry.java
deleted file mode 100644
index ae28015..0000000
--- a/modules/core/src/main/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamRegistry.java
+++ /dev/null
@@ -1,224 +0,0 @@
-/*
- * 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.ignite.marshaller.optimized;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.util.io.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-
-import java.io.*;
-import java.util.concurrent.*;
-
-/**
- * Storage for object streams.
- */
-class OptimizedObjectStreamRegistry {
-    /** Holders. */
-    private static final ThreadLocal<StreamHolder> holders = new ThreadLocal<>();
-
-    /** Holders pool. */
-    private static BlockingQueue<StreamHolder> pool;
-
-    /**
-     * Ensures singleton.
-     */
-    private OptimizedObjectStreamRegistry() {
-        // No-op.
-    }
-
-    /**
-     * Sets streams pool size.
-     *
-     * @param size Streams pool size.
-     */
-    static void poolSize(int size) {
-        if (size > 0) {
-            pool = new LinkedBlockingQueue<>(size);
-
-            for (int i = 0; i < size; i++) {
-                boolean b = pool.offer(new StreamHolder());
-
-                assert b;
-            }
-        }
-        else
-            pool = null;
-    }
-
-    /**
-     * Gets output stream.
-     *
-     * @return Object output stream.
-     * @throws org.apache.ignite.internal.IgniteInterruptedCheckedException If thread is
interrupted while trying to take holder from pool.
-     */
-    static OptimizedObjectOutputStream out() throws IgniteInterruptedCheckedException {
-        return holder().acquireOut();
-    }
-
-    /**
-     * Gets input stream.
-     *
-     * @return Object input stream.
-     * @throws org.apache.ignite.internal.IgniteInterruptedCheckedException If thread is
interrupted while trying to take holder from pool.
-     */
-    static OptimizedObjectInputStream in() throws IgniteInterruptedCheckedException {
-        return holder().acquireIn();
-    }
-
-    /**
-     * Closes and releases output stream.
-     *
-     * @param out Object output stream.
-     */
-    static void closeOut(OptimizedObjectOutputStream out) {
-        U.close(out, null);
-
-        StreamHolder holder = holders.get();
-
-        holder.releaseOut();
-
-        if (pool != null) {
-            holders.remove();
-
-            boolean b = pool.offer(holder);
-
-            assert b;
-        }
-    }
-
-    /**
-     * Closes and releases input stream.
-     *
-     * @param in Object input stream.
-     */
-    @SuppressWarnings("TypeMayBeWeakened")
-    static void closeIn(OptimizedObjectInputStream in) {
-        U.close(in, null);
-
-        StreamHolder holder = holders.get();
-
-        holder.releaseIn();
-
-        if (pool != null) {
-            holders.remove();
-
-            boolean b = pool.offer(holder);
-
-            assert b;
-        }
-    }
-
-    /**
-     * Gets holder from pool or thread local.
-     *
-     * @return Stream holder.
-     * @throws org.apache.ignite.internal.IgniteInterruptedCheckedException If thread is
interrupted while trying to take holder from pool.
-     */
-    private static StreamHolder holder() throws IgniteInterruptedCheckedException {
-        StreamHolder holder = holders.get();
-
-        if (holder == null) {
-            try {
-                holders.set(holder = pool != null ? pool.take() : new StreamHolder());
-            }
-            catch (InterruptedException e) {
-                throw new IgniteInterruptedCheckedException("Failed to take object stream
from pool (thread interrupted).", e);
-            }
-        }
-
-        return holder;
-    }
-
-    /**
-     * Streams holder.
-     */
-    private static class StreamHolder {
-        /** Output stream. */
-        private final OptimizedObjectOutputStream out = createOut();
-
-        /** Input stream. */
-        private final OptimizedObjectInputStream in = createIn();
-
-        /** Output streams counter. */
-        private int outAcquireCnt;
-
-        /** Input streams counter. */
-        private int inAcquireCnt;
-
-        /**
-         * Gets output stream.
-         *
-         * @return Object output stream.
-         */
-        OptimizedObjectOutputStream acquireOut() {
-            return outAcquireCnt++ > 0 ? createOut() : out;
-        }
-
-        /**
-         * Gets input stream.
-         *
-         * @return Object input stream.
-         */
-        OptimizedObjectInputStream acquireIn() {
-            return inAcquireCnt++ > 0 ? createIn() : in;
-        }
-
-        /**
-         * Releases output stream.
-         */
-        void releaseOut() {
-            outAcquireCnt--;
-        }
-
-        /**
-         * Releases input stream.
-         */
-        void releaseIn() {
-            inAcquireCnt--;
-        }
-
-        /**
-         * Creates output stream.
-         *
-         * @return Object output stream.
-         */
-        private OptimizedObjectOutputStream createOut() {
-            try {
-                return new OptimizedObjectOutputStream(new GridUnsafeDataOutput(4 * 1024));
-            }
-            catch (IOException e) {
-                throw new IgniteException("Failed to create object output stream.", e);
-            }
-        }
-
-        /**
-         * Creates input stream.
-         *
-         * @return Object input stream.
-         */
-        private OptimizedObjectInputStream createIn() {
-            try {
-                return new OptimizedObjectInputStream(new GridUnsafeDataInput());
-            }
-            catch (IOException e) {
-                throw new IgniteException("Failed to create object input stream.", e);
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/15b4ebb1/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index 1278556..c46c532 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -1677,7 +1677,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements
TcpDiscov
             Map<String, Object> attrs = new HashMap<>(node.getAttributes());
 
             attrs.put(IgniteNodeAttributes.ATTR_SECURITY_CREDENTIALS,
-                marsh.marshal(attrs.get(IgniteNodeAttributes.ATTR_SECURITY_CREDENTIALS)));
+                U.toArray(marsh.marshal(attrs.get(IgniteNodeAttributes.ATTR_SECURITY_CREDENTIALS))));
 
             node.setAttributes(attrs);
         }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/15b4ebb1/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamSelfTest.java
index 09adbb6..eea9a32 100644
--- a/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/marshaller/optimized/OptimizedObjectStreamSelfTest.java
@@ -239,41 +239,6 @@ public class OptimizedObjectStreamSelfTest extends GridCommonAbstractTest
{
     /**
      * @throws Exception If failed.
      */
-    public void testPool() throws Exception {
-        final TestObject obj = new TestObject();
-
-        obj.longVal = 100L;
-        obj.doubleVal = 100.0d;
-        obj.longArr = new Long[100 * 1024];
-        obj.doubleArr = new Double[100 * 1024];
-
-        Arrays.fill(obj.longArr, 100L);
-        Arrays.fill(obj.doubleArr, 100.0d);
-
-        final OptimizedMarshaller marsh = new OptimizedMarshaller();
-
-        marsh.setContext(CTX);
-
-        marsh.setPoolSize(5);
-
-        try {
-            multithreaded(new Callable<Object>() {
-                @Override public Object call() throws Exception {
-                    for (int i = 0; i < 50; i++)
-                        assertEquals(obj, marsh.unmarshal(marsh.marshal(obj), null));
-
-                    return null;
-                }
-            }, 20);
-        }
-        finally {
-            marsh.setPoolSize(0);
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
     public void testObjectWithNulls() throws Exception {
         TestObject obj = new TestObject();
 
@@ -883,7 +848,7 @@ public class OptimizedObjectStreamSelfTest extends GridCommonAbstractTest
{
      * @throws Exception If failed.
      */
     public void testReadToArray() throws Exception {
-        OptimizedObjectInputStream in = OptimizedObjectStreamRegistry.in();
+        OptimizedObjectInputStream in = new OptimizedObjectInputStream(new GridUnsafeDataInput());
 
         try {
             byte[] arr = new byte[50];
@@ -922,7 +887,7 @@ public class OptimizedObjectStreamSelfTest extends GridCommonAbstractTest
{
                 assertEquals(i < 10 ? 40 + i : 0, buf[i]);
         }
         finally {
-            OptimizedObjectStreamRegistry.closeIn(in);
+            U.closeQuiet(in);
         }
     }
 
@@ -1020,7 +985,7 @@ public class OptimizedObjectStreamSelfTest extends GridCommonAbstractTest
{
         OptimizedObjectInputStream in = null;
 
         try {
-            out = OptimizedObjectStreamRegistry.out();
+            out = new OptimizedObjectOutputStream(new GridUnsafeDataOutput(512));
 
             out.context(CTX, null, true);
 
@@ -1028,7 +993,7 @@ public class OptimizedObjectStreamSelfTest extends GridCommonAbstractTest
{
 
             byte[] arr = out.out().array();
 
-            in = OptimizedObjectStreamRegistry.in();
+            in = new OptimizedObjectInputStream(new GridUnsafeDataInput());
 
             in.context(CTX, null, getClass().getClassLoader());
 
@@ -1041,8 +1006,8 @@ public class OptimizedObjectStreamSelfTest extends GridCommonAbstractTest
{
             return (T)obj0;
         }
         finally {
-            OptimizedObjectStreamRegistry.closeOut(out);
-            OptimizedObjectStreamRegistry.closeIn(in);
+            U.closeQuiet(out);
+            U.closeQuiet(in);
         }
     }
 


Mime
View raw message