renaming wip
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/60a1a481 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/60a1a481 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/60a1a481 Branch: refs/heads/ignite-406 Commit: 60a1a481781c0a8c6533489c83adac82ec2ed4f6 Parents: b2c679e Author: Yakov Zhdanov <yzhda...@gridgain.com> Authored: Wed Mar 11 17:50:32 2015 +0300 Committer: Yakov Zhdanov <yzhda...@gridgain.com> Committed: Wed Mar 11 17:50:32 2015 +0300 ---------------------------------------------------------------------- .../ignite/codegen/MessageCodeGenerator.java | 2 +- .../ignite/internal/GridKernalContext.java | 2 +- .../ignite/internal/GridKernalContextImpl.java | 2 +- .../apache/ignite/internal/IgniteKernal.java | 2 +- .../communication/GridIoMessageFactory.java | 2 +- .../processors/cache/GridCacheAdapter.java | 2 +- .../GridDistributedCacheAdapter.java | 2 +- .../version/GridCacheRawVersionedEntry.java | 2 +- .../datastream/DataStreamerCacheUpdaters.java | 206 --- .../datastream/DataStreamerEntry.java | 170 --- .../datastream/DataStreamerFuture.java | 60 - .../processors/datastream/DataStreamerImpl.java | 1405 ------------------ .../datastream/DataStreamerProcessor.java | 307 ---- .../datastream/DataStreamerRequest.java | 451 ------ .../datastream/DataStreamerResponse.java | 166 --- .../datastream/DataStreamerUpdateJob.java | 150 -- .../internal/processors/datastream/package.html | 24 - .../datastreamer/DataStreamerCacheUpdaters.java | 206 +++ .../datastreamer/DataStreamerEntry.java | 170 +++ .../datastreamer/DataStreamerFuture.java | 60 + .../datastreamer/DataStreamerImpl.java | 1405 ++++++++++++++++++ .../datastreamer/DataStreamerProcessor.java | 307 ++++ .../datastreamer/DataStreamerRequest.java | 451 ++++++ .../datastreamer/DataStreamerResponse.java | 166 +++ .../datastreamer/DataStreamerUpdateJob.java | 150 ++ .../processors/datastreamer/package.html | 24 + .../dr/IgniteDrDataStreamerCacheUpdater.java | 2 +- .../processors/igfs/IgfsDataManager.java | 2 +- .../datastream/DataStreamerImplSelfTest.java | 205 --- .../DataStreamerProcessorSelfTest.java | 970 ------------ .../IgniteDataStreamerPerformanceTest.java | 197 --- .../datastreamer/DataStreamerImplSelfTest.java | 205 +++ .../DataStreamerProcessorSelfTest.java | 970 ++++++++++++ .../IgniteDataStreamerPerformanceTest.java | 197 +++ .../ignite/testsuites/IgniteCacheTestSuite.java | 2 +- 35 files changed, 4322 insertions(+), 4322 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java ---------------------------------------------------------------------- diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java index 07200f5..5037b8a 100644 --- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java +++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java @@ -18,7 +18,7 @@ package org.apache.ignite.codegen; import org.apache.ignite.internal.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.util.typedef.internal.*; import org.apache.ignite.lang.*; import org.apache.ignite.plugin.extensions.communication.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java index 8b6eaa9..f780676 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java @@ -36,7 +36,7 @@ import org.apache.ignite.internal.processors.clock.*; import org.apache.ignite.internal.processors.closure.*; import org.apache.ignite.internal.processors.cluster.*; import org.apache.ignite.internal.processors.continuous.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.processors.datastructures.*; import org.apache.ignite.internal.processors.hadoop.*; import org.apache.ignite.internal.processors.igfs.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java index c066c94..362a472 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java @@ -38,7 +38,7 @@ import org.apache.ignite.internal.processors.clock.*; import org.apache.ignite.internal.processors.closure.*; import org.apache.ignite.internal.processors.cluster.*; import org.apache.ignite.internal.processors.continuous.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.processors.datastructures.*; import org.apache.ignite.internal.processors.hadoop.*; import org.apache.ignite.internal.processors.igfs.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java index 92e4ced..c0b5225 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java @@ -42,7 +42,7 @@ import org.apache.ignite.internal.processors.clock.*; import org.apache.ignite.internal.processors.closure.*; import org.apache.ignite.internal.processors.cluster.*; import org.apache.ignite.internal.processors.continuous.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.processors.datastructures.*; import org.apache.ignite.internal.processors.hadoop.*; import org.apache.ignite.internal.processors.job.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java index 7be89cf..762507a 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java @@ -34,7 +34,7 @@ import org.apache.ignite.internal.processors.cache.transactions.*; import org.apache.ignite.internal.processors.cache.version.*; import org.apache.ignite.internal.processors.clock.*; import org.apache.ignite.internal.processors.continuous.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.processors.igfs.*; import org.apache.ignite.internal.processors.rest.handlers.task.*; import org.apache.ignite.internal.processors.streamer.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java index ff5e58c..2e16475 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java @@ -32,7 +32,7 @@ import org.apache.ignite.internal.processors.cache.dr.*; import org.apache.ignite.internal.processors.cache.query.*; import org.apache.ignite.internal.processors.cache.transactions.*; import org.apache.ignite.internal.processors.cache.version.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.processors.dr.*; import org.apache.ignite.internal.processors.task.*; import org.apache.ignite.internal.transactions.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java index 3d85590..792fc47 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java @@ -26,7 +26,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.*; import org.apache.ignite.internal.processors.cache.distributed.near.*; import org.apache.ignite.internal.processors.cache.transactions.*; import org.apache.ignite.internal.processors.cache.version.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.processors.task.*; import org.apache.ignite.internal.util.future.*; import org.apache.ignite.internal.util.typedef.internal.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java index 9e73e1b..986515c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheRawVersionedEntry.java @@ -20,7 +20,7 @@ package org.apache.ignite.internal.processors.cache.version; import org.apache.ignite.*; import org.apache.ignite.internal.*; import org.apache.ignite.internal.processors.cache.*; -import org.apache.ignite.internal.processors.datastream.*; +import org.apache.ignite.internal.processors.datastreamer.*; import org.apache.ignite.internal.util.typedef.internal.*; import org.apache.ignite.marshaller.*; import org.apache.ignite.plugin.extensions.communication.*; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java deleted file mode 100644 index 64f0851..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerCacheUpdaters.java +++ /dev/null @@ -1,206 +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.internal.processors.datastream; - -import org.apache.ignite.*; -import org.apache.ignite.internal.processors.cache.*; -import org.apache.ignite.internal.util.typedef.*; -import org.jetbrains.annotations.*; - -import java.util.*; - -/** - * Bundled factory for cache updaters. - */ -public class DataStreamerCacheUpdaters { - /** */ - private static final IgniteDataStreamer.Updater INDIVIDUAL = new Individual(); - - /** */ - private static final IgniteDataStreamer.Updater BATCHED = new Batched(); - - /** */ - private static final IgniteDataStreamer.Updater BATCHED_SORTED = new BatchedSorted(); - - /** - * Updates cache using independent {@link org.apache.ignite.cache.GridCache#put(Object, Object, org.apache.ignite.lang.IgnitePredicate[])} and - * {@link org.apache.ignite.cache.GridCache#remove(Object, org.apache.ignite.lang.IgnitePredicate[])} operations. Thus it is safe from deadlocks but performance - * is not the best. - * - * @return Single updater. - */ - public static <K, V> IgniteDataStreamer.Updater<K, V> individual() { - return INDIVIDUAL; - } - - /** - * Updates cache using batched methods {@link org.apache.ignite.cache.GridCache#putAll(Map, org.apache.ignite.lang.IgnitePredicate[])} and - * {@link org.apache.ignite.cache.GridCache#removeAll(Collection, org.apache.ignite.lang.IgnitePredicate[])}. Can cause deadlocks if the same keys are getting - * updated concurrently. Performance is generally better than in {@link #individual()}. - * - * @return Batched updater. - */ - public static <K, V> IgniteDataStreamer.Updater<K, V> batched() { - return BATCHED; - } - - /** - * Updates cache using batched methods {@link org.apache.ignite.cache.GridCache#putAll(Map, org.apache.ignite.lang.IgnitePredicate[])} and - * {@link org.apache.ignite.cache.GridCache#removeAll(Collection, org.apache.ignite.lang.IgnitePredicate[])}. Keys are sorted in natural order and if all updates - * use the same rule deadlock can not happen. Performance is generally better than in {@link #individual()}. - * - * @return Batched sorted updater. - */ - public static <K extends Comparable<?>, V> IgniteDataStreamer.Updater<K, V> batchedSorted() { - return BATCHED_SORTED; - } - - /** - * Updates cache. - * - * @param cache Cache. - * @param rmvCol Keys to remove. - * @param putMap Entries to put. - * @throws IgniteException If failed. - */ - protected static <K, V> void updateAll(IgniteCache<K, V> cache, @Nullable Set<K> rmvCol, - Map<K, V> putMap) { - assert rmvCol != null || putMap != null; - - // Here we assume that there are no key duplicates, so the following calls are valid. - if (rmvCol != null) - ((IgniteCacheProxy<K, V>)cache).removeAll(rmvCol); - - if (putMap != null) - cache.putAll(putMap); - } - - /** - * Simple cache updater implementation. Updates keys one by one thus is not dead lock prone. - */ - private static class Individual<K, V> implements IgniteDataStreamer.Updater<K, V>, InternalUpdater { - /** */ - private static final long serialVersionUID = 0L; - - /** {@inheritDoc} */ - @Override public void update(IgniteCache<K, V> cache, Collection<Map.Entry<K, V>> entries) { - assert cache != null; - assert !F.isEmpty(entries); - - for (Map.Entry<K, V> entry : entries) { - K key = entry.getKey(); - - assert key != null; - - V val = entry.getValue(); - - if (val == null) - cache.remove(key); - else - cache.put(key, val); - } - } - } - - /** - * Batched updater. Updates cache using batch operations thus is dead lock prone. - */ - private static class Batched<K, V> implements IgniteDataStreamer.Updater<K, V>, InternalUpdater { - /** */ - private static final long serialVersionUID = 0L; - - /** {@inheritDoc} */ - @Override public void update(IgniteCache<K, V> cache, Collection<Map.Entry<K, V>> entries) { - assert cache != null; - assert !F.isEmpty(entries); - - Map<K, V> putAll = null; - Set<K> rmvAll = null; - - for (Map.Entry<K, V> entry : entries) { - K key = entry.getKey(); - - assert key != null; - - V val = entry.getValue(); - - if (val == null) { - if (rmvAll == null) - rmvAll = new HashSet<>(); - - rmvAll.add(key); - } - else { - if (putAll == null) - putAll = new HashMap<>(); - - putAll.put(key, val); - } - } - - updateAll(cache, rmvAll, putAll); - } - } - - /** - * Batched updater. Updates cache using batch operations thus is dead lock prone. - */ - private static class BatchedSorted<K, V> implements IgniteDataStreamer.Updater<K, V>, InternalUpdater { - /** */ - private static final long serialVersionUID = 0L; - - /** {@inheritDoc} */ - @Override public void update(IgniteCache<K, V> cache, Collection<Map.Entry<K, V>> entries) { - assert cache != null; - assert !F.isEmpty(entries); - - Map<K, V> putAll = null; - Set<K> rmvAll = null; - - for (Map.Entry<K, V> entry : entries) { - K key = entry.getKey(); - - assert key instanceof Comparable; - - V val = entry.getValue(); - - if (val == null) { - if (rmvAll == null) - rmvAll = new TreeSet<>(); - - rmvAll.add(key); - } - else { - if (putAll == null) - putAll = new TreeMap<>(); - - putAll.put(key, val); - } - } - - updateAll(cache, rmvAll, putAll); - } - } - - /** - * Marker interface for updaters which do not need to unwrap cache objects. - */ - public static interface InternalUpdater { - // No-op. - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java deleted file mode 100644 index 2aa11e5..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerEntry.java +++ /dev/null @@ -1,170 +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.internal.processors.datastream; - -import org.apache.ignite.internal.processors.cache.*; -import org.apache.ignite.internal.util.tostring.*; -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.plugin.extensions.communication.*; - -import java.nio.*; -import java.util.*; - -/** - * - */ -public class DataStreamerEntry implements Map.Entry<KeyCacheObject, CacheObject>, Message { - /** */ - private static final long serialVersionUID = 0L; - - /** */ - @GridToStringInclude - protected KeyCacheObject key; - - /** */ - @GridToStringInclude - protected CacheObject val; - - /** - * - */ - public DataStreamerEntry() { - // No-op. - } - - /** - * @param key Key. - * @param val Value. - */ - public DataStreamerEntry(KeyCacheObject key, CacheObject val) { - this.key = key; - this.val = val; - } - - /** {@inheritDoc} */ - @Override public KeyCacheObject getKey() { - return key; - } - - /** {@inheritDoc} */ - @Override public CacheObject getValue() { - return val; - } - - /** {@inheritDoc} */ - @Override public CacheObject setValue(CacheObject val) { - CacheObject old = this.val; - - this.val = val; - - return old; - } - - /** - * @param ctx Cache context. - * @return Map entry unwrapping internal key and value. - */ - public <K, V> Map.Entry<K, V> toEntry(final GridCacheContext ctx) { - return new Map.Entry<K, V>() { - @Override public K getKey() { - return key.value(ctx.cacheObjectContext(), false); - } - - @Override public V setValue(V val) { - throw new UnsupportedOperationException(); - } - - @Override public V getValue() { - return val != null ? val.<V>value(ctx.cacheObjectContext(), false) : null; - } - }; - } - - /** {@inheritDoc} */ - @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) { - writer.setBuffer(buf); - - if (!writer.isHeaderWritten()) { - if (!writer.writeHeader(directType(), fieldsCount())) - return false; - - writer.onHeaderWritten(); - } - - switch (writer.state()) { - case 0: - if (!writer.writeMessage("key", key)) - return false; - - writer.incrementState(); - - case 1: - if (!writer.writeMessage("val", val)) - return false; - - writer.incrementState(); - - } - - return true; - } - - /** {@inheritDoc} */ - @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) { - reader.setBuffer(buf); - - if (!reader.beforeMessageRead()) - return false; - - switch (reader.state()) { - case 0: - key = reader.readMessage("key"); - - if (!reader.isLastRead()) - return false; - - reader.incrementState(); - - case 1: - val = reader.readMessage("val"); - - if (!reader.isLastRead()) - return false; - - reader.incrementState(); - - } - - return true; - } - - /** {@inheritDoc} */ - @Override public byte directType() { - return 95; - } - - /** {@inheritDoc} */ - @Override public byte fieldsCount() { - return 2; - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(DataStreamerEntry.class, this); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java deleted file mode 100644 index d16887d..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerFuture.java +++ /dev/null @@ -1,60 +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.internal.processors.datastream; - -import org.apache.ignite.*; -import org.apache.ignite.internal.util.future.*; -import org.apache.ignite.internal.util.tostring.*; -import org.apache.ignite.internal.util.typedef.internal.*; - -/** - * Data streamer future. - */ -class DataStreamerFuture extends GridFutureAdapter<Object> { - /** */ - private static final long serialVersionUID = 0L; - - /** Data loader. */ - @GridToStringExclude - private DataStreamerImpl dataLdr; - - /** - * @param dataLdr Data streamer. - */ - DataStreamerFuture(DataStreamerImpl dataLdr) { - assert dataLdr != null; - - this.dataLdr = dataLdr; - } - - /** {@inheritDoc} */ - @Override public boolean cancel() throws IgniteCheckedException { - if (onCancelled()) { - dataLdr.closeEx(true); - - return true; - } - - return false; - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(DataStreamerFuture.class, this, super.toString()); - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java deleted file mode 100644 index 691c74f..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerImpl.java +++ /dev/null @@ -1,1405 +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.internal.processors.datastream; - -import org.apache.ignite.*; -import org.apache.ignite.cluster.*; -import org.apache.ignite.events.*; -import org.apache.ignite.internal.*; -import org.apache.ignite.internal.cluster.*; -import org.apache.ignite.internal.managers.communication.*; -import org.apache.ignite.internal.managers.deployment.*; -import org.apache.ignite.internal.managers.eventstorage.*; -import org.apache.ignite.internal.processors.affinity.*; -import org.apache.ignite.internal.processors.cache.*; -import org.apache.ignite.internal.processors.cache.distributed.dht.*; -import org.apache.ignite.internal.processors.cache.version.*; -import org.apache.ignite.internal.processors.cacheobject.*; -import org.apache.ignite.internal.processors.dr.*; -import org.apache.ignite.internal.util.*; -import org.apache.ignite.internal.util.future.*; -import org.apache.ignite.internal.util.lang.*; -import org.apache.ignite.internal.util.tostring.*; -import org.apache.ignite.internal.util.typedef.*; -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.lang.*; -import org.jdk8.backport.*; -import org.jetbrains.annotations.*; - -import java.util.*; -import java.util.Map.*; -import java.util.concurrent.*; -import java.util.concurrent.atomic.*; - -import static org.apache.ignite.events.EventType.*; -import static org.apache.ignite.internal.GridTopic.*; -import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*; - -/** - * Data streamer implementation. - */ -@SuppressWarnings("unchecked") -public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed { - /** Isolated updater. */ - private static final Updater ISOLATED_UPDATER = new IsolatedUpdater(); - - /** Cache updater. */ - private Updater<K, V> updater = ISOLATED_UPDATER; - - /** */ - private byte[] updaterBytes; - - /** Max remap count before issuing an error. */ - private static final int DFLT_MAX_REMAP_CNT = 32; - - /** Log reference. */ - private static final AtomicReference<IgniteLogger> logRef = new AtomicReference<>(); - - /** Logger. */ - private static IgniteLogger log; - - /** Cache name ({@code null} for default cache). */ - private final String cacheName; - - - /** Per-node buffer size. */ - @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized") - private int bufSize = DFLT_PER_NODE_BUFFER_SIZE; - - /** */ - private int parallelOps = DFLT_MAX_PARALLEL_OPS; - - /** */ - private long autoFlushFreq; - - /** Mapping. */ - @GridToStringInclude - private ConcurrentMap<UUID, Buffer> bufMappings = new ConcurrentHashMap8<>(); - - /** Discovery listener. */ - private final GridLocalEventListener discoLsnr; - - /** Context. */ - private final GridKernalContext ctx; - - /** */ - private final IgniteCacheObjectProcessor cacheObjProc; - - /** */ - private final CacheObjectContext cacheObjCtx; - - /** Communication topic for responses. */ - private final Object topic; - - /** */ - private byte[] topicBytes; - - /** {@code True} if data loader has been cancelled. */ - private volatile boolean cancelled; - - /** Active futures of this data loader. */ - @GridToStringInclude - private final Collection<IgniteInternalFuture<?>> activeFuts = new GridConcurrentHashSet<>(); - - /** Closure to remove from active futures. */ - @GridToStringExclude - private final IgniteInClosure<IgniteInternalFuture<?>> rmvActiveFut = new IgniteInClosure<IgniteInternalFuture<?>>() { - @Override public void apply(IgniteInternalFuture<?> t) { - boolean rmv = activeFuts.remove(t); - - assert rmv; - } - }; - - /** Job peer deploy aware. */ - private volatile GridPeerDeployAware jobPda; - - /** Deployment class. */ - private Class<?> depCls; - - /** Future to track loading finish. */ - private final GridFutureAdapter<?> fut; - - /** Public API future to track loading finish. */ - private final IgniteFuture<?> publicFut; - - /** Busy lock. */ - private final GridSpinBusyLock busyLock = new GridSpinBusyLock(); - - /** Closed flag. */ - private final AtomicBoolean closed = new AtomicBoolean(); - - /** */ - private volatile long lastFlushTime = U.currentTimeMillis(); - - /** */ - private final DelayQueue<DataStreamerImpl<K, V>> flushQ; - - /** */ - private boolean skipStore; - - /** */ - private int maxRemapCnt = DFLT_MAX_REMAP_CNT; - - /** Whether a warning at {@link DataStreamerImpl#allowOverwrite()} printed */ - private static boolean isWarningPrinted; - - /** - * @param ctx Grid kernal context. - * @param cacheName Cache name. - * @param flushQ Flush queue. - */ - public DataStreamerImpl( - final GridKernalContext ctx, - @Nullable final String cacheName, - DelayQueue<DataStreamerImpl<K, V>> flushQ - ) { - assert ctx != null; - - this.ctx = ctx; - this.cacheObjProc = ctx.cacheObjects(); - - if (log == null) - log = U.logger(ctx, logRef, DataStreamerImpl.class); - - ClusterNode node = F.first(ctx.grid().cluster().forCacheNodes(cacheName).nodes()); - - if (node == null) - throw new IllegalStateException("Cache doesn't exist: " + cacheName); - - this.cacheObjCtx = ctx.cacheObjects().contextForCache(node, cacheName); - this.cacheName = cacheName; - this.flushQ = flushQ; - - discoLsnr = new GridLocalEventListener() { - @Override public void onEvent(Event evt) { - assert evt.type() == EVT_NODE_FAILED || evt.type() == EVT_NODE_LEFT; - - DiscoveryEvent discoEvt = (DiscoveryEvent)evt; - - UUID id = discoEvt.eventNode().id(); - - // Remap regular mappings. - final Buffer buf = bufMappings.remove(id); - - if (buf != null) { - // Only async notification is possible since - // discovery thread may be trapped otherwise. - ctx.closure().callLocalSafe( - new Callable<Object>() { - @Override public Object call() throws Exception { - buf.onNodeLeft(); - - return null; - } - }, - true /* system pool */ - ); - } - } - }; - - ctx.event().addLocalEventListener(discoLsnr, EVT_NODE_FAILED, EVT_NODE_LEFT); - - // Generate unique topic for this loader. - topic = TOPIC_DATASTREAM.topic(IgniteUuid.fromUuid(ctx.localNodeId())); - - ctx.io().addMessageListener(topic, new GridMessageListener() { - @Override public void onMessage(UUID nodeId, Object msg) { - assert msg instanceof DataStreamerResponse; - - DataStreamerResponse res = (DataStreamerResponse)msg; - - if (log.isDebugEnabled()) - log.debug("Received data load response: " + res); - - Buffer buf = bufMappings.get(nodeId); - - if (buf != null) - buf.onResponse(res); - - else if (log.isDebugEnabled()) - log.debug("Ignoring response since node has left [nodeId=" + nodeId + ", "); - } - }); - - if (log.isDebugEnabled()) - log.debug("Added response listener within topic: " + topic); - - fut = new DataStreamerFuture(this); - - publicFut = new IgniteFutureImpl<>(fut); - } - - /** - * @return Cache object context. - */ - public CacheObjectContext cacheObjectContext() { - return cacheObjCtx; - } - - /** - * Enters busy lock. - */ - private void enterBusy() { - if (!busyLock.enterBusy()) - throw new IllegalStateException("Data streamer has been closed."); - } - - /** - * Leaves busy lock. - */ - private void leaveBusy() { - busyLock.leaveBusy(); - } - - /** {@inheritDoc} */ - @Override public IgniteFuture<?> future() { - return publicFut; - } - - /** - * @return Internal future. - */ - public IgniteInternalFuture<?> internalFuture() { - return fut; - } - - /** {@inheritDoc} */ - @Override public void deployClass(Class<?> depCls) { - this.depCls = depCls; - } - - /** {@inheritDoc} */ - @Override public void updater(Updater<K, V> updater) { - A.notNull(updater, "updater"); - - this.updater = updater; - } - - /** {@inheritDoc} */ - @Override public boolean allowOverwrite() { - boolean allow = updater != ISOLATED_UPDATER; - - if (!allow && !isWarningPrinted) { - synchronized (this) { - if (!isWarningPrinted) { - log.warning("Data streamer will not overwrite existing cache entries for better performance " + - "(to change, set allowOverwrite to true)"); - - isWarningPrinted = true; - } - } - } - - return allow; - } - - /** {@inheritDoc} */ - @Override public void allowOverwrite(boolean allow) { - if (allow == allowOverwrite()) - return; - - ClusterNode node = F.first(ctx.grid().cluster().forCacheNodes(cacheName).nodes()); - - if (node == null) - throw new IgniteException("Failed to get node for cache: " + cacheName); - - updater = allow ? DataStreamerCacheUpdaters.<K, V>individual() : ISOLATED_UPDATER; - } - - /** {@inheritDoc} */ - @Override public boolean skipStore() { - return skipStore; - } - - /** {@inheritDoc} */ - @Override public void skipStore(boolean skipStore) { - this.skipStore = skipStore; - } - - /** {@inheritDoc} */ - @Override @Nullable public String cacheName() { - return cacheName; - } - - /** {@inheritDoc} */ - @Override public int perNodeBufferSize() { - return bufSize; - } - - /** {@inheritDoc} */ - @Override public void perNodeBufferSize(int bufSize) { - A.ensure(bufSize > 0, "bufSize > 0"); - - this.bufSize = bufSize; - } - - /** {@inheritDoc} */ - @Override public int perNodeParallelOperations() { - return parallelOps; - } - - /** {@inheritDoc} */ - @Override public void perNodeParallelOperations(int parallelOps) { - this.parallelOps = parallelOps; - } - - /** {@inheritDoc} */ - @Override public long autoFlushFrequency() { - return autoFlushFreq; - } - - /** {@inheritDoc} */ - @Override public void autoFlushFrequency(long autoFlushFreq) { - A.ensure(autoFlushFreq >= 0, "autoFlushFreq >= 0"); - - long old = this.autoFlushFreq; - - if (autoFlushFreq != old) { - this.autoFlushFreq = autoFlushFreq; - - if (autoFlushFreq != 0 && old == 0) - flushQ.add(this); - else if (autoFlushFreq == 0) - flushQ.remove(this); - } - } - - /** {@inheritDoc} */ - @Override public IgniteFuture<?> addData(Map<K, V> entries) throws IllegalStateException { - A.notNull(entries, "entries"); - - return addData(entries.entrySet()); - } - - /** {@inheritDoc} */ - @Override public IgniteFuture<?> addData(Collection<? extends Map.Entry<K, V>> entries) { - A.notEmpty(entries, "entries"); - - enterBusy(); - - try { - GridFutureAdapter<Object> resFut = new GridFutureAdapter<>(); - - resFut.listen(rmvActiveFut); - - activeFuts.add(resFut); - - Collection<KeyCacheObject> keys = null; - - if (entries.size() > 1) { - keys = new GridConcurrentHashSet<>(entries.size(), U.capacity(entries.size()), 1); - - for (Map.Entry<K, V> entry : entries) - keys.add(cacheObjProc.toCacheKeyObject(cacheObjCtx, entry.getKey(), true)); - } - - Collection<? extends DataStreamerEntry> entries0 = F.viewReadOnly(entries, new C1<Entry<K, V>, DataStreamerEntry>() { - @Override public DataStreamerEntry apply(Entry<K, V> e) { - KeyCacheObject key = cacheObjProc.toCacheKeyObject(cacheObjCtx, e.getKey(), true); - CacheObject val = cacheObjProc.toCacheObject(cacheObjCtx, e.getValue(), true); - - return new DataStreamerEntry(key, val); - } - }); - - load0(entries0, resFut, keys, 0); - - return new IgniteFutureImpl<>(resFut); - } - catch (IgniteException e) { - return new IgniteFinishedFutureImpl<>(e); - } - finally { - leaveBusy(); - } - } - - /** - * @param key Key. - * @param val Value. - * @return Future. - */ - public IgniteFuture<?> addDataInternal(KeyCacheObject key, CacheObject val) { - return addDataInternal(Collections.singleton(new DataStreamerEntry(key, val))); - } - - /** - * @param key Key. - * @return Future. - */ - public IgniteFuture<?> removeDataInternal(KeyCacheObject key) { - return addDataInternal(Collections.singleton(new DataStreamerEntry(key, null))); - } - - /** - * @param entries Entries. - * @return Future. - */ - public IgniteFuture<?> addDataInternal(Collection<? extends DataStreamerEntry> entries) { - enterBusy(); - - GridFutureAdapter<Object> resFut = new GridFutureAdapter<>(); - - try { - resFut.listen(rmvActiveFut); - - activeFuts.add(resFut); - - Collection<KeyCacheObject> keys = null; - - if (entries.size() > 1) { - keys = new GridConcurrentHashSet<>(entries.size(), U.capacity(entries.size()), 1); - - for (DataStreamerEntry entry : entries) - keys.add(entry.getKey()); - } - - load0(entries, resFut, keys, 0); - - return new IgniteFutureImpl<>(resFut); - } - catch (Throwable e) { - resFut.onDone(e); - - if (e instanceof Error) - throw e; - - return new IgniteFinishedFutureImpl<>(e); - } - finally { - leaveBusy(); - } - } - - /** {@inheritDoc} */ - @Override public IgniteFuture<?> addData(Map.Entry<K, V> entry) { - A.notNull(entry, "entry"); - - return addData(F.asList(entry)); - } - - /** {@inheritDoc} */ - @Override public IgniteFuture<?> addData(K key, V val) { - A.notNull(key, "key"); - - KeyCacheObject key0 = cacheObjProc.toCacheKeyObject(cacheObjCtx, key, true); - CacheObject val0 = cacheObjProc.toCacheObject(cacheObjCtx, val, true); - - return addDataInternal(Collections.singleton(new DataStreamerEntry(key0, val0))); - } - - /** {@inheritDoc} */ - @Override public IgniteFuture<?> removeData(K key) { - return addData(key, null); - } - - /** - * @param entries Entries. - * @param resFut Result future. - * @param activeKeys Active keys. - * @param remaps Remaps count. - */ - private void load0( - Collection<? extends DataStreamerEntry> entries, - final GridFutureAdapter<Object> resFut, - @Nullable final Collection<KeyCacheObject> activeKeys, - final int remaps - ) { - assert entries != null; - - Map<ClusterNode, Collection<DataStreamerEntry>> mappings = new HashMap<>(); - - boolean initPda = ctx.deploy().enabled() && jobPda == null; - - for (DataStreamerEntry entry : entries) { - List<ClusterNode> nodes; - - try { - KeyCacheObject key = entry.getKey(); - - assert key != null; - - if (initPda) { - jobPda = new DataStreamerPda(key.value(cacheObjCtx, false), - entry.getValue() != null ? entry.getValue().value(cacheObjCtx, false) : null, - updater); - - initPda = false; - } - - nodes = nodes(key); - } - catch (IgniteCheckedException e) { - resFut.onDone(e); - - return; - } - - if (F.isEmpty(nodes)) { - resFut.onDone(new ClusterTopologyException("Failed to map key to node " + - "(no nodes with cache found in topology) [infos=" + entries.size() + - ", cacheName=" + cacheName + ']')); - - return; - } - - for (ClusterNode node : nodes) { - Collection<DataStreamerEntry> col = mappings.get(node); - - if (col == null) - mappings.put(node, col = new ArrayList<>()); - - col.add(entry); - } - } - - for (final Map.Entry<ClusterNode, Collection<DataStreamerEntry>> e : mappings.entrySet()) { - final UUID nodeId = e.getKey().id(); - - Buffer buf = bufMappings.get(nodeId); - - if (buf == null) { - Buffer old = bufMappings.putIfAbsent(nodeId, buf = new Buffer(e.getKey())); - - if (old != null) - buf = old; - } - - final Collection<DataStreamerEntry> entriesForNode = e.getValue(); - - IgniteInClosure<IgniteInternalFuture<?>> lsnr = new IgniteInClosure<IgniteInternalFuture<?>>() { - @Override public void apply(IgniteInternalFuture<?> t) { - try { - t.get(); - - if (activeKeys != null) { - for (DataStreamerEntry e : entriesForNode) - activeKeys.remove(e.getKey()); - - if (activeKeys.isEmpty()) - resFut.onDone(); - } - else { - assert entriesForNode.size() == 1; - - // That has been a single key, - // so complete result future right away. - resFut.onDone(); - } - } - catch (IgniteCheckedException e1) { - if (log.isDebugEnabled()) - log.debug("Future finished with error [nodeId=" + nodeId + ", err=" + e1 + ']'); - - if (cancelled) { - resFut.onDone(new IgniteCheckedException("Data streamer has been cancelled: " + - DataStreamerImpl.this, e1)); - } - else if (remaps + 1 > maxRemapCnt) { - resFut.onDone(new IgniteCheckedException("Failed to finish operation (too many remaps): " - + remaps), e1); - } - else - load0(entriesForNode, resFut, activeKeys, remaps + 1); - } - } - }; - - GridFutureAdapter<?> f; - - try { - f = buf.update(entriesForNode, lsnr); - } - catch (IgniteInterruptedCheckedException e1) { - resFut.onDone(e1); - - return; - } - - if (ctx.discovery().node(nodeId) == null) { - if (bufMappings.remove(nodeId, buf)) - buf.onNodeLeft(); - - if (f != null) - f.onDone(new ClusterTopologyCheckedException("Failed to wait for request completion " + - "(node has left): " + nodeId)); - } - } - } - - /** - * @param key Key to map. - * @return Nodes to send requests to. - * @throws IgniteCheckedException If failed. - */ - private List<ClusterNode> nodes(KeyCacheObject key) throws IgniteCheckedException { - GridAffinityProcessor aff = ctx.affinity(); - - return !allowOverwrite() ? aff.mapKeyToPrimaryAndBackups(cacheName, key) : - Collections.singletonList(aff.mapKeyToNode(cacheName, key)); - } - - /** - * Performs flush. - * - * @throws IgniteCheckedException If failed. - */ - private void doFlush() throws IgniteCheckedException { - lastFlushTime = U.currentTimeMillis(); - - List<IgniteInternalFuture> activeFuts0 = null; - - int doneCnt = 0; - - for (IgniteInternalFuture<?> f : activeFuts) { - if (!f.isDone()) { - if (activeFuts0 == null) - activeFuts0 = new ArrayList<>((int)(activeFuts.size() * 1.2)); - - activeFuts0.add(f); - } - else { - f.get(); - - doneCnt++; - } - } - - if (activeFuts0 == null || activeFuts0.isEmpty()) - return; - - while (true) { - Queue<IgniteInternalFuture<?>> q = null; - - for (Buffer buf : bufMappings.values()) { - IgniteInternalFuture<?> flushFut = buf.flush(); - - if (flushFut != null) { - if (q == null) - q = new ArrayDeque<>(bufMappings.size() * 2); - - q.add(flushFut); - } - } - - if (q != null) { - assert !q.isEmpty(); - - boolean err = false; - - for (IgniteInternalFuture fut = q.poll(); fut != null; fut = q.poll()) { - try { - fut.get(); - } - catch (IgniteCheckedException e) { - if (log.isDebugEnabled()) - log.debug("Failed to flush buffer: " + e); - - err = true; - } - } - - if (err) - // Remaps needed - flush buffers. - continue; - } - - doneCnt = 0; - - for (int i = 0; i < activeFuts0.size(); i++) { - IgniteInternalFuture f = activeFuts0.get(i); - - if (f == null) - doneCnt++; - else if (f.isDone()) { - f.get(); - - doneCnt++; - - activeFuts0.set(i, null); - } - else - break; - } - - if (doneCnt == activeFuts0.size()) - return; - } - } - - /** {@inheritDoc} */ - @SuppressWarnings("ForLoopReplaceableByForEach") - @Override public void flush() throws IgniteException { - enterBusy(); - - try { - doFlush(); - } - catch (IgniteCheckedException e) { - throw U.convertException(e); - } - finally { - leaveBusy(); - } - } - - /** - * Flushes every internal buffer if buffer was flushed before passed in - * threshold. - * <p> - * Does not wait for result and does not fail on errors assuming that this method - * should be called periodically. - */ - @Override public void tryFlush() throws IgniteInterruptedException { - if (!busyLock.enterBusy()) - return; - - try { - for (Buffer buf : bufMappings.values()) - buf.flush(); - - lastFlushTime = U.currentTimeMillis(); - } - catch (IgniteInterruptedCheckedException e) { - throw U.convertException(e); - } - finally { - leaveBusy(); - } - } - - /** - * @param cancel {@code True} to close with cancellation. - * @throws IgniteException If failed. - */ - @Override public void close(boolean cancel) throws IgniteException { - try { - closeEx(cancel); - } - catch (IgniteCheckedException e) { - throw U.convertException(e); - } - } - - /** - * @param cancel {@code True} to close with cancellation. - * @throws IgniteCheckedException If failed. - */ - public void closeEx(boolean cancel) throws IgniteCheckedException { - if (!closed.compareAndSet(false, true)) - return; - - busyLock.block(); - - if (log.isDebugEnabled()) - log.debug("Closing data streamer [ldr=" + this + ", cancel=" + cancel + ']'); - - IgniteCheckedException e = null; - - try { - // Assuming that no methods are called on this loader after this method is called. - if (cancel) { - cancelled = true; - - for (Buffer buf : bufMappings.values()) - buf.cancelAll(); - } - else - doFlush(); - - ctx.event().removeLocalEventListener(discoLsnr); - - ctx.io().removeMessageListener(topic); - } - catch (IgniteCheckedException e0) { - e = e0; - } - - fut.onDone(null, e); - - if (e != null) - throw e; - } - - /** - * @return {@code true} If the loader is closed. - */ - boolean isClosed() { - return fut.isDone(); - } - - /** {@inheritDoc} */ - @Override public void close() throws IgniteException { - close(false); - } - - /** - * @return Max remap count. - */ - public int maxRemapCount() { - return maxRemapCnt; - } - - /** - * @param maxRemapCnt New max remap count. - */ - public void maxRemapCount(int maxRemapCnt) { - this.maxRemapCnt = maxRemapCnt; - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(DataStreamerImpl.class, this); - } - - /** {@inheritDoc} */ - @Override public long getDelay(TimeUnit unit) { - return unit.convert(nextFlushTime() - U.currentTimeMillis(), TimeUnit.MILLISECONDS); - } - - /** - * @return Next flush time. - */ - private long nextFlushTime() { - return lastFlushTime + autoFlushFreq; - } - - /** {@inheritDoc} */ - @Override public int compareTo(Delayed o) { - return nextFlushTime() > ((DataStreamerImpl)o).nextFlushTime() ? 1 : -1; - } - - /** - * - */ - private class Buffer { - /** Node. */ - private final ClusterNode node; - - /** Active futures. */ - private final Collection<IgniteInternalFuture<Object>> locFuts; - - /** Buffered entries. */ - private List<DataStreamerEntry> entries; - - /** */ - @GridToStringExclude - private GridFutureAdapter<Object> curFut; - - /** Local node flag. */ - private final boolean isLocNode; - - /** ID generator. */ - private final AtomicLong idGen = new AtomicLong(); - - /** Active futures. */ - private final ConcurrentMap<Long, GridFutureAdapter<Object>> reqs; - - /** */ - private final Semaphore sem; - - /** Closure to signal on task finish. */ - @GridToStringExclude - private final IgniteInClosure<IgniteInternalFuture<Object>> signalC = new IgniteInClosure<IgniteInternalFuture<Object>>() { - @Override public void apply(IgniteInternalFuture<Object> t) { - signalTaskFinished(t); - } - }; - - /** - * @param node Node. - */ - Buffer(ClusterNode node) { - assert node != null; - - this.node = node; - - locFuts = new GridConcurrentHashSet<>(); - reqs = new ConcurrentHashMap8<>(); - - // Cache local node flag. - isLocNode = node.equals(ctx.discovery().localNode()); - - entries = newEntries(); - curFut = new GridFutureAdapter<>(); - curFut.listen(signalC); - - sem = new Semaphore(parallelOps); - } - - /** - * @param newEntries Infos. - * @param lsnr Listener for the operation future. - * @throws IgniteInterruptedCheckedException If failed. - * @return Future for operation. - */ - @Nullable GridFutureAdapter<?> update(Iterable<DataStreamerEntry> newEntries, - IgniteInClosure<IgniteInternalFuture<?>> lsnr) throws IgniteInterruptedCheckedException { - List<DataStreamerEntry> entries0 = null; - GridFutureAdapter<Object> curFut0; - - synchronized (this) { - curFut0 = curFut; - - curFut0.listen(lsnr); - - for (DataStreamerEntry entry : newEntries) - entries.add(entry); - - if (entries.size() >= bufSize) { - entries0 = entries; - - entries = newEntries(); - curFut = new GridFutureAdapter<>(); - curFut.listen(signalC); - } - } - - if (entries0 != null) { - submit(entries0, curFut0); - - if (cancelled) - curFut0.onDone(new IgniteCheckedException("Data streamer has been cancelled: " + DataStreamerImpl.this)); - } - - return curFut0; - } - - /** - * @return Fresh collection with some space for outgrowth. - */ - private List<DataStreamerEntry> newEntries() { - return new ArrayList<>((int)(bufSize * 1.2)); - } - - /** - * @return Future if any submitted. - * - * @throws IgniteInterruptedCheckedException If thread has been interrupted. - */ - @Nullable IgniteInternalFuture<?> flush() throws IgniteInterruptedCheckedException { - List<DataStreamerEntry> entries0 = null; - GridFutureAdapter<Object> curFut0 = null; - - synchronized (this) { - if (!entries.isEmpty()) { - entries0 = entries; - curFut0 = curFut; - - entries = newEntries(); - curFut = new GridFutureAdapter<>(); - curFut.listen(signalC); - } - } - - if (entries0 != null) - submit(entries0, curFut0); - - // Create compound future for this flush. - GridCompoundFuture<Object, Object> res = null; - - for (IgniteInternalFuture<Object> f : locFuts) { - if (res == null) - res = new GridCompoundFuture<>(); - - res.add(f); - } - - for (IgniteInternalFuture<Object> f : reqs.values()) { - if (res == null) - res = new GridCompoundFuture<>(); - - res.add(f); - } - - if (res != null) - res.markInitialized(); - - return res; - } - - /** - * Increments active tasks count. - * - * @throws IgniteInterruptedCheckedException If thread has been interrupted. - */ - private void incrementActiveTasks() throws IgniteInterruptedCheckedException { - U.acquire(sem); - } - - /** - * @param f Future that finished. - */ - private void signalTaskFinished(IgniteInternalFuture<Object> f) { - assert f != null; - - sem.release(); - } - - /** - * @param entries Entries to submit. - * @param curFut Current future. - * @throws IgniteInterruptedCheckedException If interrupted. - */ - private void submit(final Collection<DataStreamerEntry> entries, final GridFutureAdapter<Object> curFut) - throws IgniteInterruptedCheckedException { - assert entries != null; - assert !entries.isEmpty(); - assert curFut != null; - - incrementActiveTasks(); - - IgniteInternalFuture<Object> fut; - - if (isLocNode) { - fut = ctx.closure().callLocalSafe( - new DataStreamerUpdateJob(ctx, log, cacheName, entries, false, skipStore, updater), false); - - locFuts.add(fut); - - fut.listen(new IgniteInClosure<IgniteInternalFuture<Object>>() { - @Override public void apply(IgniteInternalFuture<Object> t) { - try { - boolean rmv = locFuts.remove(t); - - assert rmv; - - curFut.onDone(t.get()); - } - catch (IgniteCheckedException e) { - curFut.onDone(e); - } - } - }); - } - else { - try { - for (DataStreamerEntry e : entries) { - e.getKey().prepareMarshal(cacheObjCtx); - - CacheObject val = e.getValue(); - - if (val != null) - val.prepareMarshal(cacheObjCtx); - } - - if (updaterBytes == null) { - assert updater != null; - - updaterBytes = ctx.config().getMarshaller().marshal(updater); - } - - if (topicBytes == null) - topicBytes = ctx.config().getMarshaller().marshal(topic); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to marshal (request will not be sent).", e); - - return; - } - - GridDeployment dep = null; - GridPeerDeployAware jobPda0 = null; - - if (ctx.deploy().enabled()) { - try { - jobPda0 = jobPda; - - assert jobPda0 != null; - - dep = ctx.deploy().deploy(jobPda0.deployClass(), jobPda0.classLoader()); - - GridCacheAdapter<Object, Object> cache = ctx.cache().internalCache(cacheName); - - if (cache != null) - cache.context().deploy().onEnter(); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to deploy class (request will not be sent): " + jobPda0.deployClass(), e); - - return; - } - - if (dep == null) - U.warn(log, "Failed to deploy class (request will be sent): " + jobPda0.deployClass()); - } - - long reqId = idGen.incrementAndGet(); - - fut = curFut; - - reqs.put(reqId, (GridFutureAdapter<Object>)fut); - - DataStreamerRequest req = new DataStreamerRequest( - reqId, - topicBytes, - cacheName, - updaterBytes, - entries, - true, - skipStore, - dep != null ? dep.deployMode() : null, - dep != null ? jobPda0.deployClass().getName() : null, - dep != null ? dep.userVersion() : null, - dep != null ? dep.participants() : null, - dep != null ? dep.classLoaderId() : null, - dep == null); - - try { - ctx.io().send(node, TOPIC_DATASTREAM, req, PUBLIC_POOL); - - if (log.isDebugEnabled()) - log.debug("Sent request to node [nodeId=" + node.id() + ", req=" + req + ']'); - } - catch (IgniteCheckedException e) { - if (ctx.discovery().alive(node) && ctx.discovery().pingNode(node.id())) - ((GridFutureAdapter<Object>)fut).onDone(e); - else - ((GridFutureAdapter<Object>)fut).onDone(new ClusterTopologyCheckedException("Failed to send " + - "request (node has left): " + node.id())); - } - } - } - - /** - * - */ - void onNodeLeft() { - assert !isLocNode; - assert bufMappings.get(node.id()) != this; - - if (log.isDebugEnabled()) - log.debug("Forcibly completing futures (node has left): " + node.id()); - - Exception e = new ClusterTopologyCheckedException("Failed to wait for request completion " + - "(node has left): " + node.id()); - - for (GridFutureAdapter<Object> f : reqs.values()) - f.onDone(e); - - // Make sure to complete current future. - GridFutureAdapter<Object> curFut0; - - synchronized (this) { - curFut0 = curFut; - } - - curFut0.onDone(e); - } - - /** - * @param res Response. - */ - void onResponse(DataStreamerResponse res) { - if (log.isDebugEnabled()) - log.debug("Received data load response: " + res); - - GridFutureAdapter<?> f = reqs.remove(res.requestId()); - - if (f == null) { - if (log.isDebugEnabled()) - log.debug("Future for request has not been found: " + res.requestId()); - - return; - } - - Throwable err = null; - - byte[] errBytes = res.errorBytes(); - - if (errBytes != null) { - try { - GridPeerDeployAware jobPda0 = jobPda; - - err = ctx.config().getMarshaller().unmarshal( - errBytes, - jobPda0 != null ? jobPda0.classLoader() : U.gridClassLoader()); - } - catch (IgniteCheckedException e) { - f.onDone(null, new IgniteCheckedException("Failed to unmarshal response.", e)); - - return; - } - } - - f.onDone(null, err); - - if (log.isDebugEnabled()) - log.debug("Finished future [fut=" + f + ", reqId=" + res.requestId() + ", err=" + err + ']'); - } - - /** - * - */ - void cancelAll() { - IgniteCheckedException err = new IgniteCheckedException("Data streamer has been cancelled: " + DataStreamerImpl.this); - - for (IgniteInternalFuture<?> f : locFuts) { - try { - f.cancel(); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to cancel mini-future.", e); - } - } - - for (GridFutureAdapter<?> f : reqs.values()) - f.onDone(err); - } - - /** {@inheritDoc} */ - @Override public String toString() { - int size; - - synchronized (this) { - size = entries.size(); - } - - return S.toString(Buffer.class, this, - "entriesCnt", size, - "locFutsSize", locFuts.size(), - "reqsSize", reqs.size()); - } - } - - /** - * Data streamer peer-deploy aware. - */ - private class DataStreamerPda implements GridPeerDeployAware { - /** */ - private static final long serialVersionUID = 0L; - - /** Deploy class. */ - private Class<?> cls; - - /** Class loader. */ - private ClassLoader ldr; - - /** Collection of objects to detect deploy class and class loader. */ - private Collection<Object> objs; - - /** - * Constructs data streamer peer-deploy aware. - * - * @param objs Collection of objects to detect deploy class and class loader. - */ - private DataStreamerPda(Object... objs) { - this.objs = Arrays.asList(objs); - } - - /** {@inheritDoc} */ - @Override public Class<?> deployClass() { - if (cls == null) { - Class<?> cls0 = null; - - if (depCls != null) - cls0 = depCls; - else { - for (Iterator<Object> it = objs.iterator(); (cls0 == null || U.isJdk(cls0)) && it.hasNext();) { - Object o = it.next(); - - if (o != null) - cls0 = U.detectClass(o); - } - - if (cls0 == null || U.isJdk(cls0)) - cls0 = DataStreamerImpl.class; - } - - assert cls0 != null : "Failed to detect deploy class [objs=" + objs + ']'; - - cls = cls0; - } - - return cls; - } - - /** {@inheritDoc} */ - @Override public ClassLoader classLoader() { - if (ldr == null) { - ClassLoader ldr0 = deployClass().getClassLoader(); - - // Safety. - if (ldr0 == null) - ldr0 = U.gridClassLoader(); - - assert ldr0 != null : "Failed to detect classloader [objs=" + objs + ']'; - - ldr = ldr0; - } - - return ldr; - } - } - - /** - * Isolated updater which only loads entry initial value. - */ - private static class IsolatedUpdater implements Updater<KeyCacheObject, CacheObject>, - DataStreamerCacheUpdaters.InternalUpdater { - /** */ - private static final long serialVersionUID = 0L; - - /** {@inheritDoc} */ - @Override public void update(IgniteCache<KeyCacheObject, CacheObject> cache, - Collection<Map.Entry<KeyCacheObject, CacheObject>> entries) { - IgniteCacheProxy<KeyCacheObject, CacheObject> proxy = (IgniteCacheProxy<KeyCacheObject, CacheObject>)cache; - - GridCacheAdapter<KeyCacheObject, CacheObject> internalCache = proxy.context().cache(); - - if (internalCache.isNear()) - internalCache = internalCache.context().near().dht(); - - GridCacheContext cctx = internalCache.context(); - - long topVer = cctx.affinity().affinityTopologyVersion(); - - GridCacheVersion ver = cctx.versions().next(topVer); - - for (Map.Entry<KeyCacheObject, CacheObject> e : entries) { - try { - e.getKey().finishUnmarshal(cctx.cacheObjectContext(), cctx.deploy().globalLoader()); - - GridCacheEntryEx entry = internalCache.entryEx(e.getKey(), topVer); - - entry.unswap(true, false); - - entry.initialValue(e.getValue(), - ver, - CU.TTL_ETERNAL, - CU.EXPIRE_TIME_ETERNAL, - false, - topVer, - GridDrType.DR_LOAD); - - cctx.evicts().touch(entry, topVer); - } - catch (GridDhtInvalidPartitionException | GridCacheEntryRemovedException ignored) { - // No-op. - } - catch (IgniteCheckedException ex) { - IgniteLogger log = cache.unwrap(Ignite.class).log(); - - U.error(log, "Failed to set initial value for cache entry: " + e, ex); - } - } - } - } -} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/60a1a481/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java deleted file mode 100644 index ee12f7a..0000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastream/DataStreamerProcessor.java +++ /dev/null @@ -1,307 +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.internal.processors.datastream; - -import org.apache.ignite.*; -import org.apache.ignite.internal.*; -import org.apache.ignite.internal.managers.communication.*; -import org.apache.ignite.internal.managers.deployment.*; -import org.apache.ignite.internal.processors.*; -import org.apache.ignite.internal.util.*; -import org.apache.ignite.internal.util.typedef.*; -import org.apache.ignite.internal.util.typedef.internal.*; -import org.apache.ignite.internal.util.worker.*; -import org.apache.ignite.marshaller.*; -import org.apache.ignite.thread.*; -import org.jetbrains.annotations.*; - -import java.util.*; -import java.util.concurrent.*; - -import static org.apache.ignite.internal.GridTopic.*; -import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*; - -/** - * - */ -public class DataStreamerProcessor<K, V> extends GridProcessorAdapter { - /** Loaders map (access is not supposed to be highly concurrent). */ - private Collection<DataStreamerImpl> ldrs = new GridConcurrentHashSet<>(); - - /** Busy lock. */ - private final GridSpinBusyLock busyLock = new GridSpinBusyLock(); - - /** Flushing thread. */ - private Thread flusher; - - /** */ - private final DelayQueue<DataStreamerImpl<K, V>> flushQ = new DelayQueue<>(); - - /** Marshaller. */ - private final Marshaller marsh; - - /** - * @param ctx Kernal context. - */ - public DataStreamerProcessor(GridKernalContext ctx) { - super(ctx); - - ctx.io().addMessageListener(TOPIC_DATASTREAM, new GridMessageListener() { - @Override public void onMessage(UUID nodeId, Object msg) { - assert msg instanceof DataStreamerRequest; - - processRequest(nodeId, (DataStreamerRequest)msg); - } - }); - - marsh = ctx.config().getMarshaller(); - } - - /** {@inheritDoc} */ - @Override public void start() throws IgniteCheckedException { - if (ctx.config().isDaemon()) - return; - - flusher = new IgniteThread(new GridWorker(ctx.gridName(), "grid-data-loader-flusher", log) { - @Override protected void body() throws InterruptedException { - while (!isCancelled()) { - DataStreamerImpl<K, V> ldr = flushQ.take(); - - if (!busyLock.enterBusy()) - return; - - try { - if (ldr.isClosed()) - continue; - - ldr.tryFlush(); - - flushQ.offer(ldr); - } - finally { - busyLock.leaveBusy(); - } - } - } - }); - - flusher.start(); - - if (log.isDebugEnabled()) - log.debug("Started data streamer processor."); - } - - /** {@inheritDoc} */ - @Override public void onKernalStop(boolean cancel) { - if (ctx.config().isDaemon()) - return; - - ctx.io().removeMessageListener(TOPIC_DATASTREAM); - - busyLock.block(); - - U.interrupt(flusher); - U.join(flusher, log); - - for (DataStreamerImpl<?, ?> ldr : ldrs) { - if (log.isDebugEnabled()) - log.debug("Closing active data streamer on grid stop [ldr=" + ldr + ", cancel=" + cancel + ']'); - - try { - ldr.closeEx(cancel); - } - catch (IgniteInterruptedCheckedException e) { - U.warn(log, "Interrupted while waiting for completion of the data streamer: " + ldr, e); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to close data streamer: " + ldr, e); - } - } - - if (log.isDebugEnabled()) - log.debug("Stopped data streamer processor."); - } - - /** - * @param cacheName Cache name ({@code null} for default cache). - * @return Data loader. - */ - public DataStreamerImpl<K, V> dataStreamer(@Nullable String cacheName) { - if (!busyLock.enterBusy()) - throw new IllegalStateException("Failed to create data streamer (grid is stopping)."); - - try { - final DataStreamerImpl<K, V> ldr = new DataStreamerImpl<>(ctx, cacheName, flushQ); - - ldrs.add(ldr); - - ldr.internalFuture().listen(new CI1<IgniteInternalFuture<?>>() { - @Override public void apply(IgniteInternalFuture<?> f) { - boolean b = ldrs.remove(ldr); - - assert b : "Loader has not been added to set: " + ldr; - - if (log.isDebugEnabled()) - log.debug("Loader has been completed: " + ldr); - } - }); - - return ldr; - } - finally { - busyLock.leaveBusy(); - } - } - - /** - * @param nodeId Sender ID. - * @param req Request. - */ - private void processRequest(UUID nodeId, DataStreamerRequest req) { - if (!busyLock.enterBusy()) { - if (log.isDebugEnabled()) - log.debug("Ignoring data load request (node is stopping): " + req); - - return; - } - - try { - if (log.isDebugEnabled()) - log.debug("Processing data load request: " + req); - - Object topic; - - try { - topic = marsh.unmarshal(req.responseTopicBytes(), null); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to unmarshal topic from request: " + req, e); - - return; - } - - ClassLoader clsLdr; - - if (req.forceLocalDeployment()) - clsLdr = U.gridClassLoader(); - else { - GridDeployment dep = ctx.deploy().getGlobalDeployment( - req.deploymentMode(), - req.sampleClassName(), - req.sampleClassName(), - req.userVersion(), - nodeId, - req.classLoaderId(), - req.participants(), - null); - - if (dep == null) { - sendResponse(nodeId, - topic, - req.requestId(), - new IgniteCheckedException("Failed to get deployment for request [sndId=" + nodeId + - ", req=" + req + ']'), - false); - - return; - } - - clsLdr = dep.classLoader(); - } - - IgniteDataStreamer.Updater<K, V> updater; - - try { - updater = marsh.unmarshal(req.updaterBytes(), clsLdr); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to unmarshal message [nodeId=" + nodeId + ", req=" + req + ']', e); - - sendResponse(nodeId, topic, req.requestId(), e, false); - - return; - } - - Collection<DataStreamerEntry> col = req.entries(); - - DataStreamerUpdateJob job = new DataStreamerUpdateJob(ctx, - log, - req.cacheName(), - col, - req.ignoreDeploymentOwnership(), - req.skipStore(), - updater); - - Exception err = null; - - try { - job.call(); - } - catch (Exception e) { - U.error(log, "Failed to finish update job.", e); - - err = e; - } - - sendResponse(nodeId, topic, req.requestId(), err, req.forceLocalDeployment()); - } - finally { - busyLock.leaveBusy(); - } - } - - /** - * @param nodeId Node ID. - * @param resTopic Response topic. - * @param reqId Request ID. - * @param err Error. - * @param forceLocDep Force local deployment. - */ - private void sendResponse(UUID nodeId, Object resTopic, long reqId, @Nullable Throwable err, - boolean forceLocDep) { - byte[] errBytes; - - try { - errBytes = err != null ? marsh.marshal(err) : null; - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to marshal message.", e); - - return; - } - - DataStreamerResponse res = new DataStreamerResponse(reqId, errBytes, forceLocDep); - - try { - ctx.io().send(nodeId, resTopic, res, PUBLIC_POOL); - } - catch (IgniteCheckedException e) { - if (ctx.discovery().alive(nodeId)) - U.error(log, "Failed to respond to node [nodeId=" + nodeId + ", res=" + res + ']', e); - else if (log.isDebugEnabled()) - log.debug("Node has left the grid: " + nodeId); - } - } - - /** {@inheritDoc} */ - @Override public void printMemoryStats() { - X.println(">>>"); - X.println(">>> Data streamer processor memory stats [grid=" + ctx.gridName() + ']'); - X.println(">>> ldrsSize: " + ldrs.size()); - } -}
