BazelCon 2024 报名现已开放！

此页面由 Cloud Translation API 翻译。

Skyframe StateMachine 指南

7.3 · 7.2 · 7.1 · 7.0 · 6.5

概览

Skyframe StateMachine 是位于堆上的解构函数对象。当所需值无法立即提供，而是异步计算时，它支持灵活且无冗余的评估¹。StateMachine 在等待期间无法占用线程资源，而必须暂停和恢复。因此，解构会公开显式重新进入点，以便跳过之前的计算。

StateMachine 可用于表达序列、分支、结构化逻辑并发，并且专为 Skyframe 互动而量身定制。StateMachine 可组合成更大的 StateMachine 并共享子 StateMachine。并发始终是结构上的层次结构，并且完全是逻辑性的。每个并发子任务都在单个共享父 SkyFunction 线程中运行。

简介

本部分简要介绍了 java.com.google.devtools.build.skyframe.state 软件包中提供的 StateMachine 的动机和用途。

Skyframe 重启简介

Skyframe 是一个用于对依赖关系图执行并行评估的框架。图中的每个节点都对应于 SkyFunction 的求值，其中 SkyKey 用于指定参数，SkyValue 用于指定结果。计算模型的设计使得 SkyFunction 可以按 SkyKey 查找 SkyValue，从而触发对其他 SkyFunction 的递归并行评估。如果请求的 SkyValue 因计算的某个子图不完整而尚未准备就绪，请求的 SkyFunction 会观察 null getValue 响应，并应返回 null（而非 SkyValue），表明它因缺少输入而未完成。当之前请求的所有 SkyValue 都变为可用时，Skyframe 会重启 SkyFunction。

在引入 SkyKeyComputeState 之前，处理重启的传统方式是完全重新运行计算。尽管这具有二次复杂性，但以这种方式编写的函数最终会完成，因为每次重新运行，返回 null 的查询更少。借助 SkyKeyComputeState，您可以将手动指定的检查点数据与 SkyFunction 相关联，从而显著减少重新计算次数。

StateMachine 是位于 SkyKeyComputeState 中的对象，通过公开挂起和恢复执行钩子，在 SkyFunction 重启时消除几乎所有重新计算（假设 SkyKeyComputeState 不会从缓存中移除）。

`SkyKeyComputeState` 中的有状态计算

从面向对象的设计角度来看，考虑在 SkyKeyComputeState 中存储计算对象（而非纯数据值）是明智之举。在 Java 中，对具有行为的对象的最低描述是函数接口，事实证明这已经足够了。StateMachine 具有以下奇怪递归定义²。

@FunctionalInterface
public interface StateMachine {
  StateMachine step(Tasks tasks) throws InterruptedException;
}

Tasks 接口类似于 SkyFunction.Environment，但它专为异步而设计，并添加了对逻辑并发子任务的支持³。

step 的返回值是另一个 StateMachine，允许以归纳方式指定一系列步骤。StateMachine 完成后，step 会返回 DONE。例如：

class HelloWorld implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    System.out.println("hello");
    return this::step2;  // The next step is HelloWorld.step2.
  }

  private StateMachine step2(Tasks tasks) {
     System.out.println("world");
     // DONE is special value defined in the `StateMachine` interface signaling
     // that the computation is done.
     return DONE;
  }
}

描述了 StateMachine，输出如下所示。

hello
world

请注意，由于 step2 满足 StateMachine 的函数接口定义，因此方法引用 this::step2 也是一个 StateMachine。在 StateMachine 中指定下一个状态的最常用方法是使用方法引用。

暂停和恢复

直观地说，将计算分解为 StateMachine 步（而不是单体式函数）可提供挂起和恢复计算所需的钩子。当 StateMachine.step 返回时，会有一个显式的挂起点。返回的 StateMachine 值指定的接续点是一个显式恢复点。因此，可以避免重新计算，因为计算可以从上次中断的地方继续。

回调、接续和异步计算

从技术层面来说，StateMachine 可用作接续，用于确定要执行的后续计算。StateMachine 可以通过从 step 函数返回来自愿挂起，而不是进行阻塞，这会将控制权转回 Driver 实例。然后，Driver 可以切换到就绪的 StateMachine，或将控制权交还给 Skyframe。

传统上，回调和接续被混为一谈。不过，StateMachine 会区分这两者。

回调 - 描述异步计算结果的存储位置。
接续 - 指定下一个执行状态。

调用异步操作时需要回调，这意味着实际操作不会在调用方法后立即发生，就像 SkyValue 查找一样。回调应尽可能简单。

接续是 StateMachine 的 StateMachine 返回值，用于封装所有异步计算解析后进行的复杂执行。这种结构化方法有助于将回调的复杂性控制在可管理的范围内。

Tasks

Tasks 接口为 StateMachine 提供了用于通过 SkyKey 查找 SkyValues 并调度并发子任务的 API。

interface Tasks {
  void enqueue(StateMachine subtask);

  void lookUp(SkyKey key, Consumer<SkyValue> sink);

  <E extends Exception>
  void lookUp(SkyKey key, Class<E> exceptionClass, ValueOrExceptionSink<E> sink);

  // lookUp overloads for 2 and 3 exception types exist, but are elided here.
}

SkyValue 查询

StateMachine 使用 Tasks.lookUp 重载来查找 SkyValue。它们类似于 SkyFunction.Environment.getValue 和 SkyFunction.Environment.getValueOrThrow，并且具有类似的异常处理语义。该实现不会立即执行查找，而是会尽可能多地批量执行⁴查找。该值可能不会立即可用（例如，需要重启 Skyframe），因此调用方会使用回调指定如何处理结果值。

StateMachine 处理器（Driver 和桥接到 SkyFrame）可确保在下一个状态开始之前值可用。示例如下。

class DoesLookup implements StateMachine, Consumer<SkyValue> {
  private Value value;

  @Override
  public StateMachine step(Tasks tasks) {
    tasks.lookUp(new Key(), (Consumer<SkyValue>) this);
    return this::processValue;
  }

  // The `lookUp` call in `step` causes this to be called before `processValue`.
  @Override  // Implementation of Consumer<SkyValue>.
  public void accept(SkyValue value) {
    this.value = (Value)value;
  }

  private StateMachine processValue(Tasks tasks) {
    System.out.println(value);  // Prints the string representation of `value`.
    return DONE;
  }
}

在上面的示例中，第一步是查找 new Key()，并将 this 作为使用方传递。之所以能实现这一点，是因为 DoesLookup 实现了 Consumer<SkyValue>。

根据协定，在下一个状态 DoesLookup.processValue 开始之前，DoesLookup.step 的所有查找都已完成。因此，在 processValue 中对其进行访问时，value 可用。

子任务

Tasks.enqueue 请求执行逻辑上并发的子任务。子任务也是 StateMachine，可以执行常规 StateMachine 可以执行的任何操作，包括递归创建更多子任务或查找 SkyValue。与 lookUp 非常相似，状态机驱动程序会确保所有子任务均已完成，然后再继续执行下一步。示例如下。

class Subtasks implements StateMachine {
  private int i = 0;

  @Override
  public StateMachine step(Tasks tasks) {
    tasks.enqueue(new Subtask1());
    tasks.enqueue(new Subtask2());
    // The next step is Subtasks.processResults. It won't be called until both
    // Subtask1 and Subtask 2 are complete.
    return this::processResults;
  }

  private StateMachine processResults(Tasks tasks) {
    System.out.println(i);  // Prints "3".
    return DONE;  // Subtasks is done.
  }

  private class Subtask1 implements StateMachine {
    @Override
    public StateMachine step(Tasks tasks) {
      i += 1;
      return DONE;  // Subtask1 is done.
    }
  }

  private class Subtask2 implements StateMachine {
    @Override
    public StateMachine step(Tasks tasks) {
      i += 2;
      return DONE;  // Subtask2 is done.
    }
  }
}

虽然 Subtask1 和 Subtask2 在逻辑上是并发的，但所有内容都在单个线程中运行，因此 i 的“并发”更新不需要任何同步。

结构化并发

由于每个 lookUp 和 enqueue 都必须在进入下一个状态之前进行解析，这意味着并发自然受限于树结构。可以创建分层⁵ 并发，如以下示例所示。

结构化并发

通过 UML 很难判断并发结构是否构成了树。有一个其他视图，可以更好地显示树结构。

非结构化并发

结构化并发更易于推理。

组合和控制流模式

本部分介绍了如何组合多个 StateMachine 的示例，以及某些控制流问题的解决方案。

顺序状态

这是最常见且最简单的控制流模式。SkyKeyComputeState 中的有状态计算中显示了此示例。

分支

您可以使用常规 Java 控制流返回不同的值，从而实现 StateMachine 中的分支状态，如以下示例所示。

class Branch implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    // Returns different state machines, depending on condition.
    if (shouldUseA()) {
      return this::performA;
    }
    return this::performB;
  }
  …
}

某些分支返回 DONE 表示提前完成的情况很常见。

高级顺序组合

由于 StateMachine 控制结构是无内存的，因此有时将 StateMachine 定义作为子任务共享可能会很不方便。设 M₁ 和 M₂ 是共享 StateMachine S 的 StateMachine 实例，其中 M₁ 和 M₂ 分别为序列 <A, S, B> 和 <X, S, Y>。问题在于，S 在完成后不知道是继续执行 B 还是 Y，并且 StateMachine 无法保留调用堆栈。本部分将介绍实现此目的的一些方法。

`StateMachine` 作为终端序列元素

这并不能解决提出的初始问题。仅当共享的 StateMachine 是序列中的终端时，才会演示顺序组合。

// S is the shared state machine.
class S implements StateMachine { … }

class M1 implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    performA();
    return new S();
  }
}

class M2 implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    performX();
    return new S();
  }
}

即使 S 本身就是一个复杂的状态机，这种方法也适用。

序列化组合的子任务

由于队列中的子任务保证会在下一个状态之前完成，因此有时可以稍微滥用⁶子任务机制。

class M1 implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    performA();
    // S starts after `step` returns and by contract must complete before `doB`
    // begins. It is effectively sequential, inducing the sequence < A, S, B >.
    tasks.enqueue(new S());
    return this::doB;
  }

  private StateMachine doB(Tasks tasks) {
    performB();
    return DONE;
  }
}

class M2 implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    performX();
    // Similarly, this induces the sequence < X, S, Y>.
    tasks.enqueue(new S());
    return this::doY;
  }

  private StateMachine doY(Tasks tasks) {
    performY();
    return DONE;
  }
}

`runAfter` 注入

有时，由于在 S 执行之前必须完成其他并行子任务或 Tasks.lookUp 调用，因此无法滥用 Tasks.enqueue。在这种情况下，将 runAfter 参数注入 S 可用于告知 S 后续步骤。

class S implements StateMachine {
  // Specifies what to run after S completes.
  private final StateMachine runAfter;

  @Override
  public StateMachine step(Tasks tasks) {
    … // Performs some computations.
    return this::processResults;
  }

  @Nullable
  private StateMachine processResults(Tasks tasks) {
    … // Does some additional processing.

    // Executes the state machine defined by `runAfter` after S completes.
    return runAfter;
  }
}

class M1 implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    performA();
    // Passes `this::doB` as the `runAfter` parameter of S, resulting in the
    // sequence < A, S, B >.
    return new S(/* runAfter= */ this::doB);
  }

  private StateMachine doB(Tasks tasks) {
    performB();
    return DONE;
  }
}

class M2 implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks) {
    performX();
    // Passes `this::doY` as the `runAfter` parameter of S, resulting in the
    // sequence < X, S, Y >.
    return new S(/* runAfter= */ this::doY);
  }

  private StateMachine doY(Tasks tasks) {
    performY();
    return DONE;
  }
}

这种方法比滥用子任务更简洁。不过，过于宽松地采用此方法（例如，使用 runAfter 嵌套多个 StateMachine）才是通向 Callback Hell 的通道。最好改用普通顺序状态来拆分顺序 runAfter。

  return new S(/* runAfter= */ new T(/* runAfter= */ this::nextStep))

可以替换为以下代码。

  private StateMachine step1(Tasks tasks) {
     doStep1();
     return new S(/* runAfter= */ this::intermediateStep);
  }

  private StateMachine intermediateStep(Tasks tasks) {
    return new T(/* runAfter= */ this::nextStep);
  }

禁止访问替代方案：`runAfterUnlessError`

在早期草稿中，我们曾考虑过一种 runAfterUnlessError，它会在出现错误时提前终止。之所以这样做，是因为错误通常会被检查两次，一次由具有 runAfter 引用的 StateMachine 进行检查，一次由 runAfter 机器本身进行检查。

经过一番考虑，我们认为代码的一致性比删除重复的错误检查更加重要。如果 runAfter 机制的运作方式与始终需要进行错误检查的 tasks.enqueue 机制不一致，就会令人困惑。

直接委托

每当发生正式状态转换时，主 Driver 循环都会推进。根据协定，推进状态意味着在下一个状态执行之前，所有之前加入队列的 SkyValue 查找和子任务都会解析。有时，代理 StateMachine 的逻辑会导致相位提前变得没有必要或适得其反。例如，如果委托方的第一个 step 执行的 SkyKey 查找可以与委托状态的查找并行执行，那么相应阶段的提前会使它们顺序执行。更合理的做法是执行直接委托，如以下示例所示。

class Parent implements StateMachine {
  @Override
  public StateMachine step(Tasks tasks ) {
    tasks.lookUp(new Key1(), this);
    // Directly delegates to `Delegate`.
    //
    // The (valid) alternative:
    //   return new Delegate(this::afterDelegation);
    // would cause `Delegate.step` to execute after `step` completes which would
    // cause lookups of `Key1` and `Key2` to be sequential instead of parallel.
    return new Delegate(this::afterDelegation).step(tasks);
  }

  private StateMachine afterDelegation(Tasks tasks) {
    …
  }
}

class Delegate implements StateMachine {
  private final StateMachine runAfter;

  Delegate(StateMachine runAfter) {
    this.runAfter = runAfter;
  }

  @Override
  public StateMachine step(Tasks tasks) {
    tasks.lookUp(new Key2(), this);
    return …;
  }

  // Rest of implementation.
  …

  private StateMachine complete(Tasks tasks) {
    …
    return runAfter;
  }
}

数据流

之前的讨论重点是管理控制流。本部分介绍了数据值的传播。

实现 `Tasks.lookUp` 回调

SkyValue 查找中提供了实现 Tasks.lookUp 回调的示例。本部分提供处理多个 SkyValues 的理由并推荐了几种方法。

`Tasks.lookUp` 回调

Tasks.lookUp 方法将回调 sink 作为参数。

  void lookUp(SkyKey key, Consumer<SkyValue> sink);

惯用的方法是使用 Java lambda 来实现这一点：

  tasks.lookUp(key, value -> myValue = (MyValueClass)value);

其中 myValue 是执行查找的 StateMachine 实例的成员变量。不过，与在 StateMachine 实现中实现 Consumer<SkyValue> 接口相比，lambda 需要额外分配内存。当有多个模糊的查找时，lambda 仍然很有用。

此外，还有错误处理 Tasks.lookUp 过载，它类似于 SkyFunction.Environment.getValueOrThrow。

  <E extends Exception> void lookUp(
      SkyKey key, Class<E> exceptionClass, ValueOrExceptionSink<E> sink);

  interface ValueOrExceptionSink<E extends Exception> {
    void acceptValueOrException(@Nullable SkyValue value, @Nullable E exception);
  }

实现示例如下所示。

class PerformLookupWithError extends StateMachine, ValueOrExceptionSink<MyException> {
  private MyValue value;
  private MyException error;

  @Override
  public StateMachine step(Tasks tasks) {
    tasks.lookUp(new MyKey(), MyException.class, ValueOrExceptionSink<MyException>) this);
    return this::processResult;
  }

  @Override
  public acceptValueOrException(@Nullable SkyValue value, @Nullable MyException exception) {
    if (value != null) {
      this.value = (MyValue)value;
      return;
    }
    if (exception != null) {
      this.error = exception;
      return;
    }
    throw new IllegalArgumentException("Both parameters were unexpectedly null.");
  }

  private StateMachine processResult(Tasks tasks) {
    if (exception != null) {
      // Handles the error.
      …
      return DONE;
    }
    // Processes `value`, which is non-null.
    …
  }
}

与没有错误处理的查询一样，让 StateMachine 类直接实现回调可以节省 lamba 的内存分配。

错误处理提供了更多详细信息，但从本质上讲，错误和正常值的传播没有太大区别。

使用多个 SkyValue

通常需要进行多次 SkyValue 查询。大多数情况下，一种有效的方法是开启 SkyValue 类型。以下是从原型设计正式版代码中简化的示例。

  @Nullable
  private StateMachine fetchConfigurationAndPackage(Tasks tasks) {
    var configurationKey = configuredTarget.getConfigurationKey();
    if (configurationKey != null) {
      tasks.lookUp(configurationKey, (Consumer<SkyValue>) this);
    }

    var packageId = configuredTarget.getLabel().getPackageIdentifier();
    tasks.lookUp(PackageValue.key(packageId), (Consumer<SkyValue>) this);

    return this::constructResult;
  }

  @Override  // Implementation of `Consumer<SkyValue>`.
  public void accept(SkyValue value) {
    if (value instanceof BuildConfigurationValue) {
      this.configurationValue = (BuildConfigurationValue) value;
      return;
    }
    if (value instanceof PackageValue) {
      this.pkg = ((PackageValue) value).getPackage();
      return;
    }
    throw new IllegalArgumentException("unexpected value: " + value);
  }

Consumer<SkyValue> 回调实现可以明确共享，因为值类型不同。否则，可以回退到基于 lambda 的实现或实现适当回调的完整内部类实例。

在 `StateMachine` 之间传播值

到目前为止，本文档仅介绍了如何排列子任务中的工作，但子任务还需要将值报告给调用方。由于子任务在逻辑上是异步的，因此系统会使用回调将其结果传达给调用方。为了实现这一点，子任务定义了一个通过其构造函数注入的接收器接口。

class BarProducer implements StateMachine {
  // Callers of BarProducer implement the following interface to accept its
  // results. Exactly one of the two methods will be called by the time
  // BarProducer completes.
  interface ResultSink {
    void acceptBarValue(Bar value);
    void acceptBarError(BarException exception);
  }

  private final ResultSink sink;

  BarProducer(ResultSink sink) {
     this.sink = sink;
  }

  … // StateMachine steps that end with this::complete.

  private StateMachine complete(Tasks tasks) {
    if (hasError()) {
      sink.acceptBarError(getError());
      return DONE;
    }
    sink.acceptBarValue(getValue());
    return DONE;
  }
}

调用方 StateMachine 将如下所示。

class Caller implements StateMachine, BarProducer.ResultSink {
  interface ResultSink {
    void acceptCallerValue(Bar value);
    void acceptCallerError(BarException error);
  }

  private final ResultSink sink;

  private Bar value;

  Caller(ResultSink sink) {
    this.sink = sink;
  }

  @Override
  @Nullable
  public StateMachine step(Tasks tasks) {
    tasks.enqueue(new BarProducer((BarProducer.ResultSink) this));
    return this::processResult;
  }

  @Override
  public void acceptBarValue(Bar value) {
    this.value = value;
  }

  @Override
  public void acceptBarError(BarException error) {
    sink.acceptCallerError(error);
  }

  private StateMachine processResult(Tasks tasks) {
    // Since all enqueued subtasks resolve before `processResult` starts, one of
    // the `BarResultSink` callbacks must have been called by this point.
    if (value == null) {
      return DONE;  // There was a previously reported error.
    }
    var finalResult = computeResult(value);
    sink.acceptCallerValue(finalResult);
    return DONE;
  }
}

上面的示例展示了以下几点。Caller 必须将其结果传播回来并定义自己的 Caller.ResultSink。Caller 会实现 BarProducer.ResultSink 回调。恢复后，processResult 会检查 value 是否为 null，以确定是否发生了错误。接受子任务或 SkyValue 查询的输出后，这是一种常见的行为模式。

请注意，acceptBarError 的实现会按照错误上报的要求，立即将结果转发给 Caller.ResultSink。

如需了解顶级 StateMachine 的替代方案，请参阅 Driver 和桥接到 SkyFunctions。

错误处理

Tasks.lookUp 回调和在 StateMachines 之间传播值中已经提供了一些错误处理示例。系统不会抛出 InterruptedException 以外的异常，而是作为值通过回调传递。此类回调通常具有析取或语义，只会传递值或错误之一。

下一部分将介绍与 Skyframe 错误处理之间的一种微妙但非常重要的交互。

错误上报 (--nokeep_going)

在错误上报期间，即使请求的所有 SkyValue 都不可用，SkyFunction 也可能会重启。在这种情况下，由于 Tasks API 协定，系统将永远无法达到后续状态。但是，StateMachine 应仍会传播异常。

无论是否达到下一个状态，都必须进行传播，因此错误处理回调必须执行此任务。对于内部 StateMachine，这可以通过调用父回调来实现。

在与 SkyFunction 接口的顶级 StateMachine 中，可以通过调用 ValueOrExceptionProducer 的 setException 方法来实现此目的。然后，ValueOrExceptionProducer.tryProduceValue 将抛出异常，即使缺少 SkyValue 也是如此。

如果直接使用 Driver，请务必检查 SkyFunction 中是否已传播错误，即使机器尚未完成处理也是如此。

事件处理

对于需要发出事件的 SkyFunction，系统会将 StoredEventHandler 注入 SkyKeyComputeState，并进一步注入需要它们的 StateMachine。过去，需要使用 StoredEventHandler，因为 Skyframe 会丢弃某些事件，除非重播，但随后此问题得到了修复。保留了 StoredEventHandler 注入，因为它简化了从错误处理回调发出的事件的实现。

`Driver` 和与 SkyFunctions 的桥接

Driver 负责管理 StateMachine 的执行，从指定的根 StateMachine 开始。由于 StateMachine 能够以递归方式将子任务 StateMachine 加入队列，因此单个 Driver 可以管理大量子任务。这些子任务会创建一个树形结构，这是结构化并发的结果。Driver 会跨子任务批量执行 SkyValue 查找，以提高效率。

围绕 Driver 构建了许多类，并提供了以下 API。

public final class Driver {
  public Driver(StateMachine root);
  public boolean drive(SkyFunction.Environment env) throws InterruptedException;
}

Driver 接受单个根 StateMachine 作为参数。调用 Driver.drive 会在不重启 Skyframe 的情况下，尽可能执行 StateMachine。当 StateMachine 完成时，它会返回 true，否则返回 false，表示并非所有值均可用。

Driver 可保持 StateMachine 的并发状态，它非常适合嵌入 SkyKeyComputeState 中。

正在直接实例化 `Driver`

StateMachine 实现通常通过回调传达其结果。您可以直接实例化 Driver，如以下示例所示。

Driver 嵌入在 SkyKeyComputeState 实现中，以及稍后定义的相应 ResultSink 的实现中。在顶层，State 对象是适合计算结果的接收器，因为它的存在时间肯定比 Driver 更长。

class State implements SkyKeyComputeState, ResultProducer.ResultSink {
  // The `Driver` instance, containing the full tree of all `StateMachine`
  // states. Responsible for calling `StateMachine.step` implementations when
  // asynchronous values are available and performing batched SkyFrame lookups.
  //
  // Non-null while `result` is being computed.
  private Driver resultProducer;

  // Variable for storing the result of the `StateMachine`
  //
  // Will be non-null after the computation completes.
  //
  private ResultType result;

  // Implements `ResultProducer.ResultSink`.
  //
  // `ResultProducer` propagates its final value through a callback that is
  // implemented here.
  @Override
  public void acceptResult(ResultType result) {
    this.result = result;
  }
}

以下代码会绘制 ResultProducer。

class ResultProducer implements StateMachine {
  interface ResultSink {
    void acceptResult(ResultType value);
  }

  private final Parameters parameters;
  private final ResultSink sink;

  … // Other internal state.

  ResultProducer(Parameters parameters, ResultSink sink) {
    this.parameters = parameters;
    this.sink = sink;
  }

  @Override
  public StateMachine step(Tasks tasks) {
    …  // Implementation.
    return this::complete;
  }

  private StateMachine complete(Tasks tasks) {
    sink.acceptResult(getResult());
    return DONE;
  }
}

然后，用于延迟计算结果的代码可能如下所示。

@Nullable
private Result computeResult(State state, Skyfunction.Environment env)
    throws InterruptedException {
  if (state.result != null) {
    return state.result;
  }
  if (state.resultProducer == null) {
    state.resultProducer = new Driver(new ResultProducer(
      new Parameters(), (ResultProducer.ResultSink)state));
  }
  if (state.resultProducer.drive(env)) {
    // Clears the `Driver` instance as it is no longer needed.
    state.resultProducer = null;
  }
  return state.result;
}

嵌入 `Driver`

如果 StateMachine 生成值且未引发任何异常，则嵌入 Driver 是另一种可能的实现方式，如下例所示。

class ResultProducer implements StateMachine {
  private final Parameters parameters;
  private final Driver driver;

  private ResultType result;

  ResultProducer(Parameters parameters) {
    this.parameters = parameters;
    this.driver = new Driver(this);
  }

  @Nullable  // Null when a Skyframe restart is needed.
  public ResultType tryProduceValue( SkyFunction.Environment env)
      throws InterruptedException {
    if (!driver.drive(env)) {
      return null;
    }
    return result;
  }

  @Override
  public StateMachine step(Tasks tasks) {
    …  // Implementation.
}

SkyFunction 的代码可能如下所示（其中 State 是 SkyKeyComputeState 的函数专用类型）。

@Nullable  // Null when a Skyframe restart is needed.
Result computeResult(SkyFunction.Environment env, State state)
    throws InterruptedException {
  if (state.result != null) {
    return state.result;
  }
  if (state.resultProducer == null) {
    state.resultProducer = new ResultProducer(new Parameters());
  }
  var result = state.resultProducer.tryProduceValue(env);
  if (result == null) {
    return null;
  }
  state.resultProducer = null;
  return state.result = result;
}

在 StateMachine 实现中嵌入 Driver 更适合 Skyframe 的同步编码风格。

可能会产生异常的 StateMachine

否则，您可以使用可嵌入 SkyKeyComputeState 的 ValueOrExceptionProducer 和 ValueOrException2Producer 类，这些类具有与同步 SkyFunction 代码匹配的同步 API。

ValueOrExceptionProducer 抽象类包含以下方法。

public abstract class ValueOrExceptionProducer<V, E extends Exception>
    implements StateMachine {
  @Nullable
  public final V tryProduceValue(Environment env)
      throws InterruptedException, E {
    …  // Implementation.
  }

  protected final void setValue(V value)  {  … // Implementation. }
  protected final void setException(E exception) {  … // Implementation. }
}

它包含一个嵌入的 Driver 实例，与嵌入驱动程序中的 ResultProducer 类非常相似，并且以类似的方式与 SkyFunction 交互。实现会在发生上述任一情况时调用 setValue 或 setException，而不是定义 ResultSink。如果同时发生这两种情况，则例外情况优先。tryProduceValue 方法可将异步回调代码桥接到同步代码，并在设置异常时抛出异常。

如前所述，在错误上报过程中，即使机器尚未完成，也可能会发生错误，因为并非所有输入都处于可用状态。为此，tryProduceValue 会抛出任何已设置的异常，即使在机器完成之前也是如此。

结语：最终移除回调

StateMachine 是一种高效但大量使用样板代码的方式来执行异步计算。接续（特别是以传递给 ListenableFuture 的 Runnable 的形式）在 Bazel 代码的某些部分很常见，但在分析 SkyFunction 中并不普遍。分析主要受 CPU 限制，并且没有适用于磁盘 I/O 的高效异步 API。最终，最好优化掉回调，因为它们有学习曲线，并且会影响可读性。

最有前景的替代方案之一是 Java 虚拟线程。所有内容都替换为同步阻塞调用，而无需编写回调。之所以能这么做，是因为与平台线程不同，占用虚拟线程资源应该是低成本的。不过，即使使用虚拟线程，用线程创建和同步基元替换简单的同步操作也是代价太高。我们从 StateMachine 迁移到了 Java 虚拟线程，但它们的速度慢了几个数量级，导致端到端分析延迟时间几乎增加了 3 倍。由于虚拟线程仍处于预览版阶段，因此在性能提升后，我们可能会稍后执行此迁移。

另一种可考虑的方法是等待 Loom 协程（如果有的话）。这样做的好处在于，您或许可以通过使用协作式多任务处理来减少同步开销。

如果所有其他方法都行不通，低级字节码重写也是一种可行的替代方案。通过充分优化，您或许可以实现接近手写回调代码的性能。

附录

回调地狱

回调地狱是使用回调的异步代码中臭名昭著的问题。这是因为后续步骤的接续嵌套在前一步中。如果步骤有很多，这种嵌套可能会非常深层。如果与控制流结合使用，代码将变得不可管理。

class CallbackHell implements StateMachine {
  @Override
  public StateMachine step(Tasks task) {
    doA();
    return (t, l) -> {
      doB();
      return (t1, l2) -> {
        doC();
        return DONE;
      };
    };
  }
}

嵌套实现的一个优势是，可以保留外部步骤的堆栈帧。在 Java 中，捕获的 lambda 变量必须实际上是最终变量，因此使用此类变量可能会很麻烦。通过将方法引用作为接续（而非 lambda）返回，可以避免深层嵌套，如下所示。

class CallbackHellAvoided implements StateMachine {
  @Override
  public StateMachine step(Tasks task) {
    doA();
    return this::step2;
  }

  private StateMachine step2(Tasks tasks) {
    doB();
    return this::step3;
  }

  private StateMachine step3(Tasks tasks) {
    doC();
    return DONE;
  }
}

如果 runAfter 注入模式的使用过于密集，也可能会发生回调地狱，但通过将注入与顺序步骤穿插，可以避免这种情况。

示例：链式 SkyValue 查找

通常的情况是，应用逻辑需要 SkyValue 查询的从属链，例如，如果第二个 SkyKey 依赖于第一个 SkyValue。粗略地考虑一下，这会导致复杂的深层嵌套回调结构。

private ValueType1 value1;
private ValueType2 value2;

private StateMachine step1(...) {
  tasks.lookUp(key1, (Consumer<SkyValue>) this);  // key1 has type KeyType1.
  return this::step2;
}

@Override
public void accept(SkyValue value) {
  this.value1 = (ValueType1) value;
}

private StateMachine step2(...) {
  KeyType2 key2 = computeKey(value1);
  tasks.lookup(key2, this::acceptValueType2);
  return this::step3;
}

private void acceptValueType2(SkyValue value) {
  this.value2 = (ValueType2) value;
}

不过，由于接续项被指定为方法引用，因此在状态转换期间，代码看起来是过程化的：step2 紧随 step1 之后。请注意，此处使用 lambda 分配 value2。这使代码的顺序与计算从上到下的顺序匹配。

其他提示

可读性：执行顺序

为了提高可读性，请尽量按执行顺序保留 StateMachine.step 实现，并将回调实现紧跟在代码中传递它们的位置后面。在控制流分支的情况下，这并不总是可行的。在这种情况下，提供更多说明可能会很有帮助。

在示例：链接的 SkyValue 查询中，为实现这一点，创建了一个中间方法引用。这样做会牺牲一点性能来换取可读性，在本例中，这可能值得。

代际假设

中等生命周期的 Java 对象会破坏 Java 垃圾回收器的代假设，后者旨在处理生命周期非常短或永久存在的对象。根据定义，SkyKeyComputeState 中的对象违反了这个假设。此类对象包含根位于 Driver 的所有仍在运行的 StateMachine 构建的树，在挂起等待异步计算完成时具有中间生命周期。

在 JDK19 中，这种情况似乎不太严重，但在使用 StateMachine 时，有时可能会发现 GC 时间增加，即使实际生成的垃圾量大幅减少也是如此。由于 StateMachine 具有中间生命周期，因此它们可能会被提升到老年代，导致老年代更快地填满，从而需要进行更昂贵的主要 GC 或完整 GC 来进行清理。

初始预防措施是尽量减少使用 StateMachine 变量，但这并不总是可行的，例如，如果需要在多个状态中使用某个值。在可能的情况下，本地堆栈 step 变量是新生代变量，并且会进行高效的 GC。

对于 StateMachine 变量，将任务拆分为子任务并遵循在 StateMachine 之间传播值的建议模式也很有帮助。请注意，遵循此模式时，只有子 StateMachine 会引用父 StateMachine，反之亦然。这意味着，当子项使用结果回调完成并使用结果回调更新父项时，子项会自然地超出范围，并符合进行 GC 的条件。

最后，在某些情况下，早期状态需要 StateMachine 变量，但后期状态不需要。确定不再需要大型对象后，将其引用设为 null 会很有帮助。

命名状态

为方法命名时，通常可以根据该方法中发生的行为来命名方法。目前尚不清楚如何在 StateMachine 中执行此操作，因为没有堆栈。例如，假设方法 foo 调用子方法 bar。在 StateMachine 中，可将其转换为状态序列 foo，后跟 bar。foo 不再包含行为 bar。因此，状态的方法名称的范围往往较窄，可能会反映本地行为。

并发树形图

以下是结构化并发中图表的另一种视图，该视图更能直观地描述树结构。这些块构成了一棵小树。

结构化并发 3D

这与 SkyFrame 在值不可用时从头开始重启的惯例相反。 ↩
请注意，step 允许抛出 InterruptedException，但示例省略了这一点。Bazel 代码中有一些低级方法会抛出此异常，该异常会向上传播到运行 StateMachine 的 Driver（稍后会介绍）。可以不将其声明为在不需要时抛出。↩
并发子任务的动机是 ConfiguredTargetFunction，它会为每个依赖项执行独立的工作。每个依赖项都有自己的独立 StateMachine，而不是操纵复杂的数据结构来一次性处理所有依赖项，从而导致效率低下。↩
单个步骤中的多个 tasks.lookUp 调用会一起进行批处理。通过并发子任务中发生的查找，可以创建其他批处理。 ↩
这在概念上与 Java 的结构化并发 jeps/428 类似。 ↩
这样做类似于生成线程并加入该线程以实现顺序组合。 ↩

Skyframe StateMachine 指南

概览

简介

Skyframe 重启简介

SkyKeyComputeState 中的有状态计算

回调、接续和异步计算

Tasks

SkyValue 查询

子任务

结构化并发

组合和控制流模式

顺序状态

分支

高级顺序组合

StateMachine 作为终端序列元素

序列化组合的子任务

runAfter 注入

禁止访问替代方案：runAfterUnlessError

直接委托

数据流

实现 Tasks.lookUp 回调

Tasks.lookUp 回调

使用多个 SkyValue

在 StateMachine 之间传播值

错误处理

错误上报 (--nokeep_going)

事件处理

Driver 和与 SkyFunctions 的桥接

正在直接实例化 Driver

嵌入 Driver

可能会产生异常的 StateMachine

结语：最终移除回调

附录

回调地狱

示例：链式 SkyValue 查找

其他提示

可读性：执行顺序

代际假设

命名状态

并发树形图

`SkyKeyComputeState` 中的有状态计算

`StateMachine` 作为终端序列元素

`runAfter` 注入

禁止访问替代方案：`runAfterUnlessError`

实现 `Tasks.lookUp` 回调

`Tasks.lookUp` 回调

在 `StateMachine` 之间传播值

`Driver` 和与 SkyFunctions 的桥接

正在直接实例化 `Driver`

嵌入 `Driver`