public SimpleSlot getSlotForTask ( CoLocationConstraint constraint , Iterable < TaskManagerLocation > locationPreferences ) { synchronized ( lock ) { if ( constraint . isAssignedAndAlive ( ) ) { // the shared slot of the co-location group is initialized and set we allocate a sub-slot final SharedSlot shared = constraint . getSharedSlot ( ) ; SimpleSlot subslot = shared . allocateSubSlot ( null ) ; subslot . setLocality ( Locality . LOCAL ) ; return subslot ; } else if ( constraint . isAssigned ( ) ) { // we had an assignment before. SharedSlot previous = constraint . getSharedSlot ( ) ; if ( previous == null ) { throw new IllegalStateException ( "Bug: Found assigned co-location constraint without a slot." ) ; } TaskManagerLocation location = previous . getTaskManagerLocation ( ) ; Tuple2 < SharedSlot , Locality > p = getSharedSlotForTask ( constraint . getGroupId ( ) , Collections . singleton ( location ) , true ) ; if ( p == null ) { return null ; } else { SharedSlot newSharedSlot = p . f0 ; // allocate the co-location group slot inside the shared slot SharedSlot constraintGroupSlot = newSharedSlot . allocateSharedSlot ( constraint . getGroupId ( ) ) ; if ( constraintGroupSlot != null ) { constraint . setSharedSlot ( constraintGroupSlot ) ; // the sub slots in the co location constraint slot have no group that they belong to // (other than the co-location-constraint slot) SimpleSlot subSlot = constraintGroupSlot . allocateSubSlot ( null ) ; subSlot . setLocality ( Locality . LOCAL ) ; return subSlot ; } else { // could not allocate the co-location-constraint shared slot return null ; } } } else { // the location constraint has not been associated with a shared slot, yet. // grab a new slot and initialize the constraint with that one. // preferred locations are defined by the vertex Tuple2 < SharedSlot , Locality > p = getSharedSlotForTask ( constraint . getGroupId ( ) , locationPreferences , false ) ; if ( p == null ) { // could not get a shared slot for this co-location-group return null ; } else { final SharedSlot availableShared = p . f0 ; final Locality l = p . f1 ; // allocate the co-location group slot inside the shared slot SharedSlot constraintGroupSlot = availableShared . allocateSharedSlot ( constraint . getGroupId ( ) ) ; // IMPORTANT: We do not lock the location, yet, since we cannot be sure that the // caller really sticks with the slot we picked! constraint . setSharedSlot ( constraintGroupSlot ) ; // the sub slots in the co location constraint slot have no group that they belong to // (other than the co-location-constraint slot) SimpleSlot sub = constraintGroupSlot . allocateSubSlot ( null ) ; sub . setLocality ( l ) ; return sub ; } } } }
Gets a slot for a task that has a co - location constraint . This method tries to grab a slot form the location - constraint s shared slot . If that slot has not been initialized then the method tries to grab another slot that is available for the location - constraint - group .