@Test public void getWorkerMultipleShards() { sConf.set(PropertyKey.USER_UFS_BLOCK_READ_LOCATION_POLICY_DETERMINISTIC_HASH_SHARDS, 2); DeterministicHashPolicy policy2 = (DeterministicHashPolicy) BlockLocationPolicy.Factory.create( DeterministicHashPolicy.class.getCanonicalName(), sConf); Set<String> addresses1 = new HashSet<>(); Set<String> addresses2 = new HashSet<>(); for (int i = 0; i < 100; i++) { addresses1.add(policy2.getWorker( GetWorkerOptions.defaults().setBlockWorkerInfos(mWorkerInfos) .setBlockInfo(new BlockInfo().setBlockId(1) .setLength(2 * (long) Constants.GB))).getHost()); addresses2.add(policy2.getWorker( GetWorkerOptions.defaults().setBlockWorkerInfos(mWorkerInfos) .setBlockInfo(new BlockInfo().setBlockId(1) .setLength(2 * (long) Constants.GB))).getHost()); } assertEquals(2, addresses1.size()); assertEquals(2, addresses2.size()); assertEquals(addresses1, addresses2); }
@Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> workerInfos = Lists.newArrayList(options.getBlockWorkerInfos()); workerInfos.sort((o1, o2) -> o1.getNetAddress().toString().compareToIgnoreCase(o2.getNetAddress().toString())); HashMap<WorkerNetAddress, BlockWorkerInfo> blockWorkerInfoMap = new HashMap<>(); for (BlockWorkerInfo workerInfo : options.getBlockWorkerInfos()) { blockWorkerInfoMap.put(workerInfo.getNetAddress(), workerInfo); } List<WorkerNetAddress> workers = new ArrayList<>(); int hv = Math.abs(mHashFunc.newHasher().putLong(options.getBlockInfo().getBlockId()).hash().asInt()); int index = hv % workerInfos.size(); for (BlockWorkerInfo blockWorkerInfoUnused : workerInfos) { WorkerNetAddress candidate = workerInfos.get(index).getNetAddress(); BlockWorkerInfo workerInfo = blockWorkerInfoMap.get(candidate); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength()) { workers.add(candidate); if (workers.size() >= mShards) { break; } } index = (index + 1) % workerInfos.size(); } return workers.isEmpty() ? null : workers.get(mRandom.nextInt(workers.size())); }
DeterministicHashPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> workerInfos = Lists.newArrayList(options.getBlockWorkerInfos()); workerInfos.sort((o1, o2) -> o1.getNetAddress().toString().compareToIgnoreCase(o2.getNetAddress().toString())); HashMap<WorkerNetAddress, BlockWorkerInfo> blockWorkerInfoMap = new HashMap<>(); for (BlockWorkerInfo workerInfo : options.getBlockWorkerInfos()) { blockWorkerInfoMap.put(workerInfo.getNetAddress(), workerInfo); } List<WorkerNetAddress> workers = new ArrayList<>(); int hv = Math.abs(mHashFunc.newHasher().putLong(options.getBlockInfo().getBlockId()).hash().asInt()); int index = hv % workerInfos.size(); for (BlockWorkerInfo blockWorkerInfoUnused : workerInfos) { WorkerNetAddress candidate = workerInfos.get(index).getNetAddress(); BlockWorkerInfo workerInfo = blockWorkerInfoMap.get(candidate); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength()) { workers.add(candidate); if (workers.size() >= mShards) { break; } } index = (index + 1) % workerInfos.size(); } return workers.isEmpty() ? null : workers.get(mRandom.nextInt(workers.size())); } }
DeterministicHashPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> workerInfos = Lists.newArrayList(options.getBlockWorkerInfos()); workerInfos.sort((o1, o2) -> o1.getNetAddress().toString().compareToIgnoreCase(o2.getNetAddress().toString())); HashMap<WorkerNetAddress, BlockWorkerInfo> blockWorkerInfoMap = new HashMap<>(); for (BlockWorkerInfo workerInfo : options.getBlockWorkerInfos()) { blockWorkerInfoMap.put(workerInfo.getNetAddress(), workerInfo); } List<WorkerNetAddress> workers = new ArrayList<>(); int hv = Math.abs(mHashFunc.newHasher().putLong(options.getBlockInfo().getBlockId()).hash().asInt()); int index = hv % workerInfos.size(); for (BlockWorkerInfo blockWorkerInfoUnused : workerInfos) { WorkerNetAddress candidate = workerInfos.get(index).getNetAddress(); BlockWorkerInfo workerInfo = blockWorkerInfoMap.get(candidate); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength()) { workers.add(candidate); if (workers.size() >= mShards) { break; } } index = (index + 1) % workerInfos.size(); } return workers.isEmpty() ? null : workers.get(mRandom.nextInt(workers.size())); } DeterministicHashPolicy(AlluxioConfiguration conf); }
DeterministicHashPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> workerInfos = Lists.newArrayList(options.getBlockWorkerInfos()); workerInfos.sort((o1, o2) -> o1.getNetAddress().toString().compareToIgnoreCase(o2.getNetAddress().toString())); HashMap<WorkerNetAddress, BlockWorkerInfo> blockWorkerInfoMap = new HashMap<>(); for (BlockWorkerInfo workerInfo : options.getBlockWorkerInfos()) { blockWorkerInfoMap.put(workerInfo.getNetAddress(), workerInfo); } List<WorkerNetAddress> workers = new ArrayList<>(); int hv = Math.abs(mHashFunc.newHasher().putLong(options.getBlockInfo().getBlockId()).hash().asInt()); int index = hv % workerInfos.size(); for (BlockWorkerInfo blockWorkerInfoUnused : workerInfos) { WorkerNetAddress candidate = workerInfos.get(index).getNetAddress(); BlockWorkerInfo workerInfo = blockWorkerInfoMap.get(candidate); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength()) { workers.add(candidate); if (workers.size() >= mShards) { break; } } index = (index + 1) % workerInfos.size(); } return workers.isEmpty() ? null : workers.get(mRandom.nextInt(workers.size())); } DeterministicHashPolicy(AlluxioConfiguration conf); @Override WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); }
DeterministicHashPolicy implements BlockLocationPolicy { @Override public WorkerNetAddress getWorker(GetWorkerOptions options) { List<BlockWorkerInfo> workerInfos = Lists.newArrayList(options.getBlockWorkerInfos()); workerInfos.sort((o1, o2) -> o1.getNetAddress().toString().compareToIgnoreCase(o2.getNetAddress().toString())); HashMap<WorkerNetAddress, BlockWorkerInfo> blockWorkerInfoMap = new HashMap<>(); for (BlockWorkerInfo workerInfo : options.getBlockWorkerInfos()) { blockWorkerInfoMap.put(workerInfo.getNetAddress(), workerInfo); } List<WorkerNetAddress> workers = new ArrayList<>(); int hv = Math.abs(mHashFunc.newHasher().putLong(options.getBlockInfo().getBlockId()).hash().asInt()); int index = hv % workerInfos.size(); for (BlockWorkerInfo blockWorkerInfoUnused : workerInfos) { WorkerNetAddress candidate = workerInfos.get(index).getNetAddress(); BlockWorkerInfo workerInfo = blockWorkerInfoMap.get(candidate); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength()) { workers.add(candidate); if (workers.size() >= mShards) { break; } } index = (index + 1) % workerInfos.size(); } return workers.isEmpty() ? null : workers.get(mRandom.nextInt(workers.size())); } DeterministicHashPolicy(AlluxioConfiguration conf); @Override WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString();  }