/*
 * Licensed to the University of California, Berkeley 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 tachyon.worker.block;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.google.common.collect.Lists;

import tachyon.worker.BlockStoreLocation;

/**
 * Represents the delta of the block store within one heartbeat period. For now, newly committed
 * blocks do not pass through this master communication mechanism, instead it is synchronise through
 * {@link tachyon.worker.block.BlockDataManager#commitBlock(long, long)}. This class is thread safe.
 */
public class BlockHeartbeatReporter implements BlockMetaEventListener {
  /** Lock for operations on the removed and added block collections */
  private final Object mLock;

  /** List of blocks that were removed in the last heartbeat period */
  private List<Long> mRemovedBlocks;
  /** Map of storage dirs to a list of blocks that were added in the last heartbeat period */
  private Map<Long, List<Long>> mAddedBlocks;

  public BlockHeartbeatReporter() {
    mLock = new Object();
    mRemovedBlocks = new ArrayList<Long>(100);
    mAddedBlocks = new HashMap<Long, List<Long>>();
  }

  /**
   * Generates the report of the block store delta in the last heartbeat period. Calling this method
   * marks the end of a period and the start of a new heartbeat period.
   *
   * @return the block store delta report for the last heartbeat period
   */
  public BlockHeartbeatReport generateReport() {
    synchronized (mLock) {
      // Copy added and removed blocks
      Map<Long, List<Long>> addedBlocks = new HashMap<Long, List<Long>>(mAddedBlocks);
      List<Long> removedBlocks = new ArrayList<Long>(mRemovedBlocks);
      // Clear added and removed blocks
      mAddedBlocks.clear();
      mRemovedBlocks.clear();
      return new BlockHeartbeatReport(addedBlocks, removedBlocks);
    }
  }

  @Override
  public void preCommitBlock(long userId, long blockId, BlockStoreLocation location) {
    // Do nothing
  }

  // TODO: Add this functionality back when block creation between client and master is changed
  @Override
  public void postCommitBlock(long userId, long blockId, BlockStoreLocation location) {
    // Long storageDirId = location.getStorageDirId();
    // synchronized (mLock) {
    // addBlockToAddedBlocks(blockId, storageDirId);
    // }
  }

  @Override
  public void preMoveBlockByClient(long userId, long blockId, BlockStoreLocation oldLocation,
      BlockStoreLocation newLocation) {
    // Do nothing
  }

  @Override
  public void postMoveBlockByClient(long userId, long blockId, BlockStoreLocation oldLocation,
      BlockStoreLocation newLocation) {
    Long storageDirId = newLocation.getStorageDirId();
    synchronized (mLock) {
      // Add the block to the removed block list to remove the previous location
      // TODO: We should have a better mechanism to indicate block movement
      mRemovedBlocks.add(blockId);
      // Remove the block from our list of added blocks in this heartbeat, if it was added to
      // prevent adding the block twice.
      removeBlockFromAddedBlocks(blockId);
      // Add the block back with the new storagedir.
      addBlockToAddedBlocks(blockId, storageDirId);
    }
  }

  @Override
  public void preRemoveBlockByClient(long userId, long blockId) {
    // Do nothing
  }

  @Override
  public void postRemoveBlockByClient(long userId, long blockId) {
    // Do nothing
  }

  @Override
  public void preRemoveBlockByWorker(long userId, long blockId) {
    // Do nothing
  }

  @Override
  public void postRemoveBlockByWorker(long userId, long blockId) {
    synchronized (mLock) {
      // Remove the block from list of added blocks, in case it was added in this heartbeat period.
      removeBlockFromAddedBlocks(blockId);
      // Add to the list of removed blocks in this heartbeat period.
      if (!mRemovedBlocks.contains(blockId)) {
        mRemovedBlocks.add(blockId);
      }
    }
  }

  @Override
  public void preMoveBlockByWorker(long userId, long blockId, BlockStoreLocation oldLocation,
      BlockStoreLocation newLocation) {
    // Do nothing
  }

  @Override
  public void postMoveBlockByWorker(long userId, long blockId, BlockStoreLocation oldLocation,
      BlockStoreLocation newLocation) {
    Long storageDirId = newLocation.getStorageDirId();
    synchronized (mLock) {
      // Add the block to the removed block list to remove the previous location
      // TODO: We should have a better mechanism to indicate block movement
      mRemovedBlocks.add(blockId);
      // Remove the block from our list of added blocks in this heartbeat, if it was added to
      // prevent adding the block twice.
      removeBlockFromAddedBlocks(blockId);
      // Add the block back with the new storagedir.
      addBlockToAddedBlocks(blockId, storageDirId);
    }
  }


  @Override
  public void preAbortBlock(long userId, long blockId) {
    // Do nothing
  }

  @Override
  public void postAbortBlock(long userId, long blockId) {
    // Do nothing
  }

  /**
   * Adds a block to the list of added blocks in this heartbeat period.
   *
   * @param blockId The id of the block to add
   * @param storageDirId The storage directory id containing the block
   */
  private void addBlockToAddedBlocks(long blockId, long storageDirId) {
    if (mAddedBlocks.containsKey(storageDirId)) {
      mAddedBlocks.get(storageDirId).add(blockId);
    } else {
      mAddedBlocks.put(storageDirId, Lists.newArrayList(blockId));
    }
  }

  /**
   * Removes the block from the added blocks map, if it exists.
   *
   * @param blockId The block to remove
   */
  private void removeBlockFromAddedBlocks(long blockId) {
    for (List<Long> blockList : mAddedBlocks.values()) {
      if (blockList.contains(blockId)) {
        blockList.remove(blockId);
      }
    }
  }
}

@@ -4142,37 +4142,356 @@
 ckId) %7B%0A    
-// Do nothing
+synchronized (mLock) %7B%0A      // Remove the block from list of added blocks, in case it was added in this heartbeat period.%0A      removeBlockFromAddedBlocks(blockId);%0A      // Add to the list of removed blocks in this heartbeat period.%0A      if (!mRemovedBlocks.contains(blockId)) %7B%0A        mRemovedBlocks.add(blockId);%0A      %7D%0A    %7D
 %0A  %7D%0A%0A  @Ove

Fix a bug and report blocks removed by clients to master also