vlambda博客
学习文章列表

public class RoundRobinVolumeChoosingPolicy<V extends FsVolumeSpi> implements VolumeChoosingPolicy<V> { public static final Log LOG = LogFactory.getLog(RoundRobinVolumeChoosingPolicy.class);
 private int curVolume = 0;
 @Override public synchronized V chooseVolume(final List<V> volumes, long blockSize) throws IOException {
 if(volumes.size() < 1) { throw new DiskOutOfSpaceException("No more available volumes"); }


 if(curVolume >= volumes.size()) { curVolume = 0; }
 int startVolume = curVolume; long maxAvailable = 0;
 while (true) { final V volume = volumes.get(curVolume); curVolume = (curVolume + 1) % volumes.size(); long availableVolumeSize = volume.getAvailable(); if (availableVolumeSize > blockSize) { return volume; }
 if (availableVolumeSize > maxAvailable) { maxAvailable = availableVolumeSize; }
 if (curVolume == startVolume) { throw new DiskOutOfSpaceException("Out of space: " + "The volume with the most available space (=" + maxAvailable + " B) is less than the block size (=" + blockSize + " B)."); } } }}

public class AvailableSpaceVolumeChoosingPolicy<V extends FsVolumeSpi> implements VolumeChoosingPolicy<V>, Configurable {
 private static final Log LOG = LogFactory.getLog(AvailableSpaceVolumeChoosingPolicy.class);
 private final Random random;
 private long balancedSpaceThreshold = DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_THRESHOLD_DEFAULT; private float balancedPreferencePercent = DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_DEFAULT;
 AvailableSpaceVolumeChoosingPolicy(Random random) { this.random = random; }
 public AvailableSpaceVolumeChoosingPolicy() { this(new Random()); }
 @Override public synchronized void setConf(Configuration conf) { balancedSpaceThreshold = conf.getLong( DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_THRESHOLD_KEY, DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_THRESHOLD_DEFAULT); balancedPreferencePercent = conf.getFloat( DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY, DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_DEFAULT);
 LOG.info("Available space volume choosing policy initialized: " + DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_THRESHOLD_KEY + " = " + balancedSpaceThreshold + ", " + DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY + " = " + balancedPreferencePercent);
 if (balancedPreferencePercent > 1.0) { LOG.warn("The value of " + DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY + " is greater than 1.0 but should be in the range 0.0 - 1.0"); }
 if (balancedPreferencePercent < 0.5) { LOG.warn("The value of " + DFS_DATANODE_AVAILABLE_SPACE_VOLUME_CHOOSING_POLICY_BALANCED_SPACE_PREFERENCE_FRACTION_KEY + " is less than 0.5 so volumes with less available disk space will receive more block allocations"); } }
 @Override public synchronized Configuration getConf() {
 return null; }
 private final VolumeChoosingPolicy<V> roundRobinPolicyBalanced = new RoundRobinVolumeChoosingPolicy<V>();
 private final VolumeChoosingPolicy<V> roundRobinPolicyHighAvailable = new RoundRobinVolumeChoosingPolicy<V>();
 private final VolumeChoosingPolicy<V> roundRobinPolicyLowAvailable = new RoundRobinVolumeChoosingPolicy<V>();
 @Override public synchronized V chooseVolume(List<V> volumes, long replicaSize) throws IOException { if (volumes.size() < 1) { throw new DiskOutOfSpaceException("No more available volumes"); }
 AvailableSpaceVolumeList volumesWithSpaces = new AvailableSpaceVolumeList(volumes);
 if (volumesWithSpaces.areAllVolumesWithinFreeSpaceThreshold()) {

 V volume = roundRobinPolicyBalanced.chooseVolume(volumes, replicaSize); if (LOG.isDebugEnabled()) { LOG.debug("All volumes are within the configured free space balance " + "threshold. Selecting " + volume + " for write of block size " + replicaSize); } return volume; } else { V volume = null;

 long mostAvailableAmongLowVolumes = volumesWithSpaces .getMostAvailableSpaceAmongVolumesWithLowAvailableSpace();
 List<V> highAvailableVolumes = extractVolumesFromPairs( volumesWithSpaces.getVolumesWithHighAvailableSpace()); List<V> lowAvailableVolumes = extractVolumesFromPairs( volumesWithSpaces.getVolumesWithLowAvailableSpace());
 float preferencePercentScaler = (highAvailableVolumes.size() * balancedPreferencePercent) + (lowAvailableVolumes.size() * (1 - balancedPreferencePercent));
 float scaledPreferencePercent = (highAvailableVolumes.size() * balancedPreferencePercent) / preferencePercentScaler;
 if (mostAvailableAmongLowVolumes < replicaSize || random.nextFloat() < scaledPreferencePercent) { volume = roundRobinPolicyHighAvailable.chooseVolume( highAvailableVolumes, replicaSize); if (LOG.isDebugEnabled()) { LOG.debug("Volumes are imbalanced. Selecting " + volume + " from high available space volumes for write of block size " + replicaSize); } } else { volume = roundRobinPolicyLowAvailable.chooseVolume( lowAvailableVolumes, replicaSize); if (LOG.isDebugEnabled()) { LOG.debug("Volumes are imbalanced. Selecting " + volume + " from low available space volumes for write of block size " + replicaSize); } } return volume; } }}