@Test public void getWorkerNoneEligibleAfterCache() { List<BlockWorkerInfo> workerInfoList = new ArrayList<>(); workerInfoList.add(new BlockWorkerInfo(new WorkerNetAddress().setHost("worker1") .setRpcPort(PORT).setDataPort(PORT).setWebPort(PORT), Constants.GB, 0)); RoundRobinPolicy policy = new RoundRobinPolicy(ConfigurationTestUtils.defaults()); GetWorkerOptions options = GetWorkerOptions.defaults().setBlockWorkerInfos(workerInfoList) .setBlockInfo(new BlockInfo().setLength((long) Constants.MB)); assertNotNull(policy.getWorker(options)); options.setBlockWorkerInfos(new ArrayList<>()); assertNull(policy.getWorker(options)); }
@Override @Nullable public WorkerNetAddress getWorker(GetWorkerOptions options) { Set<WorkerNetAddress> eligibleAddresses = new HashSet<>(); for (BlockWorkerInfo info : options.getBlockWorkerInfos()) { eligibleAddresses.add(info.getNetAddress()); } WorkerNetAddress address = mBlockLocationCache.get(options.getBlockInfo().getBlockId()); if (address != null && eligibleAddresses.contains(address)) { return address; } else { address = null; } if (!mInitialized) { mWorkerInfoList = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(mWorkerInfoList); mIndex = 0; mInitialized = true; } for (int i = 0; i < mWorkerInfoList.size(); i++) { WorkerNetAddress candidate = mWorkerInfoList.get(mIndex).getNetAddress(); BlockWorkerInfo workerInfo = findBlockWorkerInfo(options.getBlockWorkerInfos(), candidate); mIndex = (mIndex + 1) % mWorkerInfoList.size(); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength() && eligibleAddresses.contains(candidate)) { address = candidate; break; } } mBlockLocationCache.put(options.getBlockInfo().getBlockId(), address); return address; }
RoundRobinPolicy implements BlockLocationPolicy { @Override @Nullable public WorkerNetAddress getWorker(GetWorkerOptions options) { Set<WorkerNetAddress> eligibleAddresses = new HashSet<>(); for (BlockWorkerInfo info : options.getBlockWorkerInfos()) { eligibleAddresses.add(info.getNetAddress()); } WorkerNetAddress address = mBlockLocationCache.get(options.getBlockInfo().getBlockId()); if (address != null && eligibleAddresses.contains(address)) { return address; } else { address = null; } if (!mInitialized) { mWorkerInfoList = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(mWorkerInfoList); mIndex = 0; mInitialized = true; } for (int i = 0; i < mWorkerInfoList.size(); i++) { WorkerNetAddress candidate = mWorkerInfoList.get(mIndex).getNetAddress(); BlockWorkerInfo workerInfo = findBlockWorkerInfo(options.getBlockWorkerInfos(), candidate); mIndex = (mIndex + 1) % mWorkerInfoList.size(); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength() && eligibleAddresses.contains(candidate)) { address = candidate; break; } } mBlockLocationCache.put(options.getBlockInfo().getBlockId(), address); return address; } }
RoundRobinPolicy implements BlockLocationPolicy { @Override @Nullable public WorkerNetAddress getWorker(GetWorkerOptions options) { Set<WorkerNetAddress> eligibleAddresses = new HashSet<>(); for (BlockWorkerInfo info : options.getBlockWorkerInfos()) { eligibleAddresses.add(info.getNetAddress()); } WorkerNetAddress address = mBlockLocationCache.get(options.getBlockInfo().getBlockId()); if (address != null && eligibleAddresses.contains(address)) { return address; } else { address = null; } if (!mInitialized) { mWorkerInfoList = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(mWorkerInfoList); mIndex = 0; mInitialized = true; } for (int i = 0; i < mWorkerInfoList.size(); i++) { WorkerNetAddress candidate = mWorkerInfoList.get(mIndex).getNetAddress(); BlockWorkerInfo workerInfo = findBlockWorkerInfo(options.getBlockWorkerInfos(), candidate); mIndex = (mIndex + 1) % mWorkerInfoList.size(); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength() && eligibleAddresses.contains(candidate)) { address = candidate; break; } } mBlockLocationCache.put(options.getBlockInfo().getBlockId(), address); return address; } RoundRobinPolicy(AlluxioConfiguration conf); }
RoundRobinPolicy implements BlockLocationPolicy { @Override @Nullable public WorkerNetAddress getWorker(GetWorkerOptions options) { Set<WorkerNetAddress> eligibleAddresses = new HashSet<>(); for (BlockWorkerInfo info : options.getBlockWorkerInfos()) { eligibleAddresses.add(info.getNetAddress()); } WorkerNetAddress address = mBlockLocationCache.get(options.getBlockInfo().getBlockId()); if (address != null && eligibleAddresses.contains(address)) { return address; } else { address = null; } if (!mInitialized) { mWorkerInfoList = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(mWorkerInfoList); mIndex = 0; mInitialized = true; } for (int i = 0; i < mWorkerInfoList.size(); i++) { WorkerNetAddress candidate = mWorkerInfoList.get(mIndex).getNetAddress(); BlockWorkerInfo workerInfo = findBlockWorkerInfo(options.getBlockWorkerInfos(), candidate); mIndex = (mIndex + 1) % mWorkerInfoList.size(); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength() && eligibleAddresses.contains(candidate)) { address = candidate; break; } } mBlockLocationCache.put(options.getBlockInfo().getBlockId(), address); return address; } RoundRobinPolicy(AlluxioConfiguration conf); @Override @Nullable WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); }
RoundRobinPolicy implements BlockLocationPolicy { @Override @Nullable public WorkerNetAddress getWorker(GetWorkerOptions options) { Set<WorkerNetAddress> eligibleAddresses = new HashSet<>(); for (BlockWorkerInfo info : options.getBlockWorkerInfos()) { eligibleAddresses.add(info.getNetAddress()); } WorkerNetAddress address = mBlockLocationCache.get(options.getBlockInfo().getBlockId()); if (address != null && eligibleAddresses.contains(address)) { return address; } else { address = null; } if (!mInitialized) { mWorkerInfoList = Lists.newArrayList(options.getBlockWorkerInfos()); Collections.shuffle(mWorkerInfoList); mIndex = 0; mInitialized = true; } for (int i = 0; i < mWorkerInfoList.size(); i++) { WorkerNetAddress candidate = mWorkerInfoList.get(mIndex).getNetAddress(); BlockWorkerInfo workerInfo = findBlockWorkerInfo(options.getBlockWorkerInfos(), candidate); mIndex = (mIndex + 1) % mWorkerInfoList.size(); if (workerInfo != null && workerInfo.getCapacityBytes() >= options.getBlockInfo().getLength() && eligibleAddresses.contains(candidate)) { address = candidate; break; } } mBlockLocationCache.put(options.getBlockInfo().getBlockId(), address); return address; } RoundRobinPolicy(AlluxioConfiguration conf); @Override @Nullable WorkerNetAddress getWorker(GetWorkerOptions options); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString();  }