WIP on tasks.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d6e5028d Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d6e5028d Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d6e5028d Branch: refs/heads/ignite-4565-ddl Commit: d6e5028dbe30bc0aa31b194b70e9deb77dec890a Parents: 04b49c7 Author: devozerov <voze...@gridgain.com> Authored: Mon Mar 20 11:19:04 2017 +0300 Committer: devozerov <voze...@gridgain.com> Committed: Mon Mar 20 11:19:04 2017 +0300 ---------------------------------------------------------------------- .../processors/query/GridQueryProcessor.java | 61 ++++--------------- .../query/ddl/task/IndexingCacheStartTask.java | 62 ++++++++++++++++++++ .../query/ddl/task/IndexingCacheStopTask.java | 44 ++++++++++++++ .../processors/query/ddl/task/IndexingTask.java | 26 ++++++++ 4 files changed, 142 insertions(+), 51 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/d6e5028d/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java index 58f577e..339b878 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java @@ -65,6 +65,8 @@ import org.apache.ignite.internal.processors.query.ddl.AbstractIndexOperation; import org.apache.ignite.internal.processors.query.ddl.CreateIndexOperation; import org.apache.ignite.internal.processors.query.ddl.IndexAcceptDiscoveryMessage; import org.apache.ignite.internal.processors.query.ddl.IndexProposeDiscoveryMessage; +import org.apache.ignite.internal.processors.query.ddl.task.IndexingCacheStartTask; +import org.apache.ignite.internal.processors.query.ddl.task.IndexingTask; import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor; import org.apache.ignite.internal.util.GridSpinBusyLock; import org.apache.ignite.internal.util.future.GridFinishedFuture; @@ -1261,7 +1263,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { */ private class DynamicIndexManagerWorker extends GridWorker { /** Tasks queue. */ - private final LinkedBlockingQueue<DynamicIndexTask> tasks = new LinkedBlockingQueue<>(); + private final LinkedBlockingQueue<IndexingTask> tasks = new LinkedBlockingQueue<>(); /** Alive nodes. */ private Collection<ClusterNode> aliveNodes; @@ -1294,7 +1296,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { * * @param task Task. */ - public void submit(DynamicIndexTask task) { + public void submit(IndexingTask task) { tasks.add(task); } @@ -1305,7 +1307,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { * @param initIdxStates Initial index states. */ public void onCacheStart(String space, QueryIndexStates initIdxStates) { - submit(new CacheStartTask(space, initIdxStates)); + submit(new IndexingCacheStartTask(space, initIdxStates)); } /** @@ -1361,52 +1363,9 @@ public class GridQueryProcessor extends GridProcessorAdapter { } /** - * Marker interface for index-related tasks. - */ - private static interface DynamicIndexTask { - // No-op. - } - - /** - * Cache start task. - */ - private static class CacheStartTask implements DynamicIndexTask { - /** Space. */ - private final String space; - - /** Initial index states. */ - private final QueryIndexStates initIdxStates; - - /** - * Constructor. - * - * @param space Space. - * @param initIdxStates Initial index states. - */ - public CacheStartTask(String space, QueryIndexStates initIdxStates) { - this.space = space; - this.initIdxStates = initIdxStates; - } - - /** - * @return Space. - */ - public String space() { - return space; - } - - /** - * @return Initial index states. - */ - public QueryIndexStates initialIndexStates() { - return initIdxStates; - } - } - - /** * Change index task. */ - private static class ChangeIndexTask implements DynamicIndexTask { + private static class ChangeIndexingTask implements IndexingTask { /** Operation. */ private final AbstractIndexOperation op; @@ -1415,7 +1374,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { * * @param op Operation. */ - public ChangeIndexTask(AbstractIndexOperation op) { + public ChangeIndexingTask(AbstractIndexOperation op) { this.op = op; } @@ -1430,7 +1389,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { /** * Node leave task. */ - private static class NodeLeaveTask implements DynamicIndexTask { + private static class NodeLeaveTask implements IndexingTask { /** Node ID. */ private final UUID nodeId; @@ -1454,7 +1413,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { /** * Type removal task (either due to cache stop or due to type undeploy). */ - private static class TypeRemoveTask implements DynamicIndexTask { + private static class TypeRemoveTask implements IndexingTask { /** Type descriptor. */ private final QueryTypeDescriptorImpl typeDesc; @@ -1478,7 +1437,7 @@ public class GridQueryProcessor extends GridProcessorAdapter { /** * Operation status message received. */ - private static class OperationStatusTask implements DynamicIndexTask { + private static class OperationStatusTask implements IndexingTask { /** Node ID. */ private final UUID nodeId; http://git-wip-us.apache.org/repos/asf/ignite/blob/d6e5028d/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStartTask.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStartTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStartTask.java new file mode 100644 index 0000000..1c6cf8c --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStartTask.java @@ -0,0 +1,62 @@ +/* + * 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.query.ddl.task; + +import org.apache.ignite.internal.processors.query.QueryIndexStates; +import org.apache.ignite.internal.util.typedef.internal.S; + +/** + * Indexing cache start task. + */ +public class IndexingCacheStartTask implements IndexingTask { + /** Space. */ + private final String space; + + /** Initial index states. */ + private final QueryIndexStates initIdxStates; + + /** + * Constructor. + * + * @param space Space. + * @param initIdxStates Initial index states. + */ + public IndexingCacheStartTask(String space, QueryIndexStates initIdxStates) { + this.space = space; + this.initIdxStates = initIdxStates; + } + + /** + * @return Space. + */ + public String space() { + return space; + } + + /** + * @return Initial index states. + */ + public QueryIndexStates initialIndexStates() { + return initIdxStates; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(IndexingCacheStartTask.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/d6e5028d/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStopTask.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStopTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStopTask.java new file mode 100644 index 0000000..3a77cd9 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingCacheStopTask.java @@ -0,0 +1,44 @@ +/* + * 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.query.ddl.task; + +import org.apache.ignite.internal.util.typedef.internal.S; + +/** + * Indexing cache stop task. + */ +public class IndexingCacheStopTask implements IndexingTask { + /** Space. */ + private final String space; + + public IndexingCacheStopTask(String space) { + this.space = space; + } + + /** + * @return Space. + */ + public String space() { + return space; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(IndexingCacheStopTask.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/d6e5028d/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingTask.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingTask.java new file mode 100644 index 0000000..fa253c7 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/ddl/task/IndexingTask.java @@ -0,0 +1,26 @@ +/* + * 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.query.ddl.task; + +/** + * Marker interface for dynamic indexing change tasks. Concrete implementation are essentially a value objects, + * with actual handling being performed in index worker thread. + */ +public interface IndexingTask { + // No-op. +}