AvailableSpaceVolumeChoosingPolicy implements VolumeChoosingPolicy<V>, Configurable { @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; } } AvailableSpaceVolumeChoosingPolicy(Random random);  AvailableSpaceVolumeChoosingPolicy(); @Override synchronized void setConf(Configuration conf); @Override synchronized Configuration getConf(); @Override synchronized V chooseVolume(List<V> volumes,
      long replicaSize);  }
@Test(timeout=60000) public void testTwoUnbalancedVolumes() throws Exception { @SuppressWarnings("unchecked") final AvailableSpaceVolumeChoosingPolicy<FsVolumeSpi> policy = ReflectionUtils.newInstance(AvailableSpaceVolumeChoosingPolicy.class, null); initPolicy(policy, 1.0f); List<FsVolumeSpi> volumes = new ArrayList<FsVolumeSpi>(); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(0).getAvailable()).thenReturn(1024L * 1024L); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(1).getAvailable()).thenReturn(1024L * 1024L * 3); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); }
@Test(timeout=60000) public void testThreeUnbalancedVolumes() throws Exception { @SuppressWarnings("unchecked") final AvailableSpaceVolumeChoosingPolicy<FsVolumeSpi> policy = ReflectionUtils.newInstance(AvailableSpaceVolumeChoosingPolicy.class, null); List<FsVolumeSpi> volumes = new ArrayList<FsVolumeSpi>(); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(0).getAvailable()).thenReturn(1024L * 1024L); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(1).getAvailable()).thenReturn(1024L * 1024L * 3); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(2).getAvailable()).thenReturn(1024L * 1024L * 3); initPolicy(policy, 1.0f); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100)); initPolicy(policy, 0.0f); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100)); }
@Test(timeout=60000) public void testFourUnbalancedVolumes() throws Exception { @SuppressWarnings("unchecked") final AvailableSpaceVolumeChoosingPolicy<FsVolumeSpi> policy = ReflectionUtils.newInstance(AvailableSpaceVolumeChoosingPolicy.class, null); List<FsVolumeSpi> volumes = new ArrayList<FsVolumeSpi>(); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(0).getAvailable()).thenReturn(1024L * 1024L); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(1).getAvailable()).thenReturn(1024L * 1024L + 1); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(2).getAvailable()).thenReturn(1024L * 1024L * 3); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(3).getAvailable()).thenReturn(1024L * 1024L * 3); initPolicy(policy, 1.0f); Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(3), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(3), policy.chooseVolume(volumes, 100)); initPolicy(policy, 0.0f); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); }
@Test(timeout=60000) public void testNotEnoughSpaceOnSelectedVolume() throws Exception { @SuppressWarnings("unchecked") final AvailableSpaceVolumeChoosingPolicy<FsVolumeSpi> policy = ReflectionUtils.newInstance(AvailableSpaceVolumeChoosingPolicy.class, null); List<FsVolumeSpi> volumes = new ArrayList<FsVolumeSpi>(); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(0).getAvailable()).thenReturn(1024L * 1024L); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(1).getAvailable()).thenReturn(1024L * 1024L * 3); initPolicy(policy, 0.0f); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 1024L * 1024L * 2)); }
@Test(timeout=60000) public void testAvailableSpaceChanges() throws Exception { @SuppressWarnings("unchecked") final AvailableSpaceVolumeChoosingPolicy<FsVolumeSpi> policy = ReflectionUtils.newInstance(AvailableSpaceVolumeChoosingPolicy.class, null); initPolicy(policy, 1.0f); List<FsVolumeSpi> volumes = new ArrayList<FsVolumeSpi>(); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(0).getAvailable()).thenReturn(1024L * 1024L); volumes.add(Mockito.mock(FsVolumeSpi.class)); Mockito.when(volumes.get(1).getAvailable()) .thenReturn(1024L * 1024L * 3) .thenReturn(1024L * 1024L * 3) .thenReturn(1024L * 1024L * 3) .thenReturn(1024L * 1024L * 1); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100)); }