AsterixDB changes for:
1. fix "writerCount!=0" issue
2. protect deactivate/activate call of an index by its opTracker.
Note that this change includes https://asterix-gerrit.ics.uci.edu/#/c/269/.
Change-Id: I4669d49fd1cf9b058f7e0da233ad45c6c78b5797
Reviewed-on: https://asterix-gerrit.ics.uci.edu/279
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <hubailmor@gmail.com>
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/DatasetLifecycleManager.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/DatasetLifecycleManager.java
index 9a91289..8476174 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/DatasetLifecycleManager.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/DatasetLifecycleManager.java
@@ -145,7 +145,10 @@
flushAndWaitForIO(dsInfo, iInfo);
if (iInfo.isOpen) {
- iInfo.index.deactivate(false);
+ ILSMOperationTracker indexOpTracker = iInfo.index.getOperationTracker();
+ synchronized (indexOpTracker) {
+ iInfo.index.deactivate(false);
+ }
}
dsInfo.indexes.remove(resourceID);
@@ -195,7 +198,10 @@
dsInfo.isOpen = true;
dsInfo.touch();
if (!iInfo.isOpen) {
- iInfo.index.activate();
+ ILSMOperationTracker opTracker = iInfo.index.getOperationTracker();
+ synchronized (opTracker) {
+ iInfo.index.activate();
+ }
iInfo.isOpen = true;
}
iInfo.touch();
@@ -530,7 +536,10 @@
for (IndexInfo iInfo : dsInfo.indexes.values()) {
if (iInfo.isOpen) {
- iInfo.index.deactivate(false);
+ ILSMOperationTracker opTracker = iInfo.index.getOperationTracker();
+ synchronized (opTracker) {
+ iInfo.index.deactivate(false);
+ }
iInfo.isOpen = false;
}
assert iInfo.referenceCount == 0;
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/PrimaryIndexOperationTracker.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/PrimaryIndexOperationTracker.java
index 7a56600..67a61a8 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/PrimaryIndexOperationTracker.java
@@ -3,9 +3,9 @@
* Licensed 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 from
- *
+ *
* 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.
@@ -32,6 +32,7 @@
import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndex;
import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIndexInternal;
+import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
import edu.uci.ics.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMOperationType;
@@ -72,7 +73,7 @@
@Override
public synchronized void completeOperation(ILSMIndex index, LSMOperationType opType,
ISearchOperationCallback searchCallback, IModificationOperationCallback modificationCallback)
- throws HyracksDataException {
+ throws HyracksDataException {
if (opType == LSMOperationType.MODIFICATION || opType == LSMOperationType.FORCE_MODIFICATION) {
decrementNumActiveOperations(modificationCallback);
if (numActiveOperations.get() == 0) {
@@ -86,7 +87,7 @@
}
public void flushIfRequested() throws HyracksDataException {
- // If we need a flush, and this is the last completing operation, then schedule the flush,
+ // If we need a flush, and this is the last completing operation, then schedule the flush,
// or if there is a flush scheduled by the checkpoint (flushOnExit), then schedule it
boolean needsFlush = false;
@@ -105,8 +106,12 @@
if (needsFlush || flushOnExit) {
//Make the current mutable components READABLE_UNWRITABLE to stop coming modify operations from entering them until the current flush is schedule.
for (ILSMIndex lsmIndex : indexes) {
- if (((AbstractLSMIndex) lsmIndex).getCurrentMutableComponentState() == ComponentState.READABLE_WRITABLE) {
- ((AbstractLSMIndex) lsmIndex).setCurrentMutableComponentState(ComponentState.READABLE_UNWRITABLE);
+ AbstractLSMIndex abstractLSMIndex = ((AbstractLSMIndex) lsmIndex);
+ ILSMOperationTracker opTracker = abstractLSMIndex.getOperationTracker();
+ synchronized (opTracker) {
+ if (abstractLSMIndex.getCurrentMutableComponentState() == ComponentState.READABLE_WRITABLE) {
+ abstractLSMIndex.setCurrentMutableComponentState(ComponentState.READABLE_UNWRITABLE);
+ }
}
}
@@ -155,7 +160,7 @@
}
private void incrementNumActiveOperations(IModificationOperationCallback modificationCallback) {
- //modificationCallback can be NoOpOperationCallback when redo/undo operations are executed.
+ //modificationCallback can be NoOpOperationCallback when redo/undo operations are executed.
if (modificationCallback != NoOpOperationCallback.INSTANCE) {
numActiveOperations.incrementAndGet();
((AbstractOperationCallback) modificationCallback).incrementLocalNumActiveOperations();
@@ -188,4 +193,4 @@
return flushLogCreated;
}
-}
\ No newline at end of file
+}