hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/sps/BlockStorageMovementAttemptedItems.java - hadoop - Git at Google

 /**
  * 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.namenode.sps;

 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_KEY;
 import static org.apache.hadoop.util.Time.monotonicNow;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;

 import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.server.namenode.sps.StoragePolicySatisfier.AttemptedItemInfo;
 import org.apache.hadoop.hdfs.server.namenode.sps.StoragePolicySatisfier.StorageTypeNodePair;
 import org.apache.hadoop.util.Daemon;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;

 /**
  * A monitor class for checking whether block storage movements attempt
  * completed or not. If this receives block storage movement attempt
  * status(either success or failure) from DN then it will just remove the
  * entries from tracking. If there is no DN reports about movement attempt
  * finished for a longer time period, then such items will retries automatically
  * after timeout. The default timeout would be 5 minutes.
  */
 public class BlockStorageMovementAttemptedItems {
   private static final Logger LOG =
       LoggerFactory.getLogger(BlockStorageMovementAttemptedItems.class);

   /**
    * A map holds the items which are already taken for blocks movements
    * processing and sent to DNs.
    */
   private final List<AttemptedItemInfo> storageMovementAttemptedItems;
   private Map<Block, Set<StorageTypeNodePair>> scheduledBlkLocs;
   // Maintains separate Queue to keep the movement finished blocks. This Q
   // is used to update the storageMovementAttemptedItems list asynchronously.
   private final BlockingQueue<Block> movementFinishedBlocks;
   private volatile boolean monitorRunning = true;
   private Daemon timerThread = null;
   private final Context context;
   //
   // It might take anywhere between 5 to 10 minutes before
   // a request is timed out.
   //
   private long selfRetryTimeout = 5 * 60 * 1000;

   //
   // It might take anywhere between 1 to 2 minutes before
   // a request is timed out.
   //
   private long minCheckTimeout = 1 * 60 * 1000; // minimum value
   private BlockStorageMovementNeeded blockStorageMovementNeeded;
   private final SPSService service;

   public BlockStorageMovementAttemptedItems(SPSService service,
       BlockStorageMovementNeeded unsatisfiedStorageMovementFiles,
       Context context) {
     this.service = service;
     long recheckTimeout = this.service.getConf().getLong(
         DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY,
         DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_DEFAULT);
     if (recheckTimeout > 0) {
       this.minCheckTimeout = Math.min(minCheckTimeout, recheckTimeout);
     }

     this.selfRetryTimeout = this.service.getConf().getLong(
         DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_KEY,
         DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_DEFAULT);
     this.blockStorageMovementNeeded = unsatisfiedStorageMovementFiles;
     storageMovementAttemptedItems = new ArrayList<>();
     scheduledBlkLocs = new HashMap<>();
     movementFinishedBlocks = new LinkedBlockingQueue<>();
     this.context = context;
   }

   /**
    * Add item to block storage movement attempted items map which holds the
    * tracking/blockCollection id versus time stamp.
    *
    * @param startPathId
    *          - start satisfier path identifier
    * @param fileId
    *          - file identifier
    * @param monotonicNow
    *          - time now
    * @param assignedBlocks
    *          - assigned blocks for block movement
    * @param retryCount
    *          - retry count
    */
   public void add(long startPathId, long fileId, long monotonicNow,
       Map<Block, Set<StorageTypeNodePair>> assignedBlocks, int retryCount) {
     AttemptedItemInfo itemInfo = new AttemptedItemInfo(startPathId, fileId,
         monotonicNow, assignedBlocks.keySet(), retryCount);
     synchronized (storageMovementAttemptedItems) {
       storageMovementAttemptedItems.add(itemInfo);
     }
     synchronized (scheduledBlkLocs) {
       scheduledBlkLocs.putAll(assignedBlocks);
     }
   }

   /**
    * Notify the storage movement attempt finished block.
    *
    * @param reportedDn
    *          reported datanode
    * @param type
    *          storage type
    * @param reportedBlock
    *          reported block
    */
   public void notifyReportedBlock(DatanodeInfo reportedDn, StorageType type,
       Block reportedBlock) {
     synchronized (scheduledBlkLocs) {
       if (scheduledBlkLocs.size() <= 0) {
         return;
       }
       matchesReportedBlock(reportedDn, type, reportedBlock);
     }
   }

   private void matchesReportedBlock(DatanodeInfo reportedDn, StorageType type,
       Block reportedBlock) {
     Set<StorageTypeNodePair> blkLocs = scheduledBlkLocs.get(reportedBlock);
     if (blkLocs == null) {
       return; // unknown block, simply skip.
     }

     for (StorageTypeNodePair dn : blkLocs) {
       boolean foundDn = dn.getDatanodeInfo().compareTo(reportedDn) == 0 ? true
           : false;
       boolean foundType = dn.getStorageType().equals(type);
       if (foundDn && foundType) {
         blkLocs.remove(dn);
         Block[] mFinishedBlocks = new Block[1];
         mFinishedBlocks[0] = reportedBlock;
         context.notifyMovementTriedBlocks(mFinishedBlocks);
         // All the block locations has reported.
         if (blkLocs.size() <= 0) {
           movementFinishedBlocks.add(reportedBlock);
           scheduledBlkLocs.remove(reportedBlock); // clean-up reported block
         }
         return; // found
       }
     }
     if (LOG.isDebugEnabled()) {
       LOG.debug("Reported block:{} not found in attempted blocks. Datanode:{}"
           + ", StorageType:{}", reportedBlock, reportedDn, type);
     }
   }

   /**
    * Starts the monitor thread.
    */
   public synchronized void start() {
     monitorRunning = true;
     timerThread = new Daemon(new BlocksStorageMovementAttemptMonitor());
     timerThread.setName("BlocksStorageMovementAttemptMonitor");
     timerThread.start();
   }

   /**
    * Sets running flag to false. Also, this will interrupt monitor thread and
    * clear all the queued up tasks.
    */
   public synchronized void stop() {
     monitorRunning = false;
     if (timerThread != null) {
       timerThread.interrupt();
     }
     this.clearQueues();
   }

   /**
    * Timed wait to stop monitor thread.
    */
   synchronized void stopGracefully() {
     if (timerThread == null) {
       return;
     }
     if (monitorRunning) {
       stop();
     }
     try {
       timerThread.join(3000);
     } catch (InterruptedException ie) {
     }
   }

   /**
    * A monitor class for checking block storage movement attempt status and long
    * waiting items periodically.
    */
   private class BlocksStorageMovementAttemptMonitor implements Runnable {
     @Override
     public void run() {
       while (monitorRunning) {
         try {
           blockStorageMovementReportedItemsCheck();
           blocksStorageMovementUnReportedItemsCheck();
           Thread.sleep(minCheckTimeout);
         } catch (InterruptedException ie) {
           LOG.info("BlocksStorageMovementAttemptMonitor thread "
               + "is interrupted.", ie);
         } catch (IOException ie) {
           LOG.warn("BlocksStorageMovementAttemptMonitor thread "
               + "received exception and exiting.", ie);
         }
       }
     }
   }

   @VisibleForTesting
   void blocksStorageMovementUnReportedItemsCheck() {
     synchronized (storageMovementAttemptedItems) {
       Iterator<AttemptedItemInfo> iter = storageMovementAttemptedItems
           .iterator();
       long now = monotonicNow();
       while (iter.hasNext()) {
         AttemptedItemInfo itemInfo = iter.next();
         if (now > itemInfo.getLastAttemptedOrReportedTime()
             + selfRetryTimeout) {
           long file = itemInfo.getFile();
           ItemInfo candidate = new ItemInfo(itemInfo.getStartPath(), file,
               itemInfo.getRetryCount() + 1);
           blockStorageMovementNeeded.add(candidate);
           iter.remove();
           LOG.info("TrackID: {} becomes timed out and moved to needed "
               + "retries queue for next iteration.", file);
         }
       }
     }
   }

   @VisibleForTesting
   void blockStorageMovementReportedItemsCheck() throws IOException {
     // Removes all available blocks from this queue and process it.
     Collection<Block> finishedBlks = new ArrayList<>();
     movementFinishedBlocks.drainTo(finishedBlks);

     // Update attempted items list
     for (Block blk : finishedBlks) {
       synchronized (storageMovementAttemptedItems) {
         Iterator<AttemptedItemInfo> iterator = storageMovementAttemptedItems
             .iterator();
         while (iterator.hasNext()) {
           AttemptedItemInfo attemptedItemInfo = iterator.next();
           attemptedItemInfo.getBlocks().remove(blk);
           if (attemptedItemInfo.getBlocks().isEmpty()) {
             blockStorageMovementNeeded.add(new ItemInfo(
                 attemptedItemInfo.getStartPath(), attemptedItemInfo.getFile(),
                 attemptedItemInfo.getRetryCount() + 1));
             iterator.remove();
           }
         }
       }
     }
   }

   @VisibleForTesting
   public int getMovementFinishedBlocksCount() {
     return movementFinishedBlocks.size();
   }

   @VisibleForTesting
   public int getAttemptedItemsCount() {
     synchronized (storageMovementAttemptedItems) {
       return storageMovementAttemptedItems.size();
     }
   }

   public void clearQueues() {
     movementFinishedBlocks.clear();
     synchronized (storageMovementAttemptedItems) {
       storageMovementAttemptedItems.clear();
     }
     synchronized (scheduledBlkLocs) {
       scheduledBlkLocs.clear();
     }
   }
 }
	/**
	* 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.namenode.sps;

	import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_DEFAULT;
	import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY;
	import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_DEFAULT;
	import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_KEY;
	import static org.apache.hadoop.util.Time.monotonicNow;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.BlockingQueue;
	import java.util.concurrent.LinkedBlockingQueue;

	import org.apache.hadoop.fs.StorageType;
	import org.apache.hadoop.hdfs.protocol.Block;
	import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
	import org.apache.hadoop.hdfs.server.namenode.sps.StoragePolicySatisfier.AttemptedItemInfo;
	import org.apache.hadoop.hdfs.server.namenode.sps.StoragePolicySatisfier.StorageTypeNodePair;
	import org.apache.hadoop.util.Daemon;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;

	/**
	* A monitor class for checking whether block storage movements attempt
	* completed or not. If this receives block storage movement attempt
	* status(either success or failure) from DN then it will just remove the
	* entries from tracking. If there is no DN reports about movement attempt
	* finished for a longer time period, then such items will retries automatically
	* after timeout. The default timeout would be 5 minutes.
	*/
	public class BlockStorageMovementAttemptedItems {
	private static final Logger LOG =
	LoggerFactory.getLogger(BlockStorageMovementAttemptedItems.class);

	/**
	* A map holds the items which are already taken for blocks movements
	* processing and sent to DNs.
	*/
	private final List<AttemptedItemInfo> storageMovementAttemptedItems;
	private Map<Block, Set<StorageTypeNodePair>> scheduledBlkLocs;
	// Maintains separate Queue to keep the movement finished blocks. This Q
	// is used to update the storageMovementAttemptedItems list asynchronously.
	private final BlockingQueue<Block> movementFinishedBlocks;
	private volatile boolean monitorRunning = true;
	private Daemon timerThread = null;
	private final Context context;
	//
	// It might take anywhere between 5 to 10 minutes before
	// a request is timed out.
	//
	private long selfRetryTimeout = 5 * 60 * 1000;

	//
	// It might take anywhere between 1 to 2 minutes before
	// a request is timed out.
	//
	private long minCheckTimeout = 1 * 60 * 1000; // minimum value
	private BlockStorageMovementNeeded blockStorageMovementNeeded;
	private final SPSService service;

	public BlockStorageMovementAttemptedItems(SPSService service,
	BlockStorageMovementNeeded unsatisfiedStorageMovementFiles,
	Context context) {
	this.service = service;
	long recheckTimeout = this.service.getConf().getLong(
	DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_KEY,
	DFS_STORAGE_POLICY_SATISFIER_RECHECK_TIMEOUT_MILLIS_DEFAULT);
	if (recheckTimeout > 0) {
	this.minCheckTimeout = Math.min(minCheckTimeout, recheckTimeout);
	}

	this.selfRetryTimeout = this.service.getConf().getLong(
	DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_KEY,
	DFS_STORAGE_POLICY_SATISFIER_SELF_RETRY_TIMEOUT_MILLIS_DEFAULT);
	this.blockStorageMovementNeeded = unsatisfiedStorageMovementFiles;
	storageMovementAttemptedItems = new ArrayList<>();
	scheduledBlkLocs = new HashMap<>();
	movementFinishedBlocks = new LinkedBlockingQueue<>();
	this.context = context;
	}

	/**
	* Add item to block storage movement attempted items map which holds the
	* tracking/blockCollection id versus time stamp.
	*
	* @param startPathId
	* - start satisfier path identifier
	* @param fileId
	* - file identifier
	* @param monotonicNow
	* - time now
	* @param assignedBlocks
	* - assigned blocks for block movement
	* @param retryCount
	* - retry count
	*/
	public void add(long startPathId, long fileId, long monotonicNow,
	Map<Block, Set<StorageTypeNodePair>> assignedBlocks, int retryCount) {
	AttemptedItemInfo itemInfo = new AttemptedItemInfo(startPathId, fileId,
	monotonicNow, assignedBlocks.keySet(), retryCount);
	synchronized (storageMovementAttemptedItems) {
	storageMovementAttemptedItems.add(itemInfo);
	}
	synchronized (scheduledBlkLocs) {
	scheduledBlkLocs.putAll(assignedBlocks);
	}
	}

	/**
	* Notify the storage movement attempt finished block.
	*
	* @param reportedDn
	* reported datanode
	* @param type
	* storage type
	* @param reportedBlock
	* reported block
	*/
	public void notifyReportedBlock(DatanodeInfo reportedDn, StorageType type,
	Block reportedBlock) {
	synchronized (scheduledBlkLocs) {
	if (scheduledBlkLocs.size() <= 0) {
	return;
	}
	matchesReportedBlock(reportedDn, type, reportedBlock);
	}
	}

	private void matchesReportedBlock(DatanodeInfo reportedDn, StorageType type,
	Block reportedBlock) {
	Set<StorageTypeNodePair> blkLocs = scheduledBlkLocs.get(reportedBlock);
	if (blkLocs == null) {
	return; // unknown block, simply skip.
	}

	for (StorageTypeNodePair dn : blkLocs) {
	boolean foundDn = dn.getDatanodeInfo().compareTo(reportedDn) == 0 ? true
	: false;
	boolean foundType = dn.getStorageType().equals(type);
	if (foundDn && foundType) {
	blkLocs.remove(dn);
	Block[] mFinishedBlocks = new Block[1];
	mFinishedBlocks[0] = reportedBlock;
	context.notifyMovementTriedBlocks(mFinishedBlocks);
	// All the block locations has reported.
	if (blkLocs.size() <= 0) {
	movementFinishedBlocks.add(reportedBlock);
	scheduledBlkLocs.remove(reportedBlock); // clean-up reported block
	}
	return; // found
	}
	}
	if (LOG.isDebugEnabled()) {
	LOG.debug("Reported block:{} not found in attempted blocks. Datanode:{}"
	+ ", StorageType:{}", reportedBlock, reportedDn, type);
	}
	}

	/**
	* Starts the monitor thread.
	*/
	public synchronized void start() {
	monitorRunning = true;
	timerThread = new Daemon(new BlocksStorageMovementAttemptMonitor());
	timerThread.setName("BlocksStorageMovementAttemptMonitor");
	timerThread.start();
	}

	/**
	* Sets running flag to false. Also, this will interrupt monitor thread and
	* clear all the queued up tasks.
	*/
	public synchronized void stop() {
	monitorRunning = false;
	if (timerThread != null) {
	timerThread.interrupt();
	}
	this.clearQueues();
	}

	/**
	* Timed wait to stop monitor thread.
	*/
	synchronized void stopGracefully() {
	if (timerThread == null) {
	return;
	}
	if (monitorRunning) {
	stop();
	}
	try {
	timerThread.join(3000);
	} catch (InterruptedException ie) {
	}
	}

	/**
	* A monitor class for checking block storage movement attempt status and long
	* waiting items periodically.
	*/
	private class BlocksStorageMovementAttemptMonitor implements Runnable {
	@Override
	public void run() {
	while (monitorRunning) {
	try {
	blockStorageMovementReportedItemsCheck();
	blocksStorageMovementUnReportedItemsCheck();
	Thread.sleep(minCheckTimeout);
	} catch (InterruptedException ie) {
	LOG.info("BlocksStorageMovementAttemptMonitor thread "
	+ "is interrupted.", ie);
	} catch (IOException ie) {
	LOG.warn("BlocksStorageMovementAttemptMonitor thread "
	+ "received exception and exiting.", ie);
	}
	}
	}
	}

	@VisibleForTesting
	void blocksStorageMovementUnReportedItemsCheck() {
	synchronized (storageMovementAttemptedItems) {
	Iterator<AttemptedItemInfo> iter = storageMovementAttemptedItems
	.iterator();
	long now = monotonicNow();
	while (iter.hasNext()) {
	AttemptedItemInfo itemInfo = iter.next();
	if (now > itemInfo.getLastAttemptedOrReportedTime()
	+ selfRetryTimeout) {
	long file = itemInfo.getFile();
	ItemInfo candidate = new ItemInfo(itemInfo.getStartPath(), file,
	itemInfo.getRetryCount() + 1);
	blockStorageMovementNeeded.add(candidate);
	iter.remove();
	LOG.info("TrackID: {} becomes timed out and moved to needed "
	+ "retries queue for next iteration.", file);
	}
	}
	}
	}

	@VisibleForTesting
	void blockStorageMovementReportedItemsCheck() throws IOException {
	// Removes all available blocks from this queue and process it.
	Collection<Block> finishedBlks = new ArrayList<>();
	movementFinishedBlocks.drainTo(finishedBlks);

	// Update attempted items list
	for (Block blk : finishedBlks) {
	synchronized (storageMovementAttemptedItems) {
	Iterator<AttemptedItemInfo> iterator = storageMovementAttemptedItems
	.iterator();
	while (iterator.hasNext()) {
	AttemptedItemInfo attemptedItemInfo = iterator.next();
	attemptedItemInfo.getBlocks().remove(blk);
	if (attemptedItemInfo.getBlocks().isEmpty()) {
	blockStorageMovementNeeded.add(new ItemInfo(
	attemptedItemInfo.getStartPath(), attemptedItemInfo.getFile(),
	attemptedItemInfo.getRetryCount() + 1));
	iterator.remove();
	}
	}
	}
	}
	}

	@VisibleForTesting
	public int getMovementFinishedBlocksCount() {
	return movementFinishedBlocks.size();
	}

	@VisibleForTesting
	public int getAttemptedItemsCount() {
	synchronized (storageMovementAttemptedItems) {
	return storageMovementAttemptedItems.size();
	}
	}

	public void clearQueues() {
	movementFinishedBlocks.clear();
	synchronized (storageMovementAttemptedItems) {
	storageMovementAttemptedItems.clear();
	}
	synchronized (scheduledBlkLocs) {
	scheduledBlkLocs.clear();
	}
	}
	}