AvailableSpaceVolumeChoosingPolicy implements VolumeChoosingPolicy<V>, Configurable { @Override public V chooseVolume(List<V> volumes, long replicaSize, String storageId) throws IOException { if (volumes.size() < 1) { throw new DiskOutOfSpaceException("No more available volumes"); } StorageType storageType = volumes.get(0).getStorageType(); int index = storageType != null ? storageType.ordinal() : StorageType.DEFAULT.ordinal(); synchronized (syncLocks[index]) { return doChooseVolume(volumes, replicaSize, storageId); } } AvailableSpaceVolumeChoosingPolicy(Random random);  AvailableSpaceVolumeChoosingPolicy(); @Override void setConf(Configuration conf); @Override Configuration getConf(); @Override V chooseVolume(List<V> volumes, long replicaSize, String storageId);  }
@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, null)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100, null)); }
@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, null)); Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100, null)); initPolicy(policy, 0.0f); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100, null)); }
@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, null)); Assert.assertEquals(volumes.get(3), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(2), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(3), policy.chooseVolume(volumes, 100, null)); initPolicy(policy, 0.0f); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(0), policy.chooseVolume(volumes, 100, null)); Assert.assertEquals(volumes.get(1), policy.chooseVolume(volumes, 100, null)); }
@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, null)); }
@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, null)); }