import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private static Resource.Builder setValue(Resource.Builder builder, Value value) {
    builder.setType(value.getType());
    switch (value.getType()) {
    case SCALAR:
        builder.setScalar(value.getScalar());
        return builder;
    case RANGES:
        builder.setRanges(value.getRanges());
        return builder;
    case SET:
        builder.setSet(value.getSet());
        return builder;
    default:
        throw new IllegalArgumentException(String.format("Unsupported spec value type: %s", value.getType()));
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private void freeMergedResource(MesosResource mesosResource) {
    if (mesosResource.getResourceId().isPresent()) {
        dynamicallyReservedPoolByResourceId.remove(mesosResource.getResourceId().get());
        logger.info("Freed resource: {}", !dynamicallyReservedPoolByResourceId
                .containsKey(mesosResource.getResourceId().get()));
    }

    String previousRole = mesosResource.getPreviousRole();
    Map<String, Value> pool = reservableMergedPoolByRole.get(previousRole);
    if (pool == null) {
        pool = new HashMap<>();
    }

    Value currValue = pool.get(mesosResource.getName());
    if (currValue == null) {
        currValue = ValueUtils.getZero(mesosResource.getType());
    }

    Value updatedValue = ValueUtils.add(currValue, mesosResource.getValue());
    pool.put(mesosResource.getName(), updatedValue);
    reservableMergedPoolByRole.put(previousRole, pool);
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private Map<String, Map<String, Value>> getReservableMergedPool(Collection<MesosResource> mesosResources) {
    Map<String, List<MesosResource>> rolePool = new HashMap<>();
    for (MesosResource mesosResource : getMergedResources(mesosResources)) {
        if (!mesosResource.hasResourceId()) {
            String role = mesosResource.getRole();
            List<MesosResource> resources = rolePool.get(role);
            if (resources == null) {
                resources = new ArrayList<>();
            }

            resources.add(mesosResource);
            rolePool.put(role, resources);
        }
    }

    Map<String, Map<String, Value>> roleResourcePool = new HashMap<>();
    for (Map.Entry<String, List<MesosResource>> entry : rolePool.entrySet()) {
        roleResourcePool.put(entry.getKey(), getResourcePool(entry.getValue()));
    }

    return roleResourcePool;
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public static Value getValue(Resource resource) {
    Type type = resource.getType();
    Value.Builder builder = Value.newBuilder();
    builder.setType(type);

    switch (type) {
        case SCALAR:
            return builder.setScalar(resource.getScalar()).build();
        case RANGES:
            return builder.setRanges(resource.getRanges()).build();
        case SET:
            return builder.setSet(resource.getSet()).build();
        default:
            return null;
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public static Value add(Value val1, Value val2) {
    Type type1 = val1.getType();
    Type type2 = val2.getType();

    if (type1 != type2) {
        return null;
    }

    switch (type1) {
        case SCALAR:
            Value.Scalar scalar = add(val1.getScalar(), val2.getScalar());
            return Value.newBuilder().setType(type1).setScalar(scalar).build();
        case RANGES:
            Value.Ranges ranges = add(val1.getRanges(), val2.getRanges());
            return Value.newBuilder().setType(type1).setRanges(ranges).build();
        default:
            return null;
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public static Value subtract(Value val1, Value val2) {
    Type type1 = val1.getType();
    Type type2 = val2.getType();

    if (type1 != type2) {
        return null;
    }

    switch (type1) {
        case SCALAR:
            Value.Scalar scalar = subtract(val1.getScalar(), val2.getScalar());
            return Value.newBuilder().setType(type1).setScalar(scalar).build();
        case RANGES:
            Value.Ranges ranges = subtract(val1.getRanges(), val2.getRanges());
            return Value.newBuilder().setType(type1).setRanges(ranges).build();
        default:
            return null;
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public static Integer compare(Value val1, Value val2) {
    Type type1 = val1.getType();
    Type type2 = val2.getType();

    if (type1 != type2) {
        return null;
    }

    switch (type1) {
        case SCALAR:
            return compare(val1.getScalar(), val2.getScalar());
        case RANGES:
            return compare(val1.getRanges(), val2.getRanges());
        default:
            return null;
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
/**
 * Picks a port number from the given offer's port resources
 *
 * @param taskId the TaskID to register the picked ports with
 * @param offer  the offer from which's resources to pick a port
 * @return a port number
 */
public long pickAndRegisterPortNumber(TaskID taskId, Offer offer) {
    for (Resource resource : offer.getResourcesList()) {
        if (!resource.getName().equals("ports")) continue;

        List<Value.Range> offeredRanges = resource.getRanges().getRangeList();
        for (Value.Range portRange : offeredRanges) {
            long begin = portRange.getBegin();
            long end = portRange.getEnd();
            for (long port = begin; port < end; port++) {
                if (!usedPortNumbers.values().contains(port)) {
                    usedPortNumbers.put(taskId, port);
                    LOGGER.info("Task {} received port {}.", taskId, port);
                    return port;
                }
            }
        }
    }
    LOGGER.warn("Offer {} had no unused port to offer! Task {} received no port!", offer.getId().getValue(), taskId.getValue());
    return -1;
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private List<Resource> getExecutorResources() {
  return Arrays.asList(Resource.newBuilder()
      .setName("cpus")
      .setType(Value.Type.SCALAR)
      .setScalar(Value.Scalar.newBuilder()
          .setValue(hbaseFrameworkConfig.getExecutorCpus()).build())
      .setRole(hbaseFrameworkConfig.getHbaseRole())
      .build(),
      Resource
          .newBuilder()
          .setName("mem")
          .setType(Value.Type.SCALAR)
          .setScalar(Value.Scalar
              .newBuilder()
              .setValue(hbaseFrameworkConfig.getExecutorHeap()
                  * hbaseFrameworkConfig.getJvmOverhead()).build())
          .setRole(hbaseFrameworkConfig.getHbaseRole())
          .build());
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private List<Resource> getTaskResources(String taskName) {
  return Arrays.asList(Resource.newBuilder()
      .setName("cpus")
      .setType(Value.Type.SCALAR)
      .setScalar(Value.Scalar.newBuilder()
          .setValue(hbaseFrameworkConfig.getTaskCpus(taskName)).build())
      .setRole(hbaseFrameworkConfig.getHbaseRole())
      .build(),
      Resource.newBuilder()
          .setName("mem")
          .setType(Value.Type.SCALAR)
          .setScalar(Value.Scalar.newBuilder()
              .setValue(hbaseFrameworkConfig.getTaskHeapSize(taskName) *
                  hbaseFrameworkConfig.getJvmOverhead()).build())
          .setRole(hbaseFrameworkConfig.getHbaseRole())
          .build());
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
/**
 * Builds a TaskInfo from the given jobspec
 * @param job - JobSpec to TaskInfo-ify
 * @param offer - offer add extra data (SlaveId)
 * @return TaskInfo fully formed
 */
private TaskInfo getTaskInfo(JobSpec job,Offer offer) {
    TaskID taskId = TaskID.newBuilder().setValue(job.getJob().getId()).build();
    TaskInfo info = TaskInfo.newBuilder().setName("task " + taskId.getValue())
             .setTaskId(taskId)
             .setSlaveId(offer.getSlaveId())
             .addResources(Resource.newBuilder()
                           .setName("cpus")
                           .setType(Value.Type.SCALAR)
                           .setScalar(Value.Scalar.newBuilder().setValue(job.getJob().getLoadValue()*1.0)))
             .addResources(Resource.newBuilder()
                           .setName("mem")
                           .setType(Value.Type.SCALAR)
                           .setScalar(Value.Scalar.newBuilder().setValue(job.getJob().getLoadValue()*1024.0)))
             .setExecutor(ExecutorInfo.newBuilder(executor)).setData(MesosUtilities.jobSpecToByteString(job)).build();
    return info;
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private ResourceBuilder(String resourceName, Value value, String preReservedRole) {
    this.resourceName = resourceName;
    this.value = value;
    this.preReservedRole = preReservedRole;
    this.role = Optional.empty();
    this.principal = Optional.empty();
    this.resourceId = Optional.empty();
    this.diskContainerPath = Optional.empty();
    this.diskPersistenceId = Optional.empty();
    this.diskMountInfo = Optional.empty();
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public Optional<MesosResource> consumeReserved(String name, Value value, String resourceId) {
    MesosResource mesosResource = dynamicallyReservedPoolByResourceId.get(resourceId);

    if (mesosResource != null) {
        if (mesosResource.isAtomic()) {
            if (sufficientValue(value, mesosResource.getValue())) {
                dynamicallyReservedPoolByResourceId.remove(resourceId);
            } else {
                logger.warn("Reserved atomic quantity of {} is insufficient: desired {}, reserved {}",
                        name,
                        TextFormat.shortDebugString(value),
                        TextFormat.shortDebugString(mesosResource.getValue()));
                return Optional.empty();
            }
        } else {
            Value availableValue = mesosResource.getValue();
            if (ValueUtils.compare(availableValue, value) > 0) {
                // Update the value in pool with the remaining unclaimed resource amount
                Resource remaining = ResourceBuilder.fromExistingResource(mesosResource.getResource())
                        .setValue(ValueUtils.subtract(availableValue, value))
                        .build();
                dynamicallyReservedPoolByResourceId.put(resourceId, new MesosResource(remaining));
                // Return only the claimed resource amount from this reservation
            } else {
                dynamicallyReservedPoolByResourceId.remove(resourceId);
            }
        }
    } else {
        logger.warn("Failed to find reserved {} resource with ID: {}. Reserved resource IDs are: {}",
                name,
                resourceId,
                dynamicallyReservedPoolByResourceId.keySet());
    }

    return Optional.ofNullable(mesosResource);
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public Optional<MesosResource> consumeAtomic(String resourceName, Value value) {
    List<MesosResource> atomicResources = unreservedAtomicPool.get(resourceName);
    List<MesosResource> filteredResources = new ArrayList<>();
    Optional<MesosResource> sufficientResource = Optional.empty();

    if (atomicResources != null) {
        for (MesosResource atomicResource : atomicResources) {
            if (!sufficientResource.isPresent() && sufficientValue(value, atomicResource.getValue())) {
                sufficientResource = Optional.of(atomicResource);
                // do NOT break: ensure filteredResources is fully populated
            } else {
                filteredResources.add(atomicResource);
            }
        }
    }

    if (filteredResources.isEmpty()) {
        unreservedAtomicPool.remove(resourceName);
    } else {
        unreservedAtomicPool.put(resourceName, filteredResources);
    }

    if (!sufficientResource.isPresent()) {
        if (atomicResources == null) {
            logger.info("Offer lacks any atomic resources named {}", resourceName);
        } else {
            logger.info("Offered quantity in all {} instances of {} is insufficient: desired {}",
                    atomicResources.size(),
                    resourceName,
                    value);
        }
    }

    return sufficientResource;
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public Optional<MesosResource> consumeReservableMerged(String name, Value desiredValue, String preReservedRole) {
    Map<String, Value> pool = reservableMergedPoolByRole.get(preReservedRole);
    if (pool == null) {
        logger.info("No unreserved resources available for role '{}'. Reservable roles are: {}",
                preReservedRole, reservableMergedPoolByRole.keySet());
        return Optional.empty();
    }

    Value availableValue = pool.get(name);

    if (sufficientValue(desiredValue, availableValue)) {
        pool.put(name, ValueUtils.subtract(availableValue, desiredValue));
        reservableMergedPoolByRole.put(preReservedRole, pool);

        Resource.Builder builder = ResourceBuilder.fromUnreservedValue(name, desiredValue).build().toBuilder();
        if (Capabilities.getInstance().supportsPreReservedResources() &&
                !preReservedRole.equals(Constants.ANY_ROLE)) {
            builder.addReservations(
                    Resource.ReservationInfo.newBuilder()
                            .setRole(preReservedRole)
                            .setType(Resource.ReservationInfo.Type.STATIC));
        }

        return Optional.of(new MesosResource(builder.build()));
    } else {
        if (availableValue == null) {
            logger.info("Offer lacks any unreserved {} resources for role {}", name, preReservedRole);
        } else {
            logger.info("Offered quantity of {} for role {} is insufficient: desired {}, offered {}",
                    name,
                    preReservedRole,
                    TextFormat.shortDebugString(desiredValue),
                    TextFormat.shortDebugString(availableValue));
        }
        return Optional.empty();
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private static boolean sufficientValue(Value desired, Value available) {
    if (desired == null) {
        return true;
    } else if (available == null) {
        return false;
    }

    Value difference = ValueUtils.subtract(desired, available);
    return ValueUtils.compare(difference, ValueUtils.getZero(desired.getType())) <= 0;
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private static Map<String, Value> getResourcePool(Collection<MesosResource> mesosResources) {
    Map<String, Value> pool = new HashMap<>();
    for (MesosResource mesosResource : mesosResources) {
        String name = mesosResource.getName();
        Value currValue = pool.get(name);

        if (currValue == null) {
            currValue = ValueUtils.getZero(mesosResource.getType());
        }

        pool.put(name, ValueUtils.add(currValue, mesosResource.getValue()));
    }

    return pool;
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
public static Value getZero(Value.Type type) {
    switch (type) {
        case SCALAR:
            Value.Scalar scalar = Value.Scalar.newBuilder().setValue(0).build();
            return Value.newBuilder().setType(type).setScalar(scalar).build();
        case RANGES:
            Value.Ranges ranges = Value.Ranges.newBuilder().addAllRange(Collections.emptyList()).build();
            return Value.newBuilder().setType(type).setRanges(ranges).build();
        default:
            return null;
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
/**
 * Minimum to keep required field errors away.
 */
public static void addResource(Offer.Builder o, String name, String role) {
    Resource.Builder b = o.addResourcesBuilder().setType(Value.Type.RANGES).setName(name);
    if (role != null) {
        b.setRole(role);
    }
}
 

import org.apache.mesos.Protos.Value; //导入依赖的package包/类
private static Offer offerWithAttribute(String name, String value) {
    Protos.Resource resource = ResourceBuilder.fromUnreservedValue(
            "cpus",
            Protos.Value.newBuilder()
                    .setType(Protos.Value.Type.SCALAR)
                    .setScalar(Protos.Value.Scalar.newBuilder().setValue(1.0))
                    .build())
            .build();
    Offer.Builder offerBuilder = OfferTestUtils.getCompleteOffer(resource).toBuilder();
    offerBuilder.addAttributesBuilder()
            .setName(name)
            .setType(Value.Type.TEXT)
            .getTextBuilder().setValue(value);
    return offerBuilder.build();
}