import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public void execute(
    ExecuteRequest request, StreamObserver<Operation> responseObserver) {
  Instance instance;
  try {
    instance = instances.get(request.getInstanceName());
  } catch (InstanceNotFoundException ex) {
    responseObserver.onError(BuildFarmInstances.toStatusException(ex));
    return;
  }

  instance.execute(
      request.getAction(),
      request.getSkipCacheLookup(),
      request.getTotalInputFileCount(),
      request.getTotalInputFileBytes(),
      (operation) -> {
        responseObserver.onNext(operation);
        responseObserver.onCompleted();
      });
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
private MemoryInstance(
    String name,
    MemoryInstanceConfig config,
    ContentAddressableStorage contentAddressableStorage,
    Map<String, Operation> outstandingOperations) {
  super(
      name,
      contentAddressableStorage,
      /*actionCache=*/ new DelegateCASMap<Digest, ActionResult>(contentAddressableStorage, ActionResult.parser()),
      outstandingOperations,
      /*completedOperations=*/ new DelegateCASMap<String, Operation>(contentAddressableStorage, Operation.parser()));
  this.config = config;
  watchers = new ConcurrentHashMap<String, List<Function<Operation, Boolean>>>();
  streams = new HashMap<String, ByteStringStreamSource>();
  queuedOperations = new ArrayList<Operation>();
  workers = new ArrayList<Worker>();
  requeuers = new HashMap<String, Watchdog>();
  operationTimeoutDelays = new HashMap<String, Watchdog>();
  this.outstandingOperations = outstandingOperations;
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
protected Operation createOperation(Action action) {
  String name = createOperationName(UUID.randomUUID().toString());

  watchers.put(name, new ArrayList<Function<Operation, Boolean>>());

  Digest actionDigest = Digests.computeDigest(action.toByteString());

  ExecuteOperationMetadata metadata = ExecuteOperationMetadata.newBuilder()
      .setActionDigest(actionDigest)
      .build();

  Operation.Builder operationBuilder = Operation.newBuilder()
      .setName(name)
      .setDone(false)
      .setMetadata(Any.pack(metadata));

  return operationBuilder.build();
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
protected boolean matchOperation(Operation operation) {
  ImmutableList.Builder<Worker> rejectedWorkers = new ImmutableList.Builder<>();
  boolean dispatched = false;
  synchronized(workers) {
    while (!dispatched && !workers.isEmpty()) {
      Worker worker = workers.remove(0);
      if (!satisfiesRequirements(worker.getPlatform(), operation)) {
        rejectedWorkers.add(worker);
      } else {
        // worker onMatch false return indicates inviability
        if (dispatched = worker.apply(operation)) {
          onDispatched(operation);
        }
      }
    }
    Iterables.addAll(workers, rejectedWorkers.build());
  }
  return dispatched;
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public void match(Platform platform, boolean requeueOnFailure, Function<Operation, Boolean> onMatch) {
  synchronized(queuedOperations) {
    ImmutableList.Builder<Operation> rejectedOperations = new ImmutableList.Builder<Operation>();
    boolean matched = false;
    while (!matched && !queuedOperations.isEmpty()) {
      Operation operation = queuedOperations.remove(0);
      if (satisfiesRequirements(platform, operation)) {
        matched = true;
        if (onMatch.apply(operation)) {
          onDispatched(operation);
        } else if (!requeueOnFailure) {
          rejectedOperations.add(operation);
        }
      } else {
        rejectedOperations.add(operation);
      }
    }
    Iterables.addAll(queuedOperations, rejectedOperations.build());
    if (!matched) {
      synchronized(workers) {
        workers.add(new Worker(platform, onMatch));
      }
    }
  }
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public boolean putOperation(Operation operation) {
  if (isCancelled(operation)) {
    throw new IllegalStateException();
  }
  if (isExecuting(operation) &&
      !outstandingOperations.containsKey(operation.getName())) {
    return false;
  }
  if (isQueued(operation)) {
    if (!matchOperation(operation)) {
      enqueueOperation(operation);
    }
  } else {
    updateOperationWatchers(operation);
  }
  return true;
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public String listOperations(
    int pageSize, String pageToken, String filter,
    ImmutableList.Builder<Operation> operations) {
  if (pageSize == 0) {
    pageSize = getListOperationsDefaultPageSize();
  } else if (getListOperationsMaxPageSize() > 0 &&
      pageSize > getListOperationsMaxPageSize()) {
    pageSize = getListOperationsMaxPageSize();
  }

  // FIXME filter?
  TokenizableIterator<Operation> iter = createOperationsIterator(pageToken);

  while (iter.hasNext() && pageSize != 0) {
    Operation operation = iter.next();
    operations.add(operation);
    if (pageSize > 0) {
      pageSize--;
    }
  }
  return iter.toNextPageToken();
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public boolean pollOperation(
    String operationName,
    ExecuteOperationMetadata.Stage stage) {
  if (stage != ExecuteOperationMetadata.Stage.QUEUED
      && stage != ExecuteOperationMetadata.Stage.EXECUTING) {
    return false;
  }
  Operation operation = getOperation(operationName);
  if (operation == null) {
    return false;
  }
  ExecuteOperationMetadata metadata = expectExecuteOperationMetadata(operation);
  if (metadata == null) {
    return false;
  }
  // stage limitation to {QUEUED, EXECUTING} above is required
  if (metadata.getStage() != stage) {
    return false;
  }
  return true;
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
private void requeue(Operation operation) {
  try {
    ExecuteOperationMetadata metadata =
        operation.getMetadata().unpack(ExecuteOperationMetadata.class);

    ExecuteOperationMetadata executingMetadata = metadata.toBuilder()
        .setStage(ExecuteOperationMetadata.Stage.QUEUED)
        .build();

    operation = operation.toBuilder()
        .setMetadata(Any.pack(executingMetadata))
        .build();
    putOperation(operation);
  } catch(InvalidProtocolBufferException ex) {
    // operation is dropped on the floor
  }
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public void put(
    Operation operation,
    StreamObserver<com.google.rpc.Status> responseObserver) {
  Instance instance;
  try {
    instance = instances.getFromOperationName(operation.getName());
  } catch (InstanceNotFoundException ex) {
    responseObserver.onError(BuildFarmInstances.toStatusException(ex));
    return;
  }

  boolean ok = instance.putOperation(operation);
  Code code = ok ? Code.OK : Code.UNAVAILABLE;
  responseObserver.onNext(com.google.rpc.Status.newBuilder()
      .setCode(code.getNumber())
      .build());
  responseObserver.onCompleted();
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
private boolean fetch(Operation operation) {
  ExecuteOperationMetadata metadata;
  Action action;
  try {
    metadata = operation.getMetadata().unpack(ExecuteOperationMetadata.class);
    action = Action.parseFrom(worker.instance.getBlob(metadata.getActionDigest()));
  } catch (InvalidProtocolBufferException ex) {
    return false;
  }
  Path execDir = worker.root.resolve(operation.getName());
  output.offer(new OperationContext(
      operation,
      execDir,
      metadata,
      action,
      new ArrayList<>()));
  return true;
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
private void waitForOperation(BigtableClusterAdminClient client, String operationName,
    int maxSeconds) {
  GetOperationRequest request = GetOperationRequest.newBuilder().setName(operationName).build();
  for (int i = 0; i < maxSeconds; i++) {
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    Operation response = client.getOperation(request);
    if (response.getError() != null) {
      return;
    }
  }
  throw new IllegalStateException(String.format(
    "Waited %d seconds and operation was not complete", maxSeconds));
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
private @Nullable ExecuteResponse getOperationResponse(Operation op) throws IOException {
  if (op.getResultCase() == Operation.ResultCase.ERROR) {
    handleStatus(op.getError());
  }
  if (op.getDone()) {
    Preconditions.checkState(op.getResultCase() != Operation.ResultCase.RESULT_NOT_SET);
    try {
      ExecuteResponse resp = op.getResponse().unpack(ExecuteResponse.class);
      if (resp.hasStatus()) {
        handleStatus(resp.getStatus());
      }
      return resp;
    } catch (InvalidProtocolBufferException e) {
      throw new IOException(e);
    }
  }
  return null;
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
protected void updateOperationWatchers(Operation operation) {
  synchronized(watchers) {
    List<Function<Operation, Boolean>> operationWatchers =
        watchers.get(operation.getName());
    if (operationWatchers != null) {
      if (operation.getDone()) {
        watchers.remove(operation.getName());
      }
      long unfilteredWatcherCount = operationWatchers.size();
      super.updateOperationWatchers(operation);
      ImmutableList.Builder<Function<Operation, Boolean>> filteredWatchers = new ImmutableList.Builder<>();
      long filteredWatcherCount = 0;
      for (Function<Operation, Boolean> watcher : operationWatchers) {
        if (watcher.apply(operation)) {
          filteredWatchers.add(watcher);
          filteredWatcherCount++;
        }
      }
      if (!operation.getDone() && filteredWatcherCount != unfilteredWatcherCount) {
        operationWatchers = new ArrayList<>();
        Iterables.addAll(operationWatchers, filteredWatchers.build());
        watchers.put(operation.getName(), operationWatchers);
      }
    } else {
      throw new IllegalStateException();
    }
  }
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
public AbstractServerInstance(
    String name,
    ContentAddressableStorage contentAddressableStorage,
    Map<Digest, ActionResult> actionCache,
    Map<String, Operation> outstandingOperations,
    Map<String, Operation> completedOperations) {
  this.name = name;
  this.contentAddressableStorage = contentAddressableStorage;
  this.actionCache = actionCache;
  this.outstandingOperations = outstandingOperations;
  this.completedOperations = completedOperations;
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
protected ExecuteOperationMetadata expectExecuteOperationMetadata(
    Operation operation) {
  Preconditions.checkState(
      operation.getMetadata().is(ExecuteOperationMetadata.class));
  try {
    return operation.getMetadata().unpack(ExecuteOperationMetadata.class);
  } catch(InvalidProtocolBufferException ex) {
    return null;
  }
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
protected Action expectAction(Operation operation) {
  try {
    return Action.parseFrom(getBlob(
        expectExecuteOperationMetadata(operation).getActionDigest()));
  } catch(InvalidProtocolBufferException ex) {
    return null;
  }
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
protected void updateOperationWatchers(Operation operation) {
  if (operation.getDone()) {
    synchronized(operationLock(operation.getName())) {
      completedOperations.put(operation.getName(), operation);
      outstandingOperations.remove(operation.getName());
    }
  } else {
    outstandingOperations.put(operation.getName(), operation);
  }
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public Operation getOperation(String name) {
  synchronized(operationLock(name)) {
    Operation operation = completedOperations.get(name);
    if (operation == null) {
      operation = outstandingOperations.get(name);
    }
    return operation;
  }
}
 

import com.google.longrunning.Operation; //导入依赖的package包/类
@Override
public void deleteOperation(String name) {
  synchronized(operationLock(name)) {
    Operation deletedOperation = completedOperations.remove(name);
    if (deletedOperation == null &&
        outstandingOperations.containsKey(name)) {
      throw new IllegalStateException();
    }
  }
}