You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by am...@apache.org on 2016/09/12 09:13:33 UTC
[1/3] asterixdb git commit: Small Cleanup Towards Fixing LifeCycle
Issues
Repository: asterixdb
Updated Branches:
refs/heads/master dd3660520 -> 11a02942f
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndex.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndex.java
index 8ea1f88..be68903 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndex.java
@@ -26,7 +26,7 @@ import org.apache.hyracks.storage.common.buffercache.IBufferCache;
* This interface describes the operations common to all indexes. Indexes
* implementing this interface can easily reuse existing index operators for
* dataflow. Users must perform operations on an via an {@link IIndexAccessor}.
- * During dataflow, the lifecycle of IIndexes are handled through an {@link IIndexLifecycleManager}.
+ * During dataflow, the lifecycle of IIndexes are handled through an {@link IResourceLifecycleManager}.
*/
public interface IIndex {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManager.java
deleted file mode 100644
index 4c00b93..0000000
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManager.java
+++ /dev/null
@@ -1,38 +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.hyracks.storage.am.common.api;
-
-import java.util.List;
-
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.common.IResourceMemoryManager;
-
-public interface IIndexLifecycleManager extends IResourceMemoryManager {
- public List<IIndex> getOpenIndexes();
-
- public void register(String resourcePath, IIndex index) throws HyracksDataException;
-
- public void open(String resourcePath) throws HyracksDataException;
-
- public void close(String resourcePath) throws HyracksDataException;
-
- public IIndex getIndex(String resourcePath) throws HyracksDataException;
-
- public void unregister(String resourcePath) throws HyracksDataException;
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManagerProvider.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManagerProvider.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManagerProvider.java
index 127ee83..5378610 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManagerProvider.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IIndexLifecycleManagerProvider.java
@@ -23,5 +23,5 @@ import java.io.Serializable;
import org.apache.hyracks.api.context.IHyracksTaskContext;
public interface IIndexLifecycleManagerProvider extends Serializable {
- public IIndexLifecycleManager getLifecycleManager(IHyracksTaskContext ctx);
+ public IResourceLifecycleManager<IIndex> getLifecycleManager(IHyracksTaskContext ctx);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IResourceLifecycleManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IResourceLifecycleManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IResourceLifecycleManager.java
new file mode 100644
index 0000000..1d18a8f
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IResourceLifecycleManager.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.common.api;
+
+import java.util.List;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.common.IResourceMemoryManager;
+
+/**
+ * The base interface for resource management
+ *
+ * @param <R>
+ * resource class
+ */
+public interface IResourceLifecycleManager<R> extends IResourceMemoryManager {
+ /**
+ * get a list of all resources which are opened
+ *
+ * @return
+ */
+ public List<R> getOpenResources();
+
+ /**
+ * Registers a resource.
+ *
+ * @param resourceId
+ * @param resource
+ * @throws HyracksDataException
+ * if a resource is already registered with this resourceId
+ */
+ public void register(String resourceId, R resource) throws HyracksDataException;
+
+ /**
+ * Opens a resource. The resource is moved to the open state
+ *
+ * @param resourceId
+ * @throws HyracksDataException
+ * if no resource with the resourceId exists
+ */
+ public void open(String resourceId) throws HyracksDataException;
+
+ /**
+ * closes a resource and free up its allocated resources
+ *
+ * @param resourceId
+ * @throws HyracksDataException
+ */
+ public void close(String resourceId) throws HyracksDataException;
+
+ /**
+ * gets the resource registered with this id
+ *
+ * @param resourceId
+ * @return the resource if registered or null
+ * @throws HyracksDataException
+ */
+ public R get(String resourceId) throws HyracksDataException;
+
+ /**
+ * unregister a resource removing its resources in memory and on disk
+ *
+ * @param resourceId
+ * @throws HyracksDataException
+ */
+ public void unregister(String resourceId) throws HyracksDataException;
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
index 2e45c7c..202c0a8 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexDataflowHelper.java
@@ -26,7 +26,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.storage.am.common.api.IIndex;
import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrame;
import org.apache.hyracks.storage.am.common.util.IndexFileNameUtil;
import org.apache.hyracks.storage.common.file.ILocalResourceFactory;
@@ -38,7 +38,7 @@ public abstract class IndexDataflowHelper implements IIndexDataflowHelper {
protected final IIndexOperatorDescriptor opDesc;
protected final IHyracksTaskContext ctx;
- protected final IIndexLifecycleManager lcManager;
+ protected final IResourceLifecycleManager<IIndex> lcManager;
protected final ILocalResourceRepository localResourceRepository;
protected final IResourceIdFactory resourceIdFactory;
protected final FileReference file;
@@ -72,8 +72,9 @@ public abstract class IndexDataflowHelper implements IIndexDataflowHelper {
@Override
public void create() throws HyracksDataException {
synchronized (lcManager) {
- index = lcManager.getIndex(resourcePath);
+ index = lcManager.get(resourcePath);
if (index != null) {
+ //how is this right?????????? <needs to be fixed>
lcManager.unregister(resourcePath);
} else {
index = createIndexInstance();
@@ -109,7 +110,7 @@ public abstract class IndexDataflowHelper implements IIndexDataflowHelper {
throw new HyracksDataException("Index does not have a valid resource ID. Has it been created yet?");
}
- index = lcManager.getIndex(resourcePath);
+ index = lcManager.get(resourcePath);
if (index == null) {
index = createIndexInstance();
lcManager.register(resourcePath, index);
@@ -128,7 +129,7 @@ public abstract class IndexDataflowHelper implements IIndexDataflowHelper {
@Override
public void destroy() throws HyracksDataException {
synchronized (lcManager) {
- index = lcManager.getIndex(resourcePath);
+ index = lcManager.get(resourcePath);
if (index != null) {
lcManager.unregister(resourcePath);
} else {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexLifecycleManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexLifecycleManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexLifecycleManager.java
index 5953abe..224283d 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexLifecycleManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexLifecycleManager.java
@@ -30,9 +30,9 @@ import java.util.Map.Entry;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
import org.apache.hyracks.storage.am.common.api.IIndex;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
-public class IndexLifecycleManager implements IIndexLifecycleManager, ILifeCycleComponent {
+public class IndexLifecycleManager implements IResourceLifecycleManager<IIndex>, ILifeCycleComponent {
private static final long DEFAULT_MEMORY_BUDGET = 1024 * 1024 * 100; // 100 megabytes
private final Map<String, IndexInfo> indexInfos;
@@ -131,6 +131,7 @@ public class IndexLifecycleManager implements IIndexLifecycleManager, ILifeCycle
}
}
+ @Override
public String toString() {
return "{index: " + index + ", isOpen: " + isOpen + ", refCount: " + referenceCount + ", lastAccess: "
+ lastAccess + "}";
@@ -138,7 +139,7 @@ public class IndexLifecycleManager implements IIndexLifecycleManager, ILifeCycle
}
@Override
- public List<IIndex> getOpenIndexes() {
+ public List<IIndex> getOpenResources() {
List<IIndex> openIndexes = new ArrayList<IIndex>();
for (IndexInfo i : indexInfos.values()) {
if (i.isOpen) {
@@ -165,6 +166,7 @@ public class IndexLifecycleManager implements IIndexLifecycleManager, ILifeCycle
}
}
+ @Override
public void dumpState(OutputStream os) throws IOException {
StringBuilder sb = new StringBuilder();
@@ -212,7 +214,7 @@ public class IndexLifecycleManager implements IIndexLifecycleManager, ILifeCycle
}
@Override
- public IIndex getIndex(String resourcePath) throws HyracksDataException {
+ public IIndex get(String resourcePath) throws HyracksDataException {
IndexInfo info = indexInfos.get(resourcePath);
return info == null ? null : info.index;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeDataflowHelper.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeDataflowHelper.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeDataflowHelper.java
index ab5be15..ef8f646 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeDataflowHelper.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeDataflowHelper.java
@@ -59,7 +59,7 @@ public class ExternalBTreeDataflowHelper extends LSMBTreeDataflowHelper {
synchronized (lcManager) {
if (index == null) {
try {
- index = lcManager.getIndex(resourcePath);
+ index = lcManager.get(resourcePath);
} catch (HyracksDataException e) {
return null;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyDataflowHelper.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyDataflowHelper.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyDataflowHelper.java
index 6e0beb3..e7a1161 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyDataflowHelper.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyDataflowHelper.java
@@ -60,7 +60,7 @@ public class ExternalBTreeWithBuddyDataflowHelper extends AbstractLSMIndexDatafl
synchronized (lcManager) {
if (index == null) {
try {
- index = lcManager.getIndex(resourcePath);
+ index = lcManager.get(resourcePath);
} catch (HyracksDataException e) {
return null;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMMergePolicyFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMMergePolicyFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMMergePolicyFactory.java
index d2d86e6..c047777 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMMergePolicyFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMMergePolicyFactory.java
@@ -23,7 +23,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
public interface ILSMMergePolicyFactory extends Serializable {
// Having two methods to create the merge policy with two different signatures is hacky, but we do it now
@@ -32,7 +32,7 @@ public interface ILSMMergePolicyFactory extends Serializable {
// in asterix and cannot be seen in hyracks. Thus we pass IHyracksTaskContext and let the merge policy do the casting.
public ILSMMergePolicy createMergePolicy(Map<String, String> configuration, IHyracksTaskContext ctx);
- public ILSMMergePolicy createMergePolicy(Map<String, String> configuration, IIndexLifecycleManager ilcm);
+ public ILSMMergePolicy createMergePolicy(Map<String, String> configuration, IResourceLifecycleManager ilcm);
public String getName();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ConstantMergePolicyFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ConstantMergePolicyFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ConstantMergePolicyFactory.java
index 08672a5..e75aa7f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ConstantMergePolicyFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ConstantMergePolicyFactory.java
@@ -24,7 +24,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
@@ -53,7 +53,7 @@ public class ConstantMergePolicyFactory implements ILSMMergePolicyFactory {
}
@Override
- public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IIndexLifecycleManager ilcm) {
+ public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IResourceLifecycleManager ilcm) {
ILSMMergePolicy policy = new ConstantMergePolicy();
policy.configure(properties);
return policy;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoMergePolicyFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoMergePolicyFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoMergePolicyFactory.java
index b3b35f4..052e9af 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoMergePolicyFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoMergePolicyFactory.java
@@ -25,7 +25,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
@@ -54,7 +54,7 @@ public class NoMergePolicyFactory implements ILSMMergePolicyFactory {
}
@Override
- public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IIndexLifecycleManager ilcm) {
+ public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IResourceLifecycleManager ilcm) {
ILSMMergePolicy policy = new NoMergePolicy();
policy.configure(properties);
return policy;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/PrefixMergePolicyFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/PrefixMergePolicyFactory.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/PrefixMergePolicyFactory.java
index 465a5f4..e1e05f6 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/PrefixMergePolicyFactory.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/PrefixMergePolicyFactory.java
@@ -25,7 +25,7 @@ import java.util.Map;
import java.util.Set;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
@@ -55,7 +55,7 @@ public class PrefixMergePolicyFactory implements ILSMMergePolicyFactory {
}
@Override
- public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IIndexLifecycleManager ilcm) {
+ public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IResourceLifecycleManager ilcm) {
ILSMMergePolicy policy = new PrefixMergePolicy();
policy.configure(properties);
return policy;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeDataflowHelper.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeDataflowHelper.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeDataflowHelper.java
index 3503df5..9731371 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeDataflowHelper.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeDataflowHelper.java
@@ -77,7 +77,7 @@ public class ExternalRTreeDataflowHelper extends LSMRTreeDataflowHelper {
synchronized (lcManager) {
if (index == null) {
try {
- index = lcManager.getIndex(resourcePath);
+ index = lcManager.get(resourcePath);
} catch (HyracksDataException e) {
return null;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestIndexLifecycleManagerProvider.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestIndexLifecycleManagerProvider.java b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestIndexLifecycleManagerProvider.java
index f48778e..409b49b 100644
--- a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestIndexLifecycleManagerProvider.java
+++ b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestIndexLifecycleManagerProvider.java
@@ -19,16 +19,17 @@
package org.apache.hyracks.test.support;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IIndex;
import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
public class TestIndexLifecycleManagerProvider implements IIndexLifecycleManagerProvider {
private static final long serialVersionUID = 1L;
@Override
- public IIndexLifecycleManager getLifecycleManager(IHyracksTaskContext ctx) {
- return TestStorageManagerComponentHolder.getIndexLifecycleManager(ctx);
+ public IResourceLifecycleManager<IIndex> getLifecycleManager(IHyracksTaskContext ctx) {
+ return TestStorageManagerComponentHolder.getIndexLifecycleManager();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
index 482dccd..ee5f320 100644
--- a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
+++ b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestStorageManagerComponentHolder.java
@@ -28,7 +28,8 @@ import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksException;
import org.apache.hyracks.api.io.IODeviceHandle;
import org.apache.hyracks.control.nc.io.IOManager;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IIndex;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.dataflow.IndexLifecycleManager;
import org.apache.hyracks.storage.common.buffercache.BufferCache;
import org.apache.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
@@ -51,13 +52,14 @@ public class TestStorageManagerComponentHolder {
private static IFileMapProvider fileMapProvider;
private static IOManager ioManager;
private static ILocalResourceRepository localResourceRepository;
- private static IIndexLifecycleManager lcManager;
+ private static IResourceLifecycleManager<IIndex> lcManager;
private static ResourceIdFactory resourceIdFactory;
private static int pageSize;
private static int numPages;
private static int maxOpenFiles;
private final static ThreadFactory threadFactory = new ThreadFactory() {
+ @Override
public Thread newThread(Runnable r) {
return new Thread(r);
}
@@ -73,7 +75,7 @@ public class TestStorageManagerComponentHolder {
lcManager = null;
}
- public synchronized static IIndexLifecycleManager getIndexLifecycleManager(IHyracksTaskContext ctx) {
+ public synchronized static IResourceLifecycleManager<IIndex> getIndexLifecycleManager() {
if (lcManager == null) {
lcManager = new IndexLifecycleManager();
}
@@ -110,7 +112,8 @@ public class TestStorageManagerComponentHolder {
public synchronized static ILocalResourceRepository getLocalResourceRepository(IHyracksTaskContext ctx) {
if (localResourceRepository == null) {
try {
- ILocalResourceRepositoryFactory localResourceRepositoryFactory = new TransientLocalResourceRepositoryFactory();
+ ILocalResourceRepositoryFactory localResourceRepositoryFactory =
+ new TransientLocalResourceRepositoryFactory();
localResourceRepository = localResourceRepositoryFactory.createRepository();
} catch (HyracksException e) {
//In order not to change the IStorageManagerInterface due to the test code, throw runtime exception.
[3/3] asterixdb git commit: Small Cleanup Towards Fixing LifeCycle
Issues
Posted by am...@apache.org.
Small Cleanup Towards Fixing LifeCycle Issues
Before this change, dataset lifecycle manager was providing a set
of functionalities that are loosly related to management of datasets
and indexes. However, it was not clear what the possible states of
a dataset or an index and what is the responsibility of each object.
This change takes the first step towards fixing this area. Indexes
of a dataset are now grouped together under a single lifecycle class
A resource aka dataset must be created outside the lifecycle manager
and registered with it before it can be assigned resources (memory)
and before it can be used by any operation. This is still not the
case.
Change-Id: I84005a33837725f41ae63297a3711215dccce1d8
Reviewed-on: https://asterix-gerrit.ics.uci.edu/1148
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: abdullah alamoudi <ba...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/11a02942
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/11a02942
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/11a02942
Branch: refs/heads/master
Commit: 11a02942f73fb982762ac2be3bc5642bda1eafdc
Parents: dd36605
Author: Abdullah Alamoudi <ba...@gmail.com>
Authored: Mon Sep 12 10:17:03 2016 +0300
Committer: abdullah alamoudi <ba...@gmail.com>
Committed: Mon Sep 12 02:12:40 2016 -0700
----------------------------------------------------------------------
.../active/message/ActiveManagerMessage.java | 8 +-
.../active/message/ActivePartitionMessage.java | 8 +-
...rixAppRuntimeContextProviderForRecovery.java | 6 -
.../app/nc/AsterixNCAppRuntimeContext.java | 14 +-
.../asterix/messaging/CCMessageBroker.java | 5 +-
.../asterix/messaging/NCMessageBroker.java | 55 +-
.../common/api/IAsterixAppRuntimeContext.java | 2 -
.../common/api/IDatasetLifecycleManager.java | 18 +-
.../common/context/BaseOperationTracker.java | 1 -
.../context/CorrelatedPrefixMergePolicy.java | 4 +-
.../CorrelatedPrefixMergePolicyFactory.java | 4 +-
.../asterix/common/context/DatasetInfo.java | 187 +++++++
.../common/context/DatasetLifecycleManager.java | 532 ++++++-------------
.../asterix/common/context/DatasetResource.java | 142 +++++
.../context/DatasetVirtualBufferCaches.java | 83 +++
.../asterix/common/context/IndexInfo.java | 45 ++
.../org/apache/asterix/common/context/Info.java | 53 ++
.../context/PrimaryIndexOperationTracker.java | 1 -
.../messaging/AbstractApplicationMessage.java | 36 --
.../messaging/api/IApplicationMessage.java | 24 +-
.../api/IApplicationMessageCallback.java | 30 --
.../common/messaging/api/ICCMessageBroker.java | 1 -
.../common/messaging/api/INCMessageBroker.java | 4 +-
.../IAsterixAppRuntimeContextProvider.java | 3 -
.../apache/asterix/metadata/MetadataNode.java | 14 +-
.../metadata/bootstrap/MetadataBootstrap.java | 10 +-
.../management/ReplicationChannel.java | 4 +-
.../message/AbstractFailbackPlanMessage.java | 4 +-
.../message/CompleteFailbackRequestMessage.java | 10 +-
.../CompleteFailbackResponseMessage.java | 8 +-
...PreparePartitionsFailbackRequestMessage.java | 10 +-
...reparePartitionsFailbackResponseMessage.java | 4 +-
.../runtime/message/ReplicaEventMessage.java | 8 +-
.../message/ReportMaxResourceIdMessage.java | 10 +-
.../ReportMaxResourceIdRequestMessage.java | 8 +-
.../message/ResourceIdRequestMessage.java | 10 +-
.../ResourceIdRequestResponseMessage.java | 19 +-
.../TakeoverMetadataNodeRequestMessage.java | 10 +-
.../TakeoverMetadataNodeResponseMessage.java | 8 +-
.../TakeoverPartitionsRequestMessage.java | 14 +-
.../TakeoverPartitionsResponseMessage.java | 8 +-
.../transaction/GlobalResourceIdFactory.java | 21 +-
.../runtime/util/AsterixClusterProperties.java | 2 +-
...dexModificationOperationCallbackFactory.java | 6 +-
...dexModificationOperationCallbackFactory.java | 6 +-
...dexModificationOperationCallbackFactory.java | 6 +-
...dexModificationOperationCallbackFactory.java | 6 +-
.../UpsertOperationCallbackFactory.java | 6 +-
.../service/recovery/CheckpointThread.java | 28 +-
.../service/recovery/RecoveryManager.java | 4 +-
.../transaction/TransactionSubsystem.java | 9 +-
.../locking/TestRuntimeContextProvider.java | 10 +-
.../apache/hyracks/api/messages/IMessage.java | 3 -
.../hyracks/api/messages/IMessageBroker.java | 3 -
.../helper/IndexLifecycleManagerProvider.java | 4 +-
.../examples/btree/helper/RuntimeContext.java | 6 +-
.../hyracks/storage/am/common/api/IIndex.java | 2 +-
.../am/common/api/IIndexLifecycleManager.java | 38 --
.../api/IIndexLifecycleManagerProvider.java | 2 +-
.../common/api/IResourceLifecycleManager.java | 83 +++
.../am/common/dataflow/IndexDataflowHelper.java | 11 +-
.../common/dataflow/IndexLifecycleManager.java | 10 +-
.../dataflow/ExternalBTreeDataflowHelper.java | 2 +-
.../ExternalBTreeWithBuddyDataflowHelper.java | 2 +-
.../lsm/common/api/ILSMMergePolicyFactory.java | 4 +-
.../impls/ConstantMergePolicyFactory.java | 4 +-
.../lsm/common/impls/NoMergePolicyFactory.java | 4 +-
.../common/impls/PrefixMergePolicyFactory.java | 4 +-
.../dataflow/ExternalRTreeDataflowHelper.java | 2 +-
.../TestIndexLifecycleManagerProvider.java | 7 +-
.../TestStorageManagerComponentHolder.java | 11 +-
71 files changed, 962 insertions(+), 769 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActiveManagerMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActiveManagerMessage.java b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActiveManagerMessage.java
index 50fa257..392bec8 100644
--- a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActiveManagerMessage.java
+++ b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActiveManagerMessage.java
@@ -22,12 +22,12 @@ import java.io.Serializable;
import org.apache.asterix.active.ActiveManager;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
import org.apache.hyracks.control.nc.NodeControllerService;
-public class ActiveManagerMessage extends AbstractApplicationMessage {
+public class ActiveManagerMessage implements IApplicationMessage {
public static final byte STOP_ACTIVITY = 0x00;
private static final long serialVersionUID = 1L;
@@ -62,7 +62,7 @@ public class ActiveManagerMessage extends AbstractApplicationMessage {
}
@Override
- public String type() {
- return "ACTIVE_MANAGER_MESSAGE";
+ public String toString() {
+ return ActiveManagerMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActivePartitionMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActivePartitionMessage.java b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActivePartitionMessage.java
index 02affc4..fc67d3c 100644
--- a/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActivePartitionMessage.java
+++ b/asterixdb/asterix-active/src/main/java/org/apache/asterix/active/message/ActivePartitionMessage.java
@@ -22,12 +22,12 @@ import java.io.Serializable;
import org.apache.asterix.active.ActiveLifecycleListener;
import org.apache.asterix.active.ActiveRuntimeId;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.service.IControllerService;
-public class ActivePartitionMessage extends AbstractApplicationMessage {
+public class ActivePartitionMessage implements IApplicationMessage {
public static final byte ACTIVE_RUNTIME_REGISTERED = 0x00;
public static final byte ACTIVE_RUNTIME_DEREGISTERED = 0x01;
@@ -70,7 +70,7 @@ public class ActivePartitionMessage extends AbstractApplicationMessage {
}
@Override
- public String type() {
- return "ACTIVE_ENTITY_TO_CC_MESSAGE";
+ public String toString() {
+ return ActivePartitionMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixAppRuntimeContextProviderForRecovery.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixAppRuntimeContextProviderForRecovery.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixAppRuntimeContextProviderForRecovery.java
index 265025f..ec0aaa8 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixAppRuntimeContextProviderForRecovery.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixAppRuntimeContextProviderForRecovery.java
@@ -30,7 +30,6 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
import org.apache.hyracks.storage.common.file.IFileMapProvider;
import org.apache.hyracks.storage.common.file.ILocalResourceRepository;
-import org.apache.hyracks.storage.common.file.IResourceIdFactory;
public class AsterixAppRuntimeContextProviderForRecovery implements IAsterixAppRuntimeContextProvider {
@@ -76,11 +75,6 @@ public class AsterixAppRuntimeContextProviderForRecovery implements IAsterixAppR
}
@Override
- public IResourceIdFactory getResourceIdFactory() {
- return asterixAppRuntimeContext.getResourceIdFactory();
- }
-
- @Override
public IIOManager getIOManager() {
return asterixAppRuntimeContext.getIOManager();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/AsterixNCAppRuntimeContext.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/AsterixNCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/AsterixNCAppRuntimeContext.java
index 9127ee5..ce20c9c 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/AsterixNCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/AsterixNCAppRuntimeContext.java
@@ -60,7 +60,6 @@ import org.apache.asterix.common.transactions.IAsterixAppRuntimeContextProvider;
import org.apache.asterix.common.transactions.IRecoveryManager;
import org.apache.asterix.common.transactions.IRecoveryManager.SystemState;
import org.apache.asterix.common.transactions.ITransactionSubsystem;
-import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
import org.apache.asterix.external.library.ExternalLibraryManager;
import org.apache.asterix.metadata.MetadataManager;
import org.apache.asterix.metadata.MetadataNode;
@@ -73,6 +72,7 @@ import org.apache.asterix.replication.recovery.RemoteRecoveryManager;
import org.apache.asterix.replication.storage.ReplicaResourcesManager;
import org.apache.asterix.runtime.transaction.GlobalResourceIdFactoryProvider;
import org.apache.asterix.runtime.util.AsterixClusterProperties;
+import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepositoryFactory;
import org.apache.asterix.transaction.management.service.transaction.TransactionSubsystem;
import org.apache.hyracks.api.application.IApplicationConfig;
@@ -105,6 +105,7 @@ public class AsterixNCAppRuntimeContext implements IAsterixAppRuntimeContext, IA
private ILSMMergePolicyFactory metadataMergePolicyFactory;
private final INCApplicationContext ncApplicationContext;
+ private final IResourceIdFactory resourceIdFactory;
private AsterixCompilerProperties compilerProperties;
private AsterixExternalProperties externalProperties;
@@ -124,7 +125,6 @@ public class AsterixNCAppRuntimeContext implements IAsterixAppRuntimeContext, IA
private ILSMIOOperationScheduler lsmIOScheduler;
private PersistentLocalResourceRepository localResourceRepository;
- private IResourceIdFactory resourceIdFactory;
private IIOManager ioManager;
private boolean isShuttingdown;
@@ -171,6 +171,7 @@ public class AsterixNCAppRuntimeContext implements IAsterixAppRuntimeContext, IA
}
allExtensions.addAll(new AsterixExtensionProperties(propertiesAccessor).getExtensions());
ncExtensionManager = new NCExtensionManager(allExtensions);
+ resourceIdFactory = new GlobalResourceIdFactoryProvider(ncApplicationContext).createResourceIdFactory();
}
@Override
@@ -193,7 +194,7 @@ public class AsterixNCAppRuntimeContext implements IAsterixAppRuntimeContext, IA
ILocalResourceRepositoryFactory persistentLocalResourceRepositoryFactory =
new PersistentLocalResourceRepositoryFactory(
- ioManager, ncApplicationContext.getNodeId(), metadataProperties);
+ ioManager, ncApplicationContext.getNodeId(), metadataProperties);
localResourceRepository = (PersistentLocalResourceRepository) persistentLocalResourceRepositoryFactory
.createRepository();
@@ -209,7 +210,6 @@ public class AsterixNCAppRuntimeContext implements IAsterixAppRuntimeContext, IA
//delete any storage data before the resource factory is initialized
localResourceRepository.deleteStorageData(true);
}
- initializeResourceIdFactory();
datasetLifecycleManager = new DatasetLifecycleManager(storageProperties, localResourceRepository,
MetadataIndexImmutableProperties.FIRST_AVAILABLE_USER_DATASET_ID, txnSubsystem.getLogManager(),
@@ -290,6 +290,7 @@ public class AsterixNCAppRuntimeContext implements IAsterixAppRuntimeContext, IA
lccm.register((ILifeCycleComponent) datasetLifecycleManager);
lccm.register((ILifeCycleComponent) txnSubsystem.getTransactionManager());
lccm.register((ILifeCycleComponent) txnSubsystem.getLockManager());
+
}
@Override
@@ -442,11 +443,6 @@ public class AsterixNCAppRuntimeContext implements IAsterixAppRuntimeContext, IA
}
@Override
- public void initializeResourceIdFactory() throws HyracksDataException {
- resourceIdFactory = new GlobalResourceIdFactoryProvider(ncApplicationContext).createResourceIdFactory();
- }
-
- @Override
public void initializeMetadata(boolean newUniverse) throws Exception {
IAsterixStateProxy proxy;
if (LOGGER.isLoggable(Level.INFO)) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/CCMessageBroker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/CCMessageBroker.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/CCMessageBroker.java
index 1345aa7..d1d7ff7 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/CCMessageBroker.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/CCMessageBroker.java
@@ -22,7 +22,6 @@ import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.common.messaging.api.ICCMessageBroker;
import org.apache.hyracks.api.messages.IMessage;
@@ -41,9 +40,9 @@ public class CCMessageBroker implements ICCMessageBroker {
@Override
public void receivedMessage(IMessage message, String nodeId) throws Exception {
- AbstractApplicationMessage absMessage = (AbstractApplicationMessage) message;
+ IApplicationMessage absMessage = (IApplicationMessage) message;
if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Received message: " + absMessage.type());
+ LOGGER.info("Received message: " + absMessage);
}
absMessage.handle(ccs);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
index 9851b61..1beff82 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
@@ -20,19 +20,14 @@ package org.apache.asterix.messaging;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.config.MessagingProperties;
import org.apache.asterix.common.memory.ConcurrentFramePool;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
import org.apache.asterix.common.messaging.api.IApplicationMessage;
-import org.apache.asterix.common.messaging.api.IApplicationMessageCallback;
import org.apache.asterix.common.messaging.api.INCMessageBroker;
import org.apache.hyracks.api.comm.IChannelControlBlock;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -44,8 +39,6 @@ public class NCMessageBroker implements INCMessageBroker {
private static final Logger LOGGER = Logger.getLogger(NCMessageBroker.class.getName());
private final NodeControllerService ncs;
- private final AtomicLong messageId = new AtomicLong(0);
- private final Map<Long, IApplicationMessageCallback> callbacks;
private final IAsterixAppRuntimeContext appContext;
private final LinkedBlockingQueue<IApplicationMessage> receivedMsgsQ;
private final ConcurrentFramePool messagingFramePool;
@@ -54,7 +47,6 @@ public class NCMessageBroker implements INCMessageBroker {
public NCMessageBroker(NodeControllerService ncs, MessagingProperties messagingProperties) {
this.ncs = ncs;
appContext = (IAsterixAppRuntimeContext) ncs.getApplicationContext().getApplicationObject();
- callbacks = new ConcurrentHashMap<>();
maxMsgSize = messagingProperties.getFrameSize();
int messagingMemoryBudget = messagingProperties.getFrameSize() * messagingProperties.getFrameCount();
messagingFramePool = new ConcurrentFramePool(ncs.getId(), messagingMemoryBudget,
@@ -65,27 +57,15 @@ public class NCMessageBroker implements INCMessageBroker {
}
@Override
- public void sendMessageToCC(IApplicationMessage message, IApplicationMessageCallback callback) throws Exception {
- registerMsgCallback(message, callback);
- try {
- ncs.sendApplicationMessageToCC(JavaSerializationUtils.serialize(message), null);
- } catch (Exception e) {
- handleMsgDeliveryFailure(message);
- throw e;
- }
+ public void sendMessageToCC(IApplicationMessage message) throws Exception {
+ ncs.sendApplicationMessageToCC(JavaSerializationUtils.serialize(message), null);
}
@Override
- public void sendMessageToNC(String nodeId, IApplicationMessage message, IApplicationMessageCallback callback)
+ public void sendMessageToNC(String nodeId, IApplicationMessage message)
throws Exception {
- registerMsgCallback(message, callback);
- try {
- IChannelControlBlock messagingChannel = ncs.getMessagingNetworkManager().getMessagingChannel(nodeId);
- sendMessageToChannel(messagingChannel, message);
- } catch (Exception e) {
- handleMsgDeliveryFailure(message);
- throw e;
- }
+ IChannelControlBlock messagingChannel = ncs.getMessagingNetworkManager().getMessagingChannel(nodeId);
+ sendMessageToChannel(messagingChannel, message);
}
@Override
@@ -95,14 +75,9 @@ public class NCMessageBroker implements INCMessageBroker {
@Override
public void receivedMessage(IMessage message, String nodeId) throws Exception {
- AbstractApplicationMessage absMessage = (AbstractApplicationMessage) message;
+ IApplicationMessage absMessage = (IApplicationMessage) message;
if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Received message: " + absMessage.type());
- }
- //if the received message is a response to a sent message, deliver it to the sender
- IApplicationMessageCallback callback = callbacks.remove(absMessage.getId());
- if (callback != null) {
- callback.deliverMessageResponse(absMessage);
+ LOGGER.info("Received message: " + absMessage);
}
absMessage.handle(ncs);
}
@@ -111,18 +86,6 @@ public class NCMessageBroker implements INCMessageBroker {
return messagingFramePool;
}
- private void registerMsgCallback(IApplicationMessage message, IApplicationMessageCallback callback) {
- if (callback != null) {
- long uniqueMessageId = messageId.incrementAndGet();
- message.setId(uniqueMessageId);
- callbacks.put(uniqueMessageId, callback);
- }
- }
-
- private void handleMsgDeliveryFailure(IApplicationMessage message) {
- callbacks.remove(message.getId());
- }
-
private void sendMessageToChannel(IChannelControlBlock ccb, IApplicationMessage msg) throws IOException {
byte[] serializedMsg = JavaSerializationUtils.serialize(msg);
if (serializedMsg.length > maxMsgSize) {
@@ -161,8 +124,8 @@ public class NCMessageBroker implements INCMessageBroker {
Thread.currentThread().interrupt();
} catch (Exception e) {
if (LOGGER.isLoggable(Level.WARNING) && msg != null) {
- LOGGER.log(Level.WARNING, "Could not process message with id: " + msg.getId() + " and type: "
- + msg.type(), e);
+ LOGGER.log(Level.WARNING, "Could not process message : "
+ + msg, e);
} else {
if (LOGGER.isLoggable(Level.WARNING)) {
LOGGER.log(Level.WARNING, "Could not process message", e);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IAsterixAppRuntimeContext.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IAsterixAppRuntimeContext.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IAsterixAppRuntimeContext.java
index 046d5c1..c009152 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IAsterixAppRuntimeContext.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IAsterixAppRuntimeContext.java
@@ -86,8 +86,6 @@ public interface IAsterixAppRuntimeContext {
public ILibraryManager getLibraryManager();
- public void initializeResourceIdFactory() throws HyracksDataException;
-
/**
* Exports the metadata node to the metadata RMI port.
*
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
index 552ce22..04e30ff 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
@@ -20,22 +20,22 @@ package org.apache.asterix.common.api;
import java.util.List;
-import org.apache.asterix.common.context.DatasetLifecycleManager.DatasetInfo;
-import org.apache.asterix.common.context.DatasetLifecycleManager.IndexInfo;
+import org.apache.asterix.common.context.DatasetInfo;
+import org.apache.asterix.common.context.IndexInfo;
+import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.common.api.IIndex;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
-public interface IDatasetLifecycleManager extends IIndexLifecycleManager {
+public interface IDatasetLifecycleManager extends IResourceLifecycleManager<IIndex> {
/**
- * @param datasetID
- * @param resourceID
+ * @param datasetId
+ * @param indexId
* @return The corresponding index, or null if it is not found in the registered indexes.
* @throws HyracksDataException
*/
- IIndex getIndex(int datasetID, long resourceID) throws HyracksDataException;
+ IIndex getIndex(int datasetId, long indexId) throws HyracksDataException;
/**
* Flushes all open datasets synchronously.
@@ -75,7 +75,7 @@ public interface IDatasetLifecycleManager extends IIndexLifecycleManager {
* @param datasetID
* @return
*/
- ILSMOperationTracker getOperationTracker(int datasetID);
+ PrimaryIndexOperationTracker getOperationTracker(int datasetID);
/**
* creates (if necessary) and returns the dataset virtual buffer caches.
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java
index 21500b7..5c1d094 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/BaseOperationTracker.java
@@ -18,7 +18,6 @@
*/
package org.apache.asterix.common.context;
-import org.apache.asterix.common.context.DatasetLifecycleManager.DatasetInfo;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.ISearchOperationCallback;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
index 3d112ef..70339f3 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicy.java
@@ -27,7 +27,7 @@ import java.util.Set;
import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IndexException;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
@@ -45,7 +45,7 @@ public class CorrelatedPrefixMergePolicy implements ILSMMergePolicy {
private final IDatasetLifecycleManager datasetLifecycleManager;
private final int datasetID;
- public CorrelatedPrefixMergePolicy(IIndexLifecycleManager datasetLifecycleManager, int datasetID) {
+ public CorrelatedPrefixMergePolicy(IResourceLifecycleManager datasetLifecycleManager, int datasetID) {
this.datasetLifecycleManager = (DatasetLifecycleManager) datasetLifecycleManager;
this.datasetID = datasetID;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
index ce405fc..3b65123 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
@@ -27,7 +27,7 @@ import java.util.Set;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
@@ -61,7 +61,7 @@ public class CorrelatedPrefixMergePolicyFactory implements ILSMMergePolicyFactor
}
@Override
- public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IIndexLifecycleManager ilcm) {
+ public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IResourceLifecycleManager ilcm) {
ILSMMergePolicy policy = new CorrelatedPrefixMergePolicy(ilcm, datasetID);
policy.configure(properties);
return policy;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java
new file mode 100644
index 0000000..6a2cc56
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetInfo.java
@@ -0,0 +1,187 @@
+/*
+ * 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.asterix.common.context;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+
+public class DatasetInfo extends Info implements Comparable<DatasetInfo> {
+ private final Map<Long, IndexInfo> indexes;
+ private final int datasetID;
+ private long lastAccess;
+ private int numActiveIOOps;
+ private boolean isExternal;
+ private boolean isRegistered;
+ private boolean memoryAllocated;
+ private boolean durable;
+
+ public DatasetInfo(int datasetID) {
+ this.indexes = new HashMap<>();
+ this.setLastAccess(-1);
+ this.datasetID = datasetID;
+ this.setRegistered(false);
+ this.setMemoryAllocated(false);
+ }
+
+ @Override
+ public void touch() {
+ super.touch();
+ setLastAccess(System.currentTimeMillis());
+ }
+
+ @Override
+ public void untouch() {
+ super.untouch();
+ setLastAccess(System.currentTimeMillis());
+ }
+
+ public synchronized void declareActiveIOOperation() {
+ setNumActiveIOOps(getNumActiveIOOps() + 1);
+ }
+
+ public synchronized void undeclareActiveIOOperation() {
+ setNumActiveIOOps(getNumActiveIOOps() - 1);
+ //notify threads waiting on this dataset info
+ notifyAll();
+ }
+
+ public synchronized Set<ILSMIndex> getDatasetIndexes() {
+ Set<ILSMIndex> datasetIndexes = new HashSet<>();
+ for (IndexInfo iInfo : getIndexes().values()) {
+ if (iInfo.isOpen()) {
+ datasetIndexes.add(iInfo.getIndex());
+ }
+ }
+
+ return datasetIndexes;
+ }
+
+ @Override
+ public int compareTo(DatasetInfo i) {
+ // sort by (isOpen, referenceCount, lastAccess) ascending, where true < false
+ //
+ // Example sort order:
+ // -------------------
+ // (F, 0, 70) <-- largest
+ // (F, 0, 60)
+ // (T, 10, 80)
+ // (T, 10, 70)
+ // (T, 9, 90)
+ // (T, 0, 100) <-- smallest
+ if (isOpen() && !i.isOpen()) {
+ return -1;
+ } else if (!isOpen() && i.isOpen()) {
+ return 1;
+ } else {
+ if (getReferenceCount() < i.getReferenceCount()) {
+ return -1;
+ } else if (getReferenceCount() > i.getReferenceCount()) {
+ return 1;
+ } else {
+ if (getLastAccess() < i.getLastAccess()) {
+ return -1;
+ } else if (getLastAccess() > i.getLastAccess()) {
+ return 1;
+ } else {
+ return 0;
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof DatasetInfo) {
+ return datasetID == ((DatasetInfo) obj).datasetID;
+ }
+ return false;
+ };
+
+ @Override
+ public int hashCode() {
+ return datasetID;
+ }
+
+ @Override
+ public String toString() {
+ return "DatasetID: " + getDatasetID() + ", isOpen: " + isOpen() + ", refCount: " + getReferenceCount()
+ + ", lastAccess: " + getLastAccess() + ", isRegistered: " + isRegistered() + ", memoryAllocated: "
+ + isMemoryAllocated() + ", isDurable: " + isDurable();
+ }
+
+ public boolean isDurable() {
+ return durable;
+ }
+
+ public int getNumActiveIOOps() {
+ return numActiveIOOps;
+ }
+
+ public void setNumActiveIOOps(int numActiveIOOps) {
+ this.numActiveIOOps = numActiveIOOps;
+ }
+
+ public boolean isExternal() {
+ return isExternal;
+ }
+
+ public void setExternal(boolean isExternal) {
+ this.isExternal = isExternal;
+ }
+
+ public Map<Long, IndexInfo> getIndexes() {
+ return indexes;
+ }
+
+ public boolean isRegistered() {
+ return isRegistered;
+ }
+
+ public void setRegistered(boolean isRegistered) {
+ this.isRegistered = isRegistered;
+ }
+
+ public void setDurable(boolean durable) {
+ this.durable = durable;
+ }
+
+ public int getDatasetID() {
+ return datasetID;
+ }
+
+ public boolean isMemoryAllocated() {
+ return memoryAllocated;
+ }
+
+ public void setMemoryAllocated(boolean memoryAllocated) {
+ this.memoryAllocated = memoryAllocated;
+ }
+
+ public long getLastAccess() {
+ return lastAccess;
+ }
+
+ public void setLastAccess(long lastAccess) {
+ this.lastAccess = lastAccess;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
index f4eec05..698dfb4 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
@@ -22,11 +22,9 @@ import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.asterix.common.api.ILocalResourceMetadata;
@@ -40,22 +38,16 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
import org.apache.hyracks.storage.am.common.api.IIndex;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
-import org.apache.hyracks.storage.am.lsm.common.impls.MultitenantVirtualBufferCache;
-import org.apache.hyracks.storage.am.lsm.common.impls.VirtualBufferCache;
-import org.apache.hyracks.storage.common.buffercache.ResourceHeapBufferAllocator;
import org.apache.hyracks.storage.common.file.ILocalResourceRepository;
import org.apache.hyracks.storage.common.file.LocalResource;
public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeCycleComponent {
+ private final Map<Integer, DatasetResource> datasets = new ConcurrentHashMap<>();
private final AsterixStorageProperties storageProperties;
- private final Map<Integer, DatasetVirtualBufferCaches> datasetVirtualBufferCachesMap;
- private final Map<Integer, ILSMOperationTracker> datasetOpTrackers;
- private final Map<Integer, DatasetInfo> datasetInfos;
private final ILocalResourceRepository resourceRepository;
private final int firstAvilableUserDatasetID;
private final long capacity;
@@ -63,7 +55,7 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
private final ILogManager logManager;
private final LogRecord logRecord;
private final int numPartitions;
- private boolean stopped = false;
+ private volatile boolean stopped = false;
public DatasetLifecycleManager(AsterixStorageProperties storageProperties,
ILocalResourceRepository resourceRepository, int firstAvilableUserDatasetID, ILogManager logManager,
@@ -73,16 +65,13 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
this.resourceRepository = resourceRepository;
this.firstAvilableUserDatasetID = firstAvilableUserDatasetID;
this.numPartitions = numPartitions;
- datasetVirtualBufferCachesMap = new HashMap<>();
- datasetOpTrackers = new HashMap<Integer, ILSMOperationTracker>();
- datasetInfos = new HashMap<Integer, DatasetInfo>();
capacity = storageProperties.getMemoryComponentGlobalBudget();
used = 0;
logRecord = new LogRecord();
}
@Override
- public synchronized IIndex getIndex(String resourcePath) throws HyracksDataException {
+ public synchronized IIndex get(String resourcePath) throws HyracksDataException {
validateDatasetLifecycleManagerState();
int datasetID = getDIDfromResourcePath(resourcePath);
long resourceID = getResourceIDfromResourcePath(resourcePath);
@@ -92,15 +81,11 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
@Override
public synchronized IIndex getIndex(int datasetID, long resourceID) throws HyracksDataException {
validateDatasetLifecycleManagerState();
- DatasetInfo dsInfo = datasetInfos.get(datasetID);
- if (dsInfo == null) {
- return null;
- }
- IndexInfo iInfo = dsInfo.indexes.get(resourceID);
- if (iInfo == null) {
+ DatasetResource datasetResource = datasets.get(datasetID);
+ if (datasetResource == null) {
return null;
}
- return iInfo.index;
+ return datasetResource.getIndex(resourceID);
}
@Override
@@ -108,20 +93,11 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
validateDatasetLifecycleManagerState();
int did = getDIDfromResourcePath(resourcePath);
long resourceID = getResourceIDfromResourcePath(resourcePath);
- DatasetInfo dsInfo = datasetInfos.get(did);
- if (dsInfo == null) {
- dsInfo = getDatasetInfo(did);
- }
- if (!dsInfo.isRegistered) {
- dsInfo.isExternal = !index.hasMemoryComponents();
- dsInfo.isRegistered = true;
- dsInfo.durable = ((ILSMIndex) index).isDurable();
+ DatasetResource datasetResource = datasets.get(did);
+ if (datasetResource == null) {
+ datasetResource = getDatasetLifecycle(did);
}
-
- if (dsInfo.indexes.containsKey(resourceID)) {
- throw new HyracksDataException("Index with resource ID " + resourceID + " already exists.");
- }
- dsInfo.indexes.put(resourceID, new IndexInfo((ILSMIndex) index, dsInfo.datasetID, resourceID));
+ datasetResource.register(resourceID, index);
}
public int getDIDfromResourcePath(String resourcePath) throws HyracksDataException {
@@ -146,24 +122,25 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
int did = getDIDfromResourcePath(resourcePath);
long resourceID = getResourceIDfromResourcePath(resourcePath);
- DatasetInfo dsInfo = datasetInfos.get(did);
- IndexInfo iInfo = dsInfo == null ? null : dsInfo.indexes.get(resourceID);
+ DatasetResource dsr = datasets.get(did);
+ IndexInfo iInfo = dsr == null ? null : dsr.getIndexInfo(resourceID);
- if (dsInfo == null || iInfo == null) {
+ if (dsr == null || iInfo == null) {
throw new HyracksDataException("Index with resource ID " + resourceID + " does not exist.");
}
- PrimaryIndexOperationTracker opTracker = (PrimaryIndexOperationTracker) datasetOpTrackers.get(dsInfo.datasetID);
- if (iInfo.referenceCount != 0 || (opTracker != null && opTracker.getNumActiveOperations() != 0)) {
+ PrimaryIndexOperationTracker opTracker = dsr.getOpTracker();
+ if (iInfo.getReferenceCount() != 0 || (opTracker != null && opTracker.getNumActiveOperations() != 0)) {
throw new HyracksDataException("Cannot remove index while it is open. (Dataset reference count = "
- + iInfo.referenceCount + ", Operation tracker number of active operations = "
+ + iInfo.getReferenceCount() + ", Operation tracker number of active operations = "
+ opTracker.getNumActiveOperations() + ")");
}
// TODO: use fine-grained counters, one for each index instead of a single counter per dataset.
// First wait for any ongoing IO operations
+ DatasetInfo dsInfo = dsr.getDatasetInfo();
synchronized (dsInfo) {
- while (dsInfo.numActiveIOOps > 0) {
+ while (dsInfo.getNumActiveIOOps() > 0) {
try {
//notification will come from DatasetInfo class (undeclareActiveIOOperation)
dsInfo.wait();
@@ -173,16 +150,17 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
}
- if (iInfo.isOpen) {
- ILSMOperationTracker indexOpTracker = iInfo.index.getOperationTracker();
+ if (iInfo.isOpen()) {
+ ILSMOperationTracker indexOpTracker = iInfo.getIndex().getOperationTracker();
synchronized (indexOpTracker) {
- iInfo.index.deactivate(false);
+ iInfo.getIndex().deactivate(false);
}
}
- dsInfo.indexes.remove(resourceID);
- if (dsInfo.referenceCount == 0 && dsInfo.isOpen && dsInfo.indexes.isEmpty() && !dsInfo.isExternal) {
- removeDatasetFromCache(dsInfo.datasetID);
+ dsInfo.getIndexes().remove(resourceID);
+ if (dsInfo.getReferenceCount() == 0 && dsInfo.isOpen() && dsInfo.getIndexes().isEmpty()
+ && !dsInfo.isExternal()) {
+ removeDatasetFromCache(dsInfo.getDatasetID());
}
}
@@ -192,29 +170,28 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
int did = getDIDfromResourcePath(resourcePath);
long resourceID = getResourceIDfromResourcePath(resourcePath);
- DatasetInfo dsInfo = datasetInfos.get(did);
- if (dsInfo == null || !dsInfo.isRegistered) {
+ DatasetResource dsr = datasets.get(did);
+ DatasetInfo dsInfo = dsr.getDatasetInfo();
+ if (dsInfo == null || !dsInfo.isRegistered()) {
throw new HyracksDataException(
"Failed to open index with resource ID " + resourceID + " since it does not exist.");
}
- IndexInfo iInfo = dsInfo.indexes.get(resourceID);
+ IndexInfo iInfo = dsInfo.getIndexes().get(resourceID);
if (iInfo == null) {
throw new HyracksDataException(
"Failed to open index with resource ID " + resourceID + " since it does not exist.");
}
- if (!dsInfo.isOpen && !dsInfo.isExternal) {
- initializeDatasetVirtualBufferCache(did);
- }
- dsInfo.isOpen = true;
- dsInfo.touch();
- if (!iInfo.isOpen) {
- ILSMOperationTracker opTracker = iInfo.index.getOperationTracker();
+ dsr.open(true);
+ dsr.touch();
+
+ if (!iInfo.isOpen()) {
+ ILSMOperationTracker opTracker = iInfo.getIndex().getOperationTracker();
synchronized (opTracker) {
- iInfo.index.activate();
+ iInfo.getIndex().activate();
}
- iInfo.isOpen = true;
+ iInfo.setOpen(true);
}
iInfo.touch();
}
@@ -225,14 +202,15 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
* that is not being used (refcount == 0) and has been least recently used, excluding metadata datasets.
* The sort order defined for DatasetInfo maintains this. See DatasetInfo.compareTo().
*/
- List<DatasetInfo> datasetInfosList = new ArrayList<DatasetInfo>(datasetInfos.values());
- Collections.sort(datasetInfosList);
- for (DatasetInfo dsInfo : datasetInfosList) {
- PrimaryIndexOperationTracker opTracker = (PrimaryIndexOperationTracker) datasetOpTrackers
- .get(dsInfo.datasetID);
- if (opTracker != null && opTracker.getNumActiveOperations() == 0 && dsInfo.referenceCount == 0
- && dsInfo.isOpen && dsInfo.datasetID >= firstAvilableUserDatasetID) {
- closeDataset(dsInfo);
+ List<DatasetResource> datasetsResources = new ArrayList<>(datasets.values());
+ Collections.sort(datasetsResources);
+ for (DatasetResource dsr : datasetsResources) {
+ PrimaryIndexOperationTracker opTracker = dsr.getOpTracker();
+ if (opTracker != null && opTracker.getNumActiveOperations() == 0
+ && dsr.getDatasetInfo().getReferenceCount() == 0
+ && dsr.getDatasetInfo().isOpen()
+ && dsr.getDatasetInfo().getDatasetID() >= getFirstAvilableUserDatasetID()) {
+ closeDataset(dsr.getDatasetInfo());
return true;
}
}
@@ -240,15 +218,15 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
private static void flushAndWaitForIO(DatasetInfo dsInfo, IndexInfo iInfo) throws HyracksDataException {
- if (iInfo.isOpen) {
- ILSMIndexAccessor accessor = iInfo.index.createAccessor(NoOpOperationCallback.INSTANCE,
+ if (iInfo.isOpen()) {
+ ILSMIndexAccessor accessor = iInfo.getIndex().createAccessor(NoOpOperationCallback.INSTANCE,
NoOpOperationCallback.INSTANCE);
- accessor.scheduleFlush(iInfo.index.getIOOperationCallback());
+ accessor.scheduleFlush(iInfo.getIndex().getIOOperationCallback());
}
// Wait for the above flush op.
synchronized (dsInfo) {
- while (dsInfo.numActiveIOOps > 0) {
+ while (dsInfo.getNumActiveIOOps() > 0) {
try {
//notification will come from DatasetInfo class (undeclareActiveIOOperation)
dsInfo.wait();
@@ -259,52 +237,64 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
}
- @Override
- public DatasetInfo getDatasetInfo(int datasetID) {
- synchronized (datasetInfos) {
- DatasetInfo dsInfo = datasetInfos.get(datasetID);
- if (dsInfo == null) {
- dsInfo = new DatasetInfo(datasetID);
- datasetInfos.put(datasetID, dsInfo);
+ public DatasetResource getDatasetLifecycle(int did) {
+ DatasetResource dsr = datasets.get(did);
+ if (dsr != null) {
+ return dsr;
+ }
+ synchronized (datasets) {
+ dsr = datasets.get(did);
+ if (dsr == null) {
+ DatasetInfo dsInfo = new DatasetInfo(did);
+ PrimaryIndexOperationTracker opTracker = new PrimaryIndexOperationTracker(did, logManager, dsInfo);
+ DatasetVirtualBufferCaches vbcs = new DatasetVirtualBufferCaches(did, storageProperties,
+ getFirstAvilableUserDatasetID(),
+ getNumPartitions());
+ dsr = new DatasetResource(dsInfo, opTracker, vbcs);
+ datasets.put(did, dsr);
}
- return dsInfo;
+ return dsr;
}
}
@Override
+ public DatasetInfo getDatasetInfo(int datasetID) {
+ return getDatasetLifecycle(datasetID).getDatasetInfo();
+ }
+
+ @Override
public synchronized void close(String resourcePath) throws HyracksDataException {
validateDatasetLifecycleManagerState();
int did = getDIDfromResourcePath(resourcePath);
long resourceID = getResourceIDfromResourcePath(resourcePath);
-
- DatasetInfo dsInfo = datasetInfos.get(did);
- if (dsInfo == null) {
+ DatasetResource dsr = datasets.get(did);
+ if (dsr == null) {
throw new HyracksDataException("No index found with resourceID " + resourceID);
}
- IndexInfo iInfo = dsInfo.indexes.get(resourceID);
+ IndexInfo iInfo = dsr.getIndexInfo(resourceID);
if (iInfo == null) {
throw new HyracksDataException("No index found with resourceID " + resourceID);
}
iInfo.untouch();
- dsInfo.untouch();
+ dsr.untouch();
}
@Override
- public synchronized List<IIndex> getOpenIndexes() {
+ public synchronized List<IIndex> getOpenResources() {
List<IndexInfo> openIndexesInfo = getOpenIndexesInfo();
List<IIndex> openIndexes = new ArrayList<IIndex>();
for (IndexInfo iInfo : openIndexesInfo) {
- openIndexes.add(iInfo.index);
+ openIndexes.add(iInfo.getIndex());
}
return openIndexes;
}
@Override
public synchronized List<IndexInfo> getOpenIndexesInfo() {
- List<IndexInfo> openIndexesInfo = new ArrayList<IndexInfo>();
- for (DatasetInfo dsInfo : datasetInfos.values()) {
- for (IndexInfo iInfo : dsInfo.indexes.values()) {
- if (iInfo.isOpen) {
+ List<IndexInfo> openIndexesInfo = new ArrayList<>();
+ for (DatasetResource dsr : datasets.values()) {
+ for (IndexInfo iInfo : dsr.getIndexes().values()) {
+ if (iInfo.isOpen()) {
openIndexesInfo.add(iInfo);
}
}
@@ -313,46 +303,23 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
private DatasetVirtualBufferCaches getVirtualBufferCaches(int datasetID) {
- synchronized (datasetVirtualBufferCachesMap) {
- DatasetVirtualBufferCaches vbcs = datasetVirtualBufferCachesMap.get(datasetID);
- if (vbcs == null) {
- vbcs = initializeDatasetVirtualBufferCache(datasetID);
- }
- return vbcs;
- }
+ return getDatasetLifecycle(datasetID).getVirtualBufferCaches();
}
@Override
public List<IVirtualBufferCache> getVirtualBufferCaches(int datasetID, int ioDeviceNum) {
DatasetVirtualBufferCaches dvbcs = getVirtualBufferCaches(datasetID);
- return dvbcs.getVirtualBufferCaches(ioDeviceNum);
+ return dvbcs.getVirtualBufferCaches(this, ioDeviceNum);
}
private void removeDatasetFromCache(int datasetID) throws HyracksDataException {
deallocateDatasetMemory(datasetID);
- datasetInfos.remove(datasetID);
- datasetVirtualBufferCachesMap.remove(datasetID);
- datasetOpTrackers.remove(datasetID);
- }
-
- private DatasetVirtualBufferCaches initializeDatasetVirtualBufferCache(int datasetID) {
- synchronized (datasetVirtualBufferCachesMap) {
- DatasetVirtualBufferCaches dvbcs = new DatasetVirtualBufferCaches(datasetID);
- datasetVirtualBufferCachesMap.put(datasetID, dvbcs);
- return dvbcs;
- }
+ datasets.remove(datasetID);
}
@Override
- public ILSMOperationTracker getOperationTracker(int datasetID) {
- synchronized (datasetOpTrackers) {
- ILSMOperationTracker opTracker = datasetOpTrackers.get(datasetID);
- if (opTracker == null) {
- opTracker = new PrimaryIndexOperationTracker(datasetID, logManager, getDatasetInfo(datasetID));
- datasetOpTrackers.put(datasetID, opTracker);
- }
- return opTracker;
- }
+ public PrimaryIndexOperationTracker getOperationTracker(int datasetID) {
+ return datasets.get(datasetID).getOpTracker();
}
private void validateDatasetLifecycleManagerState() throws HyracksDataException {
@@ -361,145 +328,6 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
}
}
- private static abstract class Info {
- protected int referenceCount;
- protected boolean isOpen;
-
- public Info() {
- referenceCount = 0;
- isOpen = false;
- }
-
- public void touch() {
- ++referenceCount;
- }
-
- public void untouch() {
- --referenceCount;
- }
- }
-
- public static class IndexInfo extends Info {
- private final ILSMIndex index;
- private final long resourceId;
- private final int datasetId;
-
- public IndexInfo(ILSMIndex index, int datasetId, long resourceId) {
- this.index = index;
- this.datasetId = datasetId;
- this.resourceId = resourceId;
- }
-
- public ILSMIndex getIndex() {
- return index;
- }
-
- public long getResourceId() {
- return resourceId;
- }
-
- public int getDatasetId() {
- return datasetId;
- }
- }
-
- public static class DatasetInfo extends Info implements Comparable<DatasetInfo> {
- private final Map<Long, IndexInfo> indexes;
- private final int datasetID;
- private long lastAccess;
- private int numActiveIOOps;
- private boolean isExternal;
- private boolean isRegistered;
- private boolean memoryAllocated;
- private boolean durable;
-
- public DatasetInfo(int datasetID) {
- this.indexes = new HashMap<Long, IndexInfo>();
- this.lastAccess = -1;
- this.datasetID = datasetID;
- this.isRegistered = false;
- this.memoryAllocated = false;
- }
-
- @Override
- public void touch() {
- super.touch();
- lastAccess = System.currentTimeMillis();
- }
-
- @Override
- public void untouch() {
- super.untouch();
- lastAccess = System.currentTimeMillis();
- }
-
- public synchronized void declareActiveIOOperation() {
- numActiveIOOps++;
- }
-
- public synchronized void undeclareActiveIOOperation() {
- numActiveIOOps--;
- //notify threads waiting on this dataset info
- notifyAll();
- }
-
- public synchronized Set<ILSMIndex> getDatasetIndexes() {
- Set<ILSMIndex> datasetIndexes = new HashSet<ILSMIndex>();
- for (IndexInfo iInfo : indexes.values()) {
- if (iInfo.isOpen) {
- datasetIndexes.add(iInfo.index);
- }
- }
-
- return datasetIndexes;
- }
-
- @Override
- public int compareTo(DatasetInfo i) {
- // sort by (isOpen, referenceCount, lastAccess) ascending, where true < false
- //
- // Example sort order:
- // -------------------
- // (F, 0, 70) <-- largest
- // (F, 0, 60)
- // (T, 10, 80)
- // (T, 10, 70)
- // (T, 9, 90)
- // (T, 0, 100) <-- smallest
- if (isOpen && !i.isOpen) {
- return -1;
- } else if (!isOpen && i.isOpen) {
- return 1;
- } else {
- if (referenceCount < i.referenceCount) {
- return -1;
- } else if (referenceCount > i.referenceCount) {
- return 1;
- } else {
- if (lastAccess < i.lastAccess) {
- return -1;
- } else if (lastAccess > i.lastAccess) {
- return 1;
- } else {
- return 0;
- }
- }
- }
-
- }
-
- @Override
- public String toString() {
- return "DatasetID: " + datasetID + ", isOpen: " + isOpen + ", refCount: " + referenceCount
- + ", lastAccess: " + lastAccess + ", isRegistered: " + isRegistered + ", memoryAllocated: "
- + memoryAllocated + ", isDurable: " + durable;
- }
-
- public boolean isDurable() {
- return durable;
- }
- }
-
@Override
public synchronized void start() {
used = 0;
@@ -507,30 +335,29 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
@Override
public synchronized void flushAllDatasets() throws HyracksDataException {
- for (DatasetInfo dsInfo : datasetInfos.values()) {
- flushDatasetOpenIndexes(dsInfo, false);
+ for (DatasetResource dsr : datasets.values()) {
+ flushDatasetOpenIndexes(dsr.getDatasetInfo(), false);
}
}
@Override
public synchronized void flushDataset(int datasetId, boolean asyncFlush) throws HyracksDataException {
- DatasetInfo datasetInfo = datasetInfos.get(datasetId);
- if (datasetInfo != null) {
- flushDatasetOpenIndexes(datasetInfo, asyncFlush);
+ DatasetResource dsr = datasets.get(datasetId);
+ if (dsr != null) {
+ flushDatasetOpenIndexes(dsr.getDatasetInfo(), asyncFlush);
}
}
@Override
public synchronized void scheduleAsyncFlushForLaggingDatasets(long targetLSN) throws HyracksDataException {
//schedule flush for datasets with min LSN (Log Serial Number) < targetLSN
- for (DatasetInfo dsInfo : datasetInfos.values()) {
- PrimaryIndexOperationTracker opTracker = (PrimaryIndexOperationTracker) getOperationTracker(
- dsInfo.datasetID);
+ for (DatasetResource dsr : datasets.values()) {
+ PrimaryIndexOperationTracker opTracker = dsr.getOpTracker();
synchronized (opTracker) {
- for (IndexInfo iInfo : dsInfo.indexes.values()) {
- AbstractLSMIOOperationCallback ioCallback = (AbstractLSMIOOperationCallback) iInfo.index
+ for (IndexInfo iInfo : dsr.getIndexes().values()) {
+ AbstractLSMIOOperationCallback ioCallback = (AbstractLSMIOOperationCallback) iInfo.getIndex()
.getIOOperationCallback();
- if (!(((AbstractLSMIndex) iInfo.index).isCurrentMutableComponentEmpty()
+ if (!(((AbstractLSMIndex) iInfo.getIndex()).isCurrentMutableComponentEmpty()
|| ioCallback.hasPendingFlush() || opTracker.isFlushLogCreated()
|| opTracker.isFlushOnExit())) {
long firstLSN = ioCallback.getFirstLSN();
@@ -552,10 +379,10 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
* This method can only be called asynchronously safely if we're sure no modify operation will take place until the flush is scheduled
*/
private void flushDatasetOpenIndexes(DatasetInfo dsInfo, boolean asyncFlush) throws HyracksDataException {
- if (!dsInfo.isExternal && dsInfo.durable) {
+ if (!dsInfo.isExternal() && dsInfo.isDurable()) {
synchronized (logRecord) {
- TransactionUtil.formFlushLogRecord(logRecord, dsInfo.datasetID, null, logManager.getNodeId(),
- dsInfo.indexes.size());
+ TransactionUtil.formFlushLogRecord(logRecord, dsInfo.getDatasetID(), null, logManager.getNodeId(),
+ dsInfo.getIndexes().size());
try {
logManager.log(logRecord);
} catch (ACIDException e) {
@@ -569,22 +396,22 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
throw new HyracksDataException(e);
}
}
- for (IndexInfo iInfo : dsInfo.indexes.values()) {
+ for (IndexInfo iInfo : dsInfo.getIndexes().values()) {
//update resource lsn
- AbstractLSMIOOperationCallback ioOpCallback = (AbstractLSMIOOperationCallback) iInfo.index
+ AbstractLSMIOOperationCallback ioOpCallback = (AbstractLSMIOOperationCallback) iInfo.getIndex()
.getIOOperationCallback();
ioOpCallback.updateLastLSN(logRecord.getLSN());
}
}
if (asyncFlush) {
- for (IndexInfo iInfo : dsInfo.indexes.values()) {
- ILSMIndexAccessor accessor = iInfo.index.createAccessor(NoOpOperationCallback.INSTANCE,
+ for (IndexInfo iInfo : dsInfo.getIndexes().values()) {
+ ILSMIndexAccessor accessor = iInfo.getIndex().createAccessor(NoOpOperationCallback.INSTANCE,
NoOpOperationCallback.INSTANCE);
- accessor.scheduleFlush(iInfo.index.getIOOperationCallback());
+ accessor.scheduleFlush(iInfo.getIndex().getIOOperationCallback());
}
} else {
- for (IndexInfo iInfo : dsInfo.indexes.values()) {
+ for (IndexInfo iInfo : dsInfo.getIndexes().values()) {
// TODO: This is not efficient since we flush the indexes sequentially.
// Think of a way to allow submitting the flush requests concurrently. We don't do them concurrently because this
// may lead to a deadlock scenario between the DatasetLifeCycleManager and the PrimaryIndexOperationTracker.
@@ -596,7 +423,7 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
private void closeDataset(DatasetInfo dsInfo) throws HyracksDataException {
// First wait for any ongoing IO operations
synchronized (dsInfo) {
- while (dsInfo.numActiveIOOps > 0) {
+ while (dsInfo.getNumActiveIOOps() > 0) {
try {
dsInfo.wait();
} catch (InterruptedException e) {
@@ -609,34 +436,33 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
} catch (Exception e) {
throw new HyracksDataException(e);
}
- for (IndexInfo iInfo : dsInfo.indexes.values()) {
- if (iInfo.isOpen) {
- ILSMOperationTracker opTracker = iInfo.index.getOperationTracker();
+ for (IndexInfo iInfo : dsInfo.getIndexes().values()) {
+ if (iInfo.isOpen()) {
+ ILSMOperationTracker opTracker = iInfo.getIndex().getOperationTracker();
synchronized (opTracker) {
- iInfo.index.deactivate(false);
+ iInfo.getIndex().deactivate(false);
}
- iInfo.isOpen = false;
+ iInfo.setOpen(false);
}
- assert iInfo.referenceCount == 0;
}
- removeDatasetFromCache(dsInfo.datasetID);
- dsInfo.isOpen = false;
+ removeDatasetFromCache(dsInfo.getDatasetID());
+ dsInfo.setOpen(false);
}
@Override
public synchronized void closeAllDatasets() throws HyracksDataException {
- List<DatasetInfo> openDatasets = new ArrayList<>(datasetInfos.values());
- for (DatasetInfo dsInfo : openDatasets) {
- closeDataset(dsInfo);
+ ArrayList<DatasetResource> openDatasets = new ArrayList<>(datasets.values());
+ for (DatasetResource dsr : openDatasets) {
+ closeDataset(dsr.getDatasetInfo());
}
}
@Override
public synchronized void closeUserDatasets() throws HyracksDataException {
- List<DatasetInfo> openDatasets = new ArrayList<>(datasetInfos.values());
- for (DatasetInfo dsInfo : openDatasets) {
- if (dsInfo.datasetID >= firstAvilableUserDatasetID) {
- closeDataset(dsInfo);
+ ArrayList<DatasetResource> openDatasets = new ArrayList<>(datasets.values());
+ for (DatasetResource dsr : openDatasets) {
+ if (dsr.getDatasetID() >= getFirstAvilableUserDatasetID()) {
+ closeDataset(dsr.getDatasetInfo());
}
}
}
@@ -652,9 +478,7 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
closeAllDatasets();
- datasetVirtualBufferCachesMap.clear();
- datasetOpTrackers.clear();
- datasetInfos.clear();
+ datasets.clear();
stopped = true;
}
@@ -673,57 +497,43 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
sb.append("[Datasets]\n");
sb.append(String.format(dsHeaderFormat, "DatasetID", "Open", "Reference Count", "Last Access"));
- for (DatasetInfo dsInfo : datasetInfos.values()) {
+ for (DatasetResource dsr : datasets.values()) {
+ DatasetInfo dsInfo = dsr.getDatasetInfo();
sb.append(
- String.format(dsFormat, dsInfo.datasetID, dsInfo.isOpen, dsInfo.referenceCount, dsInfo.lastAccess));
+ String.format(dsFormat, dsInfo.getDatasetID(), dsInfo.isOpen(), dsInfo.getReferenceCount(),
+ dsInfo.getLastAccess()));
}
sb.append("\n");
sb.append("[Indexes]\n");
sb.append(String.format(idxHeaderFormat, "DatasetID", "ResourceID", "Open", "Reference Count", "Index"));
- for (DatasetInfo dsInfo : datasetInfos.values()) {
- for (Map.Entry<Long, IndexInfo> entry : dsInfo.indexes.entrySet()) {
+ for (DatasetResource dsr : datasets.values()) {
+ DatasetInfo dsInfo = dsr.getDatasetInfo();
+ for (Map.Entry<Long, IndexInfo> entry : dsInfo.getIndexes().entrySet()) {
IndexInfo iInfo = entry.getValue();
- sb.append(String.format(idxFormat, dsInfo.datasetID, entry.getKey(), iInfo.isOpen, iInfo.referenceCount,
- iInfo.index));
+ sb.append(String.format(idxFormat, dsInfo.getDatasetID(), entry.getKey(), iInfo.isOpen(),
+ iInfo.getReferenceCount(),
+ iInfo.getIndex()));
}
}
-
outputStream.write(sb.toString().getBytes());
}
- private synchronized void allocateDatasetMemory(int datasetId) throws HyracksDataException {
- DatasetInfo dsInfo = datasetInfos.get(datasetId);
- if (dsInfo == null) {
+ private synchronized void deallocateDatasetMemory(int datasetId) throws HyracksDataException {
+ DatasetResource dsr = datasets.get(datasetId);
+ if (dsr == null) {
throw new HyracksDataException(
"Failed to allocate memory for dataset with ID " + datasetId + " since it is not open.");
}
- synchronized (dsInfo) {
- // This is not needed for external datasets' indexes since they never use the virtual buffer cache.
- if (!dsInfo.memoryAllocated && !dsInfo.isExternal) {
- long additionalSize = getVirtualBufferCaches(dsInfo.datasetID).getTotalSize();
- while (used + additionalSize > capacity) {
- if (!evictCandidateDataset()) {
- throw new HyracksDataException("Cannot allocate dataset " + dsInfo.datasetID
- + " memory since memory budget would be exceeded.");
- }
- }
- used += additionalSize;
- dsInfo.memoryAllocated = true;
- }
- }
- }
-
- private synchronized void deallocateDatasetMemory(int datasetId) throws HyracksDataException {
- DatasetInfo dsInfo = datasetInfos.get(datasetId);
+ DatasetInfo dsInfo = dsr.getDatasetInfo();
if (dsInfo == null) {
throw new HyracksDataException(
"Failed to deallocate memory for dataset with ID " + datasetId + " since it is not open.");
}
synchronized (dsInfo) {
- if (dsInfo.isOpen && dsInfo.memoryAllocated) {
- used -= getVirtualBufferCaches(dsInfo.datasetID).getTotalSize();
- dsInfo.memoryAllocated = false;
+ if (dsInfo.isOpen() && dsInfo.isMemoryAllocated()) {
+ used -= getVirtualBufferCaches(dsInfo.getDatasetID()).getTotalSize();
+ dsInfo.setMemoryAllocated(false);
}
}
}
@@ -731,54 +541,34 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
@Override
public synchronized void allocateMemory(String resourcePath) throws HyracksDataException {
//a resource name in the case of DatasetLifecycleManager is a dataset id which is passed to the ResourceHeapBufferAllocator.
- int did = Integer.parseInt(resourcePath);
- allocateDatasetMemory(did);
- }
-
- private class DatasetVirtualBufferCaches {
- private final int datasetID;
- private final Map<Integer, List<IVirtualBufferCache>> ioDeviceVirtualBufferCaches = new HashMap<>();
-
- public DatasetVirtualBufferCaches(int datasetID) {
- this.datasetID = datasetID;
- }
-
- private List<IVirtualBufferCache> initializeVirtualBufferCaches(int ioDeviceNum) {
- assert ioDeviceVirtualBufferCaches.size() < numPartitions;
- int numPages = datasetID < firstAvilableUserDatasetID
- ? storageProperties.getMetadataMemoryComponentNumPages()
- : storageProperties.getMemoryComponentNumPages();
- List<IVirtualBufferCache> vbcs = new ArrayList<>();
- for (int i = 0; i < storageProperties.getMemoryComponentsNum(); i++) {
- MultitenantVirtualBufferCache vbc = new MultitenantVirtualBufferCache(
- new VirtualBufferCache(
- new ResourceHeapBufferAllocator(DatasetLifecycleManager.this,
- Integer.toString(datasetID)),
- storageProperties.getMemoryComponentPageSize(),
- numPages / storageProperties.getMemoryComponentsNum() / numPartitions));
- vbcs.add(vbc);
- }
- ioDeviceVirtualBufferCaches.put(ioDeviceNum, vbcs);
- return vbcs;
+ int datasetId = Integer.parseInt(resourcePath);
+ DatasetResource dsr = datasets.get(datasetId);
+ if (dsr == null) {
+ throw new HyracksDataException(
+ "Failed to allocate memory for dataset with ID " + datasetId + " since it is not open.");
}
-
- public List<IVirtualBufferCache> getVirtualBufferCaches(int ioDeviceNum) {
- synchronized (ioDeviceVirtualBufferCaches) {
- List<IVirtualBufferCache> vbcs = ioDeviceVirtualBufferCaches.get(ioDeviceNum);
- if (vbcs == null) {
- vbcs = initializeVirtualBufferCaches(ioDeviceNum);
+ DatasetInfo dsInfo = dsr.getDatasetInfo();
+ synchronized (dsInfo) {
+ // This is not needed for external datasets' indexes since they never use the virtual buffer cache.
+ if (!dsInfo.isMemoryAllocated() && !dsInfo.isExternal()) {
+ long additionalSize = getVirtualBufferCaches(dsInfo.getDatasetID()).getTotalSize();
+ while (used + additionalSize > capacity) {
+ if (!evictCandidateDataset()) {
+ throw new HyracksDataException("Cannot allocate dataset " + dsInfo.getDatasetID()
+ + " memory since memory budget would be exceeded.");
+ }
}
- return vbcs;
+ used += additionalSize;
+ dsInfo.setMemoryAllocated(true);
}
}
+ }
- public long getTotalSize() {
- int numPages = datasetID < firstAvilableUserDatasetID
- ? storageProperties.getMetadataMemoryComponentNumPages()
- : storageProperties.getMemoryComponentNumPages();
-
- return storageProperties.getMemoryComponentPageSize() * numPages;
- }
+ public int getFirstAvilableUserDatasetID() {
+ return firstAvilableUserDatasetID;
}
+ public int getNumPartitions() {
+ return numPartitions;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetResource.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetResource.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetResource.java
new file mode 100644
index 0000000..01f2e2b
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetResource.java
@@ -0,0 +1,142 @@
+/*
+ * 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.asterix.common.context;
+
+import java.util.Map;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.common.api.IIndex;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+
+/**
+ * A dataset can be in one of two states { EVICTED , LOADED }.
+ * When a dataset is created, it is in the EVICTED state. In the EVICTED state, the allowed operations are:
+ * 1. DELETE: delete the dataset completely.
+ * 2. LOAD: move the dataset to the LOADED state.
+ * When a dataset is in the LOADED state, the allowed operations are:
+ * 1. OPEN: increment the open counter.
+ * 2. ALLOCATE RESOURCES (memory)
+ * 3. DEALLOCATE RESOURCES (memory)
+ * 4. CLOSE: decrement the open counter.
+ * 5. EVICT: deallocate resources and unload the dataset moving it to the EVICTED state
+ */
+public class DatasetResource implements Comparable<DatasetResource> {
+ private final DatasetInfo datasetInfo;
+ private final PrimaryIndexOperationTracker datasetPrimaryOpTracker;
+ private final DatasetVirtualBufferCaches datasetVirtualBufferCaches;
+
+ public DatasetResource(DatasetInfo datasetInfo,
+ PrimaryIndexOperationTracker datasetPrimaryOpTracker,
+ DatasetVirtualBufferCaches datasetVirtualBufferCaches) {
+ this.datasetInfo = datasetInfo;
+ this.datasetPrimaryOpTracker = datasetPrimaryOpTracker;
+ this.datasetVirtualBufferCaches = datasetVirtualBufferCaches;
+ }
+
+ public boolean isRegistered() {
+ return datasetInfo.isRegistered();
+ }
+
+ public IndexInfo getIndexInfo(long resourceID) {
+ return datasetInfo.getIndexes().get(resourceID);
+ }
+
+ public boolean isOpen() {
+ return datasetInfo.isOpen();
+ }
+
+ public boolean isExternal() {
+ return datasetInfo.isExternal();
+ }
+
+ public void open(boolean open) {
+ datasetInfo.setOpen(open);
+ }
+
+ public void touch() {
+ datasetInfo.touch();
+ }
+
+ public void untouch() {
+ datasetInfo.untouch();
+ }
+
+ public DatasetVirtualBufferCaches getVirtualBufferCaches() {
+ return datasetVirtualBufferCaches;
+ }
+
+ public IIndex getIndex(long resourceID) {
+ IndexInfo iInfo = getIndexInfo(resourceID);
+ if (iInfo == null) {
+ return null;
+ }
+ return iInfo.getIndex();
+ }
+
+ public void register(long resourceID, IIndex index) throws HyracksDataException {
+ if (!datasetInfo.isRegistered()) {
+ synchronized (datasetInfo) {
+ if (!datasetInfo.isRegistered()) {
+ datasetInfo.setExternal(!index.hasMemoryComponents());
+ datasetInfo.setRegistered(true);
+ datasetInfo.setDurable(((ILSMIndex) index).isDurable());
+ }
+ }
+ }
+ if (datasetInfo.getIndexes().containsKey(resourceID)) {
+ throw new HyracksDataException("Index with resource ID " + resourceID + " already exists.");
+ }
+ datasetInfo.getIndexes().put(resourceID,
+ new IndexInfo((ILSMIndex) index, datasetInfo.getDatasetID(), resourceID));
+ }
+
+ public DatasetInfo getDatasetInfo() {
+ return datasetInfo;
+ }
+
+ public PrimaryIndexOperationTracker getOpTracker() {
+ return datasetPrimaryOpTracker;
+ }
+
+ @Override
+ public int compareTo(DatasetResource o) {
+ return datasetInfo.compareTo(o.datasetInfo);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof DatasetResource) {
+ return datasetInfo.equals(((DatasetResource) obj).datasetInfo);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return datasetInfo.hashCode();
+ }
+
+ public Map<Long, IndexInfo> getIndexes() {
+ return datasetInfo.getIndexes();
+ }
+
+ public int getDatasetID() {
+ return datasetInfo.getDatasetID();
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetVirtualBufferCaches.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetVirtualBufferCaches.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetVirtualBufferCaches.java
new file mode 100644
index 0000000..df85509
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetVirtualBufferCaches.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.common.context;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.common.config.AsterixStorageProperties;
+import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
+import org.apache.hyracks.storage.am.lsm.common.impls.MultitenantVirtualBufferCache;
+import org.apache.hyracks.storage.am.lsm.common.impls.VirtualBufferCache;
+import org.apache.hyracks.storage.common.IResourceMemoryManager;
+import org.apache.hyracks.storage.common.buffercache.ResourceHeapBufferAllocator;
+
+public class DatasetVirtualBufferCaches {
+ private final int datasetID;
+ private final AsterixStorageProperties storageProperties;
+ private final int firstAvilableUserDatasetID;
+ private final int numPartitions;
+ private final Map<Integer, List<IVirtualBufferCache>> ioDeviceVirtualBufferCaches = new HashMap<>();
+
+ public DatasetVirtualBufferCaches(int datasetID, AsterixStorageProperties storageProperties,
+ int firstAvilableUserDatasetID, int numPartitions) {
+ this.datasetID = datasetID;
+ this.storageProperties = storageProperties;
+ this.firstAvilableUserDatasetID = firstAvilableUserDatasetID;
+ this.numPartitions = numPartitions;
+ }
+
+ public List<IVirtualBufferCache> initializeVirtualBufferCaches(IResourceMemoryManager memoryManager,
+ int ioDeviceNum) {
+ int numPages = datasetID < firstAvilableUserDatasetID
+ ? storageProperties.getMetadataMemoryComponentNumPages()
+ : storageProperties.getMemoryComponentNumPages();
+ List<IVirtualBufferCache> vbcs = new ArrayList<>();
+ for (int i = 0; i < storageProperties.getMemoryComponentsNum(); i++) {
+ MultitenantVirtualBufferCache vbc = new MultitenantVirtualBufferCache(
+ new VirtualBufferCache(
+ new ResourceHeapBufferAllocator(memoryManager,
+ Integer.toString(datasetID)),
+ storageProperties.getMemoryComponentPageSize(),
+ numPages / storageProperties.getMemoryComponentsNum() / numPartitions));
+ vbcs.add(vbc);
+ }
+ ioDeviceVirtualBufferCaches.put(ioDeviceNum, vbcs);
+ return vbcs;
+ }
+
+ public List<IVirtualBufferCache> getVirtualBufferCaches(IResourceMemoryManager memoryManager, int ioDeviceNum) {
+ synchronized (ioDeviceVirtualBufferCaches) {
+ List<IVirtualBufferCache> vbcs = ioDeviceVirtualBufferCaches.get(ioDeviceNum);
+ if (vbcs == null) {
+ vbcs = initializeVirtualBufferCaches(memoryManager, ioDeviceNum);
+ }
+ return vbcs;
+ }
+ }
+
+ public long getTotalSize() {
+ int numPages = datasetID < firstAvilableUserDatasetID
+ ? storageProperties.getMetadataMemoryComponentNumPages()
+ : storageProperties.getMemoryComponentNumPages();
+ return storageProperties.getMemoryComponentPageSize() * ((long) numPages);
+ }
+}
[2/3] asterixdb git commit: Small Cleanup Towards Fixing LifeCycle
Issues
Posted by am...@apache.org.
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/IndexInfo.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/IndexInfo.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/IndexInfo.java
new file mode 100644
index 0000000..34ccce0
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/IndexInfo.java
@@ -0,0 +1,45 @@
+/*
+ * 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.asterix.common.context;
+
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+
+public class IndexInfo extends Info {
+ private final ILSMIndex index;
+ private final long resourceId;
+ private final int datasetId;
+
+ public IndexInfo(ILSMIndex index, int datasetId, long resourceId) {
+ this.index = index;
+ this.datasetId = datasetId;
+ this.resourceId = resourceId;
+ }
+
+ public ILSMIndex getIndex() {
+ return index;
+ }
+
+ public long getResourceId() {
+ return resourceId;
+ }
+
+ public int getDatasetId() {
+ return datasetId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/Info.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/Info.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/Info.java
new file mode 100644
index 0000000..999eb34
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/Info.java
@@ -0,0 +1,53 @@
+/*
+ * 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.asterix.common.context;
+
+public abstract class Info {
+ private int referenceCount;
+ private boolean isOpen;
+
+ public Info() {
+ referenceCount = 0;
+ isOpen = false;
+ }
+
+ public void touch() {
+ ++referenceCount;
+ }
+
+ public void untouch() {
+ --referenceCount;
+ }
+
+ public int getReferenceCount() {
+ return referenceCount;
+ }
+
+ public void setReferenceCount(int referenceCount) {
+ this.referenceCount = referenceCount;
+ }
+
+ public boolean isOpen() {
+ return isOpen;
+ }
+
+ public void setOpen(boolean isOpen) {
+ this.isOpen = isOpen;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
index 71c30d5..6350d73 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
@@ -22,7 +22,6 @@ package org.apache.asterix.common.context;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.asterix.common.context.DatasetLifecycleManager.DatasetInfo;
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
import org.apache.asterix.common.transactions.AbstractOperationCallback;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/AbstractApplicationMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/AbstractApplicationMessage.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/AbstractApplicationMessage.java
deleted file mode 100644
index 5737957..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/AbstractApplicationMessage.java
+++ /dev/null
@@ -1,36 +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.asterix.common.messaging;
-
-import org.apache.asterix.common.messaging.api.IApplicationMessage;
-
-public abstract class AbstractApplicationMessage implements IApplicationMessage {
- private static final long serialVersionUID = 1L;
- protected long id;
-
- @Override
- public void setId(long id) {
- this.id = id;
- }
-
- @Override
- public long getId() {
- return id;
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessage.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessage.java
index b93082d..dbd2139 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessage.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessage.java
@@ -22,31 +22,11 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.messages.IMessage;
import org.apache.hyracks.api.service.IControllerService;
+@FunctionalInterface
public interface IApplicationMessage extends IMessage {
/**
- * Sets a unique message id that identifies this message within an NC.
- * This id is set by {@link INCMessageBroker#sendMessageToCC(IApplicationMessage, IApplicationMessageCallback)}
- * when the callback is not null to notify the sender when the response to that message is received.
- *
- * @param messageId
- */
- public void setId(long messageId);
-
- /**
- * @return The unique message id if it has been set, otherwise 0.
- */
- public long getId();
-
- /**
* handle the message upon delivery
*/
- public void handle(IControllerService cs) throws HyracksDataException;
-
- /**
- * get a string representation for the message type
- *
- * @return
- */
- public String type();
+ public void handle(IControllerService cs) throws HyracksDataException, InterruptedException;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessageCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessageCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessageCallback.java
deleted file mode 100644
index 3bad5fb..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/IApplicationMessageCallback.java
+++ /dev/null
@@ -1,30 +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.asterix.common.messaging.api;
-
-public interface IApplicationMessageCallback {
-
- /**
- * Notifies the message sender when the response has been received.
- *
- * @param message
- * The response message
- */
- public void deliverMessageResponse(IApplicationMessage message);
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/ICCMessageBroker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/ICCMessageBroker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/ICCMessageBroker.java
index 2290b93..ced2b6d 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/ICCMessageBroker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/ICCMessageBroker.java
@@ -21,7 +21,6 @@ package org.apache.asterix.common.messaging.api;
import org.apache.hyracks.api.messages.IMessageBroker;
public interface ICCMessageBroker extends IMessageBroker {
- public static final long NO_CALLBACK_MESSAGE_ID = -1;
/**
* Sends the passed message to the specified {@code nodeId}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java
index f01d0c3..707f864 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java
@@ -29,7 +29,7 @@ public interface INCMessageBroker extends IMessageBroker {
* @param callback
* @throws Exception
*/
- public void sendMessageToCC(IApplicationMessage message, IApplicationMessageCallback callback) throws Exception;
+ public void sendMessageToCC(IApplicationMessage message) throws Exception;
/**
* Sends application message from this NC to another NC.
@@ -38,7 +38,7 @@ public interface INCMessageBroker extends IMessageBroker {
* @param callback
* @throws Exception
*/
- public void sendMessageToNC(String nodeId, IApplicationMessage message, IApplicationMessageCallback callback)
+ public void sendMessageToNC(String nodeId, IApplicationMessage message)
throws Exception;
/**
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IAsterixAppRuntimeContextProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IAsterixAppRuntimeContextProvider.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IAsterixAppRuntimeContextProvider.java
index d0dd3fa..b66bcf8 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IAsterixAppRuntimeContextProvider.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IAsterixAppRuntimeContextProvider.java
@@ -27,7 +27,6 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
import org.apache.hyracks.storage.common.file.IFileMapProvider;
import org.apache.hyracks.storage.common.file.ILocalResourceRepository;
-import org.apache.hyracks.storage.common.file.IResourceIdFactory;
public interface IAsterixAppRuntimeContextProvider {
@@ -49,8 +48,6 @@ public interface IAsterixAppRuntimeContextProvider {
public ILocalResourceRepository getLocalResourceRepository();
- public IResourceIdFactory getResourceIdFactory();
-
public IIOManager getIOManager();
public IAsterixAppRuntimeContext getAppContext();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java
index c929b41..6de796e 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/MetadataNode.java
@@ -407,7 +407,7 @@ public class MetadataNode implements IMetadataNode {
throws ACIDException, HyracksDataException, IndexException {
long resourceID = metadataIndex.getResourceID();
String resourceName = metadataIndex.getFile().toString();
- ILSMIndex lsmIndex = (ILSMIndex) datasetLifecycleManager.getIndex(resourceName);
+ ILSMIndex lsmIndex = (ILSMIndex) datasetLifecycleManager.get(resourceName);
try {
datasetLifecycleManager.open(resourceName);
@@ -674,7 +674,7 @@ public class MetadataNode implements IMetadataNode {
throws ACIDException, HyracksDataException, IndexException {
long resourceID = metadataIndex.getResourceID();
String resourceName = metadataIndex.getFile().toString();
- ILSMIndex lsmIndex = (ILSMIndex) datasetLifecycleManager.getIndex(resourceName);
+ ILSMIndex lsmIndex = (ILSMIndex) datasetLifecycleManager.get(resourceName);
try {
datasetLifecycleManager.open(resourceName);
// prepare a Callback for logging
@@ -1066,7 +1066,7 @@ public class MetadataNode implements IMetadataNode {
try {
IMetadataIndex index = MetadataPrimaryIndexes.DATAVERSE_DATASET;
String resourceName = index.getFile().toString();
- IIndex indexInstance = datasetLifecycleManager.getIndex(resourceName);
+ IIndex indexInstance = datasetLifecycleManager.get(resourceName);
datasetLifecycleManager.open(resourceName);
IIndexAccessor indexAccessor = indexInstance.createAccessor(NoOpOperationCallback.INSTANCE,
NoOpOperationCallback.INSTANCE);
@@ -1088,7 +1088,7 @@ public class MetadataNode implements IMetadataNode {
datasetLifecycleManager.close(resourceName);
index = MetadataPrimaryIndexes.DATASET_DATASET;
- indexInstance = datasetLifecycleManager.getIndex(resourceName);
+ indexInstance = datasetLifecycleManager.get(resourceName);
datasetLifecycleManager.open(resourceName);
indexAccessor = indexInstance.createAccessor(NoOpOperationCallback.INSTANCE,
NoOpOperationCallback.INSTANCE);
@@ -1113,7 +1113,7 @@ public class MetadataNode implements IMetadataNode {
datasetLifecycleManager.close(resourceName);
index = MetadataPrimaryIndexes.INDEX_DATASET;
- indexInstance = datasetLifecycleManager.getIndex(resourceName);
+ indexInstance = datasetLifecycleManager.get(resourceName);
datasetLifecycleManager.open(resourceName);
indexAccessor = indexInstance.createAccessor(NoOpOperationCallback.INSTANCE,
NoOpOperationCallback.INSTANCE);
@@ -1149,7 +1149,7 @@ public class MetadataNode implements IMetadataNode {
throw new MetadataException("No file for Index " + index.getDataverseName() + "." + index.getIndexName());
}
String resourceName = index.getFile().toString();
- IIndex indexInstance = datasetLifecycleManager.getIndex(resourceName);
+ IIndex indexInstance = datasetLifecycleManager.get(resourceName);
datasetLifecycleManager.open(resourceName);
IIndexAccessor indexAccessor = indexInstance.createAccessor(NoOpOperationCallback.INSTANCE,
NoOpOperationCallback.INSTANCE);
@@ -1187,7 +1187,7 @@ public class MetadataNode implements IMetadataNode {
int mostRecentDatasetId = MetadataIndexImmutableProperties.FIRST_AVAILABLE_USER_DATASET_ID;
try {
String resourceName = MetadataPrimaryIndexes.DATASET_DATASET.getFile().toString();
- IIndex indexInstance = datasetLifecycleManager.getIndex(resourceName);
+ IIndex indexInstance = datasetLifecycleManager.get(resourceName);
datasetLifecycleManager.open(resourceName);
try {
IIndexAccessor indexAccessor = indexInstance.createAccessor(NoOpOperationCallback.INSTANCE,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
index 2629fea..55b1045 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/bootstrap/MetadataBootstrap.java
@@ -32,9 +32,9 @@ import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.asterix.common.api.ILocalResourceMetadata;
import org.apache.asterix.common.cluster.ClusterPartition;
import org.apache.asterix.common.config.AsterixMetadataProperties;
+import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.config.GlobalConfig;
import org.apache.asterix.common.config.IAsterixPropertiesProvider;
-import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.context.BaseOperationTracker;
import org.apache.asterix.common.context.CorrelatedPrefixMergePolicyFactory;
import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallbackFactory;
@@ -59,10 +59,10 @@ import org.apache.asterix.metadata.entities.Dataverse;
import org.apache.asterix.metadata.entities.FeedPolicyEntity;
import org.apache.asterix.metadata.entities.Index;
import org.apache.asterix.metadata.entities.InternalDatasetDetails;
-import org.apache.asterix.metadata.entities.Node;
-import org.apache.asterix.metadata.entities.NodeGroup;
import org.apache.asterix.metadata.entities.InternalDatasetDetails.FileStructure;
import org.apache.asterix.metadata.entities.InternalDatasetDetails.PartitioningStrategy;
+import org.apache.asterix.metadata.entities.Node;
+import org.apache.asterix.metadata.entities.NodeGroup;
import org.apache.asterix.metadata.feeds.BuiltinFeedPolicies;
import org.apache.asterix.metadata.utils.MetadataConstants;
import org.apache.asterix.om.types.BuiltinType;
@@ -319,6 +319,8 @@ public class MetadataBootstrap {
String resourceName = metadataPartitionPath + File.separator + index.getFileNameRelativePath();
FileReference file = ioManager.getAbsoluteFileRef(metadataDeviceId, resourceName);
+ // this should not be done this way. dataset lifecycle manager shouldn't return virtual buffer caches for
+ // a dataset that was not yet created
List<IVirtualBufferCache> virtualBufferCaches = runtimeContext.getDatasetLifecycleManager()
.getVirtualBufferCaches(index.getDatasetId().getId(), metadataPartition.getIODeviceNum());
ITypeTraits[] typeTraits = index.getTypeTraits();
@@ -360,7 +362,7 @@ public class MetadataBootstrap {
+ " to intialize as a new instance. (WARNING: all data will be lost.)");
}
resourceID = resource.getResourceId();
- lsmBtree = (LSMBTree) dataLifecycleManager.getIndex(absolutePath);
+ lsmBtree = (LSMBTree) dataLifecycleManager.get(absolutePath);
if (lsmBtree == null) {
lsmBtree = LSMBTreeUtils.createLSMTree(virtualBufferCaches, file, bufferCache, fileMapProvider,
typeTraits, comparatorFactories, bloomFilterKeyFields,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
index 5a88a9f..c0d8320 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
@@ -46,7 +46,7 @@ import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.asterix.common.cluster.ClusterPartition;
import org.apache.asterix.common.config.AsterixReplicationProperties;
import org.apache.asterix.common.config.IAsterixPropertiesProvider;
-import org.apache.asterix.common.context.DatasetLifecycleManager.IndexInfo;
+import org.apache.asterix.common.context.IndexInfo;
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
import org.apache.asterix.common.replication.IReplicaResourcesManager;
@@ -59,7 +59,6 @@ import org.apache.asterix.common.transactions.ILogManager;
import org.apache.asterix.common.transactions.LogRecord;
import org.apache.asterix.common.transactions.LogSource;
import org.apache.asterix.common.transactions.LogType;
-import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
import org.apache.asterix.common.utils.TransactionUtil;
import org.apache.asterix.replication.functions.ReplicaFilesRequest;
import org.apache.asterix.replication.functions.ReplicaIndexFlushRequest;
@@ -70,6 +69,7 @@ import org.apache.asterix.replication.storage.LSMComponentLSNSyncTask;
import org.apache.asterix.replication.storage.LSMComponentProperties;
import org.apache.asterix.replication.storage.LSMIndexFileProperties;
import org.apache.asterix.replication.storage.ReplicaResourcesManager;
+import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
import org.apache.hyracks.api.application.INCApplicationContext;
import org.apache.hyracks.storage.am.common.api.IMetaDataPageManager;
import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/AbstractFailbackPlanMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/AbstractFailbackPlanMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/AbstractFailbackPlanMessage.java
index 2eb1be9..7c386d5 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/AbstractFailbackPlanMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/AbstractFailbackPlanMessage.java
@@ -18,9 +18,9 @@
*/
package org.apache.asterix.runtime.message;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
-public abstract class AbstractFailbackPlanMessage extends AbstractApplicationMessage {
+public abstract class AbstractFailbackPlanMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
protected final long planId;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackRequestMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackRequestMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackRequestMessage.java
index c96bcb8..5cb1a6a 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackRequestMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackRequestMessage.java
@@ -55,7 +55,8 @@ public class CompleteFailbackRequestMessage extends AbstractFailbackPlanMessage
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
- sb.append("Plan ID: " + planId);
+ sb.append(CompleteFailbackRequestMessage.class.getSimpleName());
+ sb.append(" Plan ID: " + planId);
sb.append(" Node ID: " + nodeId);
sb.append(" Partitions: " + partitions);
return sb.toString();
@@ -78,7 +79,7 @@ public class CompleteFailbackRequestMessage extends AbstractFailbackPlanMessage
CompleteFailbackResponseMessage reponse = new CompleteFailbackResponseMessage(planId,
requestId, partitions);
try {
- broker.sendMessageToCC(reponse, null);
+ broker.sendMessageToCC(reponse);
} catch (Exception e) {
LOGGER.log(Level.SEVERE, "Failure sending message to CC", e);
hde = ExceptionUtils.suppressIntoHyracksDataException(hde, e);
@@ -88,9 +89,4 @@ public class CompleteFailbackRequestMessage extends AbstractFailbackPlanMessage
throw hde;
}
}
-
- @Override
- public String type() {
- return "COMPLETE_FAILBACK_REQUEST";
- }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackResponseMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackResponseMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackResponseMessage.java
index 3c4b58c..73e5fd2 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackResponseMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/CompleteFailbackResponseMessage.java
@@ -41,7 +41,8 @@ public class CompleteFailbackResponseMessage extends AbstractFailbackPlanMessage
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
- sb.append("Plan ID: " + planId);
+ sb.append(CompleteFailbackResponseMessage.class.getSimpleName());
+ sb.append(" Plan ID: " + planId);
sb.append(" Partitions: " + partitions);
return sb.toString();
}
@@ -50,9 +51,4 @@ public class CompleteFailbackResponseMessage extends AbstractFailbackPlanMessage
public void handle(IControllerService cs) throws HyracksDataException {
AsterixClusterProperties.INSTANCE.processCompleteFailbackResponse(this);
}
-
- @Override
- public String type() {
- return "COMPLETE_FAILBACK_RESPONSE";
- }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackRequestMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackRequestMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackRequestMessage.java
index e3b9fbe..c112366 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackRequestMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackRequestMessage.java
@@ -64,7 +64,8 @@ public class PreparePartitionsFailbackRequestMessage extends AbstractFailbackPla
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
- sb.append("Plan ID: " + planId);
+ sb.append(PreparePartitionsFailbackRequestMessage.class.getSimpleName());
+ sb.append(" Plan ID: " + planId);
sb.append(" Partitions: " + partitions);
sb.append(" releaseMetadataNode: " + releaseMetadataNode);
return sb.toString();
@@ -109,15 +110,10 @@ public class PreparePartitionsFailbackRequestMessage extends AbstractFailbackPla
PreparePartitionsFailbackResponseMessage reponse = new PreparePartitionsFailbackResponseMessage(planId,
requestId, partitions);
try {
- broker.sendMessageToCC(reponse, null);
+ broker.sendMessageToCC(reponse);
} catch (Exception e) {
LOGGER.log(Level.SEVERE, "Failed sending message to cc", e);
throw ExceptionUtils.convertToHyracksDataException(e);
}
}
-
- @Override
- public String type() {
- return "PREPARE_PARTITIONS_FAILBACK_REQUEST";
- }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackResponseMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackResponseMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackResponseMessage.java
index 2e52773..299121a 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackResponseMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/PreparePartitionsFailbackResponseMessage.java
@@ -44,7 +44,7 @@ public class PreparePartitionsFailbackResponseMessage extends AbstractFailbackPl
}
@Override
- public String type() {
- return "PREPARE_PARTITIONS_FAILBACK_RESPONSE";
+ public String toString() {
+ return PreparePartitionsFailbackResponseMessage.class.getSimpleName() + " " + partitions.toString();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReplicaEventMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReplicaEventMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReplicaEventMessage.java
index 2aa4746..fc55968 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReplicaEventMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReplicaEventMessage.java
@@ -19,7 +19,7 @@
package org.apache.asterix.runtime.message;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.common.replication.Replica;
import org.apache.asterix.common.replication.ReplicaEvent;
import org.apache.asterix.event.schema.cluster.Node;
@@ -28,7 +28,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
import org.apache.hyracks.control.nc.NodeControllerService;
-public class ReplicaEventMessage extends AbstractApplicationMessage {
+public class ReplicaEventMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private final String nodeId;
@@ -66,7 +66,7 @@ public class ReplicaEventMessage extends AbstractApplicationMessage {
}
@Override
- public String type() {
- return "REPLICA_EVENT";
+ public String toString() {
+ return ReplicaEventMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdMessage.java
index 1604e29..f9f6233 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdMessage.java
@@ -23,7 +23,7 @@ import java.util.logging.Logger;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.exceptions.ExceptionUtils;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.common.messaging.api.INCMessageBroker;
import org.apache.asterix.common.metadata.MetadataIndexImmutableProperties;
import org.apache.asterix.common.transactions.IAsterixResourceIdManager;
@@ -32,7 +32,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
import org.apache.hyracks.control.nc.NodeControllerService;
-public class ReportMaxResourceIdMessage extends AbstractApplicationMessage {
+public class ReportMaxResourceIdMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private static final Logger LOGGER = Logger.getLogger(ReportMaxResourceIdMessage.class.getName());
private final long maxResourceId;
@@ -62,7 +62,7 @@ public class ReportMaxResourceIdMessage extends AbstractApplicationMessage {
MetadataIndexImmutableProperties.FIRST_AVAILABLE_USER_DATASET_ID);
ReportMaxResourceIdMessage maxResourceIdMsg = new ReportMaxResourceIdMessage(ncs.getId(), maxResourceId);
try {
- ((INCMessageBroker) ncs.getApplicationContext().getMessageBroker()).sendMessageToCC(maxResourceIdMsg, null);
+ ((INCMessageBroker) ncs.getApplicationContext().getMessageBroker()).sendMessageToCC(maxResourceIdMsg);
} catch (Exception e) {
LOGGER.log(Level.SEVERE, "Unable to report max local resource id", e);
throw ExceptionUtils.convertToHyracksDataException(e);
@@ -70,7 +70,7 @@ public class ReportMaxResourceIdMessage extends AbstractApplicationMessage {
}
@Override
- public String type() {
- return "REPORT_MAX_RESOURCE_ID_RESPONSE";
+ public String toString() {
+ return ReportMaxResourceIdMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdRequestMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdRequestMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdRequestMessage.java
index 3e2becf..203104e 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdRequestMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ReportMaxResourceIdRequestMessage.java
@@ -18,12 +18,12 @@
*/
package org.apache.asterix.runtime.message;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
import org.apache.hyracks.control.nc.NodeControllerService;
-public class ReportMaxResourceIdRequestMessage extends AbstractApplicationMessage {
+public class ReportMaxResourceIdRequestMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
@Override
@@ -32,7 +32,7 @@ public class ReportMaxResourceIdRequestMessage extends AbstractApplicationMessag
}
@Override
- public String type() {
- return "REPORT_MAX_RESOURCE_ID_REQUEST";
+ public String toString() {
+ return ReportMaxResourceIdRequestMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java
index 5e7d808..32d3f64 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java
@@ -21,7 +21,7 @@ package org.apache.asterix.runtime.message;
import java.util.Set;
import org.apache.asterix.common.exceptions.ExceptionUtils;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.common.messaging.api.ICCMessageBroker;
import org.apache.asterix.common.transactions.IAsterixResourceIdManager;
import org.apache.asterix.runtime.util.AsterixAppContextInfo;
@@ -29,7 +29,7 @@ import org.apache.asterix.runtime.util.AsterixClusterProperties;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
-public class ResourceIdRequestMessage extends AbstractApplicationMessage {
+public class ResourceIdRequestMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private final String src;
@@ -43,7 +43,6 @@ public class ResourceIdRequestMessage extends AbstractApplicationMessage {
ICCMessageBroker broker =
(ICCMessageBroker) AsterixAppContextInfo.INSTANCE.getCCApplicationContext().getMessageBroker();
ResourceIdRequestResponseMessage reponse = new ResourceIdRequestResponseMessage();
- reponse.setId(id);
if (!AsterixClusterProperties.INSTANCE.isClusterActive()) {
reponse.setResourceId(-1);
reponse.setException(new Exception("Cannot generate global resource id when cluster is not active."));
@@ -66,7 +65,6 @@ public class ResourceIdRequestMessage extends AbstractApplicationMessage {
throws Exception {
Set<String> getParticipantNodes = AsterixClusterProperties.INSTANCE.getParticipantNodes();
ReportMaxResourceIdRequestMessage msg = new ReportMaxResourceIdRequestMessage();
- msg.setId(ICCMessageBroker.NO_CALLBACK_MESSAGE_ID);
for (String nodeId : getParticipantNodes) {
if (!resourceIdManager.reported(nodeId)) {
broker.sendApplicationMessageToNC(msg, nodeId);
@@ -75,7 +73,7 @@ public class ResourceIdRequestMessage extends AbstractApplicationMessage {
}
@Override
- public String type() {
- return "RESOURCE_ID_REQUEST";
+ public String toString() {
+ return ReportMaxResourceIdRequestMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestResponseMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestResponseMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestResponseMessage.java
index 62c2163..d4cc022 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestResponseMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestResponseMessage.java
@@ -18,11 +18,14 @@
*/
package org.apache.asterix.runtime.message;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
+import org.apache.asterix.runtime.transaction.GlobalResourceIdFactory;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
+import org.apache.hyracks.control.nc.NodeControllerService;
-public class ResourceIdRequestResponseMessage extends AbstractApplicationMessage {
+public class ResourceIdRequestResponseMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private long resourceId;
@@ -45,13 +48,15 @@ public class ResourceIdRequestResponseMessage extends AbstractApplicationMessage
}
@Override
- public void handle(IControllerService cs) throws HyracksDataException {
- // Do nothing. for this message, the callback handles it, we probably should get rid of callbacks and
- // instead, use the handle in the response to perform callback action
+ public void handle(IControllerService cs) throws HyracksDataException, InterruptedException {
+ NodeControllerService ncs = (NodeControllerService) cs;
+ IAsterixAppRuntimeContext asterixNcAppRuntimeCtx =
+ (IAsterixAppRuntimeContext) ncs.getApplicationContext().getApplicationObject();
+ ((GlobalResourceIdFactory) asterixNcAppRuntimeCtx.getResourceIdFactory()).addNewIds(this);
}
@Override
- public String type() {
- return "RESOURCE_ID_RESPONSE";
+ public String toString() {
+ return ResourceIdRequestResponseMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeRequestMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeRequestMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeRequestMessage.java
index b18a879..e877f52 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeRequestMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeRequestMessage.java
@@ -23,13 +23,13 @@ import java.util.logging.Logger;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.exceptions.ExceptionUtils;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.common.messaging.api.INCMessageBroker;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
import org.apache.hyracks.control.nc.NodeControllerService;
-public class TakeoverMetadataNodeRequestMessage extends AbstractApplicationMessage {
+public class TakeoverMetadataNodeRequestMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private static final Logger LOGGER = Logger.getLogger(TakeoverMetadataNodeRequestMessage.class.getName());
@@ -51,7 +51,7 @@ public class TakeoverMetadataNodeRequestMessage extends AbstractApplicationMessa
TakeoverMetadataNodeResponseMessage reponse = new TakeoverMetadataNodeResponseMessage(
appContext.getTransactionSubsystem().getId());
try {
- broker.sendMessageToCC(reponse, null);
+ broker.sendMessageToCC(reponse);
} catch (Exception e) {
LOGGER.log(Level.SEVERE, "Failed taking over metadata", e);
hde = ExceptionUtils.suppressIntoHyracksDataException(hde, e);
@@ -63,7 +63,7 @@ public class TakeoverMetadataNodeRequestMessage extends AbstractApplicationMessa
}
@Override
- public String type() {
- return "TAKEOVER_METADATA_NODE_REQUEST";
+ public String toString() {
+ return TakeoverMetadataNodeRequestMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeResponseMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeResponseMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeResponseMessage.java
index f7016bc..e18fc8d 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeResponseMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverMetadataNodeResponseMessage.java
@@ -18,12 +18,12 @@
*/
package org.apache.asterix.runtime.message;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.runtime.util.AsterixClusterProperties;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
-public class TakeoverMetadataNodeResponseMessage extends AbstractApplicationMessage {
+public class TakeoverMetadataNodeResponseMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private final String nodeId;
@@ -42,7 +42,7 @@ public class TakeoverMetadataNodeResponseMessage extends AbstractApplicationMess
}
@Override
- public String type() {
- return "TAKEOVER_METADATA_NODE_RESPONSE";
+ public String toString() {
+ return TakeoverMetadataNodeResponseMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsRequestMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsRequestMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsRequestMessage.java
index 3b91084..e024eed 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsRequestMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsRequestMessage.java
@@ -25,14 +25,14 @@ import java.util.logging.Logger;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.exceptions.ExceptionUtils;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.common.messaging.api.INCMessageBroker;
import org.apache.asterix.common.replication.IRemoteRecoveryManager;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
import org.apache.hyracks.control.nc.NodeControllerService;
-public class TakeoverPartitionsRequestMessage extends AbstractApplicationMessage {
+public class TakeoverPartitionsRequestMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private static final Logger LOGGER = Logger.getLogger(TakeoverPartitionsRequestMessage.class.getName());
@@ -61,7 +61,8 @@ public class TakeoverPartitionsRequestMessage extends AbstractApplicationMessage
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
- sb.append("Request ID: " + requestId);
+ sb.append(TakeoverPartitionsRequestMessage.class.getSimpleName());
+ sb.append(" Request ID: " + requestId);
sb.append(" Node ID: " + nodeId);
sb.append(" Partitions: ");
for (Integer partitionId : partitions) {
@@ -92,7 +93,7 @@ public class TakeoverPartitionsRequestMessage extends AbstractApplicationMessage
TakeoverPartitionsResponseMessage reponse = new TakeoverPartitionsResponseMessage(requestId,
appContext.getTransactionSubsystem().getId(), partitions);
try {
- broker.sendMessageToCC(reponse, null);
+ broker.sendMessageToCC(reponse);
} catch (Exception e) {
LOGGER.log(Level.SEVERE, "Failure taking over partitions", e);
hde = ExceptionUtils.suppressIntoHyracksDataException(hde, e);
@@ -103,9 +104,4 @@ public class TakeoverPartitionsRequestMessage extends AbstractApplicationMessage
}
}
}
-
- @Override
- public String type() {
- return "TAKEOVER_PARTITIONS_REQUEST";
- }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsResponseMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsResponseMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsResponseMessage.java
index 9ec71b7..1c57a2a 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsResponseMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/TakeoverPartitionsResponseMessage.java
@@ -18,12 +18,12 @@
*/
package org.apache.asterix.runtime.message;
-import org.apache.asterix.common.messaging.AbstractApplicationMessage;
+import org.apache.asterix.common.messaging.api.IApplicationMessage;
import org.apache.asterix.runtime.util.AsterixClusterProperties;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.service.IControllerService;
-public class TakeoverPartitionsResponseMessage extends AbstractApplicationMessage {
+public class TakeoverPartitionsResponseMessage implements IApplicationMessage {
private static final long serialVersionUID = 1L;
private final Integer[] partitions;
@@ -54,7 +54,7 @@ public class TakeoverPartitionsResponseMessage extends AbstractApplicationMessag
}
@Override
- public String type() {
- return "TAKEOVER_PARTITIONS_RESPONSE";
+ public String toString() {
+ return TakeoverPartitionsResponseMessage.class.getSimpleName();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/GlobalResourceIdFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/GlobalResourceIdFactory.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/GlobalResourceIdFactory.java
index 4cb65c2..a8b61f7 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/GlobalResourceIdFactory.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/GlobalResourceIdFactory.java
@@ -20,8 +20,6 @@ package org.apache.asterix.runtime.transaction;
import java.util.concurrent.LinkedBlockingQueue;
-import org.apache.asterix.common.messaging.api.IApplicationMessage;
-import org.apache.asterix.common.messaging.api.IApplicationMessageCallback;
import org.apache.asterix.common.messaging.api.INCMessageBroker;
import org.apache.asterix.runtime.message.ResourceIdRequestMessage;
import org.apache.asterix.runtime.message.ResourceIdRequestResponseMessage;
@@ -34,10 +32,10 @@ import org.apache.hyracks.storage.common.file.IResourceIdFactory;
* A resource id factory that generates unique resource ids across all NCs by requesting
* unique ids from the cluster controller.
*/
-public class GlobalResourceIdFactory implements IResourceIdFactory, IApplicationMessageCallback {
+public class GlobalResourceIdFactory implements IResourceIdFactory {
private final IApplicationContext appCtx;
- private final LinkedBlockingQueue<IApplicationMessage> resourceIdResponseQ;
+ private final LinkedBlockingQueue<ResourceIdRequestResponseMessage> resourceIdResponseQ;
private final String nodeId;
public GlobalResourceIdFactory(IApplicationContext appCtx) {
@@ -46,6 +44,10 @@ public class GlobalResourceIdFactory implements IResourceIdFactory, IApplication
this.nodeId = ((NodeControllerService) appCtx.getControllerService()).getApplicationContext().getNodeId();
}
+ public void addNewIds(ResourceIdRequestResponseMessage resourceIdResponse) throws InterruptedException {
+ resourceIdResponseQ.put(resourceIdResponse);
+ }
+
@Override
public long createId() throws HyracksDataException {
try {
@@ -54,15 +56,15 @@ public class GlobalResourceIdFactory implements IResourceIdFactory, IApplication
if (!resourceIdResponseQ.isEmpty()) {
synchronized (resourceIdResponseQ) {
if (!resourceIdResponseQ.isEmpty()) {
- reponse = (ResourceIdRequestResponseMessage) resourceIdResponseQ.take();
+ reponse = resourceIdResponseQ.take();
}
}
}
//if no response available or it has an exception, request a new one
if (reponse == null || reponse.getException() != null) {
ResourceIdRequestMessage msg = new ResourceIdRequestMessage(nodeId);
- ((INCMessageBroker) appCtx.getMessageBroker()).sendMessageToCC(msg, this);
- reponse = (ResourceIdRequestResponseMessage) resourceIdResponseQ.take();
+ ((INCMessageBroker) appCtx.getMessageBroker()).sendMessageToCC(msg);
+ reponse = resourceIdResponseQ.take();
if (reponse.getException() != null) {
throw new HyracksDataException(reponse.getException().getMessage());
}
@@ -72,9 +74,4 @@ public class GlobalResourceIdFactory implements IResourceIdFactory, IApplication
throw new HyracksDataException(e);
}
}
-
- @Override
- public void deliverMessageResponse(IApplicationMessage message) {
- resourceIdResponseQ.offer(message);
- }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/util/AsterixClusterProperties.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/util/AsterixClusterProperties.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/util/AsterixClusterProperties.java
index d201d60..d9a55ad 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/util/AsterixClusterProperties.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/util/AsterixClusterProperties.java
@@ -411,7 +411,7 @@ public class AsterixClusterProperties {
for (Integer partitionId : request.getPartitions()) {
nodePartitions.add(clusterPartitions.get(partitionId));
}
- failedTakeoverRequests.add(request.getId());
+ failedTakeoverRequests.add(request.getRequestId());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java
index 52e2818..e5f3555 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallbackFactory.java
@@ -29,7 +29,7 @@ import org.apache.asterix.common.transactions.JobId;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
@@ -59,9 +59,9 @@ public class PrimaryIndexModificationOperationCallbackFactory extends AbstractOp
throws HyracksDataException {
ITransactionSubsystem txnSubsystem = txnSubsystemProvider.getTransactionSubsystem(ctx);
- IIndexLifecycleManager indexLifeCycleManager =
+ IResourceLifecycleManager indexLifeCycleManager =
txnSubsystem.getAsterixAppRuntimeContextProvider().getDatasetLifecycleManager();
- ILSMIndex index = (ILSMIndex) indexLifeCycleManager.getIndex(resourcePath);
+ ILSMIndex index = (ILSMIndex) indexLifeCycleManager.get(resourcePath);
if (index == null) {
throw new HyracksDataException("Index(id:" + resourceId + ") is not registered.");
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java
index c6743dd..f9f2b89 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/SecondaryIndexModificationOperationCallbackFactory.java
@@ -29,7 +29,7 @@ import org.apache.asterix.common.transactions.JobId;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
@@ -55,9 +55,9 @@ public class SecondaryIndexModificationOperationCallbackFactory extends Abstract
int resourcePartition, Object resource, IHyracksTaskContext ctx, IOperatorNodePushable operatorNodePushable)
throws HyracksDataException {
ITransactionSubsystem txnSubsystem = txnSubsystemProvider.getTransactionSubsystem(ctx);
- IIndexLifecycleManager indexLifeCycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
+ IResourceLifecycleManager indexLifeCycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
.getDatasetLifecycleManager();
- ILSMIndex index = (ILSMIndex) indexLifeCycleManager.getIndex(resourcePath);
+ ILSMIndex index = (ILSMIndex) indexLifeCycleManager.get(resourcePath);
if (index == null) {
throw new HyracksDataException("Index(id:" + resourceId + ") is not registered.");
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java
index 8c91c1a..df7d65d 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetPrimaryIndexModificationOperationCallbackFactory.java
@@ -29,7 +29,7 @@ import org.apache.asterix.common.transactions.JobId;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
@@ -53,9 +53,9 @@ public class TempDatasetPrimaryIndexModificationOperationCallbackFactory extends
int resourcePartition, Object resource, IHyracksTaskContext ctx, IOperatorNodePushable operatorNodePushable)
throws HyracksDataException {
ITransactionSubsystem txnSubsystem = txnSubsystemProvider.getTransactionSubsystem(ctx);
- IIndexLifecycleManager indexLifeCycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
+ IResourceLifecycleManager indexLifeCycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
.getDatasetLifecycleManager();
- ILSMIndex index = (ILSMIndex) indexLifeCycleManager.getIndex(resourcePath);
+ ILSMIndex index = (ILSMIndex) indexLifeCycleManager.get(resourcePath);
if (index == null) {
throw new HyracksDataException("Index(id:" + resourceId + ") is not registered.");
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java
index 3e11531..39e916b 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/TempDatasetSecondaryIndexModificationOperationCallbackFactory.java
@@ -29,7 +29,7 @@ import org.apache.asterix.common.transactions.JobId;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
@@ -53,9 +53,9 @@ public class TempDatasetSecondaryIndexModificationOperationCallbackFactory exten
int resourcePartition, Object resource, IHyracksTaskContext ctx, IOperatorNodePushable operatorNodePushable)
throws HyracksDataException {
ITransactionSubsystem txnSubsystem = txnSubsystemProvider.getTransactionSubsystem(ctx);
- IIndexLifecycleManager indexLifeCycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
+ IResourceLifecycleManager indexLifeCycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
.getDatasetLifecycleManager();
- ILSMIndex index = (ILSMIndex) indexLifeCycleManager.getIndex(resourcePath);
+ ILSMIndex index = (ILSMIndex) indexLifeCycleManager.get(resourcePath);
if (index == null) {
throw new HyracksDataException("Index(id:" + resourceId + ") is not registered.");
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
index 5bf1505..85cdc89 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/UpsertOperationCallbackFactory.java
@@ -28,7 +28,7 @@ import org.apache.asterix.common.transactions.JobId;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
@@ -55,9 +55,9 @@ public class UpsertOperationCallbackFactory extends AbstractOperationCallbackFac
throws HyracksDataException {
ITransactionSubsystem txnSubsystem = txnSubsystemProvider.getTransactionSubsystem(ctx);
- IIndexLifecycleManager indexLifeCycleManager =
+ IResourceLifecycleManager indexLifeCycleManager =
txnSubsystem.getAsterixAppRuntimeContextProvider().getDatasetLifecycleManager();
- ILSMIndex index = (ILSMIndex) indexLifeCycleManager.getIndex(resourceName);
+ ILSMIndex index = (ILSMIndex) indexLifeCycleManager.get(resourceName);
if (index == null) {
throw new HyracksDataException("Index(id:" + resourceId + ") is not registered.");
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointThread.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointThread.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointThread.java
index c3cf3e0..851289e 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointThread.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/CheckpointThread.java
@@ -18,21 +18,25 @@
*/
package org.apache.asterix.transaction.management.service.recovery;
+import java.io.IOError;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.transactions.ILogManager;
import org.apache.asterix.common.transactions.IRecoveryManager;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
public class CheckpointThread extends Thread {
+ private static final Logger LOGGER = Logger.getLogger(CheckpointThread.class.getName());
private long lsnThreshold;
private long checkpointTermInSecs;
private final ILogManager logManager;
private final IRecoveryManager recoveryMgr;
- public CheckpointThread(IRecoveryManager recoveryMgr, IIndexLifecycleManager indexLifecycleManager, ILogManager logManager,
+ public CheckpointThread(IRecoveryManager recoveryMgr, ILogManager logManager,
long lsnThreshold, long checkpointTermInSecs) {
this.recoveryMgr = recoveryMgr;
this.logManager = logManager;
@@ -45,24 +49,25 @@ public class CheckpointThread extends Thread {
Thread.currentThread().setName("Checkpoint Thread");
- long currentCheckpointAttemptMinLSN = -1;
+ long currentCheckpointAttemptMinLSN;
long lastCheckpointLSN = -1;
- long currentLogLSN = 0;
- long targetCheckpointLSN = 0;
+ long currentLogLSN;
+ long targetCheckpointLSN;
while (true) {
try {
sleep(checkpointTermInSecs * 1000);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
//ignore
}
-
- if(lastCheckpointLSN == -1)
- {
+ if (lastCheckpointLSN == -1) {
try {
- //Since the system just started up after sharp checkpoint, last checkpoint LSN is considered as the min LSN of the current log partition
+ //Since the system just started up after sharp checkpoint,
+ //last checkpoint LSN is considered as the min LSN of the current log partition
lastCheckpointLSN = logManager.getReadableSmallestLSN();
} catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Error getting smallest readable LSN", e);
lastCheckpointLSN = 0;
}
}
@@ -82,13 +87,12 @@ public class CheckpointThread extends Thread {
currentCheckpointAttemptMinLSN = recoveryMgr.checkpoint(false, targetCheckpointLSN);
//checkpoint was completed at target LSN or above
- if(currentCheckpointAttemptMinLSN >= targetCheckpointLSN)
- {
+ if (currentCheckpointAttemptMinLSN >= targetCheckpointLSN) {
lastCheckpointLSN = currentCheckpointAttemptMinLSN;
}
} catch (ACIDException | HyracksDataException e) {
- throw new Error("failed to checkpoint", e);
+ throw new IOError(e);
}
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
index 0183b29..eb22c22 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
@@ -367,7 +367,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
String resourceAbsolutePath =
partitionIODevicePath + File.separator + localResource.getResourceName();
localResource.setResourcePath(resourceAbsolutePath);
- index = (ILSMIndex) datasetLifecycleManager.getIndex(resourceAbsolutePath);
+ index = (ILSMIndex) datasetLifecycleManager.get(resourceAbsolutePath);
if (index == null) {
//#. create index instance and register to indexLifeCycleManager
localResourceMetadata = (ILocalResourceMetadata) localResource.getResourceObject();
@@ -568,7 +568,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
public long getLocalMinFirstLSN() throws HyracksDataException {
IDatasetLifecycleManager datasetLifecycleManager =
txnSubsystem.getAsterixAppRuntimeContextProvider().getDatasetLifecycleManager();
- List<IIndex> openIndexList = datasetLifecycleManager.getOpenIndexes();
+ List<IIndex> openIndexList = datasetLifecycleManager.getOpenResources();
long firstLSN;
//the min first lsn can only be the current append or smaller
long minFirstLSN = logMgr.getAppendLSN();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionSubsystem.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionSubsystem.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionSubsystem.java
index f7ed355..6b25542 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionSubsystem.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/TransactionSubsystem.java
@@ -81,8 +81,7 @@ public class TransactionSubsystem implements ITransactionSubsystem {
this.recoveryManager = new RecoveryManager(this);
if (asterixAppRuntimeContextProvider != null) {
- this.checkpointThread = new CheckpointThread(recoveryManager,
- asterixAppRuntimeContextProvider.getDatasetLifecycleManager(), logManager,
+ this.checkpointThread = new CheckpointThread(recoveryManager, logManager,
this.txnProperties.getCheckpointLSNThreshold(), this.txnProperties.getCheckpointPollFrequency());
this.checkpointThread.start();
} else {
@@ -95,22 +94,27 @@ public class TransactionSubsystem implements ITransactionSubsystem {
}
}
+ @Override
public ILogManager getLogManager() {
return logManager;
}
+ @Override
public ILockManager getLockManager() {
return lockManager;
}
+ @Override
public ITransactionManager getTransactionManager() {
return transactionManager;
}
+ @Override
public IRecoveryManager getRecoveryManager() {
return recoveryManager;
}
+ @Override
public IAsterixAppRuntimeContextProvider getAsterixAppRuntimeContextProvider() {
return asterixAppRuntimeContextProvider;
}
@@ -119,6 +123,7 @@ public class TransactionSubsystem implements ITransactionSubsystem {
return txnProperties;
}
+ @Override
public String getId() {
return id;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/locking/TestRuntimeContextProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/locking/TestRuntimeContextProvider.java b/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/locking/TestRuntimeContextProvider.java
index 3a4a56a..887532c 100644
--- a/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/locking/TestRuntimeContextProvider.java
+++ b/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/locking/TestRuntimeContextProvider.java
@@ -18,6 +18,8 @@
*/
package org.apache.asterix.transaction.management.service.locking;
+import static org.mockito.Mockito.mock;
+
import java.util.concurrent.Executors;
import org.apache.asterix.common.api.AsterixThreadExecutor;
@@ -31,9 +33,6 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
import org.apache.hyracks.storage.common.file.IFileMapProvider;
import org.apache.hyracks.storage.common.file.ILocalResourceRepository;
-import org.apache.hyracks.storage.common.file.ResourceIdFactory;
-
-import static org.mockito.Mockito.mock;
class TestRuntimeContextProvider implements IAsterixAppRuntimeContextProvider {
@@ -86,11 +85,6 @@ class TestRuntimeContextProvider implements IAsterixAppRuntimeContextProvider {
}
@Override
- public ResourceIdFactory getResourceIdFactory() {
- throw new UnsupportedOperationException();
- }
-
- @Override
public IIOManager getIOManager() {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessage.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessage.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessage.java
index ad19cf0..2f5a873 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessage.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessage.java
@@ -20,9 +20,6 @@ package org.apache.hyracks.api.messages;
import java.io.Serializable;
-/**
- * @author rico
- */
public interface IMessage extends Serializable {
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessageBroker.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessageBroker.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessageBroker.java
index 2b166d2..8aaa563 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessageBroker.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/messages/IMessageBroker.java
@@ -18,9 +18,6 @@
*/
package org.apache.hyracks.api.messages;
-/**
- * @author rico
- */
public interface IMessageBroker {
public void receivedMessage(IMessage message, String nodeId) throws Exception;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/IndexLifecycleManagerProvider.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/IndexLifecycleManagerProvider.java b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/IndexLifecycleManagerProvider.java
index 5228f80..960a11c 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/IndexLifecycleManagerProvider.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/IndexLifecycleManagerProvider.java
@@ -19,14 +19,14 @@
package org.apache.hyracks.examples.btree.helper;
import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
public enum IndexLifecycleManagerProvider implements IIndexLifecycleManagerProvider {
INSTANCE;
@Override
- public IIndexLifecycleManager getLifecycleManager(IHyracksTaskContext ctx) {
+ public IResourceLifecycleManager getLifecycleManager(IHyracksTaskContext ctx) {
return RuntimeContext.get(ctx).getIndexLifecycleManager();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/11a02942/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
index 2e96527..a4909ad 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/RuntimeContext.java
@@ -24,7 +24,7 @@ import java.util.concurrent.ThreadFactory;
import org.apache.hyracks.api.application.INCApplicationContext;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
import org.apache.hyracks.storage.am.common.dataflow.IndexLifecycleManager;
import org.apache.hyracks.storage.common.buffercache.BufferCache;
import org.apache.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
@@ -46,7 +46,7 @@ public class RuntimeContext {
private IBufferCache bufferCache;
private IFileMapManager fileMapManager;
private ILocalResourceRepository localResourceRepository;
- private IIndexLifecycleManager lcManager;
+ private IResourceLifecycleManager lcManager;
private ResourceIdFactory resourceIdFactory;
private ThreadFactory threadFactory = new ThreadFactory() {
public Thread newThread(Runnable r) {
@@ -90,7 +90,7 @@ public class RuntimeContext {
return resourceIdFactory;
}
- public IIndexLifecycleManager getIndexLifecycleManager() {
+ public IResourceLifecycleManager getIndexLifecycleManager() {
return lcManager;
}
}