@Test public void testResourceCalculationOnSlotAllocatingAndReleasing() { ResourceProfile rp1 = new ResourceProfile(1.0, 100, 100, 100, 100, 100, Collections.emptyMap()); ResourceProfile rp2 = new ResourceProfile(2.0, 200, 200, 200, 200, 200, Collections.singletonMap("gpu", new GPUResource(2.0))); ResourceProfile rp3 = new ResourceProfile(3.0, 300, 300, 300, 300, 300, Collections.singletonMap("gpu", new GPUResource(3.0))); final TestingAllocatedSlotActions allocatedSlotActions = new TestingAllocatedSlotActions(); SlotSharingManager slotSharingManager = new SlotSharingManager( SLOT_SHARING_GROUP_ID, allocatedSlotActions, SLOT_OWNER); SlotSharingManager.MultiTaskSlot unresolvedRootSlot = slotSharingManager.createRootSlot( new SlotRequestId(), new CompletableFuture<>(), new SlotRequestId()); SlotSharingManager.MultiTaskSlot leftMultiTaskSlot = unresolvedRootSlot.allocateMultiTaskSlot(new SlotRequestId(), new SlotSharingGroupId()); SlotSharingManager.SingleTaskSlot firstChild = leftMultiTaskSlot.allocateSingleTaskSlot( new SlotRequestId(), rp1, new SlotSharingGroupId(), Locality.LOCAL); SlotSharingManager.SingleTaskSlot secondChild = leftMultiTaskSlot.allocateSingleTaskSlot( new SlotRequestId(), rp2, new SlotSharingGroupId(), Locality.LOCAL); assertEquals(rp1, firstChild.getReservedResources()); assertEquals(rp2, secondChild.getReservedResources()); assertEquals(rp1.merge(rp2), leftMultiTaskSlot.getReservedResources()); assertEquals(rp1.merge(rp2), unresolvedRootSlot.getReservedResources()); SlotSharingManager.SingleTaskSlot thirdChild = unresolvedRootSlot.allocateSingleTaskSlot( new SlotRequestId(), rp3, new SlotSharingGroupId(), Locality.LOCAL); assertEquals(rp3, thirdChild.getReservedResources()); assertEquals(rp1.merge(rp2).merge(rp3), unresolvedRootSlot.getReservedResources()); secondChild.release(new Throwable("Release for testing")); assertEquals(rp1, leftMultiTaskSlot.getReservedResources()); assertEquals(rp1.merge(rp3), unresolvedRootSlot.getReservedResources()); thirdChild.release(new Throwable("Release for testing")); assertEquals(rp1, unresolvedRootSlot.getReservedResources()); firstChild.release(new Throwable("Release for testing")); assertEquals(ResourceProfile.ZERO, unresolvedRootSlot.getReservedResources()); }
@Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } SlotSharingManager(
			SlotSharingGroupId slotSharingGroupId,
			AllocatedSlotActions allocatedSlotActions,
			SlotOwner slotOwner); }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } SlotSharingManager(
			SlotSharingGroupId slotSharingGroupId,
			AllocatedSlotActions allocatedSlotActions,
			SlotOwner slotOwner); boolean isEmpty(); boolean contains(SlotRequestId slotRequestId); @Nonnull Collection<SlotSelectionStrategy.SlotInfoAndResources> listResolvedRootSlotInfo(@Nullable AbstractID groupId); @Nullable MultiTaskSlot getResolvedRootSlot(@Nonnull SlotInfo slotInfo); @Override String toString(); @VisibleForTesting Collection<MultiTaskSlot> getResolvedRootSlots(); }
SlotSharingManager { @Nonnull MultiTaskSlot createRootSlot( SlotRequestId slotRequestId, CompletableFuture<? extends SlotContext> slotContextFuture, SlotRequestId allocatedSlotRequestId) { final MultiTaskSlot rootMultiTaskSlot = new MultiTaskSlot( slotRequestId, slotContextFuture, allocatedSlotRequestId); LOG.debug("Create multi task slot [{}] in slot [{}].", slotRequestId, allocatedSlotRequestId); allTaskSlots.put(slotRequestId, rootMultiTaskSlot); unresolvedRootSlots.put(slotRequestId, rootMultiTaskSlot); slotContextFuture.whenComplete( (SlotContext slotContext, Throwable throwable) -> { if (slotContext != null) { final MultiTaskSlot resolvedRootNode = unresolvedRootSlots.remove(slotRequestId); if (resolvedRootNode != null) { final AllocationID allocationId = slotContext.getAllocationId(); LOG.trace("Fulfill multi task slot [{}] with slot [{}].", slotRequestId, allocationId); final Map<AllocationID, MultiTaskSlot> innerMap = resolvedRootSlots.computeIfAbsent( slotContext.getTaskManagerLocation(), taskManagerLocation -> new HashMap<>(4)); MultiTaskSlot previousValue = innerMap.put(allocationId, resolvedRootNode); Preconditions.checkState(previousValue == null); } } else { rootMultiTaskSlot.release(throwable); } }); return rootMultiTaskSlot; } SlotSharingManager(
			SlotSharingGroupId slotSharingGroupId,
			AllocatedSlotActions allocatedSlotActions,
			SlotOwner slotOwner); boolean isEmpty(); boolean contains(SlotRequestId slotRequestId); @Nonnull Collection<SlotSelectionStrategy.SlotInfoAndResources> listResolvedRootSlotInfo(@Nullable AbstractID groupId); @Nullable MultiTaskSlot getResolvedRootSlot(@Nonnull SlotInfo slotInfo); @Override String toString(); @VisibleForTesting Collection<MultiTaskSlot> getResolvedRootSlots();  }