class Scheduler implements InstanceListener, SlotAvailabilityListener, SlotProvider {

	@Override
	public CompletableFuture<LogicalSlot> allocateSlot(
			SlotRequestId slotRequestId,
			ScheduledUnit task,
			boolean allowQueued,
			SlotProfile slotProfile,
			Time allocationTimeout) {

		try {
			final Object ret = scheduleTask(task, allowQueued, slotProfile.getPreferredLocations());

			if (ret instanceof SimpleSlot) {
				return CompletableFuture.completedFuture((SimpleSlot) ret);
			}
			else if (ret instanceof CompletableFuture) {
				@SuppressWarnings("unchecked")
				CompletableFuture<LogicalSlot> typed = (CompletableFuture<LogicalSlot>) ret;
				return FutureUtils.orTimeout(typed, allocationTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
			}
			else {
				// this should never happen, simply guard this case with an exception
				throw new RuntimeException();
			}
		} catch (NoResourceAvailableException e) {
			return FutureUtils.completedExceptionally(e);
		}
	}

	public  Scheduler(Executor executor);

}