@Test public void testFormulateDestinationListForOutput() throws IOException { final Set<PeerStatus> collection = new HashSet<>(); collection.add(new PeerStatus(new PeerDescription("HasMedium", 1111, true), 4096, true)); collection.add(new PeerStatus(new PeerDescription("HasLots", 2222, true), 10240, true)); collection.add(new PeerStatus(new PeerDescription("HasLittle", 3333, true), 1024, true)); collection.add(new PeerStatus(new PeerDescription("HasMedium", 4444, true), 4096, true)); collection.add(new PeerStatus(new PeerDescription("HasMedium", 5555, true), 4096, true)); PeerStatusProvider peerStatusProvider = Mockito.mock(PeerStatusProvider.class); PeerSelector peerSelector = new PeerSelector(peerStatusProvider, null); final List<PeerStatus> destinations = peerSelector.formulateDestinationList(collection, TransferDirection.RECEIVE); final Map<String, Integer> selectedCounts = calculateAverageSelectedCount(collection, destinations); logger.info("selectedCounts={}", selectedCounts); assertTrue("HasLots should send lots", selectedCounts.get("HasLots") > selectedCounts.get("HasMedium")); assertTrue("HasMedium should send medium", selectedCounts.get("HasMedium") > selectedCounts.get("HasLittle")); }
List<PeerStatus> formulateDestinationList(final Set<PeerStatus> statuses, final TransferDirection direction) { final int numDestinations = Math.max(128, statuses.size()); final Map<PeerStatus, Integer> entryCountMap = new HashMap<>(); long totalFlowFileCount = 0L; for (final PeerStatus nodeInfo : statuses) { totalFlowFileCount += nodeInfo.getFlowFileCount(); } int totalEntries = 0; for (final PeerStatus nodeInfo : statuses) { final int flowFileCount = nodeInfo.getFlowFileCount(); final double percentageOfFlowFiles = Math.min(0.8D, ((double) flowFileCount / (double) totalFlowFileCount)); final double relativeWeighting = (direction == TransferDirection.SEND) ? (1 - percentageOfFlowFiles) : percentageOfFlowFiles; final int entries = Math.max(1, (int) (numDestinations * relativeWeighting)); entryCountMap.put(nodeInfo, Math.max(1, entries)); totalEntries += entries; } final List<PeerStatus> destinations = new ArrayList<>(totalEntries); for (int i = 0; i < totalEntries; i++) { destinations.add(null); } for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final PeerStatus nodeInfo = entry.getKey(); final int numEntries = entry.getValue(); int skipIndex = numEntries; for (int i = 0; i < numEntries; i++) { int n = (skipIndex * i); while (true) { final int index = n % destinations.size(); PeerStatus status = destinations.get(index); if (status == null) { status = new PeerStatus(nodeInfo.getPeerDescription(), nodeInfo.getFlowFileCount(), nodeInfo.isQueryForPeers()); destinations.set(index, status); break; } else { n++; } } } } final StringBuilder distributionDescription = new StringBuilder(); distributionDescription.append("New Weighted Distribution of Nodes:"); for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final double percentage = entry.getValue() * 100D / destinations.size(); distributionDescription.append("\n").append(entry.getKey()).append(" will receive ").append(percentage).append("% of data"); } logger.info(distributionDescription.toString()); return destinations; }
PeerSelector { List<PeerStatus> formulateDestinationList(final Set<PeerStatus> statuses, final TransferDirection direction) { final int numDestinations = Math.max(128, statuses.size()); final Map<PeerStatus, Integer> entryCountMap = new HashMap<>(); long totalFlowFileCount = 0L; for (final PeerStatus nodeInfo : statuses) { totalFlowFileCount += nodeInfo.getFlowFileCount(); } int totalEntries = 0; for (final PeerStatus nodeInfo : statuses) { final int flowFileCount = nodeInfo.getFlowFileCount(); final double percentageOfFlowFiles = Math.min(0.8D, ((double) flowFileCount / (double) totalFlowFileCount)); final double relativeWeighting = (direction == TransferDirection.SEND) ? (1 - percentageOfFlowFiles) : percentageOfFlowFiles; final int entries = Math.max(1, (int) (numDestinations * relativeWeighting)); entryCountMap.put(nodeInfo, Math.max(1, entries)); totalEntries += entries; } final List<PeerStatus> destinations = new ArrayList<>(totalEntries); for (int i = 0; i < totalEntries; i++) { destinations.add(null); } for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final PeerStatus nodeInfo = entry.getKey(); final int numEntries = entry.getValue(); int skipIndex = numEntries; for (int i = 0; i < numEntries; i++) { int n = (skipIndex * i); while (true) { final int index = n % destinations.size(); PeerStatus status = destinations.get(index); if (status == null) { status = new PeerStatus(nodeInfo.getPeerDescription(), nodeInfo.getFlowFileCount(), nodeInfo.isQueryForPeers()); destinations.set(index, status); break; } else { n++; } } } } final StringBuilder distributionDescription = new StringBuilder(); distributionDescription.append("New Weighted Distribution of Nodes:"); for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final double percentage = entry.getValue() * 100D / destinations.size(); distributionDescription.append("\n").append(entry.getKey()).append(" will receive ").append(percentage).append("% of data"); } logger.info(distributionDescription.toString()); return destinations; } }
PeerSelector { List<PeerStatus> formulateDestinationList(final Set<PeerStatus> statuses, final TransferDirection direction) { final int numDestinations = Math.max(128, statuses.size()); final Map<PeerStatus, Integer> entryCountMap = new HashMap<>(); long totalFlowFileCount = 0L; for (final PeerStatus nodeInfo : statuses) { totalFlowFileCount += nodeInfo.getFlowFileCount(); } int totalEntries = 0; for (final PeerStatus nodeInfo : statuses) { final int flowFileCount = nodeInfo.getFlowFileCount(); final double percentageOfFlowFiles = Math.min(0.8D, ((double) flowFileCount / (double) totalFlowFileCount)); final double relativeWeighting = (direction == TransferDirection.SEND) ? (1 - percentageOfFlowFiles) : percentageOfFlowFiles; final int entries = Math.max(1, (int) (numDestinations * relativeWeighting)); entryCountMap.put(nodeInfo, Math.max(1, entries)); totalEntries += entries; } final List<PeerStatus> destinations = new ArrayList<>(totalEntries); for (int i = 0; i < totalEntries; i++) { destinations.add(null); } for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final PeerStatus nodeInfo = entry.getKey(); final int numEntries = entry.getValue(); int skipIndex = numEntries; for (int i = 0; i < numEntries; i++) { int n = (skipIndex * i); while (true) { final int index = n % destinations.size(); PeerStatus status = destinations.get(index); if (status == null) { status = new PeerStatus(nodeInfo.getPeerDescription(), nodeInfo.getFlowFileCount(), nodeInfo.isQueryForPeers()); destinations.set(index, status); break; } else { n++; } } } } final StringBuilder distributionDescription = new StringBuilder(); distributionDescription.append("New Weighted Distribution of Nodes:"); for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final double percentage = entry.getValue() * 100D / destinations.size(); distributionDescription.append("\n").append(entry.getKey()).append(" will receive ").append(percentage).append("% of data"); } logger.info(distributionDescription.toString()); return destinations; } PeerSelector(final PeerStatusProvider peerStatusProvider, final File persistenceFile); }
PeerSelector { List<PeerStatus> formulateDestinationList(final Set<PeerStatus> statuses, final TransferDirection direction) { final int numDestinations = Math.max(128, statuses.size()); final Map<PeerStatus, Integer> entryCountMap = new HashMap<>(); long totalFlowFileCount = 0L; for (final PeerStatus nodeInfo : statuses) { totalFlowFileCount += nodeInfo.getFlowFileCount(); } int totalEntries = 0; for (final PeerStatus nodeInfo : statuses) { final int flowFileCount = nodeInfo.getFlowFileCount(); final double percentageOfFlowFiles = Math.min(0.8D, ((double) flowFileCount / (double) totalFlowFileCount)); final double relativeWeighting = (direction == TransferDirection.SEND) ? (1 - percentageOfFlowFiles) : percentageOfFlowFiles; final int entries = Math.max(1, (int) (numDestinations * relativeWeighting)); entryCountMap.put(nodeInfo, Math.max(1, entries)); totalEntries += entries; } final List<PeerStatus> destinations = new ArrayList<>(totalEntries); for (int i = 0; i < totalEntries; i++) { destinations.add(null); } for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final PeerStatus nodeInfo = entry.getKey(); final int numEntries = entry.getValue(); int skipIndex = numEntries; for (int i = 0; i < numEntries; i++) { int n = (skipIndex * i); while (true) { final int index = n % destinations.size(); PeerStatus status = destinations.get(index); if (status == null) { status = new PeerStatus(nodeInfo.getPeerDescription(), nodeInfo.getFlowFileCount(), nodeInfo.isQueryForPeers()); destinations.set(index, status); break; } else { n++; } } } } final StringBuilder distributionDescription = new StringBuilder(); distributionDescription.append("New Weighted Distribution of Nodes:"); for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final double percentage = entry.getValue() * 100D / destinations.size(); distributionDescription.append("\n").append(entry.getKey()).append(" will receive ").append(percentage).append("% of data"); } logger.info(distributionDescription.toString()); return destinations; } PeerSelector(final PeerStatusProvider peerStatusProvider, final File persistenceFile); void penalize(final Peer peer, final long penalizationMillis); void penalize(final PeerDescription peerDescription, final long penalizationMillis); boolean isPenalized(final PeerStatus peerStatus); void clear(); PeerStatus getNextPeerStatus(final TransferDirection direction); void refreshPeers(); void setEventReporter(EventReporter eventReporter); }
PeerSelector { List<PeerStatus> formulateDestinationList(final Set<PeerStatus> statuses, final TransferDirection direction) { final int numDestinations = Math.max(128, statuses.size()); final Map<PeerStatus, Integer> entryCountMap = new HashMap<>(); long totalFlowFileCount = 0L; for (final PeerStatus nodeInfo : statuses) { totalFlowFileCount += nodeInfo.getFlowFileCount(); } int totalEntries = 0; for (final PeerStatus nodeInfo : statuses) { final int flowFileCount = nodeInfo.getFlowFileCount(); final double percentageOfFlowFiles = Math.min(0.8D, ((double) flowFileCount / (double) totalFlowFileCount)); final double relativeWeighting = (direction == TransferDirection.SEND) ? (1 - percentageOfFlowFiles) : percentageOfFlowFiles; final int entries = Math.max(1, (int) (numDestinations * relativeWeighting)); entryCountMap.put(nodeInfo, Math.max(1, entries)); totalEntries += entries; } final List<PeerStatus> destinations = new ArrayList<>(totalEntries); for (int i = 0; i < totalEntries; i++) { destinations.add(null); } for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final PeerStatus nodeInfo = entry.getKey(); final int numEntries = entry.getValue(); int skipIndex = numEntries; for (int i = 0; i < numEntries; i++) { int n = (skipIndex * i); while (true) { final int index = n % destinations.size(); PeerStatus status = destinations.get(index); if (status == null) { status = new PeerStatus(nodeInfo.getPeerDescription(), nodeInfo.getFlowFileCount(), nodeInfo.isQueryForPeers()); destinations.set(index, status); break; } else { n++; } } } } final StringBuilder distributionDescription = new StringBuilder(); distributionDescription.append("New Weighted Distribution of Nodes:"); for (final Map.Entry<PeerStatus, Integer> entry : entryCountMap.entrySet()) { final double percentage = entry.getValue() * 100D / destinations.size(); distributionDescription.append("\n").append(entry.getKey()).append(" will receive ").append(percentage).append("% of data"); } logger.info(distributionDescription.toString()); return destinations; } PeerSelector(final PeerStatusProvider peerStatusProvider, final File persistenceFile); void penalize(final Peer peer, final long penalizationMillis); void penalize(final PeerDescription peerDescription, final long penalizationMillis); boolean isPenalized(final PeerStatus peerStatus); void clear(); PeerStatus getNextPeerStatus(final TransferDirection direction); void refreshPeers(); void setEventReporter(EventReporter eventReporter);  }