ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [06/32] incubator-ignite git commit: #ignite-180: Add processor.
Date Thu, 09 Apr 2015 11:28:30 GMT
#ignite-180: Add processor.


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

Branch: refs/heads/ignite-sprint-3
Commit: e449286923890085b02dfc9b3db44d87fb081585
Parents: 063fee4
Author: ivasilinets <ivasilinets@gridgain.com>
Authored: Tue Apr 7 12:08:20 2015 +0300
Committer: ivasilinets <ivasilinets@gridgain.com>
Committed: Tue Apr 7 12:08:20 2015 +0300

----------------------------------------------------------------------
 .../configuration/CollectionConfiguration.java  |  24 ++-
 .../internal/processors/cache/CacheType.java    |   5 -
 .../processors/cache/GridCacheProcessor.java    |  17 --
 .../datastructures/DataStructuresProcessor.java | 190 ++++++++++++++-----
 4 files changed, 161 insertions(+), 75 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e4492869/modules/core/src/main/java/org/apache/ignite/configuration/CollectionConfiguration.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/CollectionConfiguration.java
b/modules/core/src/main/java/org/apache/ignite/configuration/CollectionConfiguration.java
index a5a286f..40ab40d 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/CollectionConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/CollectionConfiguration.java
@@ -20,6 +20,8 @@ package org.apache.ignite.configuration;
 import org.apache.ignite.cache.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 
+import java.io.*;
+
 import static org.apache.ignite.cache.CacheAtomicityMode.*;
 import static org.apache.ignite.cache.CacheMemoryMode.*;
 import static org.apache.ignite.cache.CacheMode.*;
@@ -27,7 +29,7 @@ import static org.apache.ignite.cache.CacheMode.*;
 /**
  * Configuration for Ignite collections.
  */
-public class CollectionConfiguration {
+public class CollectionConfiguration implements Externalizable {
     /** Cache atomicity mode. */
     private CacheAtomicityMode atomicityMode = ATOMIC;
 
@@ -136,4 +138,24 @@ public class CollectionConfiguration {
     @Override public String toString() {
         return S.toString(CollectionConfiguration.class, this);
     }
+
+    /** {@inheritDoc} */
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeInt(atomicityMode.ordinal());
+        out.writeInt(cacheMode.ordinal());
+        out.writeObject(memoryMode);
+        out.writeInt(backups);
+        out.writeLong(offHeapMaxMem);
+        out.writeBoolean(collocated);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+        atomicityMode = CacheAtomicityMode.fromOrdinal(in.readInt());
+        cacheMode = CacheMode.fromOrdinal(in.readInt());
+        memoryMode = (CacheMemoryMode) in.readObject();
+        backups = in.readInt();
+        offHeapMaxMem = in.readLong();
+        collocated = in.readBoolean();
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e4492869/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheType.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheType.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheType.java
index d1cbfcd..e0747b9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheType.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheType.java
@@ -36,11 +36,6 @@ public enum CacheType {
     INTERNAL(false, SYSTEM_POOL),
 
     /**
-     * Internal datastructure's cache, should use separate thread pool.
-     */
-    DATASTRUCTURE(false, UTILITY_CACHE_POOL),
-
-    /**
      * Internal replicated cache, should use separate thread pool.
      */
     UTILITY(false, UTILITY_CACHE_POOL),

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e4492869/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 073dce2..c8d2ded 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -1291,23 +1291,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
     }
 
     /**
-     * Gets a collection of currentlty started caches.
-     *
-     * @return Collection of started cache names.
-     */
-    public Collection<CacheConfiguration> dataStructuresCacheNames() {
-        Collection<CacheConfiguration> res = new HashSet<>();
-        for (String name : registeredCaches.keySet()) {
-            DynamicCacheDescriptor desc = registeredCaches.get(name);
-
-            if (desc.cacheType() == CacheType.DATASTRUCTURE)
-                res.add(desc.cacheConfiguration());
-        }
-
-        return res;
-    }
-
-    /**
      * Gets cache mode.
      *
      * @param cacheName Cache name to check.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e4492869/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
index b8ace76..ed18785 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java
@@ -95,7 +95,7 @@ public final class DataStructuresProcessor extends GridProcessorAdapter
{
     private GridCacheProjectionEx<CacheDataStructuresConfigurationKey, Map<String,
DataStructureInfo>> utilityCache;
 
     /** */
-    private GridCacheProjectionEx<CacheDataStructuresCacheKey, List<CacheConfiguration>>
utilityDataCache;
+    private GridCacheProjectionEx<CacheDataStructuresCacheKey, List<CacheCollectionInfo>>
utilityDataCache;
 
     /**
      * @param ctx Context.
@@ -713,15 +713,6 @@ public final class DataStructuresProcessor extends GridProcessorAdapter
{
 
     /**
      * @param cfg Collection configuration.
-     * @param i Index.
-     * @return Cache name.
-     */
-    private String getCacheName(CollectionConfiguration cfg, int i) {
-        return "data_structures_" + i;
-    }
-
-    /**
-     * @param cfg Collection configuration.
      * @param name Cache name.
      * @return Cache configuration.
      */
@@ -742,55 +733,22 @@ public final class DataStructuresProcessor extends GridProcessorAdapter
{
      * @param cfg Collection configuration.
      * @return Cache name.
      */
-    private String findCompatibleConfiguration(CollectionConfiguration cfg) throws IgniteCheckedException
{
-        List<CacheConfiguration> cfgs = utilityDataCache.get(DATA_STRUCTURES_CACHE_KEY);
-
-        if (cfgs == null)
-            return null;
-
-        for (CacheConfiguration ccfg : cfgs) {
-            if (ccfg.getAtomicityMode() == cfg.atomicityMode() &&
-                    ccfg.getMemoryMode() == cfg.memoryMode() &&
-                    ccfg.getCacheMode() == cfg.cacheMode() &&
-                    ccfg.getBackups() == cfg.backups() &&
-                    ccfg.getOffHeapMaxMemory() == cfg.offHeapMaxMem())
-                return ccfg.getName();
-        }
-
-        return null;
-    }
-
-    /**
-     * @param cfg Collection configuration.
-     * @return Cache name.
-     */
     private String compatibleConfiguration(CollectionConfiguration cfg) throws IgniteCheckedException
{
-        String cacheName = findCompatibleConfiguration(cfg);
-
-        if (cacheName == null) {
-            try (IgniteInternalTx tx = utilityDataCache.txStartEx(PESSIMISTIC, REPEATABLE_READ))
{
-                cacheName = findCompatibleConfiguration(cfg);
+        T2<String, IgniteCheckedException> res =
+            utilityDataCache.invoke(DATA_STRUCTURES_CACHE_KEY, new AddDataCacheProcessor(cfg)).get();
 
-                if (cacheName == null) {
-                    List<CacheConfiguration> oldVal = utilityDataCache.get(DATA_STRUCTURES_CACHE_KEY);
+        IgniteCheckedException err = res.get2();
 
-                    cacheName = getCacheName(cfg, oldVal != null ? oldVal.size() : 0);
-
-                    CacheConfiguration newCfg = cacheConfiguration(cfg, cacheName);
-
-                    ctx.cache().dynamicStartCache(newCfg, cacheName, null, CacheType.DATASTRUCTURE,
false).get();
+        if (err != null)
+            throw err;
 
-                    List<CacheConfiguration> newVal = oldVal != null ? new ArrayList(oldVal)
:
-                            new ArrayList<CacheConfiguration>();
+        String cacheName = res.get1();
 
-                    newVal.add(newCfg);
+        assert cacheName != null;
 
-                    utilityDataCache.put(DATA_STRUCTURES_CACHE_KEY, newVal);
-                }
+        CacheConfiguration newCfg = cacheConfiguration(cfg, cacheName);
 
-                tx.commit();
-            }
-        }
+        ctx.cache().dynamicStartCache(newCfg, cacheName, null, CacheType.INTERNAL, false).get();
 
         return cacheName;
     }
@@ -1394,6 +1352,54 @@ public final class DataStructuresProcessor extends GridProcessorAdapter
{
     /**
      *
      */
+    static class CacheCollectionInfo implements Externalizable {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** */
+        private String cacheName;
+
+        /** */
+        private CollectionConfiguration cfg;
+
+        /**
+         * Required by {@link Externalizable}.
+         */
+        public CacheCollectionInfo() {
+            // No-op.
+        }
+
+        /**
+         * @param cacheName Collection cache name.
+         * @param cfg CollectionConfiguration.
+         */
+        public CacheCollectionInfo(String cacheName, CollectionConfiguration cfg) {
+            this.cacheName = cacheName;
+            this.cfg = cfg;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+            cfg = new CollectionConfiguration();
+            cfg.readExternal(in);
+            cacheName = U.readString(in);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeExternal(ObjectOutput out) throws IOException {
+            cfg.writeExternal(out);
+            U.writeString(out, cacheName);
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(CacheCollectionInfo.class, this);
+        }
+    }
+
+    /**
+     *
+     */
     static class QueueInfo extends CollectionInfo {
         /** */
         private static final long serialVersionUID = 0L;
@@ -1695,6 +1701,86 @@ public final class DataStructuresProcessor extends GridProcessorAdapter
{
     /**
      *
      */
+    static class AddDataCacheProcessor implements
+        EntryProcessor<CacheDataStructuresCacheKey, List<CacheCollectionInfo>,
+            T2<String, IgniteCheckedException>>, Externalizable {
+        /** */
+        private static final long serialVersionUID = 0L;
+
+        /** */
+        private CollectionConfiguration cfg;
+
+        /**
+         * @param cfg Data structure information.
+         */
+        AddDataCacheProcessor(CollectionConfiguration cfg) {
+            this.cfg = cfg;
+        }
+
+        /**
+         * Required by {@link Externalizable}.
+         */
+        public AddDataCacheProcessor() {
+            // No-op.
+        }
+
+        /** {@inheritDoc} */
+        @Override public T2<String, IgniteCheckedException> process(
+            MutableEntry<CacheDataStructuresCacheKey, List<CacheCollectionInfo>>
entry,
+            Object... args)
+        {
+            List<CacheCollectionInfo> list = entry.getValue();
+
+            if (list == null) {
+                list = new ArrayList<>();
+
+                list.add(new CacheCollectionInfo("datastructeres_0", cfg));
+
+                entry.setValue(list);
+
+                return new T2<>("datastructeres_0", null);
+            }
+
+            for (CacheCollectionInfo col : list) {
+                if (col.cfg.atomicityMode() == cfg.atomicityMode() &&
+                    col.cfg.memoryMode() == cfg.memoryMode() &&
+                    col.cfg.cacheMode() == cfg.cacheMode() &&
+                    col.cfg.backups() == cfg.backups() &&
+                    col.cfg.offHeapMaxMem() == cfg.offHeapMaxMem())
+
+                    return new T2<>(col.cacheName, null);
+            }
+
+            String newName = "datastructeres_" + list.size();
+
+            List<CacheCollectionInfo> newList = new ArrayList<>(list);
+
+            newList.add(new CacheCollectionInfo(newName, cfg));
+
+            return new T2<>(newName, null);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeExternal(ObjectOutput out) throws IOException {
+            cfg.writeExternal(out);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+            cfg = new CollectionConfiguration();
+
+            cfg.readExternal(in);
+        }
+
+        /** {@inheritDoc} */
+        @Override public String toString() {
+            return S.toString(AddDataCacheProcessor.class, this);
+        }
+    }
+
+    /**
+     *
+     */
     static class RemoveDataStructureProcessor implements
         EntryProcessor<CacheDataStructuresConfigurationKey, Map<String, DataStructureInfo>,
             T2<Boolean, IgniteCheckedException>>, Externalizable {


Mime
View raw message