/**
 * 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.hadoop.hdfs.server.datanode;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.StorageType;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.BlockListAsLongs;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.ReplicaOutputStreams;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetFactory;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
import org.apache.hadoop.util.DataChecksum;
import org.junit.Before;
import org.junit.Test;

/**
 * this class tests the methods of the  SimulatedFSDataset.
 */
public class TestSimulatedFSDataset {
  Configuration conf = null;
  static final String bpid = "BP-TEST";
  static final int NUMBLOCKS = 20;
  static final int BLOCK_LENGTH_MULTIPLIER = 79;
  static final long FIRST_BLK_ID = 1;

  private final int storageCount;

  public TestSimulatedFSDataset() {
    this(1);
  }

  protected TestSimulatedFSDataset(int storageCount) {
    this.storageCount = storageCount;
  }

  @Before
  public void setUp() throws Exception {
    conf = new HdfsConfiguration();
    SimulatedFSDataset.setFactory(conf);
  }
  
  static long blockIdToLen(long blkid) {
    return blkid * BLOCK_LENGTH_MULTIPLIER;
  }

  static int addSomeBlocks(SimulatedFSDataset fsdataset) throws IOException {
    return addSomeBlocks(fsdataset, false);
  }

  static int addSomeBlocks(SimulatedFSDataset fsdataset,
      boolean negativeBlkID) throws IOException {
    return addSomeBlocks(fsdataset, FIRST_BLK_ID, negativeBlkID);
  }

  static int addSomeBlocks(SimulatedFSDataset fsdataset, long startingBlockId,
      boolean negativeBlkID) throws IOException {
    int bytesAdded = 0;
    for (long i = startingBlockId; i < startingBlockId+NUMBLOCKS; ++i) {
      long blkID = negativeBlkID ? i * -1 : i;
      ExtendedBlock b = new ExtendedBlock(bpid, blkID, 0, 0);
      // we pass expected len as zero, - fsdataset should use the sizeof actual
      // data written
      ReplicaInPipeline bInfo = fsdataset.createRbw(
          StorageType.DEFAULT, null, b, false).getReplica();
      ReplicaOutputStreams out = bInfo.createStreams(true,
          DataChecksum.newDataChecksum(DataChecksum.Type.CRC32, 512));
      try {
        OutputStream dataOut  = out.getDataOut();
        assertEquals(0, fsdataset.getLength(b));
        for (int j=1; j <= blockIdToLen(i); ++j) {
          dataOut.write(j);
          assertEquals(j, bInfo.getBytesOnDisk()); // correct length even as we write
          bytesAdded++;
        }
      } finally {
        out.close();
      }
      b.setNumBytes(blockIdToLen(i));
      fsdataset.finalizeBlock(b, false);
      assertEquals(blockIdToLen(i), fsdataset.getLength(b));
    }
    return bytesAdded;
  }

  static void readSomeBlocks(SimulatedFSDataset fsdataset,
      boolean negativeBlkID) throws IOException {
    for (long i = FIRST_BLK_ID; i <= NUMBLOCKS; ++i) {
      long blkID = negativeBlkID ? i * -1 : i;
      ExtendedBlock b = new ExtendedBlock(bpid, blkID, 0, 0);
      assertTrue(fsdataset.isValidBlock(b));
      assertEquals(blockIdToLen(i), fsdataset.getLength(b));
      checkBlockDataAndSize(fsdataset, b, blockIdToLen(i));
    }
  }
  
  @Test
  public void testFSDatasetFactory() {
    final Configuration conf = new Configuration();
    FsDatasetSpi.Factory<?> f = FsDatasetSpi.Factory.getFactory(conf);
    assertEquals(FsDatasetFactory.class, f.getClass());
    assertFalse(f.isSimulated());

    SimulatedFSDataset.setFactory(conf);
    FsDatasetSpi.Factory<?> s = FsDatasetSpi.Factory.getFactory(conf);
    assertEquals(SimulatedFSDataset.Factory.class, s.getClass());
    assertTrue(s.isSimulated());
  }

  @Test
  public void testGetMetaData() throws IOException {
    final SimulatedFSDataset fsdataset = getSimulatedFSDataset();
    ExtendedBlock b = new ExtendedBlock(bpid, FIRST_BLK_ID, 5, 0);
    try {
      assertTrue(fsdataset.getMetaDataInputStream(b) == null);
      assertTrue("Expected an IO exception", false);
    } catch (IOException e) {
      // ok - as expected
    }
    addSomeBlocks(fsdataset); // Only need to add one but ....
    b = new ExtendedBlock(bpid, FIRST_BLK_ID, 0, 0);
    InputStream metaInput = fsdataset.getMetaDataInputStream(b);
    DataInputStream metaDataInput = new DataInputStream(metaInput);
    short version = metaDataInput.readShort();
    assertEquals(BlockMetadataHeader.VERSION, version);
    DataChecksum checksum = DataChecksum.newDataChecksum(metaDataInput);
    assertEquals(DataChecksum.Type.NULL, checksum.getChecksumType());
    assertEquals(0, checksum.getChecksumSize());  
  }


  @Test
  public void testStorageUsage() throws IOException {
    final SimulatedFSDataset fsdataset = getSimulatedFSDataset();
    assertEquals(fsdataset.getDfsUsed(), 0);
    assertEquals(fsdataset.getRemaining(), fsdataset.getCapacity());
    int bytesAdded = addSomeBlocks(fsdataset);
    assertEquals(bytesAdded, fsdataset.getDfsUsed());
    assertEquals(fsdataset.getCapacity()-bytesAdded,  fsdataset.getRemaining());
  }



  static void checkBlockDataAndSize(SimulatedFSDataset fsdataset,
      ExtendedBlock b, long expectedLen) throws IOException {
    InputStream input = fsdataset.getBlockInputStream(b);
    long lengthRead = 0;
    int data;
    while ((data = input.read()) != -1) {
      assertEquals(SimulatedFSDataset.simulatedByte(b.getLocalBlock(),
          lengthRead), (byte) (data & SimulatedFSDataset.BYTE_MASK));
      lengthRead++;
    }
    assertEquals(expectedLen, lengthRead);
  }

  @Test
  public void testWriteRead() throws IOException {
    testWriteRead(false);
    testWriteRead(true);
  }

  private void testWriteRead(boolean negativeBlkID) throws IOException {
    final SimulatedFSDataset fsdataset = getSimulatedFSDataset();
    addSomeBlocks(fsdataset, negativeBlkID);
    readSomeBlocks(fsdataset, negativeBlkID);
  }

  @Test
  public void testGetBlockReport() throws IOException {
    final SimulatedFSDataset fsdataset = getSimulatedFSDataset();
    assertBlockReportCountAndSize(fsdataset, 0);
    addSomeBlocks(fsdataset);
    assertBlockReportCountAndSize(fsdataset, NUMBLOCKS);
    assertBlockLengthInBlockReports(fsdataset);
  }
  
  @Test
  public void testInjectionEmpty() throws IOException {
    SimulatedFSDataset fsdataset = getSimulatedFSDataset(); 
    assertBlockReportCountAndSize(fsdataset, 0);
    int bytesAdded = addSomeBlocks(fsdataset);
    assertBlockReportCountAndSize(fsdataset, NUMBLOCKS);
    assertBlockLengthInBlockReports(fsdataset);
    
    // Inject blocks into an empty fsdataset
    //  - injecting the blocks we got above.
    SimulatedFSDataset sfsdataset = getSimulatedFSDataset();
    injectBlocksFromBlockReport(fsdataset, sfsdataset);
    assertBlockReportCountAndSize(fsdataset, NUMBLOCKS);
    assertBlockLengthInBlockReports(fsdataset, sfsdataset);

    assertEquals(bytesAdded, sfsdataset.getDfsUsed());
    assertEquals(sfsdataset.getCapacity()-bytesAdded, sfsdataset.getRemaining());
  }

  @Test
  public void testInjectionNonEmpty() throws IOException {
    SimulatedFSDataset fsdataset = getSimulatedFSDataset(); 
    assertBlockReportCountAndSize(fsdataset, 0);
    int bytesAdded = addSomeBlocks(fsdataset);
    assertBlockReportCountAndSize(fsdataset, NUMBLOCKS);
    assertBlockLengthInBlockReports(fsdataset);
    
    // Inject blocks into an non-empty fsdataset
    //  - injecting the blocks we got above.
    SimulatedFSDataset sfsdataset = getSimulatedFSDataset();
    // Add come blocks whose block ids do not conflict with
    // the ones we are going to inject.
    bytesAdded += addSomeBlocks(sfsdataset, NUMBLOCKS+1, false);
    assertBlockReportCountAndSize(sfsdataset, NUMBLOCKS);
    injectBlocksFromBlockReport(fsdataset, sfsdataset);
    assertBlockReportCountAndSize(sfsdataset, NUMBLOCKS * 2);
    assertBlockLengthInBlockReports(fsdataset, sfsdataset);
    assertEquals(bytesAdded, sfsdataset.getDfsUsed());
    assertEquals(sfsdataset.getCapacity()-bytesAdded,  sfsdataset.getRemaining());
    
    // Now test that the dataset cannot be created if it does not have sufficient cap
    conf.setLong(SimulatedFSDataset.CONFIG_PROPERTY_CAPACITY, 10);
 
    try {
      sfsdataset = getSimulatedFSDataset();
      sfsdataset.addBlockPool(bpid, conf);
      injectBlocksFromBlockReport(fsdataset, sfsdataset);
      assertTrue("Expected an IO exception", false);
    } catch (IOException e) {
      // ok - as expected
    }
  }

  public void checkInvalidBlock(ExtendedBlock b) {
    final SimulatedFSDataset fsdataset = getSimulatedFSDataset();
    assertFalse(fsdataset.isValidBlock(b));
    try {
      fsdataset.getLength(b);
      assertTrue("Expected an IO exception", false);
    } catch (IOException e) {
      // ok - as expected
    }
    
    try {
      fsdataset.getBlockInputStream(b);
      assertTrue("Expected an IO exception", false);
    } catch (IOException e) {
      // ok - as expected
    }
    
    try {
      fsdataset.finalizeBlock(b, false);
      assertTrue("Expected an IO exception", false);
    } catch (IOException e) {
      // ok - as expected
    }
  }
  
  @Test
  public void testInValidBlocks() throws IOException {
    final SimulatedFSDataset fsdataset = getSimulatedFSDataset();
    ExtendedBlock b = new ExtendedBlock(bpid, FIRST_BLK_ID, 5, 0);
    checkInvalidBlock(b);
    
    // Now check invlaid after adding some blocks
    addSomeBlocks(fsdataset);
    b = new ExtendedBlock(bpid, NUMBLOCKS + 99, 5, 0);
    checkInvalidBlock(b);
  }

  @Test
  public void testInvalidate() throws IOException {
    final SimulatedFSDataset fsdataset = getSimulatedFSDataset();
    int bytesAdded = addSomeBlocks(fsdataset);
    Block[] deleteBlocks = new Block[2];
    deleteBlocks[0] = new Block(1, 0, 0);
    deleteBlocks[1] = new Block(2, 0, 0);
    fsdataset.invalidate(bpid, deleteBlocks);
    checkInvalidBlock(new ExtendedBlock(bpid, deleteBlocks[0]));
    checkInvalidBlock(new ExtendedBlock(bpid, deleteBlocks[1]));
    long sizeDeleted = blockIdToLen(1) + blockIdToLen(2);
    assertEquals(bytesAdded-sizeDeleted, fsdataset.getDfsUsed());
    assertEquals(fsdataset.getCapacity()-bytesAdded+sizeDeleted,  fsdataset.getRemaining());
    
    // Now make sure the rest of the blocks are valid
    for (int i=3; i <= NUMBLOCKS; ++i) {
      Block b = new Block(i, 0, 0);
      assertTrue(fsdataset.isValidBlock(new ExtendedBlock(bpid, b)));
    }
  }

  /**
   * Inject all of the blocks returned from sourceFSDataset's block reports
   * into destinationFSDataset.
   */
  private void injectBlocksFromBlockReport(SimulatedFSDataset sourceFSDataset,
      SimulatedFSDataset destinationFSDataset) throws IOException {
    for (Map.Entry<DatanodeStorage, BlockListAsLongs> ent :
        sourceFSDataset.getBlockReports(bpid).entrySet()) {
      destinationFSDataset.injectBlocks(bpid, ent.getValue());
    }
  }

  /**
   * Assert that the number of block reports returned from fsdataset matches
   * {@code storageCount}, and that the total number of blocks is equal to
   * expectedBlockCount.
   */
  private void assertBlockReportCountAndSize(SimulatedFSDataset fsdataset,
      int expectedBlockCount) {
    Map<DatanodeStorage, BlockListAsLongs> blockReportMap =
        fsdataset.getBlockReports(bpid);
    assertEquals(storageCount, blockReportMap.size());
    int totalCount = 0;
    for (Map.Entry<DatanodeStorage, BlockListAsLongs> ent :
        blockReportMap.entrySet()) {
      totalCount += ent.getValue().getNumberOfBlocks();
    }
    assertEquals(expectedBlockCount, totalCount);
  }

  /**
   * Convenience method to call {@link #assertBlockLengthInBlockReports(
   * SimulatedFSDataset,SimulatedFSDataset)} with a null second parameter.
   */
  private void assertBlockLengthInBlockReports(SimulatedFSDataset fsdataset)
      throws IOException {
    assertBlockLengthInBlockReports(fsdataset, null);
  }

  /**
   * Assert that, for all of the blocks in the block report(s) returned from
   * fsdataset, they are not null and their length matches the expectation.
   * If otherFSDataset is non-null, additionally confirm that its idea of the
   * length of the block matches as well.
   */
  private void assertBlockLengthInBlockReports(SimulatedFSDataset fsdataset,
      SimulatedFSDataset otherFSDataset) throws IOException {
    for (Map.Entry<DatanodeStorage, BlockListAsLongs> ent :
        fsdataset.getBlockReports(bpid).entrySet()) {
      for (Block b : ent.getValue()) {
        assertNotNull(b);
        assertEquals(blockIdToLen(b.getBlockId()), b.getNumBytes());
        if (otherFSDataset != null) {
          assertEquals(blockIdToLen(b.getBlockId()), otherFSDataset
              .getLength(new ExtendedBlock(bpid, b)));
        }
      }
    }
  }

  protected SimulatedFSDataset getSimulatedFSDataset() {
    SimulatedFSDataset fsdataset = new SimulatedFSDataset(null, conf);
    fsdataset.addBlockPool(bpid, conf);
    return fsdataset;
  }

  @Test
  public void testConcurrentAddBlockPool() throws InterruptedException,
      IOException {
    final String[] bpids = {"BP-TEST1-", "BP-TEST2-"};
    final SimulatedFSDataset fsdataset = new SimulatedFSDataset(null, conf);
    class AddBlockPoolThread extends Thread {
      private int id;
      private IOException ioe;
      public AddBlockPoolThread(int id) {
        super();
        this.id = id;
      }
      public void test() throws InterruptedException, IOException {
        this.join();
        if (ioe != null) {
          throw ioe;
        }
      }
      public void run() {
        for (int i=0; i < 10000; i++) {
          // add different block pools concurrently
          String newbpid = bpids[id] + i;
          fsdataset.addBlockPool(newbpid, conf);
          // and then add a block into the pool
          ExtendedBlock block = new ExtendedBlock(newbpid,1);
          try {
            // it will throw an exception if the block pool is not found
            fsdataset.createTemporary(StorageType.DEFAULT, null, block, false);
          } catch (IOException ioe) {
            // JUnit does not capture exception in non-main thread,
            // so cache it and then let main thread throw later.
            this.ioe = ioe;
          }
          assert(fsdataset.getReplicaString(newbpid,1) != "null");
        }
      }
    };
    AddBlockPoolThread t1 = new AddBlockPoolThread(0);
    AddBlockPoolThread t2 = new AddBlockPoolThread(1);
    t1.start();
    t2.start();
    t1.test();
    t2.test();
  }
}
