/*
 * Copyright 2009-2013 by The Regents of the University of California
 * 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.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package edu.uci.ics.asterix.common.context;

import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.storage.am.common.api.IModificationOperationCallback;
import edu.uci.ics.hyracks.storage.am.common.api.ISearchOperationCallback;
import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
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.IVirtualBufferCache;
import edu.uci.ics.hyracks.storage.am.lsm.common.impls.LSMOperationType;

public class PrimaryIndexOperationTracker extends BaseOperationTracker {

    private final DatasetLifecycleManager datasetLifecycleManager;
    private final IVirtualBufferCache datasetBufferCache;
    private final int datasetID;
    // Number of active operations on a ILSMIndex instance.
    private AtomicInteger numActiveOperations;

    public PrimaryIndexOperationTracker(DatasetLifecycleManager datasetLifecycleManager, int datasetID,
            ILSMIOOperationCallbackFactory ioOpCallbackFactory) {
        super(ioOpCallbackFactory);
        this.datasetLifecycleManager = datasetLifecycleManager;
        this.numActiveOperations = new AtomicInteger(0);
        this.datasetID = datasetID;
        datasetBufferCache = datasetLifecycleManager.getVirtualBufferCache(datasetID);
    }

    @Override
    public void beforeOperation(ILSMIndex index, LSMOperationType opType, ISearchOperationCallback searchCallback,
            IModificationOperationCallback modificationCallback) throws HyracksDataException {
        numActiveOperations.incrementAndGet();
    }

    @Override
    public void afterOperation(ILSMIndex index, LSMOperationType opType, ISearchOperationCallback searchCallback,
            IModificationOperationCallback modificationCallback) throws HyracksDataException {
        // Searches are immediately considered complete, because they should not prevent the execution of flushes.
        if (opType == LSMOperationType.SEARCH || opType == LSMOperationType.NOOP || opType == LSMOperationType.FLUSH
                || opType == LSMOperationType.MERGE) {
            completeOperation(index, opType, searchCallback, modificationCallback);
        }
    }

    @Override
    public void completeOperation(ILSMIndex index, LSMOperationType opType, ISearchOperationCallback searchCallback,
            IModificationOperationCallback modificationCallback) throws HyracksDataException {
        int nActiveOps = numActiveOperations.decrementAndGet();

        if (opType != LSMOperationType.FLUSH) {
            flushIfFull(nActiveOps);
        }
    }

    private void flushIfFull(int nActiveOps) throws HyracksDataException {
        // If we need a flush, and this is the last completing operation, then schedule the flush.
        if (datasetBufferCache.isFull() && nActiveOps == 0) {
            Set<ILSMIndex> indexes = datasetLifecycleManager.getDatasetIndexes(datasetID);
            for (ILSMIndex lsmIndex : indexes) {
                ILSMIndexAccessor accessor = (ILSMIndexAccessor) lsmIndex.createAccessor(
                        NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
                accessor.scheduleFlush(((BaseOperationTracker) lsmIndex.getOperationTracker()).getIOOperationCallback());
            }

        }
    }

    public void exclusiveJobCommitted() throws HyracksDataException {
        numActiveOperations.set(0);
        flushIfFull(0);
    }

    public int getNumActiveOperations() {
        return numActiveOperations.get();
    }
}

@@ -700,58 +700,8 @@
 Set;
-%0Aimport java.util.concurrent.atomic.AtomicInteger;
 %0A%0Aim
@@ -1665,29 +1665,19 @@
 private 
-AtomicInteger
+int
  numActi
@@ -2000,28 +2000,9 @@
 s = 
-new AtomicInteger(0)
+0
 ;%0A  
@@ -2150,24 +2150,37 @@
 %0A    public 
+synchronized 
 void beforeO
@@ -2221,32 +2221,44 @@
 tionType opType,
+%0A           
  ISearchOperatio
@@ -2278,36 +2278,24 @@
 rchCallback,
-%0A           
  IModificati
@@ -2327,32 +2327,44 @@
 icationCallback)
+%0A           
  throws HyracksD
@@ -2409,26 +2409,10 @@
 ions
-.incrementAndGet()
+++
 ;%0A  
@@ -3261,24 +3261,78 @@
         int 
+nActiveOps;%0A        synchronized (this) %7B%0A            
 nActiveOps =
@@ -3355,28 +3355,21 @@
 ions
-.decrementAndGet();%0A
+--;%0A        %7D
 %0A   
@@ -3699,24 +3699,165 @@
 Ops == 0) %7B%0A
+            synchronized (this) %7B%0A                if (numActiveOperations %3E 0) %7B%0A                    return;%0A                %7D%0A            %7D%0A
             
@@ -4379,32 +4379,66 @@
 DataException %7B%0A
+        synchronized (this) %7B%0A    
         numActiv
@@ -4452,16 +4452,23 @@
 ions
-.set(0);
+ = 0;%0A        %7D
 %0A   
@@ -4506,16 +4506,29 @@
  public 
+synchronized 
 int getN
@@ -4588,14 +4588,8 @@
 ions
-.get()
 ;%0A  

Fix for issue 589.