TaskAssignor { public static <C, T extends Comparable<T>> Map<C, ClientState<T>> assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas) { long seed = 0L; for (C client : states.keySet()) { seed += client.hashCode(); } TaskAssignor<C, T> assignor = new TaskAssignor<>(states, tasks, seed); log.info("Assigning tasks to clients: {}, prevAssignmentBalanced: {}, " + "prevClientsUnchangeed: {}, tasks: {}, replicas: {}", states, assignor.prevAssignmentBalanced, assignor.prevClientsUnchanged, tasks, numStandbyReplicas); assignor.assignTasks(); if (numStandbyReplicas > 0) assignor.assignStandbyTasks(numStandbyReplicas); log.info("Assigned with: " + assignor.states); return assignor.states; } private  TaskAssignor(Map<C, ClientState<T>> states, Set<T> tasks, long randomSeed); static Map<C, ClientState<T>> assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas); void assignTasks(); void assignStandbyTasks(int numStandbyReplicas);  }
@Test public void testAssignWithoutStandby() { HashMap<Integer, ClientState<Integer>> states = new HashMap<>(); for (int i = 0; i < 6; i++) { states.put(i, new ClientState<Integer>(1d)); } Set<Integer> tasks; Map<Integer, ClientState<Integer>> assignments; int numActiveTasks; int numAssignedTasks; tasks = mkSet(0, 1, 2, 3, 4, 5); assignments = TaskAssignor.assign(states, tasks, 0); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertEquals(1, assignment.activeTasks.size()); assertEquals(1, assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size(), numAssignedTasks); tasks = mkSet(0, 1, 2, 3, 4, 5, 6, 7); assignments = TaskAssignor.assign(states, tasks, 0); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertTrue(1 <= assignment.activeTasks.size()); assertTrue(2 >= assignment.activeTasks.size()); assertTrue(1 <= assignment.assignedTasks.size()); assertTrue(2 >= assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size(), numAssignedTasks); tasks = mkSet(0, 1, 2, 3); assignments = TaskAssignor.assign(states, tasks, 0); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertTrue(0 <= assignment.activeTasks.size()); assertTrue(1 >= assignment.activeTasks.size()); assertTrue(0 <= assignment.assignedTasks.size()); assertTrue(1 >= assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size(), numAssignedTasks); }
@Test public void testAssignWithStandby() { HashMap<Integer, ClientState<Integer>> states = new HashMap<>(); for (int i = 0; i < 6; i++) { states.put(i, new ClientState<Integer>(1d)); } Set<Integer> tasks; Map<Integer, ClientState<Integer>> assignments; int numActiveTasks; int numAssignedTasks; tasks = mkSet(0, 1, 2, 3, 4, 5); numActiveTasks = 0; numAssignedTasks = 0; assignments = TaskAssignor.assign(states, tasks, 1); for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertEquals(1, assignment.activeTasks.size()); assertEquals(2, assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size() * 2, numAssignedTasks); tasks = mkSet(0, 1, 2, 3, 4, 5, 6, 7); assignments = TaskAssignor.assign(states, tasks, 1); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertTrue(1 <= assignment.activeTasks.size()); assertTrue(2 >= assignment.activeTasks.size()); assertTrue(2 <= assignment.assignedTasks.size()); assertTrue(3 >= assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size() * 2, numAssignedTasks); tasks = mkSet(0, 1, 2, 3); assignments = TaskAssignor.assign(states, tasks, 1); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertTrue(0 <= assignment.activeTasks.size()); assertTrue(1 >= assignment.activeTasks.size()); assertTrue(1 <= assignment.assignedTasks.size()); assertTrue(2 >= assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size() * 2, numAssignedTasks); tasks = mkSet(0, 1); assignments = TaskAssignor.assign(states, tasks, 1); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertTrue(0 <= assignment.activeTasks.size()); assertTrue(1 >= assignment.activeTasks.size()); assertTrue(0 <= assignment.assignedTasks.size()); assertTrue(1 >= assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size() * 2, numAssignedTasks); assignments = TaskAssignor.assign(states, tasks, 2); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertTrue(0 <= assignment.activeTasks.size()); assertTrue(1 >= assignment.activeTasks.size()); assertTrue(1 == assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size() * 3, numAssignedTasks); assignments = TaskAssignor.assign(states, tasks, 3); numActiveTasks = 0; numAssignedTasks = 0; for (ClientState<Integer> assignment : assignments.values()) { numActiveTasks += assignment.activeTasks.size(); numAssignedTasks += assignment.assignedTasks.size(); assertTrue(0 <= assignment.activeTasks.size()); assertTrue(1 >= assignment.activeTasks.size()); assertTrue(1 <= assignment.assignedTasks.size()); assertTrue(2 >= assignment.assignedTasks.size()); } assertEquals(tasks.size(), numActiveTasks); assertEquals(tasks.size() * 4, numAssignedTasks); }
@Test public void testStickiness() { List<Integer> tasks; Map<Integer, ClientState<Integer>> states; Map<Integer, ClientState<Integer>> assignments; int i; tasks = mkList(0, 1, 2, 3, 4, 5); Collections.shuffle(tasks); states = new HashMap<>(); i = 0; for (int task : tasks) { ClientState<Integer> state = new ClientState<>(1d); state.prevActiveTasks.add(task); state.prevAssignedTasks.add(task); states.put(i++, state); } assignments = TaskAssignor.assign(states, mkSet(0, 1, 2, 3, 4, 5), 0); for (int client : states.keySet()) { Set<Integer> oldActive = states.get(client).prevActiveTasks; Set<Integer> oldAssigned = states.get(client).prevAssignedTasks; Set<Integer> newActive = assignments.get(client).activeTasks; Set<Integer> newAssigned = assignments.get(client).assignedTasks; assertEquals(oldActive, newActive); assertEquals(oldAssigned, newAssigned); } tasks = mkList(0, 1, 2, 3, -1, -1); Collections.shuffle(tasks); states = new HashMap<>(); i = 0; for (int task : tasks) { ClientState<Integer> state = new ClientState<>(1d); if (task >= 0) { state.prevActiveTasks.add(task); state.prevAssignedTasks.add(task); } states.put(i++, state); } assignments = TaskAssignor.assign(states, mkSet(0, 1, 2, 3), 0); for (int client : states.keySet()) { Set<Integer> oldActive = states.get(client).prevActiveTasks; Set<Integer> oldAssigned = states.get(client).prevAssignedTasks; Set<Integer> newActive = assignments.get(client).activeTasks; Set<Integer> newAssigned = assignments.get(client).assignedTasks; assertEquals(oldActive, newActive); assertEquals(oldAssigned, newAssigned); } List<Set<Integer>> taskSets = mkList(mkSet(0, 1), mkSet(2, 3), mkSet(4, 5), mkSet(6, 7), mkSet(8, 9), mkSet(10, 11)); Collections.shuffle(taskSets); states = new HashMap<>(); i = 0; for (Set<Integer> taskSet : taskSets) { ClientState<Integer> state = new ClientState<>(1d); state.prevActiveTasks.addAll(taskSet); state.prevAssignedTasks.addAll(taskSet); states.put(i++, state); } assignments = TaskAssignor.assign(states, mkSet(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11), 0); for (int client : states.keySet()) { Set<Integer> oldActive = states.get(client).prevActiveTasks; Set<Integer> oldAssigned = states.get(client).prevAssignedTasks; Set<Integer> newActive = assignments.get(client).activeTasks; Set<Integer> newAssigned = assignments.get(client).assignedTasks; Set<Integer> intersection = new HashSet<>(); intersection.addAll(oldActive); intersection.retainAll(newActive); assertTrue(intersection.size() > 0); intersection.clear(); intersection.addAll(oldAssigned); intersection.retainAll(newAssigned); assertTrue(intersection.size() > 0); } }